(paimon) branch master updated: [core] Optimize FieldCollectAgg with equaliser when count-limit is negative (#10607)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new ad9c431aa0 [core] Optimize FieldCollectAgg with equaliser when 
count-limit is negative (#10607)
ad9c431aa0 is described below

commit ad9c431aa01aa75fac183c70f50c28c3d61371e4
Author: tsreaper <tsreape...@gmail.com>
AuthorDate: Wed Oct 21 17:25:55 2026 +0800

    [core] Optimize FieldCollectAgg with equaliser when count-limit is negative 
(#10607)
---
 .../compact/aggregate/FieldCollectAgg.java         | 61 ++++++++++++++--------
 1 file changed, 39 insertions(+), 22 deletions(-)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
 
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
index 4559f07eb9..0620f31031 100644
--- 
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
+++ 
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldCollectAgg.java
@@ -97,24 +97,19 @@ public class FieldCollectAgg extends FieldAggregator {
 
     @Override
     public Object aggReversed(Object accumulator, Object 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);
-        }
+        return agg(inputField, true, accumulator, false);
     }
 
     @Override
     public Object agg(Object accumulator, Object inputField) {
-        return agg(accumulator, inputField, false);
+        return agg(accumulator, false, inputField, true);
     }
 
-    private Object agg(Object accumulator, Object inputField, boolean 
accumulatorHasDuplicate) {
+    private Object agg(
+            Object accumulator,
+            boolean accumulatorHasDuplicate,
+            Object inputField,
+            boolean inputHasDuplicate) {
         InternalArray accumulatorArray = (InternalArray) accumulator;
         InternalArray inputArray = (InternalArray) inputField;
 
@@ -130,16 +125,21 @@ public class FieldCollectAgg extends FieldAggregator {
             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));
             List<Object> result =
                     collection instanceof List
                             ? (List<Object>) collection
@@ -149,8 +149,20 @@ public class FieldCollectAgg extends FieldAggregator {
         }
     }
 
+    private int checkDuplicateUntil(boolean hasDuplicate, Collection<Object> 
collection) {
+        if (distinct && equaliser != null) {
+            if (hasDuplicate) {
+                return Integer.MAX_VALUE;
+            } else {
+                return collection.size();
+            }
+        } else {
+            return 0;
+        }
+    }
+
     private void collect(
-            Collection<Object> collection, @Nullable InternalArray array, 
boolean removeDuplicate) {
+            Collection<Object> collection, @Nullable InternalArray array, int 
checkDuplicateUntil) {
         if (array == null) {
             return;
         }
@@ -160,21 +172,26 @@ public class FieldCollectAgg extends FieldAggregator {
             Object element =
                     elementGetter.getElementOrNull(
                             array, countLimit < 0 ? array.size() - 1 - i : i);
-            if (!(removeDuplicate && contains(collection, element))) {
+            if (!contains(collection, element, checkDuplicateUntil)) {
                 collection.add(element);
             }
         }
     }
 
-    private boolean contains(Collection<Object> list, @Nullable Object 
element) {
-        if (element == null) {
-            return list.contains(null);
+    private boolean contains(
+            Collection<Object> list, @Nullable Object element, int 
checkDuplicateUntil) {
+        if (checkDuplicateUntil <= 0) {
+            return false;
         }
 
         for (Object o : list) {
             if (equals(o, element)) {
                 return true;
             }
+            checkDuplicateUntil--;
+            if (checkDuplicateUntil <= 0) {
+                break;
+            }
         }
         return false;
     }