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

Search before asking

• I searched in the issues and found nothing similar.

Please describe the bug 🐞

The collect aggregator differs from Java FieldCollectAgg on current master in three ways.

1. fields.<col>.count-limit ([core] Support count-limit for collect agg paimon#10540) is accepted as a table option but ignored. A table with count-limit=3 keeps every element when Rust merges or compacts it, while Java keeps the first 3 (or the last 3 for -3).
2. Java aggReversed(acc, input) is agg(input, true, acc, false) since [core] Optimize FieldCollectAgg with equaliser when count-limit is negative paimon#10607: the older input goes first. Rust appends it after the accumulator, so a partial-update table with a sequence group gets a different ARRAY order from Rust and Java for the same rows. Java's FieldIgnoreRetractAgg also delegates aggReversed to the wrapped function now, and Rust still uses the default agg(input, acc) path there.
3. With distinct=true, Rust checks each element against every collected element with Arrow ==. Java uses a LinkedHashSet for non-binary scalar elements. Writing 20,000 single-element rows for one key with distinct=true takes 8.7 s on main (release build, in-memory FileIO).

Solution

Port the private agg(accumulator, accumulatorHasDuplicate, input, inputHasDuplicate) of FieldCollectAgg, including checkDuplicateUntil and the reversed collection for negative limits, read count-limit like Java's intType option, delegate aggReversed in the ignore-retract wrapper, and look up distinct elements that Java puts in a LinkedHashSet by their arrow-row encoded key.

Are you willing to submit a PR?

• I'm willing to submit a PR!