Add IntGrouper to avoid unnecessary boxing/unboxing in array-based ag…

…gregation (#4668)

* Add IntGrouper

* Fix build

* Address comments

* Add a benchmark query

benchmarks/src/main/java/io/druid/benchmark/query/GroupByBenchmark.java

@@ -266,6 +266,29 @@ private void setupQueries()
 
       basicQueries.put("filter", queryA);
     }
+
+    { // basic.singleZipf
+      final QuerySegmentSpec intervalSpec = new MultipleIntervalSegmentSpec(
+          Collections.singletonList(basicSchema.getDataInterval())
+      );
+      // Use multiple aggregators to see how the number of aggregators impact to the query performance
+      List<AggregatorFactory> queryAggs = ImmutableList.of(
+          new LongSumAggregatorFactory("sumLongSequential", "sumLongSequential"),
+          new LongSumAggregatorFactory("rows", "rows"),
+          new DoubleSumAggregatorFactory("sumFloatNormal", "sumFloatNormal"),
+          new DoubleMinAggregatorFactory("minFloatZipf", "minFloatZipf")
+      );
+      GroupByQuery queryA = GroupByQuery
+          .builder()
+          .setDataSource("blah")
+          .setQuerySegmentSpec(intervalSpec)
+          .setDimensions(ImmutableList.of(new DefaultDimensionSpec("dimZipf", null)))
+          .setAggregatorSpecs(queryAggs)
+          .setGranularity(Granularity.fromString(queryGranularity))
+          .build();
+
+      basicQueries.put("singleZipf", queryA);
+    }
     SCHEMA_QUERY_MAP.put("basic", basicQueries);
 
     // simple one column schema, for testing performance difference between querying on numeric values as Strings and

processing/src/main/java/io/druid/query/groupby/epinephelinae/BufferArrayGrouper.java

@@ -32,7 +32,6 @@
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
-import java.util.function.ToIntFunction;
 
 /**
  * A buffer grouper for array-based aggregation.  This grouper stores aggregated values in the buffer using the grouping
@@ -48,7 +47,7 @@
  * different segments cannot be currently retrieved, this grouper can be used only when performing per-segment query
  * execution.
  */
-public class BufferArrayGrouper implements Grouper<Integer>
+public class BufferArrayGrouper implements IntGrouper
 {
   private static final Logger LOG = new Logger(BufferArrayGrouper.class);
 
@@ -137,16 +136,14 @@ public boolean isInitialized()
   }
 
   @Override
-  public AggregateResult aggregate(Integer key, int dimIndex)
+  public AggregateResult aggregate(int key, int dimIndex)
   {
     Preconditions.checkArgument(
         dimIndex >= 0 && dimIndex < cardinalityWithMissingValue,
         "Invalid dimIndex[%s]",
         dimIndex
     );
 
-    Preconditions.checkNotNull(key);
-
     final int recordOffset = dimIndex * recordSize;
 
     if (recordOffset + recordSize > valBuffer.capacity()) {
@@ -209,7 +206,7 @@ public void reset()
   }
 
   @Override
-  public ToIntFunction<Integer> hashFunction()
+  public IntGrouperHashFunction hashFunction()
   {
     return key -> key + 1;
   }

processing/src/main/java/io/druid/query/groupby/epinephelinae/GroupByQueryEngineV2.java

@@ -582,7 +582,7 @@ public ArrayAggregateIterator(
     }
 
     @Override
-    protected Grouper<Integer> newGrouper()
+    protected IntGrouper newGrouper()
     {
       return new BufferArrayGrouper(
           Suppliers.ofInstance(buffer),
@@ -595,6 +595,17 @@ protected Grouper<Integer> newGrouper()
 
     @Override
     protected void aggregateSingleValueDims(Grouper<Integer> grouper)
+    {
+      aggregateSingleValueDims((IntGrouper) grouper);
+    }
+
+    @Override
+    protected void aggregateMultiValueDims(Grouper<Integer> grouper)
+    {
+      aggregateMultiValueDims((IntGrouper) grouper);
+    }
+
+    private void aggregateSingleValueDims(IntGrouper grouper)
     {
       while (!cursor.isDone()) {
         final int key;
@@ -612,8 +623,7 @@ protected void aggregateSingleValueDims(Grouper<Integer> grouper)
       }
     }
 
-    @Override
-    protected void aggregateMultiValueDims(Grouper<Integer> grouper)
+    private void aggregateMultiValueDims(IntGrouper grouper)
     {
       if (dim == null) {
         throw new ISE("dim must exist");

processing/src/main/java/io/druid/query/groupby/epinephelinae/IntGrouper.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.query.groupby.epinephelinae;
+
+import com.google.common.base.Preconditions;
+
+import java.util.function.ToIntFunction;
+
+/**
+ * {@link Grouper} specialized for the primitive int type
+ */
+public interface IntGrouper extends Grouper<Integer>
+{
+  default AggregateResult aggregate(int key)
+  {
+    return aggregate(key, hashFunction().apply(key));
+  }
+
+  AggregateResult aggregate(int key, int keyHash);
+
+  /**
+   * {@inheritDoc}
+   *
+   * @deprecated Please use {@link #aggregate(int)} instead.
+   */
+  @Deprecated
+  @Override
+  default AggregateResult aggregate(Integer key)
+  {
+    Preconditions.checkNotNull(key);
+    return aggregate(key.intValue());
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @deprecated Please use {@link #aggregate(int, int)} instead.
+   */
+  @Deprecated
+  @Override
+  default AggregateResult aggregate(Integer key, int keyHash)
+  {
+    Preconditions.checkNotNull(key);
+    return aggregate(key.intValue(), keyHash);
+  }
+
+  @Override
+  IntGrouperHashFunction hashFunction();
+
+  interface IntGrouperHashFunction extends ToIntFunction<Integer>
+  {
+    @Override
+    default int applyAsInt(Integer value)
+    {
+      return apply(value.intValue());
+    }
+
+    int apply(int value);
+  }
+}

processing/src/test/java/io/druid/query/groupby/epinephelinae/BufferArrayGrouperTest.java

@@ -43,7 +43,7 @@ public class BufferArrayGrouperTest
   public void testAggregate()
   {
     final TestColumnSelectorFactory columnSelectorFactory = GrouperTestUtil.newColumnSelectorFactory();
-    final Grouper<Integer> grouper = newGrouper(columnSelectorFactory, 1024);
+    final IntGrouper grouper = newGrouper(columnSelectorFactory, 1024);
 
     columnSelectorFactory.setRow(new MapBasedRow(0, ImmutableMap.of("value", 10L)));
     grouper.aggregate(12);