[python] Aggregate same-key rows in the write buffer of aggregation tables

[jackylee-ch]

Purpose

The read path aggregates aggregation-engine tables (SplitRead builds AggregateMergeFunction), but the write buffer degraded every aggregation table to DeduplicateMergeFunction. So duplicate keys inside a single write_arrow were silently reduced to the last row instead of aggregated — a batch of total=10/20/30 for one key produced 30, not 60. It only looked correct when each row was committed separately, because read re-aggregates across files.

FileStoreWrite._build_pk_merge_function now builds the same AggregateMergeFunction as the read path for supported aggregation tables, so same-key rows in one buffer are partially aggregated (read and compaction re-aggregate across files, matching Java's per-buffer partial aggregation). Aggregation configured with options pypaimon does not implement (retract opt-ins, sequence groups, out-of-scope aggregators such as rbm64) still falls back to deduplicate with a warning — the read-side check_supported guard raises for those, so the user gets the explicit error at read.

Tests

test_aggregation_e2e aggregates duplicate keys within one write_arrow (sum → 60, max, default last-non-null; disjoint key untouched) — returns 30 on the pre-fix dedupe path. The write-side fallback-warning test is repurposed to an unsupported-option table, since supported aggregation no longer falls back.

Written with Claude Code; verification is mine.

[Akash3121]

This is not safe when the table configures  sequence.field . The read path feeds AggregateMergeFunction  in user-sequence order through  builtin_seq_comparator , but the write buffer sorts same-key rows only by generated  _SEQUENCE_NUMBER , so this new partial aggregation permanently folds them in arrival order. For example, with  sequence.field=total , two rows in one  write_arrow - (total=100, label='hi')  followed by  (total=50, label='lo')  - are written as the single  total=50, label='lo'  row, although the existing sequence-field contract requires the  total=100  row to win. Once collapsed, read-side sorting cannot recover the discarded row. Please either order each same-key run with the configured sequence comparator before calling this merge function, or reject aggregation writes with  sequence.field until that ordering is implemented. Add a one-batch regression where the highest sequence value is written first.

I reproduced the failure directly through  KeyValueDataWriter._merge_pending_by_pk : the two rows above collapse to  {'total': 50, 'label': 'lo'} . The existing sequence-field E2E test uses separate commits, so it does not exercise the new same-buffer folding path.

[jackylee-ch]

Thanks @Akash3121, good catch. Fixed via your first option: _sort_by_primary_key now folds each same-key run in sequence.field order (honoring sequence.field.sort-order, nulls-first) before merging, reusing the same accessors split_read feeds builtin_seq_comparator, so the write buffer and read heap agree by construction. Complex/VARIANT sequence fields (already rejected on read) are dropped from the sort keys instead of crashing the write. Added a one-batch regression: two same-key rows in one write_arrow, highest-sequence first, now keeps that row for last_value. PTAL.

[JingsongLi]

This PR has real end-to-end value for supported aggregation tables: a same-key SUM batch now returns 60 instead of silently keeping the final value 30. I reviewed the write-buffer fold and AggregateMergeFunction construction and ran 76 aggregation, merge-buffer, and sequence tests locally; the head's Python and Native CI is green. git diff --check also passed.

P1 — floating sequence.field still makes the result depend on write grouping. On this exact head I created two identical aggregation tables with a DOUBLE sequence field and last_value labels. I wrote (seq=1.0, label='finite') then (seq=NaN, label='nan'). Writing both rows in one batch returned finite; committing them separately returned nan. The new _sort_by_primary_key uses Arrow floating sort, while the read comparator follows a different NaN order. Because the buffer collapses the two rows before commit, the lost winner cannot be recovered by a later read. Please integrate the Java-compatible floating sequence ordering from #10166 into this head and add this one-batch versus separate-commit regression.

Production dependency — unsupported aggregation options still write using deduplicate fallback. This head's _build_pk_merge_function explicitly permits that fallback and its test expects a warning after writing. A later read error cannot recover the discarded input. #10165 adds early rejection; please integrate or make it a required predecessor before this PR merges, then verify the combined writer construction and supported aggregation read/write path. Until these two dependencies are present on the merged head, I would not ship this change despite the green CI.

[jackylee-ch]

Superseded by #10166. That PR takes the conservative path for the aggregate engine on the write side — a deduplicate fallback with an explicit warning, and the shared merge-function dispatch raising NotImplementedError for the aggregate engine — rather than aggregating in the write buffer. That avoids the sequence.field folding-order problem raised in review here (same-key rows would otherwise fold in arrival order, not sequence order). Closing in favour of that approach. Thanks @Akash3121 for the sequence.field catch.