IGNITE-24131 Reuse GroupKey.Builder for hash aggregate (#11783)

Author: timoninmaxim

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/exp/agg/GroupKey.java

@@ -91,5 +91,12 @@ public GroupKey build() {
 
             return new GroupKey(fields);
         }
+
+        /** */
+        public void clear() {
+            Arrays.fill(fields, null);
+
+            idx = 0;
+        }
     }
 }

modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/rel/HashAggregateNode.java

@@ -24,6 +24,8 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.util.ImmutableBitSet;
@@ -244,13 +246,38 @@ private class Grouping {
         /** */
         private final RowHandler<Row> handler;
 
+        /** */
+        private GroupKey.Builder grpKeyBld;
+
+        /** */
+        private final BiFunction<GroupKey, List<AccumulatorWrapper<Row>>, List<AccumulatorWrapper<Row>>> getOrCreateGroup;
+
+        /** */
+        private final Function<GroupKey, List<AccumulatorWrapper<Row>>> createGroup;
+
         /** */
         private Grouping(byte grpId, ImmutableBitSet grpFields) {
             this.grpId = grpId;
             this.grpFields = grpFields;
 
+            grpKeyBld = GroupKey.builder(grpFields.cardinality());
             handler = context().rowHandler();
 
+            createGroup = (k) -> create();
+
+            getOrCreateGroup = (k, v) -> {
+                if (v == null) {
+                    grpKeyBld = GroupKey.builder(grpFields.cardinality());
+
+                    return create();
+                }
+                else {
+                    grpKeyBld.clear();
+
+                    return v;
+                }
+            };
+
             init();
         }
 
@@ -259,7 +286,7 @@ private void init() {
             // Initializes aggregates for case when no any rows will be added into the aggregate to have 0 as result.
             // Doesn't do it for MAP type due to we don't want send from MAP node zero results because it looks redundant.
             if (grpFields.isEmpty() && (type == AggregateType.REDUCE || type == AggregateType.SINGLE))
-                groups.put(GroupKey.EMPTY_GRP_KEY, create(GroupKey.EMPTY_GRP_KEY));
+                groups.put(GroupKey.EMPTY_GRP_KEY, create());
         }
 
         /** */
@@ -293,14 +320,10 @@ else if (type == AggregateType.MAP)
 
         /** */
         private void addOnMapper(Row row) {
-            GroupKey.Builder b = GroupKey.builder(grpFields.cardinality());
-
             for (Integer field : grpFields)
-                b.add(handler.get(field, row));
-
-            GroupKey grpKey = b.build();
+                grpKeyBld.add(handler.get(field, row));
 
-            List<AccumulatorWrapper<Row>> wrappers = groups.computeIfAbsent(grpKey, this::create);
+            List<AccumulatorWrapper<Row>> wrappers = groups.compute(grpKeyBld.build(), getOrCreateGroup);
 
             for (AccumulatorWrapper<Row> wrapper : wrappers)
                 wrapper.add(row);
@@ -315,7 +338,7 @@ private void addOnReducer(Row row) {
 
             GroupKey grpKey = (GroupKey)handler.get(1, row);
 
-            List<AccumulatorWrapper<Row>> wrappers = groups.computeIfAbsent(grpKey, this::create);
+            List<AccumulatorWrapper<Row>> wrappers = groups.computeIfAbsent(grpKey, createGroup);
             Accumulator<Row>[] accums = hasAccumulators() ? (Accumulator<Row>[])handler.get(2, row) : null;
 
             for (int i = 0; i < wrappers.size(); i++) {
@@ -388,7 +411,7 @@ private List<Row> getOnReducer(int cnt) {
         }
 
         /** */
-        private List<AccumulatorWrapper<Row>> create(GroupKey key) {
+        private List<AccumulatorWrapper<Row>> create() {
             if (accFactory == null)
                 return Collections.emptyList();