JingsongLi commented on code in PR #10540:
URL: https://github.com/apache/paimon/pull/10540#discussion_r4258647606
##########
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java:
##########
@@ -95,69 +97,82 @@ private static boolean needsEqualiser(DataType elementType)
{
@Override
public Object aggReversed(Object accumulator, Object inputField) {
- // we don't need to actually do the reverse here for this agg
- // because accumulator has been distinct, just let accumulator be
accumulator will speed up
- // distinct process
- return agg(accumulator, inputField);
+ if (countLimit != Integer.MAX_VALUE) {
+ // order is important with limit
+ return agg(inputField, accumulator, true);
+ } else {
+ // we don't need to actually do the reverse here for this agg
+ // because accumulator has been distinct, just let accumulator be
accumulator will speed
+ // up
+ // distinct process
+ return agg(accumulator, inputField, false);
+ }
}
@Override
public Object agg(Object accumulator, Object inputField) {
- if (accumulator == null && inputField == null) {
- return null;
- }
+ return agg(accumulator, inputField, false);
+ }
- if ((accumulator == null || inputField == null) && !distinct) {
- return accumulator == null ? inputField : accumulator;
- }
+ private Object agg(Object accumulator, Object inputField, boolean
accumulatorHasDuplicate) {
+ InternalArray accumulatorArray = (InternalArray) accumulator;
+ InternalArray inputArray = (InternalArray) inputField;
- if (distinct && equaliser != null) {
- List<Object> collection = new ArrayList<>();
- // do not need to distinct accumulator, because the accumulator is
always distinct, no
- // need to distinct it every time
- collect(collection, accumulator);
- collectWithEqualiser(collection, inputField);
+ if (accumulatorArray == null && inputArray == null) {
+ return null;
+ } else if (inputArray == null
+ && !distinct
+ && accumulatorArray.size() <= Math.abs(countLimit)) {
+ return accumulatorArray;
+ } else if (accumulatorArray == null
+ && !distinct
+ && inputArray.size() <= Math.abs(countLimit)) {
+ 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);
return new GenericArray(collection.toArray());
} else {
- Collection<Object> collection = distinct ? new HashSet<>() : new
ArrayList<>();
- collect(collection, accumulator);
- collect(collection, inputField);
- return new GenericArray(collection.toArray());
+ collect(collection, inputArray, removeDuplicate);
+ collect(collection, accumulatorArray, removeDuplicate);
+ List<Object> result =
+ collection instanceof List
+ ? (List<Object>) collection
+ : new ArrayList<>(collection);
+ Collections.reverse(result);
+ return new GenericArray(result.toArray());
}
}
- private void collect(Collection<Object> collection, @Nullable Object data)
{
- if (data == null) {
+ private void collect(
+ Collection<Object> collection, @Nullable InternalArray array,
boolean removeDuplicate) {
+ if (array == null) {
return;
}
- InternalArray array = (InternalArray) data;
- for (int i = 0; i < array.size(); i++) {
- collection.add(elementGetter.getElementOrNull(array, i));
- }
- }
-
- private void collectWithEqualiser(List<Object> list, Object data) {
- if (data == null) {
- return;
- }
-
- InternalArray array = (InternalArray) data;
- for (int i = 0; i < array.size(); i++) {
- Object element = elementGetter.getElementOrNull(array, i);
- if (!contains(list, element)) {
- list.add(element);
+ int limit = Math.abs(countLimit);
+ for (int i = 0; i < array.size() && collection.size() < limit; i++) {
Review Comment:
[P2] Handle Integer.MIN_VALUE before taking the absolute value
fields.<field>.count-limit=-2147483648 is accepted by the int option, but
Math.abs(Integer.MIN_VALUE) is still negative. The loop then never executes, so
even agg(null, [1, 2]) returns an empty array, with distinct either enabled or
disabled. A negative limit of this magnitude should retain every element of
such a small input, rather than silently discard them. Please compute the
magnitude as a long, including in the fast-path size checks, or explicitly
reject Integer.MIN_VALUE during configuration validation.
##########
paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java:
##########
@@ -2139,6 +2139,50 @@ private InternalRow row(
return GenericRow.of(k0, k1, BinaryString.fromString(v), seq, ts);
}
+ @Test
+ public void testFieldCollectAggWithCountLimit() {
+ DataType[] elementTypes = {DataTypes.INT(),
RowType.of(DataTypes.INT())};
+ for (DataType elementType : elementTypes) {
+ assertCollectCountLimit(elementType, false, 3, 2, 1, 2);
+ assertCollectCountLimit(elementType, true, 3, 2, 1, 3);
+ assertCollectCountLimit(elementType, false, -3, 3, 3, 1);
+ assertCollectCountLimit(elementType, true, -3, 2, 3, 1);
+ assertCollectCountLimit(elementType, false, 0);
+ assertCollectCountLimit(elementType, true, 0);
+ }
+ }
+
+ private void assertCollectCountLimit(
+ DataType elementType, boolean distinct, int countLimit, int...
expected) {
+ Map<String, String> config = new HashMap<>();
+ config.put("fields.fieldName.distinct", Boolean.toString(distinct));
+ config.put("fields.fieldName.count-limit",
Integer.toString(countLimit));
+ CoreOptions options = CoreOptions.fromMap(config);
+ ArrayType arrayType = DataTypes.ARRAY(elementType);
+ FieldCollectAgg agg = new FieldCollectAggFactory().create(arrayType,
options, "fieldName");
+ InternalArray.ElementGetter elementGetter =
InternalArray.createElementGetter(elementType);
+ Object[] expectedElements = makeArray(elementType,
expected).toObjectArray();
+
+ InternalArray input = makeArray(elementType, 2, 1, 2, 3, 3, 1);
+ InternalArray single = (InternalArray) agg.agg(null, input);
+ assertThat(unnest(single,
elementGetter)).containsExactly(expectedElements);
+
+ InternalArray older = makeArray(elementType, 2, 1, 2);
+ InternalArray newer = makeArray(elementType, 3, 3, 1);
+ InternalArray forward = (InternalArray) agg.agg(older, newer);
+ InternalArray reversed = (InternalArray) agg.aggReversed(newer, older);
+ assertThat(unnest(forward,
elementGetter)).containsExactly(expectedElements);
Review Comment:
[P1] Initialize the accumulators through the aggregator
This new test fails for ARRAY<ROW<INT>> with distinct=true and
count-limit=3: the forward result is [ROW(2), ROW(1), ROW(2)], while the
assertion expects [ROW(2), ROW(1), ROW(3)]. The existing forward implementation
assumes that its accumulator is already distinct, but older is a raw array
containing duplicates. Normal merge callers initialize the accumulator with
agg(null, input). Please initialize older that way for the forward check, and
initialize newer that way for the reversed check, so the test exercises valid
accumulator states. I reproduced this failure both in the full
FieldAggregatorTest class and by rerunning the new method without fast-build.
##########
paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java:
##########
@@ -95,69 +97,82 @@ private static boolean needsEqualiser(DataType elementType)
{
@Override
public Object aggReversed(Object accumulator, Object inputField) {
- // we don't need to actually do the reverse here for this agg
- // because accumulator has been distinct, just let accumulator be
accumulator will speed up
- // distinct process
- return agg(accumulator, inputField);
+ if (countLimit != Integer.MAX_VALUE) {
+ // order is important with limit
+ return agg(inputField, accumulator, true);
Review Comment:
[P2] Forward aggReversed through the ignore-retract wrapper
This new duplicate-aware branch is bypassed when
fields.<field>.ignore-retract=true: FieldIgnoreRetractAgg does not override
aggReversed, so the inherited implementation calls agg(inputField,
accumulator), which reaches collect with accumulatorHasDuplicate=false. This is
reachable through partial-update sequence groups. With ARRAY<ROW<INT>>,
distinct=true and count-limit=3, receiving sequence 2 with [ROW(3)] and then
sequence 1 with [ROW(1), ROW(1), ROW(2)] produces [ROW(1), ROW(1), ROW(2)] when
ignore-retract is enabled; without the wrapper it correctly produces [ROW(1),
ROW(2), ROW(3)]. The new limit therefore lets duplicates consume the
distinct-element budget and drops ROW(3). Please make FieldIgnoreRetractAgg
delegate aggReversed to its wrapped aggregator and cover this combination
through the partial-update factory.
--
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]
