diff --git a/crates/integrations/datafusion/tests/pk_tables.rs b/crates/integrations/datafusion/tests/pk_tables.rs index 83a24d24f..3e675636d 100644 --- a/crates/integrations/datafusion/tests/pk_tables.rs +++ b/crates/integrations/datafusion/tests/pk_tables.rs @@ -2674,6 +2674,57 @@ async fn test_pk_aggregation_mixed_aggregators() { assert_eq!(string_value(first_seen.as_ref(), 0), "a"); // first non-null wins } +/// Java registers `first_not_null_value` as an SPI alias of +/// `first_non_null_value`, so a schema written by Java/Flink can carry it. Both +/// CREATE TABLE and the read-side merge function must resolve it. Planning does +/// not build the merge function, so before the fix a table carrying the alias +/// planned fine and only failed once rows were pulled. +/// +/// The read side also has unit coverage in `paimon`; if this ever goes red, the +/// alias belongs in both places, not out of CREATE. +#[tokio::test] +async fn test_pk_aggregation_accepts_legacy_first_not_null_value() { + let (_tmp, sql_context) = setup_sql_context().await; + + sql_context + .sql( + "CREATE TABLE paimon.test_db.t_agg_legacy_name ( + id INT NOT NULL, first_seen STRING, + PRIMARY KEY (id) + ) WITH ( + 'bucket' = '1', + 'merge-engine' = 'aggregation', + 'fields.first_seen.aggregate-function' = 'first_not_null_value' + )", + ) + .await + .unwrap(); + + for values in ["(1, NULL)", "(1, 'b')", "(1, 'c')"] { + sql_context + .sql(&format!( + "INSERT INTO paimon.test_db.t_agg_legacy_name VALUES {values}" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + } + + let batches = sql_context + .sql("SELECT id, first_seen FROM paimon.test_db.t_agg_legacy_name") + .await + .unwrap() + .collect() + .await + .unwrap(); + + assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 1); + let first_seen = batches[0].column_by_name("first_seen").unwrap(); + assert_eq!(string_value(first_seen.as_ref(), 0), "b"); +} + /// `sequence.field` forces the named column to `last_value`, even when a /// table-level default aggregator would otherwise apply. #[tokio::test] diff --git a/crates/paimon/src/spec/aggregation.rs b/crates/paimon/src/spec/aggregation.rs index 7885a86ae..5f2b4c288 100644 --- a/crates/paimon/src/spec/aggregation.rs +++ b/crates/paimon/src/spec/aggregation.rs @@ -358,13 +358,33 @@ pub(crate) fn remove_field_scoped_options(options: &mut HashMap, const SUPPORTED_AGGREGATOR_NAMES_HINT: &str = "supported: sum, product, min, max, last_value, \ first_value, last_non_null_value, first_non_null_value, bool_and, bool_or, listagg"; +/// Java keeps `first_not_null_value` registered as an SPI alias of +/// `first_non_null_value`: `FieldFirstNonNullValueAggLegacyFactory` is listed in +/// `paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory` +/// next to the canonical factory, and both build the same +/// `FieldFirstNonNullValueAgg`. The alias survives from the switch-case to SPI +/// refactor and is deliberately undocumented, so a Java-written schema can carry +/// it even though nothing suggests it — which is why it is resolved here but +/// stays out of [`SUPPORTED_AGGREGATOR_NAMES_HINT`]. +/// +/// Matching is exact: Java's `FactoryUtil#discoverFactory` selects a factory +/// with `identifier().equals(identifier)`, so neither case nor `-`/`_` is +/// normalised. +pub(crate) fn canonical_aggregator_name(name: &str) -> &str { + match name { + "first_not_null_value" => "first_non_null_value", + other => other, + } +} + /// Whether `name` matches one of the basic-mode aggregator identifiers. Must /// stay in sync with the `match` arms in -/// `crate::table::aggregator::new_aggregator` — guarded by -/// `tests::validation_table_matches_constructors`. +/// `crate::table::aggregator::new_aggregator`. Both resolve the name through +/// [`canonical_aggregator_name`] first; `tests::validation_table_matches_constructors` +/// guards the type table below rather than this predicate. pub(crate) fn is_known_aggregator_name(name: &str) -> bool { matches!( - name, + canonical_aggregator_name(name), "sum" | "product" | "min" @@ -388,7 +408,9 @@ pub(crate) fn validate_aggregator_for_type( field_name: &str, dt: &DataType, ) -> crate::Result<()> { - let ok = match name { + // Errors below echo the caller's `name`, not the canonical one, so a user + // never reads back a function name they did not write. + let ok = match canonical_aggregator_name(name) { "sum" => matches!( dt, DataType::TinyInt(_) @@ -677,6 +699,48 @@ mod tests { ); } + /// Java registers `first_not_null_value` as an SPI alias of + /// `first_non_null_value`, so a Java-written schema can carry it on all three + /// name-keyed paths: a non-primary-key column (type-checked), a primary-key + /// column (name-only), and `fields.default-aggregate-function`. + #[test] + fn test_legacy_first_not_null_value_alias_is_accepted() { + for (key, value) in [ + ("fields.amount.aggregate-function", "first_not_null_value"), + ("fields.id.aggregate-function", "first_not_null_value"), + (FIELDS_DEFAULT_AGG_FUNCTION_OPTION, "first_not_null_value"), + ] { + let options = aggregation_options(&[(key, value)]); + AggregationConfig::new(&options) + .validate_create_mode(&pk(), &sample_fields()) + .unwrap_or_else(|err| panic!("'{key}' = '{value}' should be accepted: {err:?}")); + } + } + + /// The alias resolves by exact match, mirroring Java's + /// `FactoryUtil#discoverFactory`, which compares identifiers with `equals`. + /// Neither case nor `-`/`_` is normalised, and the reported name is the one + /// the user wrote. + #[test] + fn test_legacy_alias_near_misses_are_still_rejected() { + for name in [ + "first_not_null_valu", + "first-not-null-value", + "FIRST_NOT_NULL_VALUE", + "not_null_value", + ] { + let options = aggregation_options(&[("fields.amount.aggregate-function", name)]); + let err = AggregationConfig::new(&options) + .validate_create_mode(&pk(), &sample_fields()) + .unwrap_err(); + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains(name) && message.contains("amount")), + "'{name}' should be rejected and echoed back verbatim, got {err:?}" + ); + } + } + #[test] fn test_rejects_aggregation_on_sequence_field_for_every_merge_engine() { // Java rejects aggregation definitions on sequence fields inside @@ -791,6 +855,9 @@ mod tests { "first_value", "last_non_null_value", "first_non_null_value", + // Java's SPI alias of the entry above; both sides must resolve it + // identically, which is exactly what this test locks. + "first_not_null_value", "bool_and", "bool_or", "listagg", diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs index a165ffe07..8eac99577 100644 --- a/crates/paimon/src/spec/mod.rs +++ b/crates/paimon/src/spec/mod.rs @@ -43,7 +43,7 @@ pub(crate) use partial_update::PartialUpdateConfig; mod aggregation; pub(crate) use aggregation::{ - remove_field_scoped_options, rename_field_scoped_options, + canonical_aggregator_name, remove_field_scoped_options, rename_field_scoped_options, validate_no_aggregation_on_sequence_field, AggregationConfig, }; diff --git a/crates/paimon/src/spec/partial_update.rs b/crates/paimon/src/spec/partial_update.rs index d7fd9d73a..25d6771f5 100644 --- a/crates/paimon/src/spec/partial_update.rs +++ b/crates/paimon/src/spec/partial_update.rs @@ -350,6 +350,10 @@ impl<'a> PartialUpdateConfig<'a> { continue; }; validate_aggregator_for_type(function, field_name, field.data_type())?; + // Compared raw on purpose: Java `PartialUpdateMergeFunction` tests + // `aggFuncName.equals(FieldLastNonNullValueAggFactory.NAME)`, and that + // factory has no legacy alias — only `first_non_null_value` does — so + // there is nothing here to canonicalize. if function != "last_non_null_value" && !protected_fields.contains(field_name) { return Err(crate::Error::ConfigInvalid { message: format!( @@ -807,6 +811,33 @@ mod tests { ); } + /// The aggregate-function name helpers are shared with the aggregation merge + /// engine, so Java's `first_not_null_value` alias has to be accepted here + /// too. It is not `last_non_null_value`, so it still needs a sequence group. + #[test] + fn test_validate_aggregate_functions_accepts_legacy_first_not_null_value() { + let options = partial_update_options(&[ + ("fields.version.sequence-group", "price"), + ("fields.price.aggregate-function", "first_not_null_value"), + ]); + let config = PartialUpdateConfig::new(&options); + let fields = vec![ + DataField::new(0, "id".to_string(), DataType::Int(IntType::new())), + DataField::new(1, "version".to_string(), DataType::Int(IntType::new())), + DataField::new(2, "price".to_string(), DataType::Int(IntType::new())), + ]; + + let functions = config + .validated_aggregate_functions(&fields, &["id".to_string()]) + .unwrap(); + + assert_eq!( + functions.get("price").map(String::as_str), + Some("first_not_null_value"), + "the configured spelling must be preserved, got {functions:?}" + ); + } + #[test] fn test_validate_aggregate_functions_rejects_unknown_field() { let options = diff --git a/crates/paimon/src/table/aggregator/mod.rs b/crates/paimon/src/table/aggregator/mod.rs index 787c8d604..d92e02f36 100644 --- a/crates/paimon/src/table/aggregator/mod.rs +++ b/crates/paimon/src/table/aggregator/mod.rs @@ -96,7 +96,16 @@ pub(crate) fn new_aggregator( data_type: &DataType, table_options: &HashMap, ) -> crate::Result> { - match name { + // `canonical_aggregator_name` folds Java's undocumented SPI alias + // `first_not_null_value` onto `first_non_null_value`; the error arm still + // echoes the caller's spelling. The aggregator itself reports the canonical + // name from `FieldAggregator::name`, which costs no parity: the one Java + // message that repeats the configured identifier is the retract rejection in + // `FieldAggregator`, and retract is rejected here before an aggregator is + // ever built. Java's non-nullable diagnostic + // (`AggregateMergeFunction`: "Field can not be null") names no function + // at all, so the Rust equivalent in `sort_merge` is strictly more specific. + match crate::spec::canonical_aggregator_name(name) { "sum" => Ok(Box::new(SumAgg::new(field_name, data_type)?)), "product" => Ok(Box::new(ProductAgg::new(field_name, data_type)?)), "min" => Ok(Box::new(MinAgg::new(field_name, data_type)?)), @@ -112,9 +121,9 @@ pub(crate) fn new_aggregator( data_type, table_options, )?)), - other => Err(crate::Error::ConfigInvalid { + _ => Err(crate::Error::ConfigInvalid { message: format!( - "Unknown aggregate function '{other}' for field '{field_name}'; \ + "Unknown aggregate function '{name}' for field '{field_name}'; \ supported: sum, product, min, max, last_value, first_value, \ last_non_null_value, first_non_null_value, bool_and, bool_or, listagg" ), @@ -136,3 +145,57 @@ pub(crate) fn unsupported_type_error( ), } } + +#[cfg(test)] +mod tests { + use arrow_array::Int32Array; + + use super::*; + use crate::spec::IntType; + + /// Java's legacy alias must build the very same aggregator, not merely pass + /// validation: `FieldFirstNonNullValueAggLegacyFactory` returns a + /// `FieldFirstNonNullValueAgg`, so both names have to lock the first + /// non-null value. + #[test] + fn test_legacy_first_not_null_value_aggregates_like_the_canonical_name() { + let data_type = DataType::Int(IntType::new()); + let options = HashMap::new(); + let input = Int32Array::from(vec![None, Some(5), Some(7)]); + + let mut results = Vec::new(); + for name in ["first_non_null_value", "first_not_null_value"] { + let mut agg = new_aggregator(name, "v", &data_type, &options) + .unwrap_or_else(|err| panic!("'{name}' should construct: {err:?}")); + for row in 0..input.len() { + agg.agg(&input, row).unwrap(); + } + let out = agg.result().unwrap(); + let out = out + .as_any() + .downcast_ref::() + .expect("Int32 result"); + results.push(out.is_valid(0).then(|| out.value(0))); + } + assert_eq!(results[0], Some(5)); + assert_eq!(results[0], results[1], "alias diverged from canonical name"); + } + + /// An unknown name is echoed back exactly as written, so resolving the alias + /// never renames what the user configured. + #[test] + fn test_unknown_aggregate_function_is_echoed_verbatim() { + let err = new_aggregator( + "first_not_null_valu", + "v", + &DataType::Int(IntType::new()), + &HashMap::new(), + ) + .unwrap_err(); + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("'first_not_null_valu'")), + "expected the caller's spelling, got {err:?}" + ); + } +}