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 baac87ca fix(core): align evolution stats and aggregation flush with
Java (#839)
baac87ca is described below
commit baac87caa90f600757144e88c00f046f6c21a775
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 15 16:08:25 2026 +0800
fix(core): align evolution stats and aggregation flush with Java (#839)
---
crates/paimon/src/table/kv_file_reader.rs | 15 +-
crates/paimon/src/table/kv_file_writer.rs | 384 +++++++++++++++++++++++++++---
crates/paimon/src/table/sort_merge.rs | 48 ++--
crates/paimon/src/table/stats_filter.rs | 193 ++++++++++++---
crates/paimon/src/table/table_scan.rs | 323 +++++++++++++++++++++++--
crates/paimon/src/table/table_write.rs | 1 +
docs/src/sql.md | 4 +-
7 files changed, 866 insertions(+), 102 deletions(-)
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 3b0a31ce..f3adf84f 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -2270,10 +2270,9 @@ mod tests {
assert!(!names.contains(&"banana".to_string()));
}
- /// Aggregation (sum): inputs 10 + 20 merge to 30. `value = 30` must match
- /// the merged row (a pre-merge filter would drop both inputs);
- /// `value = 10` must match nothing (a pre-merge filter would keep the
- /// 10-input and leak it).
+ /// Each flush aggregates its duplicate keys first (30 and 9); the read
+ /// merges both files to 39. `value = 39` must match the merged row, while
+ /// `value = 10` must not leak an input row.
#[tokio::test]
async fn kv_read_aggregation_filters_on_merged_value() {
let file_io = test_file_io();
@@ -2288,17 +2287,17 @@ mod tests {
],
);
- write_commit(&table, &int_batch(vec![1], vec![Some(10)])).await;
- write_commit(&table, &int_batch(vec![1], vec![Some(20)])).await;
+ write_commit(&table, &int_batch(vec![1, 1], vec![Some(10),
Some(20)])).await;
+ write_commit(&table, &int_batch(vec![1, 1], vec![Some(4),
Some(5)])).await;
let fields = table.schema().fields().to_vec();
let match_merged = PredicateBuilder::new(&fields)
- .equal("value", Datum::Int(30))
+ .equal("value", Datum::Int(39))
.unwrap();
let batches = read_rows(&table, None, Some(match_merged)).await;
assert_eq!(int_column(&batches, "id"), vec![1]);
- assert_eq!(int_column(&batches, "value"), vec![30]);
+ assert_eq!(int_column(&batches, "value"), vec![39]);
let match_input = PredicateBuilder::new(&fields)
.equal("value", Datum::Int(10))
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index bf1cd9aa..514d7524 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -26,6 +26,7 @@
//!
//! Reference:
[org.apache.paimon.io.KeyValueDataFileWriterImpl](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/io/KeyValueDataFileWriterImpl.java)
+use crate::arrow::arrow_fields_to_paimon;
use crate::arrow::format::create_format_writer;
use crate::io::FileIO;
use crate::spec::stats::{compute_column_stats, BinaryTableStats};
@@ -35,6 +36,7 @@ use crate::spec::{
SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_NAME,
};
use crate::table::prepared_files::PreparedFiles;
+use crate::table::sort_merge::{AggregateMergeFunction, BufferedBatch,
MergeRow};
use crate::Result;
use arrow_array::{Array, BooleanArray, Int64Array, Int8Array, RecordBatch,
UInt32Array};
use arrow_ord::sort::{lexsort_to_indices, SortColumn, SortOptions};
@@ -79,7 +81,9 @@ pub(crate) struct KeyValueWriteConfig {
pub changelog_file_prefix: String,
pub changelog_file_compression: String,
pub changelog_file_format: String,
- /// Primary key column indices in the user schema.
+ /// Full primary keys, including partition columns which must not be
aggregated.
+ pub primary_keys: Vec<String>,
+ /// Trimmed primary key column indices in the user schema.
pub primary_key_indices: Vec<usize>,
/// Paimon DataTypes for each primary key column (same order as
primary_key_indices).
pub primary_key_types: Vec<DataType>,
@@ -255,7 +259,7 @@ impl KeyValueFileWriter {
// Deduplicate → keep last row per key group (highest seq)
// FirstRow → keep first row per key group (lowest seq)
// PartialUpdate → per column, keep the latest non-null value
- // Aggregation → keep all rows for read-side field-wise merge
+ // Aggregation → apply per-field aggregators in sequence order
let (data_batch, data_seq, data_indices) = match
self.config.merge_engine {
MergeEngine::PartialUpdate => {
let (merged, merged_seq) =
@@ -264,7 +268,14 @@ impl KeyValueFileWriter {
UInt32Array::from_iter_values(0..u32::try_from(merged.num_rows()).unwrap());
(merged, merged_seq, identity)
}
- MergeEngine::Deduplicate | MergeEngine::FirstRow |
MergeEngine::Aggregation => {
+ MergeEngine::Aggregation => {
+ let (merged, merged_seq) =
+ self.merge_aggregation_rows(&combined, seq_array.as_ref(),
&sorted_indices)?;
+ let identity =
+
UInt32Array::from_iter_values(0..u32::try_from(merged.num_rows()).unwrap());
+ (merged, merged_seq, identity)
+ }
+ MergeEngine::Deduplicate | MergeEngine::FirstRow => {
let selected = self.select_flush_indices(&combined,
&sorted_indices)?;
(
combined.clone(),
@@ -281,6 +292,16 @@ impl KeyValueFileWriter {
None
};
+ // Java derives file sequence bounds from emitted rows; allocation
still
+ // advances over all buffered inputs, including rows folded away.
+ let output_sequences =
data_seq.as_any().downcast_ref::<Int64Array>().unwrap();
+ let (min_output_seq, max_output_seq) = data_indices
+ .values()
+ .iter()
+ .map(|&idx| output_sequences.value(idx as usize))
+ .fold((i64::MAX, i64::MIN), |(min, max), seq| {
+ (min.min(seq), max.max(seq))
+ });
let data_file = self
.write_indexed_file(
&data_batch,
@@ -291,8 +312,8 @@ impl KeyValueFileWriter {
file_ordinal: self.written_files.len(),
file_format: &self.config.file_format,
file_compression: &self.config.file_compression,
- min_sequence_number: start_seq,
- max_sequence_number: end_seq,
+ min_sequence_number: min_output_seq,
+ max_sequence_number: max_output_seq,
delete_row_count: data_delete_row_count,
},
)
@@ -587,14 +608,108 @@ impl KeyValueFileWriter {
MergeEngine::PartialUpdate => {
unreachable!("partial-update merges rows at flush via
merge_partial_update_rows")
}
- // Aggregation keeps every row on flush and performs the per-field
- // merge on the read side.
- MergeEngine::Aggregation => Ok((0..sorted_indices.len())
- .map(|idx| sorted_indices.value(idx))
- .collect()),
+ MergeEngine::Aggregation => {
+ unreachable!("aggregation merges rows at flush via
merge_aggregation_rows")
+ }
}
}
+ fn merge_aggregation_rows(
+ &self,
+ batch: &RecordBatch,
+ seq_array: &dyn Array,
+ sorted_indices: &UInt32Array,
+ ) -> Result<(RecordBatch, Arc<dyn Array>)> {
+ let schema = batch.schema();
+ let value_kind_idx = schema
+ .fields()
+ .iter()
+ .position(|field| field.name() == VALUE_KIND_FIELD_NAME);
+ let value_kinds = value_kind_idx
+ .map(|idx| {
+ batch
+ .column(idx)
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "_VALUE_KIND column must be Int8".into(),
+ source: None,
+ })
+ })
+ .transpose()?;
+ let output_indices: Vec<_> = (0..batch.num_columns())
+ .filter(|idx| Some(*idx) != value_kind_idx)
+ .collect();
+ let output_fields: Vec<_> = output_indices
+ .iter()
+ .map(|&idx| schema.field(idx).clone())
+ .collect();
+ let output_schema = Arc::new(ArrowSchema::new(output_fields.clone()));
+ let sequence_fields: Vec<_> = self
+ .config
+ .sequence_field_indices
+ .iter()
+ .map(|&idx| schema.field(idx).name().clone())
+ .collect();
+ let merge = AggregateMergeFunction::new(
+ &self.config.table_options,
+ &self.config.table_name,
+ &arrow_fields_to_paimon(&output_fields)?,
+ &self.config.primary_keys,
+ &sequence_fields,
+ )?;
+ let rows: Vec<_> = sorted_indices
+ .values()
+ .iter()
+ .map(|&idx| MergeRow {
+ batch_idx: 0,
+ row_idx: idx as usize,
+ // Ordering is already established by the write buffer's Arrow
sort.
+ sequence_number: 0,
+ user_sequences: Vec::new(),
+ value_kind: value_kinds
+ .filter(|kinds| kinds.is_valid(idx as usize))
+ .map_or(0, |kinds| kinds.value(idx as usize)),
+ })
+ .collect();
+ let key_rows = self.convert_key_rows(batch)?;
+ let buffers = [BufferedBatch::Source(batch.clone())];
+ let mut merged = Vec::new();
+ let mut last_indices = Vec::new();
+ let mut start = 0;
+ while start < rows.len() {
+ let mut end = start + 1;
+ while end < rows.len()
+ && key_rows.row(rows[end].row_idx) ==
key_rows.row(rows[start].row_idx)
+ {
+ end += 1;
+ }
+ let group = rows[start..end].iter().collect::<Vec<_>>();
+ merged.push(merge.merge_ordered(&group, &buffers, &output_indices,
&output_schema)?);
+ // Java retains the last row in sequence-field order, which need
not
+ // have the largest arrival sequence number.
+ last_indices.push(sorted_indices.value(end - 1));
+ start = end;
+ }
+ let arrow_error = |e: arrow_schema::ArrowError|
crate::Error::DataInvalid {
+ message: format!("Failed to build merged aggregation batch: {e}"),
+ source: Some(Box::new(e)),
+ };
+ let merged =
+ arrow_select::concat::concat_batches(&output_schema,
&merged).map_err(arrow_error)?;
+ let merged = if let Some(idx) = value_kind_idx {
+ let mut columns = merged.columns().to_vec();
+ columns.insert(idx, Arc::new(Int8Array::from(vec![0;
merged.num_rows()])));
+ RecordBatch::try_new(schema, columns).map_err(arrow_error)?
+ } else {
+ merged
+ };
+ let merged_seq =
+ arrow_select::take::take(seq_array,
&UInt32Array::from(last_indices), None)
+ .map_err(arrow_error)?;
+ Ok((merged, merged_seq))
+ }
+
/// Merge same-key rows at flush for the partial-update engine, mirroring
/// Java `MergeTreeWriter#flushWriteBuffer` (the write buffer applies the
/// merge function before any file is written) with the same semantics as
@@ -836,7 +951,7 @@ mod tests {
use super::*;
use crate::io::FileIOBuilder;
use crate::spec::IntType;
- use arrow_array::{Int32Array, UInt32Array};
+ use arrow_array::{Int32Array, RecordBatchReader, StringArray, UInt32Array};
use std::collections::HashMap;
fn test_write_config(merge_engine: MergeEngine) -> KeyValueWriteConfig {
@@ -866,6 +981,7 @@ mod tests {
changelog_file_prefix: "changelog-".to_string(),
changelog_file_compression: "none".to_string(),
changelog_file_format: "parquet".to_string(),
+ primary_keys: vec!["id".into()],
primary_key_indices: vec![0],
primary_key_types: vec![DataType::Int(IntType::new())],
sequence_field_indices: vec![1],
@@ -1349,33 +1465,247 @@ mod tests {
));
}
- #[test]
- fn test_select_flush_indices_keeps_all_rows_for_aggregation_engine() {
+ #[tokio::test]
+ async fn test_flush_aggregation_merges_same_key_across_buffered_batches() {
let schema = Arc::new(ArrowSchema::new(vec![
- Arc::new(ArrowField::new("id", ArrowDataType::Int32, false)),
- Arc::new(ArrowField::new("seq", ArrowDataType::Int64, false)),
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, true),
+ ]));
+ let mut config = test_write_config(MergeEngine::Aggregation);
+ config.sequence_field_indices.clear();
+ config.write_buffer_size = i64::MAX;
+ config
+ .table_options
+ .insert("fields.value.aggregate-function".into(), "sum".into());
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ let mut writer = KeyValueFileWriter::new(io.clone(), config,
7).unwrap();
+ for value in [10, 20] {
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(Int32Array::from(vec![value])),
+ ],
+ )
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ }
+ let prepared = writer.prepare_commit().await.unwrap();
+ assert_eq!(prepared.data_files.len(), 1);
+ let file = &prepared.data_files[0];
+ assert_eq!(file.row_count, 1);
+ assert_eq!((file.min_sequence_number, file.max_sequence_number), (8,
8));
+ let physical = read_kv_file(&io, file).await;
+ assert_eq!(physical.num_rows(), 1);
+ assert_eq!(
+ physical
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values(),
+ &[30]
+ );
+ }
+
+ #[tokio::test]
+ async fn
test_flush_aggregation_sequence_partition_keys_and_input_changelog() {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("p", ArrowDataType::Int32, false),
+ ArrowField::new("seq", ArrowDataType::Int64, true),
+ ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8, false),
+ ArrowField::new("amount", ArrowDataType::Int32, true),
+ ArrowField::new("label", ArrowDataType::Utf8, true),
+ ArrowField::new("note", ArrowDataType::Utf8, true),
]));
let batch = RecordBatch::try_new(
schema,
vec![
- Arc::new(Int32Array::from(vec![1, 1])) as Arc<dyn
arrow_array::Array>,
- Arc::new(Int64Array::from(vec![10, 20])) as Arc<dyn
arrow_array::Array>,
+ Arc::new(Int32Array::from(vec![2, 1, 1, 1, 2])),
+ Arc::new(Int32Array::from(vec![7; 5])),
+ Arc::new(Int64Array::from(vec![
+ Some(30),
+ Some(20),
+ Some(10),
+ Some(20),
+ None,
+ ])),
+ Arc::new(Int8Array::from(vec![0, 2, 0, 2, 0])),
+ Arc::new(Int32Array::from(vec![None, Some(10), Some(20), None,
None])),
+ Arc::new(StringArray::from(vec![
+ "two", "later", "early", "tie", "low",
+ ])),
+ Arc::new(StringArray::from(vec![
+ None,
+ Some("kept"),
+ None,
+ None,
+ None,
+ ])),
],
)
.unwrap();
- let sorted_indices = UInt32Array::from(vec![0, 1]);
- let writer = KeyValueFileWriter::new(
- FileIOBuilder::new("memory").build().unwrap(),
- test_write_config(MergeEngine::Aggregation),
- 0,
- )
- .unwrap();
+ let mut config = test_write_config(MergeEngine::Aggregation);
+ // Only id participates in within-partition grouping, but p is also a
+ // primary key and must not be summed by the default aggregator.
+ config.primary_keys = vec!["id".into(), "p".into()];
+ config.sequence_field_indices = vec![2];
+ config.write_buffer_size = i64::MAX;
+ config.input_changelog = true;
+ for (key, value) in [
+ ("fields.default-aggregate-function", "sum"),
+ ("fields.label.aggregate-function", "listagg"),
+ ("fields.note.aggregate-function", "last_non_null_value"),
+ ] {
+ config.table_options.insert(key.into(), value.into());
+ }
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ let mut writer = KeyValueFileWriter::new(io.clone(), config,
7).unwrap();
+ writer.write(&batch.slice(0, 2)).await.unwrap();
+ writer.write(&batch.slice(2, 3)).await.unwrap();
+ let prepared = writer.prepare_commit().await.unwrap();
+ let file = &prepared.data_files[0];
+ assert_eq!(file.row_count, 2);
+ assert_eq!(
+ (file.min_sequence_number, file.max_sequence_number),
+ (7, 10)
+ );
+ assert_eq!(file.delete_row_count, Some(0));
+ let physical = read_kv_file(&io, file).await;
+ let i32_values = |name| {
+ physical
+ .column_by_name(name)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>()
+ };
+ let i64_values = |name| {
+ physical
+ .column_by_name(name)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>()
+ };
+ assert_eq!(i32_values("id"), vec![Some(1), Some(2)]);
+ assert_eq!(i32_values("p"), vec![Some(7), Some(7)]);
+ assert_eq!(i32_values("amount"), vec![Some(30), None]);
+ assert_eq!(i64_values("seq"), vec![Some(20), Some(30)]);
+ assert_eq!(
+ i64_values(SEQUENCE_NUMBER_FIELD_NAME),
+ vec![Some(10), Some(7)]
+ );
+ let labels = physical
+ .column_by_name("label")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(
+ labels.iter().collect::<Vec<_>>(),
+ vec![Some("early,later,tie"), Some("low,two")]
+ );
+ let notes = physical
+ .column_by_name("note")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ assert_eq!(notes.iter().collect::<Vec<_>>(), vec![Some("kept"), None]);
+ assert_eq!(
+ physical
+ .column_by_name(VALUE_KIND_FIELD_NAME)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap()
+ .values(),
+ &[0, 0]
+ );
+ let changelog = &prepared.changelog_files[0];
+ assert_eq!(changelog.row_count, 5);
+ let changelog = read_kv_file(&io, changelog).await;
+ assert_eq!(
+ changelog
+ .column_by_name("amount")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>(),
+ vec![Some(20), Some(10), None, None, None]
+ );
+ assert_eq!(
+ changelog
+ .column_by_name(VALUE_KIND_FIELD_NAME)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap()
+ .values(),
+ &[0, 2, 2, 0, 0]
+ );
+ // Sequence allocation advances over all consumed input, not output
rows.
+ writer.write(&batch.slice(1, 1)).await.unwrap();
+ let next = writer.prepare_commit().await.unwrap();
+ assert_eq!(
+ (
+ next.data_files[0].min_sequence_number,
+ next.data_files[0].max_sequence_number
+ ),
+ (12, 12)
+ );
+ }
- let selected = writer
- .select_flush_indices(&batch, &sorted_indices)
+ #[tokio::test]
+ async fn test_flush_aggregation_rejects_retract_before_writing_files() {
+ for kind in [1, 3] {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("seq", ArrowDataType::Int64, false),
+ ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8,
false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(Int64Array::from(vec![10, 20])),
+ Arc::new(Int8Array::from(vec![0, kind])),
+ ],
+ )
+ .unwrap();
+ let mut writer = KeyValueFileWriter::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ test_write_config(MergeEngine::Aggregation),
+ 0,
+ )
.unwrap();
+ writer.write(&batch).await.unwrap();
+ let err = writer.prepare_commit().await.err().unwrap();
+ assert!(
+ matches!(err, crate::Error::Unsupported { message } if
message.contains("DELETE or UPDATE_BEFORE"))
+ );
+ assert!(writer.written_files.is_empty());
+ assert!(writer.written_changelog_files.is_empty());
+ }
+ }
- assert_eq!(selected, vec![0, 1]);
+ async fn read_kv_file(io: &FileIO, file: &DataFileMeta) -> RecordBatch {
+ let path = format!("memory:/kv-test/bucket-0/{}", file.file_name);
+ let data = io.new_input(&path).unwrap().read().await.unwrap();
+ let reader =
+
parquet::arrow::arrow_reader::ParquetRecordBatchReader::try_new(data,
1024).unwrap();
+ let schema = reader.schema();
+ let batches = reader.map(|batch| batch.unwrap()).collect::<Vec<_>>();
+ arrow_select::concat::concat_batches(&schema, &batches).unwrap()
}
#[test]
diff --git a/crates/paimon/src/table/sort_merge.rs
b/crates/paimon/src/table/sort_merge.rs
index d1f2bdee..e63be64a 100644
--- a/crates/paimon/src/table/sort_merge.rs
+++ b/crates/paimon/src/table/sort_merge.rs
@@ -604,16 +604,17 @@ impl AggregateMergeFunction {
aggregators: Mutex::new(aggregators),
})
}
-}
-impl MergeFunction for AggregateMergeFunction {
- fn merge(
+ /// Aggregate a key group already ordered by user sequence and auto
sequence.
+ /// The write buffer uses its sorted order directly; the merge reader sorts
+ /// references from its input streams before sharing the same
implementation.
+ pub(super) fn merge_ordered(
&self,
- rows: &[MergeRow],
+ rows: &[&MergeRow],
batch_buffer: &[BufferedBatch],
source_output_col_indices: &[usize],
output_schema: &SchemaRef,
- ) -> crate::Result<MergeResult> {
+ ) -> crate::Result<RecordBatch> {
if rows.is_empty() {
return Err(Error::UnexpectedError {
message: "merge called with empty rows".to_string(),
@@ -631,15 +632,6 @@ impl MergeFunction for AggregateMergeFunction {
}
}
- // Sort row indices by user sequence (if configured), then system
- // sequence, so per-field aggregators see the canonical "ascending
- // sequence" order documented in their contracts.
- let mut ordered_row_indices: Vec<usize> = (0..rows.len()).collect();
- ordered_row_indices.sort_by(|&lhs_idx, &rhs_idx| {
- compare_sequence_order(&rows[lhs_idx], &rows[rhs_idx])
- .then_with(|| lhs_idx.cmp(&rhs_idx))
- });
-
let mut aggregators = self
.aggregators
.lock()
@@ -653,8 +645,7 @@ impl MergeFunction for AggregateMergeFunction {
}
}
- for &row_idx in &ordered_row_indices {
- let row = &rows[row_idx];
+ for row in rows {
for (col_idx, slot) in aggregators.iter_mut().enumerate() {
if let Some(agg) = slot.as_mut() {
let source_array = batch_buffer[row.batch_idx]
@@ -667,7 +658,7 @@ impl MergeFunction for AggregateMergeFunction {
// Use the last sorted row to source primary-key column values: every
// row in the group shares the same PK by construction, so any row
// works; picking the last one keeps the slice cheap to compute.
- let pk_source = &rows[*ordered_row_indices.last().unwrap()];
+ let pk_source = rows.last().unwrap();
let output_columns: Vec<ArrayRef> = aggregators
.iter()
@@ -716,7 +707,28 @@ impl MergeFunction for AggregateMergeFunction {
}
})?;
- Ok(MergeResult::MaterializedRow(batch))
+ Ok(batch)
+ }
+}
+
+impl MergeFunction for AggregateMergeFunction {
+ fn merge(
+ &self,
+ rows: &[MergeRow],
+ batch_buffer: &[BufferedBatch],
+ source_output_col_indices: &[usize],
+ output_schema: &SchemaRef,
+ ) -> crate::Result<MergeResult> {
+ let mut ordered: Vec<_> = rows.iter().collect();
+ // Stable sorting keeps input order for equal sequences.
+ ordered.sort_by(|lhs, rhs| compare_sequence_order(lhs, rhs));
+ self.merge_ordered(
+ &ordered,
+ batch_buffer,
+ source_output_col_indices,
+ output_schema,
+ )
+ .map(MergeResult::MaterializedRow)
}
}
diff --git a/crates/paimon/src/table/stats_filter.rs
b/crates/paimon/src/table/stats_filter.rs
index 010a2344..8c778b2b 100644
--- a/crates/paimon/src/table/stats_filter.rs
+++ b/crates/paimon/src/table/stats_filter.rs
@@ -355,14 +355,66 @@ fn extract_stats_datum(row: &BinaryRow, index: usize,
data_type: &DataType) -> O
///
/// In data evolution mode, a logical row can be spread across multiple files
with
/// different column sets. After `group_by_overlapping_row_id`, each group
contains
-/// files covering the same row ID range. Stats for each field come from the
file
-/// with the highest `max_sequence_number` that actually contains that field.
+/// files with overlapping row ID ranges. A field can use stats only from a
+/// unique latest provider covering the entire group, interpreted by field ID
+/// in that provider's schema.
///
/// Reference:
[DataEvolutionFileStoreScan.evolutionStats](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java)
+pub(super) async fn data_evolution_group_matches_predicates_for_table(
+ table: &Table,
+ group: &[DataFileMeta],
+ predicates: &[Predicate],
+ schema_cache: &mut HashMap<i64, Option<Arc<ResolvedStatsSchema>>>,
+) -> bool {
+ if has_always_false(predicates, &[]) {
+ return false;
+ }
+ if predicates
+ .iter()
+ .all(|p| matches!(p, Predicate::AlwaysTrue))
+ {
+ return true;
+ }
+ let mut schemas = Vec::with_capacity(group.len());
+ for file in group {
+ let Some(schema) = resolve_stats_schema(table, file.schema_id,
schema_cache).await else {
+ // Unknown file schemas cannot establish either field absence or
stats layout.
+ return true;
+ };
+ schemas.push(schema);
+ }
+ data_evolution_group_matches_resolved_predicates(
+ group,
+ predicates,
+ table.schema().fields(),
+ &schemas,
+ )
+}
+
+#[cfg(test)]
pub(super) fn data_evolution_group_matches_predicates(
group: &[DataFileMeta],
predicates: &[Predicate],
table_fields: &[DataField],
+) -> bool {
+ let schema = Arc::new(ResolvedStatsSchema {
+ file_fields: table_fields.to_vec(),
+ field_mapping: identity_field_mapping(table_fields.len()),
+ key_fields: Vec::new(),
+ });
+ data_evolution_group_matches_resolved_predicates(
+ group,
+ predicates,
+ table_fields,
+ &vec![schema; group.len()],
+ )
+}
+
+fn data_evolution_group_matches_resolved_predicates(
+ group: &[DataFileMeta],
+ predicates: &[Predicate],
+ table_fields: &[DataField],
+ schemas: &[Arc<ResolvedStatsSchema>],
) -> bool {
if predicates.is_empty() || group.is_empty() {
return true;
@@ -381,42 +433,78 @@ pub(super) fn data_evolution_group_matches_predicates(
return true;
}
- // Sort files by max_sequence_number descending so the highest-seq file
wins per field.
- let mut sorted_files: Vec<&DataFileMeta> = group.iter().collect();
- sorted_files.sort_by_key(|f| std::cmp::Reverse(f.max_sequence_number));
-
- // For each table field, find which file (index in sorted_files) provides
it.
- // Use file_data_columns (based on write_cols) to determine which file
contains
- // the field, not file_stats_columns (based on value_stats_cols) which only
- // indicates stats coverage.
- let field_sources: Vec<Option<(usize, usize)>> = {
- let per_file_columns: Vec<Vec<&str>> = sorted_files
- .iter()
- .map(|file| file_data_columns(file, table_fields))
- .collect();
- table_fields
- .iter()
- .enumerate()
- .map(|(field_idx, field)| {
- for (file_idx, cols) in per_file_columns.iter().enumerate() {
- if cols.iter().any(|c| *c == field.name()) {
- return Some((file_idx, field_idx));
- }
- }
- None
- })
- .collect()
+ let ranges: Option<Vec<_>> = group
+ .iter()
+ .map(|file| {
+ let start = file.first_row_id?;
+ let end = start.checked_add(file.row_count)?;
+ (end > start).then_some((start, end))
+ })
+ .collect();
+ let Some(ranges) = ranges else {
+ return true;
+ };
+ let start = ranges.iter().map(|range| range.0).min().unwrap();
+ let end = ranges.iter().map(|range| range.1).max().unwrap();
+ let Some(row_count) = end.checked_sub(start) else {
+ return true;
};
- // Build per-file stats without arity validation — data evolution files
- // may have fewer columns than the current table schema.
- let file_stats: Vec<FileStatsRows> = sorted_files
+ // Resolve providers by field ID in each file's schema, then choose the
latest
+ // provider. Column names in write_cols/value_stats_cols belong to that
schema.
+ let mut sorted_files: Vec<_> = group.iter().zip(schemas).collect();
+ sorted_files.sort_by_key(|(file, _)|
std::cmp::Reverse(file.max_sequence_number));
+ let per_file_columns: Vec<_> = sorted_files
.iter()
- .map(|file| FileStatsRows::from_data_file(file, table_fields))
+ .map(|(file, schema)| file_data_columns(file, &schema.file_fields))
.collect();
+ let mut field_sources = vec![None; table_fields.len()];
+ let mut usable_provider = vec![false; table_fields.len()];
+ for field_idx in 0..table_fields.len() {
+ let mut providers =
+ sorted_files
+ .iter()
+ .enumerate()
+ .filter_map(|(file_idx, (file, schema))| {
+ let old_idx = schema.field_mapping[field_idx]?;
+ per_file_columns[file_idx]
+ .contains(&schema.file_fields[old_idx].name())
+ .then_some((file_idx, *file))
+ });
+ if let Some((file_idx, file)) = providers.next() {
+ field_sources[field_idx] = Some((file_idx, field_idx));
+ let tied = providers
+ .next()
+ .is_some_and(|(_, next)| next.max_sequence_number ==
file.max_sequence_number);
+ usable_provider[field_idx] =
+ !tied && file.first_row_id == Some(start) && file.row_count ==
row_count;
+ }
+ }
- // row_count is the max across the group (overlapping row ranges).
- let row_count = group.iter().map(|f| f.row_count).max().unwrap_or(0);
+ let file_stats: Vec<FileStatsRows> = sorted_files
+ .iter()
+ .map(|(file, schema)| {
+ let mut stats = FileStatsRows::from_data_file(file,
&schema.file_fields);
+ let mapping = table_fields
+ .iter()
+ .enumerate()
+ .map(|(idx, field)| {
+ let old_idx = schema.field_mapping[idx]?;
+ // Java ignores stats for changed types. Keep the field's
provider
+ // but mark its stats unknown, rather than treating it as
absent/null.
+ if !usable_provider[idx]
+ || schema.file_fields[old_idx].data_type() !=
field.data_type()
+ || !valid_evolution_stats(&stats, old_idx,
field.data_type(), row_count)
+ {
+ return None;
+ }
+ stats.stats_index(old_idx)
+ })
+ .collect();
+ stats.stats_col_mapping = Some(mapping);
+ stats
+ })
+ .collect();
predicates.iter().all(|predicate| {
data_evolution_predicate_may_match(
@@ -429,6 +517,45 @@ pub(super) fn data_evolution_group_matches_predicates(
})
}
+/// Java DataEvolutionFileStoreScan.isValidStats: inconsistent statistics must
+/// not become evidence that an entire row-id group can be dropped.
+fn valid_evolution_stats(
+ stats: &FileStatsRows,
+ index: usize,
+ data_type: &DataType,
+ row_count: i64,
+) -> bool {
+ let Some(stats_index) = stats.stats_index(index) else {
+ return false;
+ };
+ if [&stats.min_values, &stats.max_values].iter().any(|row| {
+ row.as_ref().is_none_or(|row| {
+ stats_index >= row.arity() as usize
+ || row.data().len() <
BinaryRow::cal_fix_part_size_in_bytes(row.arity()) as usize
+ })
+ }) {
+ return false;
+ }
+ let nulls = stats.null_count(index);
+ if nulls.is_some_and(|count| count < 0 || count > row_count) {
+ return false;
+ }
+ match (
+ stats.min_value(index, data_type),
+ stats.max_value(index, data_type),
+ ) {
+ (None, None) => true,
+ (Some(min), Some(max)) => {
+ nulls != Some(row_count)
+ && matches!(
+ min.partial_cmp(&max),
+ Some(std::cmp::Ordering::Less | std::cmp::Ordering::Equal)
+ )
+ }
+ _ => false,
+ }
+}
+
/// Resolve which columns a file actually contains (for field source
resolution).
/// Uses `write_cols` if present, otherwise assumes all table fields.
/// This is distinct from `file_stats_columns` which resolves stats coverage.
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index b48e6229..431c8407 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -28,7 +28,7 @@ use super::kv_file_reader::retain_primary_key_conjuncts;
use super::partition_filter::PartitionFilter;
use super::row_position_selection::RowPositionSelection;
use super::stats_filter::{
- data_evolution_group_matches_predicates,
data_file_matches_predicates_for_table,
+ data_evolution_group_matches_predicates_for_table,
data_file_matches_predicates_for_table,
data_file_matches_predicates_with_key_stats, group_by_overlapping_row_id,
FileStatsRows,
ResolvedStatsSchema,
};
@@ -2037,6 +2037,7 @@ impl<'a> PaimonTableScan<'a> {
// Group by (partition, bucket), decomposing entries to avoid cloning
partition.
let groups = group_data_files_by_partition_bucket(entries);
+ let mut stats_schema_cache = HashMap::new();
let snapshot_id = snapshot.id();
let base_path = table_path.trim_end_matches('/');
@@ -2126,16 +2127,19 @@ impl<'a> PaimonTableScan<'a> {
row_id_groups
} else {
let before = row_id_groups.len();
- let groups = row_id_groups
- .into_iter()
- .filter(|group| {
- data_evolution_group_matches_predicates(
- group,
- &self.data_predicates,
- self.table.schema().fields(),
- )
- })
- .collect::<Vec<_>>();
+ let mut groups = Vec::with_capacity(before);
+ for group in row_id_groups {
+ if data_evolution_group_matches_predicates_for_table(
+ self.table,
+ &group,
+ &self.data_predicates,
+ &mut stats_schema_cache,
+ )
+ .await
+ {
+ groups.push(group);
+ }
+ }
if let Some(trace) = trace.as_deref_mut() {
trace.data_evolution_groups_pruned_by_stats += before
- groups.len();
}
@@ -3602,6 +3606,291 @@ mod tests {
assert_eq!(file_names_from_files(&pruned), vec!["old-id.parquet"]);
}
+ #[tokio::test]
+ async fn test_data_evolution_stats_use_file_schema_after_evolution() {
+ let old_schema = TableSchema::new(
+ 0,
+ &PaimonSchema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("a", DataType::Int(IntType::new()))
+ .column("b", DataType::Int(IntType::new()))
+ .build()
+ .unwrap(),
+ );
+ let cases = [
+ (
+ "drop",
+ vec![SchemaChange::drop_column("a".into())],
+ "b",
+ Datum::Int(20),
+ ),
+ (
+ "reorder",
+ vec![SchemaChange::update_column_position(
+ ColumnMove::move_first("b".into()),
+ )],
+ "b",
+ Datum::Int(20),
+ ),
+ (
+ "rename",
+ vec![SchemaChange::rename_column("b".into(),
"renamed".into())],
+ "renamed",
+ Datum::Int(20),
+ ),
+ (
+ "widen",
+ vec![SchemaChange::update_column_type(
+ "a".into(),
+ DataType::Double(crate::spec::DoubleType::new()),
+ )],
+ "a",
+ Datum::Double(10.0),
+ ),
+ ];
+ for (name, changes, field, literal) in cases {
+ let schema = old_schema.apply_changes(changes).unwrap();
+ let table =
data_evolution_test_table(&format!("memory:/de_stats_{name}"), schema);
+ write_schema_file(&table, &old_schema).await;
+ for dense in [false, true] {
+ let mut file = make_evo_file("old.parquet", 10, 1, 1, Some(0));
+ let mut stats = BinaryRowBuilder::new(if dense { 2 } else { 3
});
+ if dense {
+ stats.write_int(0, 10);
+ stats.write_int(1, 20);
+ file.write_cols = Some(vec!["a".into(), "b".into()]);
+ file.value_stats_cols = Some(vec!["a".into(), "b".into()]);
+ } else {
+ stats.write_int(0, 1);
+ stats.write_int(1, 10);
+ stats.write_int(2, 20);
+ }
+ let stats = stats.build_serialized();
+ file.value_stats = BinaryTableStats::new(
+ stats.clone(),
+ stats,
+ vec![Some(0); if dense { 2 } else { 3 }],
+ );
+ let entries = vec![ManifestEntry::new(
+ FileKind::Add,
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ -1,
+ file,
+ 0,
+ )];
+ let pb = PredicateBuilder::new(table.schema().fields());
+ for (predicate, expected) in [
+ (pb.equal(field, literal.clone()).unwrap(), 1),
+ // Type changes cannot reuse old stats; matching unchanged
+ // fields can still reject this file after rename/reorder.
+ (
+ pb.equal(
+ field,
+ match literal {
+ Datum::Double(_) => Datum::Double(999.0),
+ _ => Datum::Int(999),
+ },
+ )
+ .unwrap(),
+ usize::from(name == "widen"),
+ ),
+ ] {
+ let scan =
+ PaimonTableScan::new(&table, None, vec![predicate],
None, None, None);
+ let plan = scan
+ .plan_snapshot_from_entries(
+ diff_snapshot_with_schema(7, table.schema().id()),
+ entries.clone(),
+ None,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap();
+ assert_eq!(plan.splits().len(), expected, "{name},
dense={dense}");
+ assert_eq!(plan.snapshot_id(), Some(7));
+ }
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn test_data_evolution_stats_require_unique_full_range_provider() {
+ let table = data_evolution_test_table(
+ "memory:/de_stats_providers",
+ two_column_schema(0, "id", "value"),
+ );
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let make_file = |name: &str, first, count, seq, value| {
+ let mut file = make_evo_file(name, 10, count, seq, Some(first));
+ file.write_cols = Some(vec!["id".into()]);
+ let mut row = BinaryRowBuilder::new(1);
+ row.write_int(0, value);
+ let row = row.build_serialized();
+ file.value_stats = BinaryTableStats::new(row.clone(), row,
vec![Some(0)]);
+ file
+ };
+ for (name, files) in [
+ (
+ "partial",
+ vec![make_file("old", 0, 10, 1, 5), make_file("new", 0, 5, 2,
20)],
+ ),
+ (
+ "tied",
+ vec![
+ make_file("left", 0, 10, 2, 20),
+ make_file("right", 0, 10, 2, 5),
+ ],
+ ),
+ (
+ "chain",
+ vec![
+ make_file("left", 0, 5, 1, 20),
+ make_file("right", 4, 6, 2, 20),
+ ],
+ ),
+ ] {
+ for reverse in [false, true] {
+ let mut files = files.clone();
+ if reverse {
+ files.reverse();
+ }
+ let entries: Vec<_> = files
+ .into_iter()
+ .map(|file| {
+ ManifestEntry::new(
+ FileKind::Add,
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ -1,
+ file,
+ 0,
+ )
+ })
+ .collect();
+ for predicate in [
+ pb.equal("id", Datum::Int(5)).unwrap(),
+ pb.is_null("id").unwrap(),
+ ] {
+ let plan =
+ PaimonTableScan::new(&table, None, vec![predicate],
None, None, None)
+ .plan_snapshot_from_entries(
+ diff_snapshot_with_schema(7, 0),
+ entries.clone(),
+ None,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap();
+ assert_eq!(plan.splits().len(), 1, "{name},
reverse={reverse}");
+ assert_eq!(plan.splits()[0].data_files().len(), 2);
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn test_data_evolution_stats_reject_inconsistent_field_stats() {
+ let fields = vec![DataField::new(
+ 0,
+ "id".into(),
+ DataType::Int(IntType::new()),
+ )];
+ let predicate = PredicateBuilder::new(&fields)
+ .equal("id", Datum::Int(5))
+ .unwrap();
+ for (min, max, nulls) in [
+ (Some(10), Some(10), -1),
+ (Some(10), Some(10), 11),
+ (None, Some(10), 0),
+ (Some(10), None, 0),
+ (Some(10), Some(1), 0),
+ (Some(10), Some(10), 10),
+ ] {
+ let stats_row = |value: Option<i32>| {
+ let mut row = BinaryRowBuilder::new(1);
+ if let Some(value) = value {
+ row.write_int(0, value);
+ } else {
+ row.set_null_at(0);
+ }
+ row.build_serialized()
+ };
+ let mut file = make_evo_file("invalid", 10, 10, 1, Some(0));
+ file.value_stats =
+ BinaryTableStats::new(stats_row(min), stats_row(max),
vec![Some(nulls)]);
+ assert!(
+ data_evolution_group_matches_predicates(
+ &[file],
+ std::slice::from_ref(&predicate),
+ &fields
+ ),
+ "{min:?}, {max:?}, {nulls}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn
test_data_evolution_stats_distinguish_readded_fields_and_unknown_schemas() {
+ let old_schema = two_column_schema(0, "id", "value");
+ let schema = old_schema
+ .apply_changes(vec![
+ SchemaChange::drop_column("value".into()),
+ SchemaChange::add_column("value".into(),
DataType::Int(IntType::new())),
+ ])
+ .unwrap();
+ let table = data_evolution_test_table("memory:/de_stats_readd",
schema);
+ write_schema_file(&table, &old_schema).await;
+ let pb = PredicateBuilder::new(table.schema().fields());
+ for file_schema_id in [0, 99] {
+ let mut file = make_evo_file("old.parquet", 10, 1, 1, Some(0));
+ file.schema_id = file_schema_id;
+ // The old name exists, but it is not the re-added field's ID.
+ file.write_cols = Some(vec!["value".into()]);
+ for (predicate, expected) in [
+ (
+ pb.equal("value", Datum::Int(20)).unwrap(),
+ usize::from(file_schema_id == 99),
+ ),
+ (pb.is_null("value").unwrap(), 1),
+ (
+ pb.is_not_null("value").unwrap(),
+ usize::from(file_schema_id == 99),
+ ),
+ (Predicate::AlwaysFalse, 0),
+ ] {
+ let entries = vec![ManifestEntry::new(
+ FileKind::Add,
+ BinaryRowBuilder::new(0).build_serialized(),
+ 0,
+ -1,
+ file.clone(),
+ 0,
+ )];
+ let plan = PaimonTableScan::new(&table, None, vec![predicate],
None, None, None)
+ .plan_snapshot_from_entries(
+ diff_snapshot_with_schema(7, table.schema().id()),
+ entries,
+ None,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap();
+ assert_eq!(
+ plan.splits().len(),
+ expected,
+ "file schema {file_schema_id}"
+ );
+ }
+ }
+ }
+
#[tokio::test]
async fn test_data_evolution_pruning_matches_renamed_columns_by_field_id()
{
let schema_v0 = two_column_schema(0, "id", "old_name");
@@ -4519,12 +4808,13 @@ mod tests {
#[test]
fn test_data_evolution_group_matches_or_prunes_when_no_child_matches() {
let fields = int_field();
- let file = test_data_file_meta(
+ let mut file = test_data_file_meta(
int_stats_row(Some(10)),
int_stats_row(Some(20)),
vec![Some(0)],
5,
);
+ file.first_row_id = Some(0);
let pb = PredicateBuilder::new(&fields);
let predicate = Predicate::or(vec![
pb.less_than("id", Datum::Int(5)).unwrap(),
@@ -4541,12 +4831,13 @@ mod tests {
#[test]
fn test_data_evolution_group_matches_or_keeps_when_any_child_matches() {
let fields = int_field();
- let file = test_data_file_meta(
+ let mut file = test_data_file_meta(
int_stats_row(Some(10)),
int_stats_row(Some(20)),
vec![Some(0)],
5,
);
+ file.first_row_id = Some(0);
let pb = PredicateBuilder::new(&fields);
let predicate = Predicate::or(vec![
pb.less_than("id", Datum::Int(15)).unwrap(),
@@ -4563,12 +4854,13 @@ mod tests {
#[test]
fn test_data_evolution_group_is_not_pruned_by_a_row_id_predicate() {
let fields = int_field();
- let file = test_data_file_meta(
+ let mut file = test_data_file_meta(
int_stats_row(Some(10)),
int_stats_row(Some(20)),
vec![Some(0)],
5,
);
+ file.first_row_id = Some(0);
let row_id =
crate::spec::row_id_leaf(crate::spec::PredicateOperator::GtEq,
vec![Datum::Long(100)]);
@@ -4587,12 +4879,13 @@ mod tests {
#[test]
fn test_data_evolution_group_matches_not_prunes_when_inner_must_match() {
let fields = int_field();
- let file = test_data_file_meta(
+ let mut file = test_data_file_meta(
int_stats_row(Some(10)),
int_stats_row(Some(10)),
vec![Some(0)],
5,
);
+ file.first_row_id = Some(0);
let predicate = Predicate::negate(
PredicateBuilder::new(&fields)
.equal("id", Datum::Int(10))
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 1a0c3993..be343117 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -1002,6 +1002,7 @@ impl TableWrite {
changelog_file_prefix: self.changelog_file_prefix.clone(),
changelog_file_compression:
self.changelog_file_compression.clone(),
changelog_file_format: self.changelog_file_format.clone(),
+ primary_keys: self.table.schema().primary_keys().to_vec(),
primary_key_indices: self.primary_key_indices.clone(),
primary_key_types: self.primary_key_types.clone(),
sequence_field_indices: self.sequence_field_indices.clone(),
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 276e6c23..e55a644b 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -2372,7 +2372,9 @@ Rust currently supports `merge-engine=aggregation` in
basic mode only. It works
with fixed buckets and ordinary dynamic buckets (`'bucket' = '-1'`) when the
primary key includes all partition columns. It supports per-field aggregate
functions such as `sum`, `min`, `max`, value functions, boolean functions, and
-`listagg`, plus `fields.default-aggregate-function`.
+`listagg`, plus `fields.default-aggregate-function`. Each write-buffer flush
+aggregates duplicate primary keys before writing data files, so a file contains
+at most one row per key. Input changelog files retain the original rows.
Sequence fields are always merged with `last_value`. Defining
`fields.<sequence-field>.aggregate-function` is rejected, matching Java schema
