fix(spec): accept the legacy first_not_null_value aggregate name

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::<usize>(), 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]

crates/paimon/src/spec/aggregation.rs

@@ -358,13 +358,33 @@ pub(crate) fn remove_field_scoped_options(options: &mut HashMap<String, String>,
 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",

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,
 };
 

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 =

crates/paimon/src/table/aggregator/mod.rs

@@ -96,7 +96,16 @@ pub(crate) fn new_aggregator(
     data_type: &DataType,
     table_options: &HashMap<String, String>,
 ) -> crate::Result<Box<dyn FieldAggregator>> {
-    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 <i> 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::<Int32Array>()
+                .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:?}"
+        );
+    }
+}