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 78e7ebb3 fix(spec): accept the legacy first_not_null_value aggregate
name (#794)
78e7ebb3 is described below
commit 78e7ebb3bede40ef8783bda39721dad34a3de8d2
Author: jackylee <[email protected]>
AuthorDate: Thu Sep 10 15:41:30 2026 +0800
fix(spec): accept the legacy first_not_null_value aggregate name (#794)
---
crates/integrations/datafusion/tests/pk_tables.rs | 51 +++++++++++++++
crates/paimon/src/spec/aggregation.rs | 75 +++++++++++++++++++++--
crates/paimon/src/spec/mod.rs | 2 +-
crates/paimon/src/spec/partial_update.rs | 31 ++++++++++
crates/paimon/src/table/aggregator/mod.rs | 69 ++++++++++++++++++++-
5 files changed, 220 insertions(+), 8 deletions(-)
diff --git a/crates/integrations/datafusion/tests/pk_tables.rs
b/crates/integrations/datafusion/tests/pk_tables.rs
index 83a24d24..3e675636 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::<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]
diff --git a/crates/paimon/src/spec/aggregation.rs
b/crates/paimon/src/spec/aggregation.rs
index 7885a86a..5f2b4c28 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<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",
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index a165ffe0..8eac9957 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 d7fd9d73..25d6771f 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 787c8d60..d92e02f3 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<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:?}"
+ );
+ }
+}
