JingsongLi commented on code in PR #10607:
URL: https://github.com/apache/paimon/pull/10607#discussion_r4313689926
##########
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java:
##########
@@ -130,16 +125,21 @@ private Object agg(Object accumulator, Object inputField,
boolean accumulatorHas
return inputArray;
}
- boolean removeDuplicate = distinct && equaliser != null;
Collection<Object> collection =
distinct && equaliser == null ? new LinkedHashSet<>() : new
ArrayList<>();
if (countLimit >= 0) {
- collect(collection, accumulatorArray, removeDuplicate &&
accumulatorHasDuplicate);
- collect(collection, inputArray, removeDuplicate);
+ collect(
+ collection,
+ accumulatorArray,
+ checkDuplicateUntil(accumulatorHasDuplicate, collection));
+ collect(collection, inputArray,
checkDuplicateUntil(inputHasDuplicate, collection));
return new GenericArray(collection.toArray());
} else {
- collect(collection, inputArray, removeDuplicate);
- collect(collection, accumulatorArray, removeDuplicate);
+ collect(collection, inputArray,
checkDuplicateUntil(inputHasDuplicate, collection));
+ collect(
+ collection,
+ accumulatorArray,
+ checkDuplicateUntil(accumulatorHasDuplicate, collection));
Review Comment:
Thanks — agreed that `initRow` bypassing aggregation is an existing
invariant problem, and positive/unlimited forward aggregation already exposed
it. I rechecked the narrower negative-limit case on the exact base (`5d019f5a`)
and this head (`efaf8dfb`) with the production
`AggregateMergeFunction.factory`, `fields.a.aggregate-function=collect`,
`fields.a.distinct=true`, `fields.a.count-limit=-3`, and
`aggregation.remove-record-on-delete=true`. In separate JDK 8 runs, DELETE
`[(1),(1)]` followed by INSERT `[(2)]` returns INSERT `[(1),(2)]` on base and
INSERT `[(1),(1),(2)]` on head. The old negative-limit branch fully
deduplicated the accumulator; the new prefix-only comparison exposes the
invalid state there too. So the root invariant issue predates this PR, but this
caller-visible result regression is introduced by this optimization.
Fixing/normalizing the state in `AggregateMergeFunction` together with this
change is fine; please cover that transition before relying on the inva
riant.
The optimization itself is useful: both versions return the same 1,000 rows
for a normalized accumulator plus one overlapping row, while equaliser calls
fall from 499,501 to 1,000. All 133 selected aggregation/merge tests passed
with regular Maven checks; the DELETE→INSERT case is not covered by them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
