This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 19eb15c fix: reject aggregation on sequence fields (#409)
19eb15c is described below
commit 19eb15c5f05f5db478437c6f6b5022401355f37f
Author: QuakeWang <[email protected]>
AuthorDate: Sat Jun 27 22:37:45 2026 +0800
fix: reject aggregation on sequence fields (#409)
---
crates/integrations/datafusion/tests/pk_tables.rs | 51 ++++++++++++++------
crates/paimon/src/spec/aggregation.rs | 57 ++++++++++++-----------
crates/paimon/src/spec/schema.rs | 30 ++++++++++++
crates/paimon/src/table/sort_merge.rs | 6 +--
docs/src/sql.md | 4 ++
5 files changed, 105 insertions(+), 43 deletions(-)
diff --git a/crates/integrations/datafusion/tests/pk_tables.rs
b/crates/integrations/datafusion/tests/pk_tables.rs
index 4fd38b2..66d5de4 100644
--- a/crates/integrations/datafusion/tests/pk_tables.rs
+++ b/crates/integrations/datafusion/tests/pk_tables.rs
@@ -2514,8 +2514,8 @@ async fn test_pk_aggregation_mixed_aggregators() {
assert_eq!(first_seen.value(0), "a"); // first non-null wins
}
-/// `sequence.field` forces the named column to `last_value`, even when the
-/// user explicitly configures another aggregator for it.
+/// `sequence.field` forces the named column to `last_value`, even when a
+/// table-level default aggregator would otherwise apply.
#[tokio::test]
async fn test_pk_aggregation_sequence_field_forced_last_value() {
let (_tmp, sql_context) = setup_sql_context().await;
@@ -2530,7 +2530,7 @@ async fn
test_pk_aggregation_sequence_field_forced_last_value() {
'merge-engine' = 'aggregation',
'sequence.field' = 'ts',
'fields.amount.aggregate-function' = 'sum',
- 'fields.ts.aggregate-function' = 'sum'
+ 'fields.default-aggregate-function' = 'sum'
)",
)
.await
@@ -2566,7 +2566,7 @@ async fn
test_pk_aggregation_sequence_field_forced_last_value() {
.and_then(|c| c.as_any().downcast_ref::<Int32Array>())
.unwrap();
assert_eq!(amount.value(0), 30); // sum still applies
- assert_eq!(ts.value(0), 250); // forced last_value over sum
+ assert_eq!(ts.value(0), 250); // forced last_value over default sum
}
/// Aggregation engine reads must surface Unsupported when a DELETE/UPDATE
@@ -2778,19 +2778,15 @@ async fn
test_pk_aggregation_create_table_rejects_incompatible_type() {
);
}
-/// CREATE TABLE must accept a function/type pair that the runtime would
-/// ignore: `sequence.field` columns are forced to `last_value` and primary-key
-/// columns get no aggregator, so type compatibility is not checked for them.
+/// CREATE TABLE must reject per-field aggregation on a sequence field,
+/// matching Java schema validation.
#[tokio::test]
-async fn
test_pk_aggregation_create_table_accepts_ignored_function_on_seq_and_pk() {
+async fn test_pk_aggregation_create_table_rejects_sequence_field_function() {
let (_tmp, sql_context) = setup_sql_context().await;
- // `listagg` is incompatible with INT, but `amount` is the sequence field
- // (forced to last_value) and `id` is a PK (copied through), so both
- // configurations are usable at runtime and must pass CREATE TABLE.
- sql_context
+ let err = sql_context
.sql(
- "CREATE TABLE paimon.test_db.t_agg_seq_pk_ok (
+ "CREATE TABLE paimon.test_db.t_agg_seq_bad (
id INT NOT NULL, amount INT, v INT,
PRIMARY KEY (id)
) WITH (
@@ -2798,12 +2794,39 @@ async fn
test_pk_aggregation_create_table_accepts_ignored_function_on_seq_and_pk
'merge-engine' = 'aggregation',
'sequence.field' = 'amount',
'fields.amount.aggregate-function' = 'listagg',
+ 'fields.v.aggregate-function' = 'sum'
+ )",
+ )
+ .await
+ .expect_err("CREATE TABLE with aggregation on sequence.field should
fail");
+ let msg = format!("{err:?}");
+ assert!(
+ msg.contains("sequence field") && msg.contains("amount"),
+ "expected sequence-field aggregation rejection, got {msg}"
+ );
+}
+
+/// CREATE TABLE must accept a function/type pair that the runtime ignores for
+/// primary-key columns: PK fields are copied through, so type compatibility is
+/// not checked for them.
+#[tokio::test]
+async fn test_pk_aggregation_create_table_accepts_ignored_function_on_pk() {
+ let (_tmp, sql_context) = setup_sql_context().await;
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.t_agg_pk_ok (
+ id INT NOT NULL, v INT,
+ PRIMARY KEY (id)
+ ) WITH (
+ 'bucket' = '1',
+ 'merge-engine' = 'aggregation',
'fields.id.aggregate-function' = 'listagg',
'fields.v.aggregate-function' = 'sum'
)",
)
.await
- .expect("CREATE TABLE with runtime-ignored function/type pairs should
succeed");
+ .expect("CREATE TABLE with runtime-ignored PK function/type pair
should succeed");
}
/// All-NULL aggregation group on a nullable `sum` column should emit NULL
diff --git a/crates/paimon/src/spec/aggregation.rs
b/crates/paimon/src/spec/aggregation.rs
index c5057f4..0de0f86 100644
--- a/crates/paimon/src/spec/aggregation.rs
+++ b/crates/paimon/src/spec/aggregation.rs
@@ -165,8 +165,9 @@ impl<'a> AggregationConfig<'a> {
///
/// For `aggregate-function` keys additionally:
/// * the function name must be one of the supported aggregators
+ /// * the field must not be listed in `sequence.field` — Java rejects
+ /// aggregation definitions on sequence fields during schema validation.
/// * the function must accept the field's declared data type — except for
- /// `sequence.field` columns (forced to `last_value` at runtime) and
/// primary-key columns (no aggregator; copied through), where the
/// configured function is ignored by the merge function's priority
/// order (Java `AggregateMergeFunction#getAggFuncName`), so only the
@@ -199,9 +200,15 @@ impl<'a> AggregationConfig<'a> {
});
};
if matches!(kind, FieldScopedOptionKind::AggregateFunction) {
- let runtime_ignores_function =
- sequence_fields.contains(&col) ||
primary_keys.iter().any(|pk| pk == col);
- if runtime_ignores_function {
+ if sequence_fields.contains(&col) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Should not define aggregation on sequence field:
'{col}'."
+ ),
+ });
+ }
+
+ if primary_keys.iter().any(|pk| pk == col) {
if !is_known_aggregator_name(value) {
return Err(crate::Error::ConfigInvalid {
message: format!(
@@ -210,9 +217,10 @@ impl<'a> AggregationConfig<'a> {
),
});
}
- } else {
- validate_aggregator_for_type(value, col,
field.data_type())?;
+ continue;
}
+
+ validate_aggregator_for_type(value, col, field.data_type())?;
}
}
@@ -597,21 +605,21 @@ mod tests {
}
#[test]
- fn test_validate_create_mode_skips_type_check_for_sequence_field() {
- // `listagg` is incompatible with INT, but `amount` is a sequence
- // field, so the runtime forces `last_value` and ignores the
- // configured function — the definition is usable and must be accepted.
+ fn test_validate_create_mode_rejects_aggregation_on_sequence_field() {
+ // Java rejects aggregation definitions on sequence fields during
+ // schema validation; the runtime still forces sequence fields to
+ // last_value when reading old or externally-created metadata.
let options = aggregation_options(&[
("sequence.field", "amount"),
("fields.amount.aggregate-function", "listagg"),
]);
- let config = AggregationConfig::new(&options);
-
- assert_eq!(
- config
- .validate_create_mode(&pk(), &sample_fields())
- .unwrap(),
- Some(AggregationMode::Basic)
+ let err = AggregationConfig::new(&options)
+ .validate_create_mode(&pk(), &sample_fields())
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sequence field") &&
message.contains("amount")),
+ "expected sequence-field aggregation rejection, got {err:?}"
);
}
@@ -631,20 +639,17 @@ mod tests {
}
#[test]
- fn
test_validate_create_mode_still_rejects_unknown_function_on_sequence_field() {
- // The function name itself must stay valid even when the runtime
- // would ignore it — typos should fail fast at CREATE TABLE.
- let options = aggregation_options(&[
- ("sequence.field", "amount"),
- ("fields.amount.aggregate-function", "lisstagg"),
- ]);
+ fn test_validate_create_mode_rejects_unknown_function_on_primary_key() {
+ // Primary-key fields are copied through at runtime, but a configured
+ // aggregation function name still must be valid so typos fail fast.
+ let options = aggregation_options(&[("fields.id.aggregate-function",
"lisstagg")]);
let err = AggregationConfig::new(&options)
.validate_create_mode(&pk(), &sample_fields())
.unwrap_err();
assert!(
matches!(err, crate::Error::ConfigInvalid { ref message }
- if message.contains("lisstagg") && message.contains("amount")),
- "expected unknown-function error on sequence field, got {err:?}"
+ if message.contains("lisstagg") && message.contains("id")),
+ "expected unknown-function error on primary-key field, got {err:?}"
);
}
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 0ad8a57..812a9b9 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -2090,6 +2090,36 @@ mod tests {
);
}
+ #[test]
+ fn test_aggregation_apply_changes_rejects_sequence_field_function() {
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("seq", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("merge-engine", "aggregation")
+ .option("sequence.field", "seq")
+ .option("fields.value.aggregate-function", "sum")
+ .build()
+ .unwrap(),
+ );
+
+ let err = table_schema
+ .apply_changes(vec![crate::spec::SchemaChange::set_option(
+ "fields.seq.aggregate-function".to_string(),
+ "sum".to_string(),
+ )])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sequence field") &&
message.contains("seq")),
+ "aggregation alter should reject sequence-field aggregate
function, got {err:?}"
+ );
+ }
+
#[test]
fn test_rename_column_rewrites_field_scoped_agg_options() {
let table_schema = TableSchema::new(
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index 0655109..523f955 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -2158,13 +2158,13 @@ mod tests {
#[test]
fn test_aggregate_merge_function_sequence_field_forced_last_value() {
- // 'tag' is the sequence field; even though user configured listagg,
- // it should be forced to last_value (so the latest tag survives).
+ // 'tag' is the sequence field; even though the table-level default is
+ // listagg, it should be forced to last_value (so the latest tag
survives).
let schema = aggregation_schema();
let output_schema = aggregation_output_schema();
let options = agg_options(&[
("fields.amount.aggregate-function", "sum"),
- ("fields.tag.aggregate-function", "listagg"),
+ ("fields.default-aggregate-function", "listagg"),
("sequence.field", "tag"),
]);
let mf = AggregateMergeFunction::new(
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 7d68f5f..88b3210 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -967,6 +967,10 @@ primary key includes all partition columns. It supports
per-field aggregate
functions such as `sum`, `min`, `max`, value functions, boolean functions, and
`listagg`, plus `fields.default-aggregate-function`.
+Sequence fields are always merged with `last_value`. Defining
+`fields.<sequence-field>.aggregate-function` is rejected, matching Java schema
+validation.
+
This is not full Java feature parity. Aggregation tables do not support retract
rows (`DELETE` / `UPDATE_BEFORE`), deletion vectors, cross-partition dynamic
bucket writes, or advanced aggregation options such as `ignore-retract`,
