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 cac63da3 partial_update: support field aggregation reads (#598)
cac63da3 is described below
commit cac63da3a6ce2ab05d4f9d091237a3004d270d7f
Author: shyjsarah <[email protected]>
AuthorDate: Thu Jul 23 20:58:01 2026 +0800
partial_update: support field aggregation reads (#598)
---
crates/integrations/datafusion/tests/pk_tables.rs | 104 +++++
crates/paimon/src/spec/partial_update.rs | 373 ++++++++++++++++-
crates/paimon/src/table/aggregator/bool_agg.rs | 23 ++
crates/paimon/src/table/aggregator/listagg.rs | 68 +++-
crates/paimon/src/table/aggregator/mod.rs | 8 +
crates/paimon/src/table/aggregator/numeric.rs | 256 +++++++++++-
crates/paimon/src/table/aggregator/value.rs | 122 +++++-
crates/paimon/src/table/kv_file_reader.rs | 105 ++++-
crates/paimon/src/table/sort_merge.rs | 461 ++++++++++++++++++++--
docs/src/sql.md | 18 +-
10 files changed, 1476 insertions(+), 62 deletions(-)
diff --git a/crates/integrations/datafusion/tests/pk_tables.rs
b/crates/integrations/datafusion/tests/pk_tables.rs
index 0bad6e3b..d51b0052 100644
--- a/crates/integrations/datafusion/tests/pk_tables.rs
+++ b/crates/integrations/datafusion/tests/pk_tables.rs
@@ -38,6 +38,8 @@ use datafusion::arrow::datatypes::{
};
use paimon::catalog::Identifier;
use paimon::Catalog;
+use paimon_datafusion::PaimonTableProvider;
+use std::collections::HashMap;
use std::sync::Arc;
// ======================= Basic PK Write + Read =======================
@@ -178,6 +180,108 @@ async fn test_pk_partial_update_fixed_bucket_e2e() {
);
}
+#[tokio::test]
+async fn test_pk_partial_update_sequence_group_aggregation_read_e2e() {
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+ sql_context
+ .sql("CREATE SCHEMA paimon.test_db")
+ .await
+ .unwrap();
+
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.t_partial_update_aggregation (
+ id INT NOT NULL,
+ version INT,
+ amount INT,
+ tag STRING,
+ PRIMARY KEY (id)
+ ) WITH (
+ 'bucket' = '1',
+ 'merge-engine' = 'partial-update'
+ )",
+ )
+ .await
+ .unwrap();
+
+ for values in [
+ "(1, 10, 10, 'b'), (2, 5, 3, 'x')",
+ "(1, 9, 20, 'a'), (2, 4, 4, 'w')",
+ "(1, 11, 5, 'c')",
+ ] {
+ sql_context
+ .sql(&format!(
+ "INSERT INTO paimon.test_db.t_partial_update_aggregation
VALUES {values}"
+ ))
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ }
+
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t_partial_update_aggregation"))
+ .await
+ .unwrap()
+ .copy_with_options(HashMap::from([
+ (
+ "fields.version.sequence-group".to_string(),
+ "amount,tag".to_string(),
+ ),
+ (
+ "fields.amount.aggregate-function".to_string(),
+ "sum".to_string(),
+ ),
+ (
+ "fields.tag.aggregate-function".to_string(),
+ "listagg".to_string(),
+ ),
+ ]));
+ let provider = PaimonTableProvider::try_new(table).unwrap();
+ sql_context
+ .register_temp_table(
+ "paimon.test_db.t_partial_update_aggregation",
+ Arc::new(provider),
+ )
+ .unwrap();
+
+ let batches = sql_context
+ .sql(
+ "SELECT id
+ FROM paimon.test_db.t_partial_update_aggregation
+ WHERE amount = 7 AND tag = 'w,x'",
+ )
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ let batch = &batches[0];
+ assert_eq!(
+ batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 2
+ );
+ assert_eq!(
+ batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>(),
+ vec!["id"]
+ );
+}
+
#[tokio::test]
async fn test_pk_partial_update_ignore_delete_alias_e2e() {
let (_tmp, catalog) = create_test_env();
diff --git a/crates/paimon/src/spec/partial_update.rs
b/crates/paimon/src/spec/partial_update.rs
index f1c84e45..b8a0018e 100644
--- a/crates/paimon/src/spec/partial_update.rs
+++ b/crates/paimon/src/spec/partial_update.rs
@@ -17,6 +17,7 @@
use std::collections::{HashMap, HashSet};
+use super::aggregation::{is_known_aggregator_name,
validate_aggregator_for_type};
use crate::spec::DataField;
const MERGE_ENGINE_OPTION: &str = "merge-engine";
@@ -32,6 +33,11 @@ const FIELDS_DEFAULT_AGG_FUNCTION_OPTION: &str =
"fields.default-aggregate-funct
const FIELDS_PREFIX: &str = "fields.";
const SEQUENCE_GROUP_SUFFIX: &str = ".sequence-group";
const AGGREGATION_FUNCTION_SUFFIX: &str = ".aggregate-function";
+const LIST_AGG_DELIMITER_SUFFIX: &str = ".list-agg-delimiter";
+const IGNORE_RETRACT_SUFFIX: &str = ".ignore-retract";
+const DISTINCT_SUFFIX: &str = ".distinct";
+const NESTED_KEY_SUFFIX: &str = ".nested-key";
+const COUNT_LIMIT_SUFFIX: &str = ".count-limit";
/// Partial-update mode recognized by the current Rust implementation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -57,8 +63,8 @@ impl SequenceGroup {
/// Partial-update-specific option inspection and validation.
///
-/// Reads support basic partial update and sequence groups. Table creation and
-/// writes remain restricted to basic partial update.
+/// Reads support basic partial update, sequence groups, and field aggregation.
+/// Table creation and writes remain restricted to basic partial update.
#[derive(Debug, Clone, Copy)]
pub(crate) struct PartialUpdateConfig<'a> {
options: &'a HashMap<String, String>,
@@ -243,6 +249,120 @@ impl<'a> PartialUpdateConfig<'a> {
Ok(required)
}
+ pub(crate) fn validated_aggregate_functions(
+ &self,
+ fields: &[DataField],
+ primary_keys: &[String],
+ ) -> crate::Result<HashMap<String, String>> {
+ let groups = self.validated_sequence_groups(fields, primary_keys)?;
+ let sequence_fields: HashSet<&str> = groups
+ .iter()
+ .flat_map(|group| group.sequence_fields.iter().map(String::as_str))
+ .collect();
+ let protected_fields: HashSet<&str> = groups
+ .iter()
+ .flat_map(|group|
group.protected_fields.iter().map(String::as_str))
+ .collect();
+ let field_names: HashSet<&str> =
fields.iter().map(DataField::name).collect();
+ let primary_keys: HashSet<&str> =
primary_keys.iter().map(String::as_str).collect();
+
+ for (key, value) in self
+ .options
+ .iter()
+ .filter(|(key, _)| is_fields_option_with_suffix(key,
LIST_AGG_DELIMITER_SUFFIX))
+ {
+ let field_name = key
+ .strip_prefix(FIELDS_PREFIX)
+ .and_then(|key| key.strip_suffix(LIST_AGG_DELIMITER_SUFFIX))
+ .filter(|field| !field.is_empty())
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!("Invalid partial-update listagg option
'{key}={value}'"),
+ })?;
+ if !field_names.contains(field_name) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Aggregation field '{field_name}' referenced by
'{key}' is not declared \
+ in the table schema"
+ ),
+ });
+ }
+ }
+
+ let mut per_field = HashMap::new();
+ for (key, value) in self
+ .options
+ .iter()
+ .filter(|(key, _)| is_fields_option_with_suffix(key,
AGGREGATION_FUNCTION_SUFFIX))
+ {
+ let field_name = key
+ .strip_prefix(FIELDS_PREFIX)
+ .and_then(|key| key.strip_suffix(AGGREGATION_FUNCTION_SUFFIX))
+ .filter(|field| !field.is_empty())
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!(
+ "Invalid partial-update aggregate-function option
'{key}={value}'"
+ ),
+ })?;
+ if !field_names.contains(field_name) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Aggregation field '{field_name}' referenced by
'{key}' is not declared \
+ in the table schema"
+ ),
+ });
+ }
+ if !is_known_aggregator_name(value) {
+ validate_aggregator_for_type(
+ value,
+ field_name,
+ fields
+ .iter()
+ .find(|field| field.name() == field_name)
+ .expect("field existence checked above")
+ .data_type(),
+ )?;
+ }
+ per_field.insert(field_name, value.as_str());
+ }
+
+ let default = self
+ .options
+ .get(FIELDS_DEFAULT_AGG_FUNCTION_OPTION)
+ .map(String::as_str);
+ if let Some(default) = default {
+ if !is_known_aggregator_name(default) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Unknown aggregate function '{default}' configured via
\
+ '{FIELDS_DEFAULT_AGG_FUNCTION_OPTION}'"
+ ),
+ });
+ }
+ }
+
+ let mut functions = HashMap::new();
+ for field in fields {
+ let field_name = field.name();
+ if sequence_fields.contains(field_name) ||
primary_keys.contains(field_name) {
+ continue;
+ }
+ let Some(function) =
per_field.get(field_name).copied().or(default) else {
+ continue;
+ };
+ validate_aggregator_for_type(function, field_name,
field.data_type())?;
+ if function != "last_non_null_value" &&
!protected_fields.contains(field_name) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Must use sequence group for aggregate function
'{function}' on field \
+ '{field_name}'"
+ ),
+ });
+ }
+ functions.insert(field_name.to_string(), function.to_string());
+ }
+ Ok(functions)
+ }
+
fn validated_mode(
&self,
has_primary_keys: bool,
@@ -277,6 +397,9 @@ impl<'a> PartialUpdateConfig<'a> {
.filter(|key| {
is_unsupported_partial_update_option(key)
&& !is_fields_option_with_suffix(key,
SEQUENCE_GROUP_SUFFIX)
+ && !is_fields_option_with_suffix(key,
AGGREGATION_FUNCTION_SUFFIX)
+ && !is_fields_option_with_suffix(key,
LIST_AGG_DELIMITER_SUFFIX)
+ && key.as_str() != FIELDS_DEFAULT_AGG_FUNCTION_OPTION
})
.cloned()
.collect();
@@ -294,6 +417,11 @@ fn is_unsupported_partial_update_option(key: &str) -> bool
{
|| key == FIELDS_DEFAULT_AGG_FUNCTION_OPTION
|| is_fields_option_with_suffix(key, SEQUENCE_GROUP_SUFFIX)
|| is_fields_option_with_suffix(key, AGGREGATION_FUNCTION_SUFFIX)
+ || is_fields_option_with_suffix(key, LIST_AGG_DELIMITER_SUFFIX)
+ || is_fields_option_with_suffix(key, IGNORE_RETRACT_SUFFIX)
+ || is_fields_option_with_suffix(key, DISTINCT_SUFFIX)
+ || is_fields_option_with_suffix(key, NESTED_KEY_SUFFIX)
+ || is_fields_option_with_suffix(key, COUNT_LIMIT_SUFFIX)
}
fn is_fields_option_with_suffix(key: &str, suffix: &str) -> bool {
@@ -395,6 +523,11 @@ mod tests {
"fields.price.ignore-delete",
"fields.price.sequence-group",
"fields.price.aggregate-function",
+ "fields.price.list-agg-delimiter",
+ "fields.price.ignore-retract",
+ "fields.price.distinct",
+ "fields.price.nested-key",
+ "fields.price.count-limit",
FIELDS_DEFAULT_AGG_FUNCTION_OPTION,
] {
let options = partial_update_options(&[(key, "value")]);
@@ -433,6 +566,51 @@ mod tests {
);
}
+ #[test]
+ fn test_validate_read_mode_accepts_field_aggregation() {
+ let options =
+ partial_update_options(&[("fields.price.aggregate-function",
"last_non_null_value")]);
+ let config = PartialUpdateConfig::new(&options);
+
+ assert_eq!(
+ config.validate_read_mode(true, "default.t").unwrap(),
+ Some(PartialUpdateMode::Basic)
+ );
+ }
+
+ #[test]
+ fn test_validate_read_mode_accepts_sequence_group_field_aggregation() {
+ let options = partial_update_options(&[
+ ("fields.version.sequence-group", "price"),
+ ("fields.price.aggregate-function", "sum"),
+ ]);
+ let config = PartialUpdateConfig::new(&options);
+
+ assert_eq!(
+ config.validate_read_mode(true, "default.t").unwrap(),
+ Some(PartialUpdateMode::SequenceGroup)
+ );
+ }
+
+ #[test]
+ fn test_validate_read_mode_rejects_unsupported_aggregation_modifiers() {
+ for key in [
+ "fields.price.ignore-retract",
+ "fields.price.distinct",
+ "fields.price.nested-key",
+ "fields.price.count-limit",
+ ] {
+ let options = partial_update_options(&[(key, "value")]);
+ let config = PartialUpdateConfig::new(&options);
+ let err = config.validate_read_mode(true,
"default.t").unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message } if
message.contains(key)),
+ "expected read-time rejection to mention '{key}', got {err:?}"
+ );
+ }
+ }
+
#[test]
fn test_parse_sequence_groups() {
let options = partial_update_options(&[
@@ -564,4 +742,195 @@ mod tests {
]
);
}
+
+ #[test]
+ fn
test_validate_aggregate_functions_accepts_last_non_null_without_sequence_group()
{
+ let options =
+ partial_update_options(&[("fields.price.aggregate-function",
"last_non_null_value")]);
+ let config = PartialUpdateConfig::new(&options);
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "price".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ assert_eq!(
+ config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap(),
+ HashMap::from([("price".to_string(),
"last_non_null_value".to_string())])
+ );
+ }
+
+ #[test]
+ fn
test_validate_aggregate_functions_rejects_non_last_without_sequence_group() {
+ let options =
partial_update_options(&[("fields.price.aggregate-function", "sum")]);
+ let config = PartialUpdateConfig::new(&options);
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "price".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ let err = config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sum")
+ && message.contains("price")
+ && message.contains("sequence group")),
+ "expected missing sequence-group error, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn test_validate_aggregate_functions_rejects_unknown_function() {
+ let options = partial_update_options(&[
+ ("fields.version.sequence-group", "price"),
+ ("fields.price.aggregate-function", "sume"),
+ ]);
+ 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 err = config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sume") && message.contains("price")),
+ "expected unknown function error, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn test_validate_aggregate_functions_rejects_unknown_field() {
+ let options =
+ partial_update_options(&[("fields.prcie.aggregate-function",
"last_non_null_value")]);
+ let config = PartialUpdateConfig::new(&options);
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(1, "price".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ let err = config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("prcie")
+ && message.contains("fields.prcie.aggregate-function")),
+ "expected unknown field error, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn
test_validate_aggregate_functions_rejects_unknown_listagg_delimiter_field() {
+ let options = partial_update_options(&[
+ ("fields.version.sequence-group", "tag"),
+ ("fields.tag.aggregate-function", "listagg"),
+ ("fields.tga.list-agg-delimiter", "|"),
+ ]);
+ 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,
+ "tag".to_string(),
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ ),
+ ];
+
+ let err = config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("tga")
+ && message.contains("fields.tga.list-agg-delimiter")),
+ "expected unknown delimiter field error, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn test_validate_aggregate_functions_rejects_incompatible_type() {
+ let options = partial_update_options(&[
+ ("fields.version.sequence-group", "name"),
+ ("fields.name.aggregate-function", "sum"),
+ ]);
+ 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,
+ "name".to_string(),
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ ),
+ ];
+
+ let err = config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sum") && message.contains("name")),
+ "expected incompatible type error, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn test_validate_default_aggregate_function_applies_to_protected_fields() {
+ let options = partial_update_options(&[
+ ("fields.version.sequence-group", "price"),
+ (FIELDS_DEFAULT_AGG_FUNCTION_OPTION, "sum"),
+ ]);
+ 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())),
+ ];
+
+ assert_eq!(
+ config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap(),
+ HashMap::from([("price".to_string(), "sum".to_string())])
+ );
+ }
+
+ #[test]
+ fn test_per_field_aggregate_function_overrides_default() {
+ let options = partial_update_options(&[
+ ("fields.version.sequence-group", "price"),
+ ("fields.price.aggregate-function", "sum"),
+ (FIELDS_DEFAULT_AGG_FUNCTION_OPTION, "last_non_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())),
+ DataField::new(3, "note".to_string(),
DataType::Int(IntType::new())),
+ ];
+
+ assert_eq!(
+ config
+ .validated_aggregate_functions(&fields, &["id".to_string()])
+ .unwrap(),
+ HashMap::from([
+ ("price".to_string(), "sum".to_string()),
+ ("note".to_string(), "last_non_null_value".to_string()),
+ ])
+ );
+ }
}
diff --git a/crates/paimon/src/table/aggregator/bool_agg.rs
b/crates/paimon/src/table/aggregator/bool_agg.rs
index 31f4f3a8..9cb5b6a6 100644
--- a/crates/paimon/src/table/aggregator/bool_agg.rs
+++ b/crates/paimon/src/table/aggregator/bool_agg.rs
@@ -82,6 +82,10 @@ impl FieldAggregator for BoolAndAgg {
Ok(())
}
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ self.agg(array, row_idx)
+ }
+
fn result(&self) -> crate::Result<ArrayRef> {
Ok(Arc::new(BooleanArray::from(vec![self.acc])))
}
@@ -132,6 +136,10 @@ impl FieldAggregator for BoolOrAgg {
Ok(())
}
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ self.agg(array, row_idx)
+ }
+
fn result(&self) -> crate::Result<ArrayRef> {
Ok(Arc::new(BooleanArray::from(vec![self.acc])))
}
@@ -197,6 +205,21 @@ mod tests {
assert_eq!(collect(or_agg.result().unwrap()), None);
}
+ #[test]
+ fn test_bool_aggregators_accept_reversed_inputs() {
+ let arr = BooleanArray::from(vec![Some(true), Some(false)]);
+
+ let mut and_agg = BoolAndAgg::new("b",
&DataType::Boolean(BooleanType::new())).unwrap();
+ and_agg.agg(&arr, 0).unwrap();
+ and_agg.agg_reversed(&arr, 1).unwrap();
+ assert_eq!(collect(and_agg.result().unwrap()), Some(false));
+
+ let mut or_agg = BoolOrAgg::new("b",
&DataType::Boolean(BooleanType::new())).unwrap();
+ or_agg.agg(&arr, 1).unwrap();
+ or_agg.agg_reversed(&arr, 0).unwrap();
+ assert_eq!(collect(or_agg.result().unwrap()), Some(true));
+ }
+
#[test]
fn test_bool_and_rejects_non_boolean_type() {
let err = BoolAndAgg::new("b",
&DataType::Int(IntType::new())).unwrap_err();
diff --git a/crates/paimon/src/table/aggregator/listagg.rs
b/crates/paimon/src/table/aggregator/listagg.rs
index 945cb370..c94a6bd2 100644
--- a/crates/paimon/src/table/aggregator/listagg.rs
+++ b/crates/paimon/src/table/aggregator/listagg.rs
@@ -43,6 +43,23 @@ fn list_agg_delimiter<'a>(field_name: &str, options: &'a
HashMap<String, String>
.unwrap_or(DEFAULT_DELIMITER)
}
+fn java_is_blank(value: &str) -> bool {
+ value.chars().all(|ch| {
+ matches!(
+ ch,
+ '\u{0009}'..='\u{000d}'
+ | '\u{001c}'..='\u{0020}'
+ | '\u{1680}'
+ | '\u{2000}'..='\u{2006}'
+ | '\u{2008}'..='\u{200a}'
+ | '\u{2028}'
+ | '\u{2029}'
+ | '\u{205f}'
+ | '\u{3000}'
+ )
+ })
+}
+
#[derive(Debug)]
pub(crate) struct ListaggAgg {
field_name: String,
@@ -103,6 +120,9 @@ impl FieldAggregator for ListaggAgg {
source: None,
})?;
let v = arr.value(row_idx);
+ if java_is_blank(v) {
+ return Ok(());
+ }
match &mut self.acc {
None => self.acc = Some(v.to_string()),
Some(prev) => {
@@ -113,6 +133,32 @@ impl FieldAggregator for ListaggAgg {
Ok(())
}
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ if array.is_null(row_idx) {
+ return Ok(());
+ }
+ let arr = array
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "listagg column '{}' received non-Utf8 Arrow array {:?}",
+ self.field_name,
+ array.data_type()
+ ),
+ source: None,
+ })?;
+ let value = arr.value(row_idx);
+ if java_is_blank(value) {
+ return Ok(());
+ }
+ self.acc = Some(match self.acc.take() {
+ None => value.to_string(),
+ Some(current) => format!("{value}{}{current}", self.delimiter),
+ });
+ Ok(())
+ }
+
fn result(&self) -> crate::Result<ArrayRef> {
Ok(Arc::new(StringArray::from(vec![self.acc.clone()])))
}
@@ -140,7 +186,7 @@ mod tests {
#[test]
fn test_listagg_default_delimiter_skips_null() {
let mut agg = ListaggAgg::new("v", &varchar_type(),
&HashMap::new()).unwrap();
- let arr = StringArray::from(vec![Some("a"), None, Some("b"),
Some("c")]);
+ let arr = StringArray::from(vec![Some("a"), None, Some(" "),
Some("b"), Some("c")]);
for i in 0..arr.len() {
agg.agg(&arr, i).unwrap();
}
@@ -228,4 +274,24 @@ mod tests {
agg.reset();
assert_eq!(collect(agg.result().unwrap()), None);
}
+
+ #[test]
+ fn test_listagg_reversed_prepends_non_blank_value() {
+ let mut agg = ListaggAgg::new("v", &varchar_type(),
&HashMap::new()).unwrap();
+ let arr = StringArray::from(vec![Some("b"), Some("a"), Some(" ")]);
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ agg.agg_reversed(&arr, 2).unwrap();
+ assert_eq!(collect(agg.result().unwrap()), Some("a,b".to_string()));
+ }
+
+ #[test]
+ fn test_listagg_java_blank_semantics_preserve_non_breaking_space() {
+ let mut agg = ListaggAgg::new("v", &varchar_type(),
&HashMap::new()).unwrap();
+ let arr = StringArray::from(vec![Some(" "), Some("\u{00a0}")]);
+ for i in 0..arr.len() {
+ agg.agg(&arr, i).unwrap();
+ }
+ assert_eq!(collect(agg.result().unwrap()),
Some("\u{00a0}".to_string()));
+ }
}
diff --git a/crates/paimon/src/table/aggregator/mod.rs
b/crates/paimon/src/table/aggregator/mod.rs
index 02e441be..787c8d60 100644
--- a/crates/paimon/src/table/aggregator/mod.rs
+++ b/crates/paimon/src/table/aggregator/mod.rs
@@ -67,6 +67,14 @@ pub(crate) trait FieldAggregator: Send + Sync +
std::fmt::Debug {
/// Accumulate one input cell.
fn agg(&mut self, array: &dyn Array, row_idx: usize) -> crate::Result<()>;
+ /// Accumulate an input that sorts before the current accumulator.
+ ///
+ /// This mirrors Java `FieldAggregator#aggReversed(accumulator, input)`,
+ /// whose default semantics are `agg(input, accumulator)`. Implementations
+ /// must define this explicitly so order-sensitive aggregators cannot
+ /// silently fall back to forward accumulation.
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()>;
+
/// Materialize the current accumulator as a 1-row Arrow array.
fn result(&self) -> crate::Result<ArrayRef>;
}
diff --git a/crates/paimon/src/table/aggregator/numeric.rs
b/crates/paimon/src/table/aggregator/numeric.rs
index 8ab7b6d2..f4851193 100644
--- a/crates/paimon/src/table/aggregator/numeric.rs
+++ b/crates/paimon/src/table/aggregator/numeric.rs
@@ -25,7 +25,7 @@
//! produce misleading aggregated values. A Decimal `sum` whose result no
//! longer fits the declared precision yields a NULL cell, matching Java
//! `DecimalUtils.add` / `Decimal.fromBigDecimal` (which return null on
-//! precision overflow rather than throwing).
+//! precision or backing `i128` overflow rather than throwing).
//!
//! `min` / `max` extend to every ordered Paimon type: numerics, Decimal,
//! Date, Time, Timestamp, and Char/VarChar. Comparison is by native value
@@ -166,19 +166,37 @@ impl FieldAggregator for SumAgg {
let v = downcast::<Float64Array>(array,
&self.field_name)?.value(row_idx);
*acc = Some(acc.map_or(v, |prev| prev + v));
}
- SumState::Decimal128 { acc, .. } => {
+ SumState::Decimal128 { precision, acc, .. } => {
let v = downcast::<Decimal128Array>(array,
&self.field_name)?.value(row_idx);
- *acc = Some(match *acc {
- None => v,
- Some(prev) => prev
- .checked_add(v)
- .ok_or_else(|| overflow_error("sum",
&self.field_name))?,
- });
+ let next = match *acc {
+ None => Some(v),
+ Some(prev) => prev.checked_add(v),
+ };
+ *acc = next.filter(|value| decimal_fits_precision(*value,
*precision));
}
}
Ok(())
}
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ if array.is_null(row_idx) {
+ return Ok(());
+ }
+ match &mut self.state {
+ SumState::F32(acc) => {
+ let v = downcast::<Float32Array>(array,
&self.field_name)?.value(row_idx);
+ *acc = Some(acc.map_or(v, |prev| v + prev));
+ Ok(())
+ }
+ SumState::F64(acc) => {
+ let v = downcast::<Float64Array>(array,
&self.field_name)?.value(row_idx);
+ *acc = Some(acc.map_or(v, |prev| v + prev));
+ Ok(())
+ }
+ _ => self.agg(array, row_idx),
+ }
+ }
+
fn result(&self) -> crate::Result<ArrayRef> {
Ok(match &self.state {
SumState::I8(acc) => Arc::new(Int8Array::from(vec![*acc])),
@@ -194,8 +212,9 @@ impl FieldAggregator for SumAgg {
} => {
// Java parity: `DecimalUtils.add` -> `Decimal.fromBigDecimal`
// returns null when the summed value no longer fits the
- // declared precision, so an overflowing sum yields a NULL cell
- // rather than a silently out-of-range Decimal.
+ // declared precision (or the backing i128), so an overflowing
+ // sum yields a NULL cell rather than a silently out-of-range
+ // Decimal.
let fitted = acc.filter(|v| decimal_fits_precision(*v,
*precision));
decimal_array(*precision, *scale, fitted, "sum",
&self.field_name)?
}
@@ -328,6 +347,25 @@ impl FieldAggregator for ProductAgg {
Ok(())
}
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ if array.is_null(row_idx) {
+ return Ok(());
+ }
+ match &mut self.state {
+ ProductState::F32(acc) => {
+ let v = downcast::<Float32Array>(array,
&self.field_name)?.value(row_idx);
+ *acc = Some(acc.map_or(v, |prev| v * prev));
+ Ok(())
+ }
+ ProductState::F64(acc) => {
+ let v = downcast::<Float64Array>(array,
&self.field_name)?.value(row_idx);
+ *acc = Some(acc.map_or(v, |prev| v * prev));
+ Ok(())
+ }
+ _ => self.agg(array, row_idx),
+ }
+ }
+
fn result(&self) -> crate::Result<ArrayRef> {
Ok(match &self.state {
ProductState::I8(acc) => Arc::new(Int8Array::from(vec![*acc])),
@@ -415,6 +453,7 @@ fn agg_minmax(
row_idx: usize,
field_name: &str,
keep_smaller: bool,
+ reversed: bool,
) -> crate::Result<()> {
if array.is_null(row_idx) {
return Ok(());
@@ -425,7 +464,18 @@ fn agg_minmax(
*$acc = Some(match *$acc {
None => v,
Some(prev) => {
- if (keep_smaller && v < prev) || (!keep_smaller && v >
prev) {
+ let take_new = if keep_smaller {
+ if reversed {
+ v < prev
+ } else {
+ v <= prev
+ }
+ } else if reversed {
+ v >= prev
+ } else {
+ v > prev
+ };
+ if take_new {
v
} else {
prev
@@ -451,8 +501,16 @@ fn agg_minmax(
(false, false) => v.total_cmp(&prev),
};
let take_new = if keep_smaller {
- // Java `FieldMinAgg` returns the input on ties.
- cmp.is_le()
+ if reversed {
+ cmp.is_lt()
+ } else {
+ // Java `FieldMinAgg` returns the input on ties.
+ cmp.is_le()
+ }
+ } else if reversed {
+ // Reversed max treats the older input as the
+ // accumulator, so ties select it.
+ cmp.is_ge()
} else {
// Java `FieldMaxAgg` retains the accumulator on ties.
cmp.is_gt()
@@ -495,7 +553,13 @@ fn agg_minmax(
None => v.to_string(),
Some(prev) => {
let take_new = if keep_smaller {
- v < prev.as_str()
+ if reversed {
+ v < prev.as_str()
+ } else {
+ v <= prev.as_str()
+ }
+ } else if reversed {
+ v >= prev.as_str()
} else {
v > prev.as_str()
};
@@ -584,7 +648,25 @@ impl FieldAggregator for MinAgg {
}
fn agg(&mut self, array: &dyn Array, row_idx: usize) -> crate::Result<()> {
- agg_minmax(&mut self.state, array, row_idx, &self.field_name, true)
+ agg_minmax(
+ &mut self.state,
+ array,
+ row_idx,
+ &self.field_name,
+ true,
+ false,
+ )
+ }
+
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ agg_minmax(
+ &mut self.state,
+ array,
+ row_idx,
+ &self.field_name,
+ true,
+ true,
+ )
}
fn result(&self) -> crate::Result<ArrayRef> {
@@ -617,7 +699,25 @@ impl FieldAggregator for MaxAgg {
}
fn agg(&mut self, array: &dyn Array, row_idx: usize) -> crate::Result<()> {
- agg_minmax(&mut self.state, array, row_idx, &self.field_name, false)
+ agg_minmax(
+ &mut self.state,
+ array,
+ row_idx,
+ &self.field_name,
+ false,
+ false,
+ )
+ }
+
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ agg_minmax(
+ &mut self.state,
+ array,
+ row_idx,
+ &self.field_name,
+ false,
+ true,
+ )
}
fn result(&self) -> crate::Result<ArrayRef> {
@@ -869,6 +969,86 @@ mod tests {
);
}
+ #[test]
+ fn test_sum_decimal_recovers_after_intermediate_precision_overflow() {
+ let mut agg = sum_agg(DataType::Decimal(DecimalType::new(3,
0).unwrap()));
+ let mut builder = Decimal128Builder::with_capacity(3)
+ .with_precision_and_scale(3, 0)
+ .unwrap();
+ builder.append_value(900);
+ builder.append_value(200);
+ builder.append_value(-200);
+ let arr = builder.finish();
+
+ for i in 0..arr.len() {
+ agg.agg(&arr, i).unwrap();
+ }
+
+ let result = agg.result().unwrap();
+ assert_eq!(
+ result
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap()
+ .value(0),
+ -200
+ );
+ }
+
+ #[test]
+ fn test_sum_decimal_recovers_after_intermediate_i128_overflow() {
+ let mut agg = sum_agg(DataType::Decimal(DecimalType::new(38,
0).unwrap()));
+ let max_decimal = 10_i128.pow(38) - 1;
+ let mut builder = Decimal128Builder::with_capacity(3)
+ .with_precision_and_scale(38, 0)
+ .unwrap();
+ builder.append_value(max_decimal);
+ builder.append_value(max_decimal);
+ builder.append_value(-max_decimal);
+ let arr = builder.finish();
+
+ for i in 0..arr.len() {
+ agg.agg(&arr, i).unwrap();
+ }
+
+ let result = agg.result().unwrap();
+ assert_eq!(
+ result
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap()
+ .value(0),
+ -max_decimal
+ );
+ }
+
+ #[test]
+ fn test_sum_decimal_reversed_recovers_after_intermediate_i128_overflow() {
+ let mut agg = sum_agg(DataType::Decimal(DecimalType::new(38,
0).unwrap()));
+ let max_decimal = 10_i128.pow(38) - 1;
+ let mut builder = Decimal128Builder::with_capacity(3)
+ .with_precision_and_scale(38, 0)
+ .unwrap();
+ builder.append_value(max_decimal);
+ builder.append_value(max_decimal);
+ builder.append_value(-max_decimal);
+ let arr = builder.finish();
+
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ agg.agg_reversed(&arr, 2).unwrap();
+
+ let result = agg.result().unwrap();
+ assert_eq!(
+ result
+ .as_any()
+ .downcast_ref::<Decimal128Array>()
+ .unwrap()
+ .value(0),
+ -max_decimal
+ );
+ }
+
#[test]
fn test_product_int_aggregates() {
let mut agg = ProductAgg::new("v",
&DataType::Int(IntType::new())).unwrap();
@@ -879,6 +1059,15 @@ mod tests {
assert_eq!(collect_i32(agg.result().unwrap()), Some(24));
}
+ #[test]
+ fn test_product_reversed_aggregates_input() {
+ let mut agg = ProductAgg::new("v",
&DataType::Int(IntType::new())).unwrap();
+ let arr = Int32Array::from(vec![Some(2), Some(3)]);
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ assert_eq!(collect_i32(agg.result().unwrap()), Some(6));
+ }
+
#[test]
fn test_product_rejects_overflow() {
let mut agg = ProductAgg::new("v",
&DataType::SmallInt(SmallIntType::new())).unwrap();
@@ -1024,6 +1213,41 @@ mod tests {
}
}
+ #[test]
+ fn test_min_max_reversed_nan_ties_match_java_operand_order() {
+ let current = f32::from_bits(0xffc0_0001);
+ let older = f32::from_bits(0x7fc0_0002);
+ let arr = Float32Array::from(vec![Some(current), Some(older)]);
+
+ let mut min = min_agg(DataType::Float(FloatType::new()));
+ min.agg(&arr, 0).unwrap();
+ min.agg_reversed(&arr, 1).unwrap();
+ let min_result = min.result().unwrap();
+ assert_eq!(
+ min_result
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap()
+ .value(0)
+ .to_bits(),
+ current.to_bits()
+ );
+
+ let mut max = max_agg(DataType::Float(FloatType::new()));
+ max.agg(&arr, 0).unwrap();
+ max.agg_reversed(&arr, 1).unwrap();
+ let max_result = max.result().unwrap();
+ assert_eq!(
+ max_result
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap()
+ .value(0)
+ .to_bits(),
+ older.to_bits()
+ );
+ }
+
#[test]
fn test_min_max_all_null_returns_null() {
let mut agg = max_agg(DataType::Int(IntType::new()));
diff --git a/crates/paimon/src/table/aggregator/value.rs
b/crates/paimon/src/table/aggregator/value.rs
index 499c85fe..147b13cb 100644
--- a/crates/paimon/src/table/aggregator/value.rs
+++ b/crates/paimon/src/table/aggregator/value.rs
@@ -55,6 +55,7 @@ enum PickPolicy {
struct PickValueAgg {
policy: PickPolicy,
arrow_type: ArrowDataType,
+ initialized: bool,
/// 1-row Arrow array holding the currently-winning value; `None` means
/// no winning row has been observed yet for the current group.
winner: Option<ArrayRef>,
@@ -65,26 +66,72 @@ impl PickValueAgg {
Ok(Self {
policy,
arrow_type: paimon_type_to_arrow(data_type)?,
+ initialized: false,
winner: None,
})
}
- fn should_replace(&self, is_null: bool) -> bool {
- match self.policy {
- PickPolicy::Last => true,
- PickPolicy::First => self.winner.is_none(),
- PickPolicy::LastNonNull => !is_null,
- PickPolicy::FirstNonNull => self.winner.is_none() && !is_null,
- }
- }
-
fn reset(&mut self) {
+ self.initialized = false;
self.winner = None;
}
fn agg(&mut self, array: &dyn Array, row_idx: usize) {
- if self.should_replace(array.is_null(row_idx)) {
- self.winner = Some(array.slice(row_idx, 1));
+ let is_null = array.is_null(row_idx);
+ match self.policy {
+ PickPolicy::Last => {
+ self.winner = Some(array.slice(row_idx, 1));
+ }
+ PickPolicy::First if !self.initialized => {
+ self.initialized = true;
+ self.winner = Some(array.slice(row_idx, 1));
+ }
+ PickPolicy::LastNonNull if !is_null => {
+ self.winner = Some(array.slice(row_idx, 1));
+ }
+ PickPolicy::FirstNonNull if !self.initialized && !is_null => {
+ self.initialized = true;
+ self.winner = Some(array.slice(row_idx, 1));
+ }
+ _ => {}
+ }
+ }
+
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) {
+ let is_null = array.is_null(row_idx);
+ match self.policy {
+ // Java default: last_value.agg(input, accumulator) returns the
+ // existing accumulator.
+ PickPolicy::Last => {}
+ // Java first_value returns the older input once the aggregator has
+ // already been initialized by the current accumulator.
+ PickPolicy::First => {
+ if self.initialized {
+ self.winner = Some(array.slice(row_idx, 1));
+ } else {
+ self.initialized = true;
+ }
+ }
+ // Java last_non_null_value keeps a non-null accumulator, otherwise
+ // it falls back to the older input.
+ PickPolicy::LastNonNull => {
+ if self.winner.is_none() && !is_null {
+ self.winner = Some(array.slice(row_idx, 1));
+ }
+ }
+ // Preserve Java's stateful first_non_null_value behavior for
+ // aggReversed(accumulator, input) == agg(input, accumulator).
+ PickPolicy::FirstNonNull => {
+ let current_is_non_null = self
+ .winner
+ .as_ref()
+ .is_some_and(|winner| !winner.is_null(0));
+ if !self.initialized && current_is_non_null {
+ self.initialized = true;
+ } else {
+ self.winner = Some(array.slice(row_idx, 1));
+ }
+ }
}
}
@@ -118,6 +165,10 @@ macro_rules! pick_agg {
self.0.agg(array, row_idx);
Ok(())
}
+ fn agg_reversed(&mut self, array: &dyn Array, row_idx: usize) ->
crate::Result<()> {
+ self.0.agg_reversed(array, row_idx);
+ Ok(())
+ }
fn result(&self) -> crate::Result<ArrayRef> {
Ok(self.0.result())
}
@@ -237,4 +288,53 @@ mod tests {
agg.reset();
assert!(agg.result().unwrap().is_null(0));
}
+
+ #[test]
+ fn test_last_value_reversed_keeps_current_value() {
+ let mut agg = LastValueAgg::new("v",
&DataType::Int(IntType::new())).unwrap();
+ let arr = Int32Array::from(vec![Some(10), Some(5)]);
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ assert_eq!(collect_i32(agg.result().unwrap()), Some(10));
+ }
+
+ #[test]
+ fn test_first_value_reversed_uses_older_null() {
+ let mut agg = FirstValueAgg::new("v",
&DataType::Int(IntType::new())).unwrap();
+ let arr = Int32Array::from(vec![Some(10), None]);
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ assert_eq!(collect_i32(agg.result().unwrap()), None);
+ }
+
+ #[test]
+ fn test_last_non_null_reversed_fills_only_empty_accumulator() {
+ let mut agg = LastNonNullValueAgg::new("v",
&DataType::Int(IntType::new())).unwrap();
+ let arr = Int32Array::from(vec![None, Some(5), Some(3)]);
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ agg.agg_reversed(&arr, 2).unwrap();
+ assert_eq!(collect_i32(agg.result().unwrap()), Some(5));
+ }
+
+ #[test]
+ fn test_first_non_null_reversed_matches_java_initialized_state() {
+ let mut agg = FirstNonNullValueAgg::new("v",
&DataType::Int(IntType::new())).unwrap();
+ let arr = Int32Array::from(vec![None, Some(7), Some(5)]);
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ agg.agg_reversed(&arr, 2).unwrap();
+ assert_eq!(collect_i32(agg.result().unwrap()), Some(7));
+ }
+
+ #[test]
+ fn
test_first_non_null_reversed_after_forward_initialization_uses_older_input() {
+ let mut agg = FirstNonNullValueAgg::new("v",
&DataType::Int(IntType::new())).unwrap();
+ let arr = Int32Array::from(vec![Some(10), None, Some(5)]);
+ agg.agg(&arr, 0).unwrap();
+ agg.agg_reversed(&arr, 1).unwrap();
+ assert_eq!(collect_i32(agg.result().unwrap()), None);
+ agg.agg_reversed(&arr, 2).unwrap();
+ assert_eq!(collect_i32(agg.result().unwrap()), Some(5));
+ }
}
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index bb36e369..ec221af5 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -533,10 +533,12 @@ mod tests {
use super::*;
use crate::catalog::Identifier;
use crate::io::FileIOBuilder;
- use crate::spec::{DataType, Datum, IntType, PredicateBuilder, Schema,
TableSchema};
+ use crate::spec::{
+ DataType, Datum, IntType, PredicateBuilder, Schema, TableSchema,
VarCharType,
+ };
use crate::table::table_commit::TableCommit;
use crate::table::{Table, TableWrite};
- use arrow_array::{Array, Int32Array};
+ use arrow_array::{Array, Int32Array, StringArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use futures::TryStreamExt;
use std::sync::Arc;
@@ -1318,6 +1320,105 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn
kv_read_partial_update_sequence_group_aggregation_with_projection() {
+ let file_io = test_file_io();
+ let table_path =
"memory:/kv_partial_update_sequence_group_aggregation";
+ setup_dirs(&file_io, table_path).await;
+
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("version", DataType::Int(IntType::new()))
+ .column("value", DataType::VarChar(VarCharType::string_type()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("merge-engine", "partial-update")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default",
"kv_partial_update_sequence_group_aggregation_t"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("version", ArrowDataType::Int32, true),
+ ArrowField::new("value", ArrowDataType::Utf8, true),
+ ]));
+ let make = |id, version, value| {
+ RecordBatch::try_new(
+ arrow_schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![id])),
+ Arc::new(Int32Array::from(vec![version])),
+ Arc::new(StringArray::from(vec![value])),
+ ],
+ )
+ .unwrap()
+ };
+
+ write_commit(&table, &make(1, 10, "b")).await;
+ write_commit(&table, &make(1, 9, "a")).await;
+ write_commit(&table, &make(1, 11, "c")).await;
+ write_commit(&table, &make(2, 5, "x")).await;
+ write_commit(&table, &make(2, 6, "y")).await;
+
+ let aggregation_schema =
table.schema().copy_with_options(HashMap::from([
+ (
+ "fields.version.sequence-group".to_string(),
+ "value".to_string(),
+ ),
+ (
+ "fields.value.aggregate-function".to_string(),
+ "listagg".to_string(),
+ ),
+ ]));
+ let aggregation_table = Table::new(
+ file_io,
+ Identifier::new("default",
"kv_partial_update_sequence_group_aggregation_t"),
+ table_path.to_string(),
+ aggregation_schema,
+ None,
+ );
+
+ let batches = read_rows(&aggregation_table, Some(&["id", "value"]),
None).await;
+
+ let mut rows = batches
+ .iter()
+ .flat_map(|batch| {
+ let ids = batch
+ .column(batch.schema().index_of("id").unwrap())
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let index = batch.schema().index_of("value").unwrap();
+ let array = batch
+ .column(index)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ (0..array.len())
+ .map(|row| (ids.value(row), array.value(row).to_string()))
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+ rows.sort_by_key(|row| row.0);
+ assert_eq!(rows, vec![(1, "a,b,c".to_string()), (2,
"x,y".to_string())]);
+ for batch in batches {
+ assert_eq!(
+ batch
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>(),
+ vec!["id", "value"]
+ );
+ }
+ }
+
/// An AlwaysFalse filter on a PK table must return nothing, end to end.
/// Two layers enforce it: scan-side stats pruning treats AlwaysFalse as
/// prune-everything (plans no files), and the post-merge residual masks
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index 4a6d5835..39aba7c5 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -182,18 +182,25 @@ impl MergeFunction for DeduplicateMergeFunction {
}
}
-/// Basic partial-update merge: for each non-key column, keep the latest
-/// non-null value ordered by user sequence (if configured) then system
sequence.
+/// Partial-update merge: for each non-key column, keep the latest non-null
+/// value or apply its configured field aggregator.
+///
+/// Sequence-group aggregators use forward accumulation when the incoming
+/// group sequence is newer or equal and reversed accumulation when it is
+/// older, matching Java `PartialUpdateMergeFunction`.
///
/// DELETE / UPDATE_BEFORE rows are ignored when `ignore-delete=true` and
/// treated as unsupported otherwise.
-#[derive(Debug, Clone)]
+#[derive(Debug)]
pub(crate) struct PartialUpdateMergeFunction {
ignore_delete: bool,
sequence_groups: Vec<RuntimeSequenceGroup>,
grouped_fields: HashSet<usize>,
+ aggregators: Option<Mutex<FieldAggregatorSlots>>,
}
+type FieldAggregatorSlots = Vec<Option<Box<dyn FieldAggregator>>>;
+
#[derive(Debug, Clone)]
struct RuntimeSequenceGroup {
sequence_indices: Vec<usize>,
@@ -211,6 +218,7 @@ impl PartialUpdateMergeFunction {
ignore_delete: CoreOptions::new(table_options).ignore_delete(),
sequence_groups: Vec::new(),
grouped_fields: HashSet::new(),
+ aggregators: None,
})
}
@@ -224,6 +232,8 @@ impl PartialUpdateMergeFunction {
let config = PartialUpdateConfig::new(table_options);
config.validate_read_mode(true, table_name)?;
let groups = config.validated_sequence_groups(table_fields,
primary_keys)?;
+ let aggregate_functions =
+ config.validated_aggregate_functions(table_fields, primary_keys)?;
let field_indices: HashMap<&str, usize> = output_fields
.iter()
.enumerate()
@@ -275,11 +285,29 @@ impl PartialUpdateMergeFunction {
})
.copied()
.collect();
+ let aggregators = output_fields
+ .iter()
+ .map(|field| -> crate::Result<Option<Box<dyn FieldAggregator>>> {
+ let Some(function) = aggregate_functions.get(field.name())
else {
+ return Ok(None);
+ };
+ Ok(Some(new_aggregator(
+ function,
+ field.name(),
+ field.data_type(),
+ table_options,
+ )?))
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
Ok(Self {
ignore_delete: CoreOptions::new(table_options).ignore_delete(),
sequence_groups,
grouped_fields,
+ aggregators: aggregators
+ .iter()
+ .any(Option::is_some)
+ .then(|| Mutex::new(aggregators)),
})
}
}
@@ -309,6 +337,18 @@ impl MergeFunction for PartialUpdateMergeFunction {
vec![None; output_schema.fields().len()];
let mut group_sequence_rows: Vec<Option<(usize, usize)>> =
vec![None; self.sequence_groups.len()];
+ let mut aggregators = match &self.aggregators {
+ Some(aggregators) => Some(aggregators.lock().map_err(|e|
Error::UnexpectedError {
+ message: format!("PartialUpdateMergeFunction aggregator mutex
poisoned: {e}"),
+ source: None,
+ })?),
+ None => None,
+ };
+ if let Some(aggregators) = aggregators.as_mut() {
+ for aggregator in aggregators.iter_mut().flatten() {
+ aggregator.reset();
+ }
+ }
let mut saw_add = false;
for row_idx in ordered_row_indices {
@@ -329,7 +369,13 @@ impl MergeFunction for PartialUpdateMergeFunction {
}
let source_array = batch_buffer[row.batch_idx]
.column_for_output(output_col_idx,
source_output_col_indices);
- if !source_array.is_null(row.row_idx) {
+ if let Some(aggregator) = aggregators
+ .as_mut()
+ .and_then(|aggregators|
aggregators.get_mut(output_col_idx))
+ .and_then(Option::as_mut)
+ {
+ aggregator.agg(source_array, row.row_idx)?;
+ } else if !source_array.is_null(row.row_idx) {
*selected = Some((row.batch_idx, row.row_idx));
}
}
@@ -344,8 +390,8 @@ impl MergeFunction for PartialUpdateMergeFunction {
continue;
}
- let should_advance = match group_sequence_rows[group_idx] {
- None => true,
+ let sequence_ordering = match group_sequence_rows[group_idx] {
+ None => Ordering::Greater,
Some((current_batch_idx, current_row_idx)) =>
compare_sequence_group_rows(
row.batch_idx,
row.row_idx,
@@ -354,9 +400,28 @@ impl MergeFunction for PartialUpdateMergeFunction {
&group.sequence_indices,
batch_buffer,
source_output_col_indices,
- )?
- .is_ge(),
+ )?,
};
+ let should_advance = sequence_ordering.is_ge();
+
+ for &output_col_idx in &group.protected_indices {
+ if let Some(aggregator) = aggregators
+ .as_mut()
+ .and_then(|aggregators|
aggregators.get_mut(output_col_idx))
+ .and_then(Option::as_mut)
+ {
+ let source_array = batch_buffer[row.batch_idx]
+ .column_for_output(output_col_idx,
source_output_col_indices);
+ if should_advance {
+ aggregator.agg(source_array, row.row_idx)?;
+ } else {
+ aggregator.agg_reversed(source_array,
row.row_idx)?;
+ }
+ } else if should_advance {
+ selected_by_col[output_col_idx] = Some((row.batch_idx,
row.row_idx));
+ }
+ }
+
if !should_advance {
continue;
}
@@ -365,9 +430,6 @@ impl MergeFunction for PartialUpdateMergeFunction {
for &output_col_idx in &group.sequence_indices {
selected_by_col[output_col_idx] = Some((row.batch_idx,
row.row_idx));
}
- for &output_col_idx in &group.protected_indices {
- selected_by_col[output_col_idx] = Some((row.batch_idx,
row.row_idx));
- }
}
}
@@ -380,23 +442,29 @@ impl MergeFunction for PartialUpdateMergeFunction {
.iter()
.enumerate()
.map(|(output_col_idx, field)| {
- Ok(match selected_by_col[output_col_idx] {
- Some((batch_idx, row_idx)) => batch_buffer[batch_idx]
- .column_for_output(output_col_idx,
source_output_col_indices)
- .slice(row_idx, 1),
- None => {
- if !field.is_nullable() {
- return Err(Error::DataInvalid {
- message: format!(
- "merge-engine=partial-update produced NULL
for non-nullable field '{}'",
- field.name()
- ),
- source: None,
- });
- }
- new_null_array(field.data_type(), 1)
- }
- })
+ let column = match aggregators
+ .as_ref()
+ .and_then(|aggregators| aggregators.get(output_col_idx))
+ .and_then(Option::as_ref)
+ {
+ Some(aggregator) => aggregator.result()?,
+ None => match selected_by_col[output_col_idx] {
+ Some((batch_idx, row_idx)) => batch_buffer[batch_idx]
+ .column_for_output(output_col_idx,
source_output_col_indices)
+ .slice(row_idx, 1),
+ None => new_null_array(field.data_type(), 1),
+ },
+ };
+ if !field.is_nullable() && column.is_null(0) {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "merge-engine=partial-update produced NULL for
non-nullable field '{}'",
+ field.name()
+ ),
+ source: None,
+ });
+ }
+ Ok(column)
})
.collect::<crate::Result<Vec<_>>>()?;
@@ -2303,6 +2371,300 @@ mod tests {
assert!(batch.column(3).is_null(0));
}
+ #[tokio::test]
+ async fn
test_partial_update_aggregation_composite_sequence_accepts_partial_null_tuple()
{
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("_SEQUENCE_NUMBER", DataType::Int64, false),
+ Field::new("_VALUE_KIND", DataType::Int8, false),
+ Field::new("seq_major", DataType::Int32, true),
+ Field::new("seq_minor", DataType::Int32, true),
+ Field::new("value", DataType::Utf8, true),
+ ]));
+ let output_schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("seq_major", DataType::Int32, true),
+ Field::new("seq_minor", DataType::Int32, true),
+ Field::new("value", DataType::Utf8, true),
+ ]));
+ let output_fields = vec![
+ DataField::new(0, "pk".into(),
crate::spec::DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "seq_major".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 2,
+ "seq_minor".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 3,
+ "value".into(),
+ crate::spec::DataType::VarChar(VarCharType::string_type()),
+ ),
+ ];
+ let options = HashMap::from([
+ ("merge-engine".to_string(), "partial-update".to_string()),
+ (
+ "fields.seq_major,seq_minor.sequence-group".to_string(),
+ "value".to_string(),
+ ),
+ (
+ "fields.value.aggregate-function".to_string(),
+ "listagg".to_string(),
+ ),
+ ]);
+ let stream = stream_from_batches(vec![RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 1])),
+ Arc::new(Int64Array::from(vec![1, 2, 3])),
+ Arc::new(Int8Array::from(vec![0, 0, 0])),
+ Arc::new(Int32Array::from(vec![Some(10), Some(11), None])),
+ Arc::new(Int32Array::from(vec![Some(1), None, None])),
+ Arc::new(StringArray::from(vec![
+ Some("base"),
+ Some("partial"),
+ Some("ignored"),
+ ])),
+ ],
+ )
+ .unwrap()]);
+
+ let result = SortMergeReaderBuilder::new(
+ vec![stream],
+ schema,
+ vec![0],
+ 1,
+ 2,
+ vec![],
+ vec![3, 4, 5],
+ output_schema,
+ Box::new(
+ PartialUpdateMergeFunction::new_with_schema(
+ &options,
+ "test_table",
+ &output_fields,
+ &output_fields,
+ &["pk".to_string()],
+ )
+ .unwrap(),
+ ),
+ )
+ .build()
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ let batch = &result[0];
+ assert_eq!(
+ batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 11
+ );
+ assert!(batch.column(2).is_null(0));
+ assert_eq!(
+ batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "base,partial"
+ );
+ }
+
+ #[tokio::test]
+ async fn
test_partial_update_last_non_null_aggregation_without_sequence_group() {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("_SEQUENCE_NUMBER", DataType::Int64, false),
+ Field::new("_VALUE_KIND", DataType::Int8, false),
+ Field::new("value", DataType::Int32, true),
+ ]));
+ let output_schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("value", DataType::Int32, true),
+ ]));
+ let output_fields = vec![
+ DataField::new(0, "pk".into(),
crate::spec::DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "value".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ ];
+ let options = HashMap::from([
+ ("merge-engine".to_string(), "partial-update".to_string()),
+ (
+ "fields.value.aggregate-function".to_string(),
+ "last_non_null_value".to_string(),
+ ),
+ ]);
+ let stream = stream_from_batches(vec![RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 1])),
+ Arc::new(Int64Array::from(vec![1, 2, 3])),
+ Arc::new(Int8Array::from(vec![0, 0, 0])),
+ Arc::new(Int32Array::from(vec![Some(10), Some(30), None])),
+ ],
+ )
+ .unwrap()]);
+
+ let result = SortMergeReaderBuilder::new(
+ vec![stream],
+ schema,
+ vec![0],
+ 1,
+ 2,
+ vec![],
+ vec![3],
+ output_schema,
+ Box::new(
+ PartialUpdateMergeFunction::new_with_schema(
+ &options,
+ "test_table",
+ &output_fields,
+ &output_fields,
+ &["pk".to_string()],
+ )
+ .unwrap(),
+ ),
+ )
+ .build()
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(
+ result[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 30
+ );
+ }
+
+ #[tokio::test]
+ async fn
test_partial_update_sequence_group_listagg_preserves_group_sequence_order() {
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("_SEQUENCE_NUMBER", DataType::Int64, false),
+ Field::new("_VALUE_KIND", DataType::Int8, false),
+ Field::new("version", DataType::Int32, true),
+ Field::new("value", DataType::Utf8, true),
+ ]));
+ let output_schema = Arc::new(Schema::new(vec![
+ Field::new("pk", DataType::Int32, false),
+ Field::new("version", DataType::Int32, true),
+ Field::new("value", DataType::Utf8, true),
+ ]));
+ let output_fields = vec![
+ DataField::new(0, "pk".into(),
crate::spec::DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "version".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 2,
+ "value".into(),
+ crate::spec::DataType::VarChar(VarCharType::string_type()),
+ ),
+ ];
+ let options = HashMap::from([
+ ("merge-engine".to_string(), "partial-update".to_string()),
+ (
+ "fields.version.sequence-group".to_string(),
+ "value".to_string(),
+ ),
+ (
+ "fields.value.aggregate-function".to_string(),
+ "listagg".to_string(),
+ ),
+ ]);
+ let stream = stream_from_batches(vec![RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 1, 1, 1, 1])),
+ Arc::new(Int64Array::from(vec![1, 2, 3, 4, 5])),
+ Arc::new(Int8Array::from(vec![0, 0, 0, 0, 0])),
+ Arc::new(Int32Array::from(vec![
+ Some(10),
+ Some(12),
+ Some(11),
+ Some(12),
+ None,
+ ])),
+ Arc::new(StringArray::from(vec![
+ Some("b"),
+ Some("d"),
+ Some("c"),
+ Some("e"),
+ Some("ignored"),
+ ])),
+ ],
+ )
+ .unwrap()]);
+
+ let result = SortMergeReaderBuilder::new(
+ vec![stream],
+ schema,
+ vec![0],
+ 1,
+ 2,
+ vec![],
+ vec![3, 4],
+ output_schema,
+ Box::new(
+ PartialUpdateMergeFunction::new_with_schema(
+ &options,
+ "test_table",
+ &output_fields,
+ &output_fields,
+ &["pk".to_string()],
+ )
+ .unwrap(),
+ ),
+ )
+ .build()
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ let batch = &result[0];
+ assert_eq!(
+ batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 12
+ );
+ assert_eq!(
+ batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "c,b,d,e"
+ );
+ }
+
#[tokio::test]
async fn test_partial_update_merge_rejects_delete_like_rows() {
let schema = make_schema();
@@ -2517,6 +2879,49 @@ mod tests {
));
}
+ #[test]
+ fn test_partial_update_new_with_schema_validates_aggregate_functions() {
+ let fields = vec![
+ DataField::new(0, "pk".into(),
crate::spec::DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "version".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ DataField::new(
+ 2,
+ "value".into(),
+ crate::spec::DataType::Int(IntType::new()),
+ ),
+ ];
+ let options = HashMap::from([
+ ("merge-engine".to_string(), "partial-update".to_string()),
+ (
+ "fields.version.sequence-group".to_string(),
+ "value".to_string(),
+ ),
+ (
+ "fields.value.aggregate-function".to_string(),
+ "sume".to_string(),
+ ),
+ ]);
+
+ let err = PartialUpdateMergeFunction::new_with_schema(
+ &options,
+ "default.t",
+ &fields,
+ &fields,
+ &["pk".to_string()],
+ )
+ .unwrap_err();
+
+ assert!(
+ matches!(err, Error::ConfigInvalid { ref message }
+ if message.contains("sume") && message.contains("value")),
+ "expected schema-aware aggregate validation, got {err:?}"
+ );
+ }
+
// ---------- AggregateMergeFunction ----------
use crate::spec::{DataType as PaimonDataType, IntType, VarCharType};
diff --git a/docs/src/sql.md b/docs/src/sql.md
index a84a956e..08b76482 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -1894,8 +1894,22 @@ composite sequence fields, multiple independent groups,
and projected reads are
supported. Rows whose sequence tuple is entirely null do not update the group;
an accepted group update can set protected fields to null. Rust table creation
and writes still reject sequence-group options because write-side group merging
-is not implemented. Partial aggregation inside sequence groups and
-remove-record options are also not supported.
+is not implemented.
+
+Existing partial-update tables may also configure
+`fields.<field>.aggregate-function` or
+`fields.default-aggregate-function`. Rust supports the same aggregate-function
+set as the basic aggregation engine: `sum`, `product`, `min`, `max`,
+`last_value`, `first_value`, `last_non_null_value`,
+`first_non_null_value`, `bool_and`, `bool_or`, and `listagg`.
+`last_non_null_value` may be used without a sequence group; other functions
+require the target field to be protected by a sequence group. When an older
+group sequence arrives after a newer one, order-sensitive functions use Java
+compatible reversed aggregation. Unknown functions and incompatible
+function/type combinations fail closed. Rust table creation and writes still
+reject partial-update aggregation options.
+
+Partial-update remove-record options and retract semantics are not supported.
Rust can read fully materialized compacted files from deletion-vector-enabled
partial-update and aggregation tables. Every split must be raw-convertible,
