[core] Improve performance of dinstinct collect agg. (#3772)

apache · Jul 18, 2024 · 6da33cb · 6da33cb
1 parent f457d35
commit 6da33cb
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 2 deletions.
diff --git a/...on-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java b/...on-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -179,7 +179,7 @@ private void updateWithSequenceGroup(KeyValue kv) {
                     row.setField(
                             i, aggregator == null ? field : aggregator.agg(accumulator, field));
                 } else if (aggregator != null) {
-                    row.setField(i, aggregator.agg(field, accumulator));
+                    row.setField(i, aggregator.aggReversed(accumulator, field));
                 }
             }
         }

diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregator.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregator.java
@@ -172,6 +172,10 @@ private static FieldAggregator createFieldNestedUpdateAgg(
 
     public abstract Object agg(Object accumulator, Object inputField);
 
+    public Object aggReversed(Object accumulator, Object inputField) {
+        return agg(inputField, accumulator);
+    }
+
     /** reset the aggregator to a clean start state. */
     public void reset() {}
 

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
@@ -87,6 +87,14 @@ String name() {
         return NAME;
     }
 
+    @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
+        // dinstinct process
+        return agg(accumulator, inputField);
+    }
+
     @Override
     public Object agg(Object accumulator, Object inputField) {
         if (accumulator == null && inputField == null) {
@@ -99,7 +107,9 @@ public Object agg(Object accumulator, Object inputField) {
 
         if (equaliser != null) {
             List<Object> collection = new ArrayList<>();
-            collectWithEqualiser(collection, accumulator);
+            // 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);
             return new GenericArray(collection.toArray());
         } else {