Skip to content

fix(core): align collect aggregation with Java FieldCollectAgg - #1352

Open
LuciferYang wants to merge 7 commits into
apache:mainfrom
LuciferYang:fix/collect-java-parity-pr
Open

LuciferYang wants to merge 7 commits into
apache:mainfrom
LuciferYang:fix/collect-java-parity-pr

Conversation

@LuciferYang

@LuciferYang LuciferYang commented Nov 9, 2026 •

Copy link
Copy Markdown
Contributor

Purpose

Linked issue: close #1351

collect ignored fields.<col>.count-limit, merged aggReversed in a different order from Java master, and checked distinct membership with a linear scan. This ports the merge logic of Java FieldCollectAgg as of apache/paimon#10540 and apache/paimon#10607.

Brief change log

  • Port of the private agg(accumulator, accumulatorHasDuplicate, input, inputHasDuplicate), including checkDuplicateUntil and backwards collection for negative limits. A DELETE value is kept as it is until the next merge.
  • count-limit is read as a 32-bit integer, like Java's intType option.
  • The ignore-retract wrapper passes agg_reversed to the wrapped function, as Java does.
  • Whether an element type uses the equaliser follows Java needsEqualiser on the Paimon type. Deciding by Arrow type put VARIANT, BLOB and VECTOR on the wrong path.
  • In distinct merges and retract, elements compare like Java: NaNs are one value, 0.0 and -0.0 are two, and map entries match by key. Types that Java keeps in a LinkedHashSet use arrow-row keys in a hash set.
  • Elements are kept in a VecDeque, so a late update through agg_reversed prepends instead of rebuilding the array, except for distinct equaliser element types with a negative limit.

Tests

  • count_limit_matches_java_for_scalar_and_row_elements: the cases of Java testFieldCollectAggWithCountLimit, forward and reversed.
  • agg_reversed_puts_older_input_first, ignore_retract_reversed_*: reversed order, with and without the wrapper.
  • delete_value_*, agg_reversed_keeps_duplicates_of_a_binary_delete_value: DELETE values per Java element type.
  • distinct_*_like_java: NaN payloads, signed zero and map entry order for scalar, ROW, ARRAY and MAP elements, including retract.

Benchmarks (release build, in-memory FileIO): writing 20,000 single-element rows for one key with distinct=true takes 5.6 ms (8.7 s on main). 20,000 single-element agg_reversed calls, as a sequence group makes for late updates, take 1.6 ms without distinct (1.5 ms on main) and 6.4 ms with distinct INT elements (4.4 s on main). Distinct ROW elements still compare each update with the collected elements, as Java's equaliser does.

About 717k random steps generated by Java master's FieldCollectAgg match Rust on every step, covering INT, ROW, BYTES, DOUBLE with NaN, MAP, and ARRAY<ARRAY<MAP>> with binary inputs against apache/paimon#11110.

API and Format

No API or storage format change. Tables that set count-limit on a collect column now get capped arrays when Rust merges them. Maps nested in arrays match by key; Java's write buffer matches them by entry order until apache/paimon#11110.

Documentation

docs/src/sql.md documents distinct and count-limit for collect.

Generated-by: Claude Opus

@JingsongLi JingsongLi left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The count-limit and reverse-order fixes are useful for mixed Java/Rust tables. I ran the row-kind and primary-key compaction integration suites (128 passed) and reproduced two additional cases below. Please also update docs/src/sql.md: it currently says count-limit is unsupported. The new collect option can persist a shortened array on write or compaction, so document what positive, negative, and zero limits keep.

let element = &operand.elements[index];
if self.distinct {
let duplicate = match (&mut merged.keys, &operand.keys) {
(Some(seen), Some(rows)) => !seen.insert(rows.row(index).as_ref().into()),

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Treat different NaN payloads as one distinct value before capping the result. I wrote [NaN(payload 1), NaN(payload 2)] and then [5.0] for the same key in separate commits, with distinct=true and count-limit=2. The Rust scan returned [NaN, NaN]; Java FieldCollectAgg returned [NaN, 5.0] for the same input. Java's Double.equals canonicalizes NaN payloads, whereas Arrow row keys preserve them. Before this PR the Rust path distinguished the two NaNs, but the new cap makes that difference discard a finite value. Could the FLOAT/DOUBLE distinct keys canonicalize NaNs while preserving signed zeros and the original stored value? Please cover this with a capped collect regression.

This comment was marked as outdated.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Needs further updating

let input = self.input_operand(&values)?;
let elements = std::mem::take(&mut self.elements);
let accumulator = self.operand(elements, false)?;
let merged = self.merge(&input, &accumulator);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Avoid rebuilding the entire array for each late update. A partial-update sequence group calls this for every add below the highest sequence already seen. With the default uncapped, non-distinct collect and 20,000 single-element late updates for one key, I measured about 2.1 ms on the base and 652–674 ms on this head in an optimized aggregator probe (before file I/O). This allocates a new vector and clones all prior elements on each call, so replay of a hot key now scales quadratically. Could the uncapped, non-distinct path prepend chunks or use a deque and assemble the final array once, while retaining the corrected merge order?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 47799ec. Elements are kept in a VecDeque. Without distinct, agg and agg_reversed append or prepend the input and cut to the limit. With distinct on element types that Java merges through a LinkedHashSet, they append or prepend without merging, and the merge runs when the array is read or once it grows past twice its last merged size. 20,000 single-element late updates now take about 3 ms without distinct and about 10 ms with distinct on INT elements (2 ms and 4.4 s on main). Distinct ROW and BINARY elements still compare each update against the whole array, as Java's equaliser path does.

…g on late updates

Distinct keys and retract treat every FLOAT/DOUBLE NaN as one value like
Java Double.equals. Non-distinct and set-path collect append or prepend
without rebuilding the array; the set path deduplicates and applies the
limit when the array is read or grows past twice its merged size.
Document collect distinct and count-limit in sql.md.
@LuciferYang

Copy link
Copy Markdown
Contributor Author

Thanks for the review and the repros. Both are fixed in 47799ec. docs/src/sql.md now describes distinct and count-limit for collect (fa61ef6): what positive, negative and zero limits keep, and that they apply when two or more rows of a key are merged on flush or compaction. A key with a single row keeps its array as written, as in Java.

Java's equaliser compares FLOAT/DOUBLE fields of ROW, ARRAY and MAP
elements with Float.compare / Double.compare. Elements that both hold a
NaN are now compared after mapping every NaN to the canonical one.
Elements holding FLOAT/DOUBLE values or maps are compared field by
field: NaNs are one value, 0.0 and -0.0 are two, and map entries are
matched by key instead of position.
@JingsongLi

Copy link
Copy Markdown
Contributor

Requirement fit: SUPPORTED. The earlier NaN/count-limit issue and quadratic late-update regression are fixed. I ran 108 aggregator unit tests, 130 related integration tests, a real write/commit/compaction NaN regression, and an optimized 20,000-update probe (about 2.5–2.8 ms on this revision versus 625–655 ms on 99a2bbe).

[P2] Resolve the normal Java write-buffer compatibility boundary for nested maps (crates/paimon/src/table/aggregator/collect.rs:409-413, 446-458). There is still a reproducible cross-engine difference for a collect column of type ARRAY<ARRAY<MAP<INT,INT>>> with distinct=true and count-limit=2. Write three rows for one key, whose collect inputs contain respectively A=[{1:10,2:20}], B=[{2:20,1:10}], and C=[{3:30}]. Rust's real table write/flush/read path persists A,C. Current Java SortBufferWriteBuffer + AggregateMergeFunction persists A,B. Its buffer serializes the originally generic inputs, and the generated equaliser compares binary nested arrays by bytes when they contain no floating fields; the new recursive unordered-map equality collapses A/B instead. This can change the capped persisted value according to which engine writes the same input.

Java's direct generic-array path also returns A,C, so this is a representation-dependent Java shortcut, not a request to blindly copy that behavior. Before treating nested collect as production Java parity, please establish the intended compatibility contract, cover the actual Java buffered/persisted path in the cross-engine regression, and either resolve the upstream equaliser inconsistency with a clear version boundary or document and scope this divergence explicitly.

Java reference used: current master FieldCollectAgg blob 0620f3103136df1b255d83e686952044216b4181; generic, binary and actual write-buffer probes were compared. The other NaN/signed-zero/map semantic cases passed. Strict MkDocs build also passed.

@LuciferYang

Copy link
Copy Markdown
Contributor Author

Thanks for the write-buffer repro. The contract I'd go with: maps match by key in every representation and at any depth. That is what Java's field-wise equaliser already does for top-level MAP fields and with row-deduplicate-ignore-fields, and what this PR does.

The difference comes from Java's byte shortcut, so I fixed it there: apache/paimon#11110 (issue apache/paimon#11109) stops comparing BinaryRow / BinaryArray values by bytes when they hold maps or multisets, as #8534 did for FLOAT/DOUBLE. Equal bytes still return equal right away. As for the version boundary, collect count-limit is only on master, so with apache/paimon#11110 in, no released Java version persists the other result.

To cover the buffered path, I replayed 15k random agg / aggReversed / DELETE / retract sequences on ARRAY<ARRAY<MAP<INT,INT>>> with binary inputs, as the write buffer passes them, and 15k with generic inputs. With apache/paimon#11110, Java gives the same results for both, and Rust matches every step. Current Java master differs on 1087 of the binary-input sequences.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Bug] collect aggregation ignores count-limit and orders aggReversed differently from Java

2 participants