Reuse IndexedInts returned from DimensionSelector.getRow() implementations

processing/src/main/java/io/druid/query/dimension/ForwardingFilteredDimensionSelector.java

@@ -42,6 +42,7 @@ final class ForwardingFilteredDimensionSelector implements DimensionSelector, Id
   private final IdLookup baseIdLookup;
   private final Int2IntOpenHashMap forwardMapping;
   private final int[] reverseMapping;
+  private final ArrayBasedIndexedInts row = new ArrayBasedIndexedInts();
 
   /**
    * @param selector must return true from {@link DimensionSelector#nameLookupPossibleInAdvance()}
@@ -70,15 +71,17 @@ public IndexedInts getRow()
   {
     IndexedInts baseRow = selector.getRow();
     int baseRowSize = baseRow.size();
-    int[] result = new int[baseRowSize];
+    row.ensureSize(baseRowSize);
     int resultSize = 0;
     for (int i = 0; i < baseRowSize; i++) {
       int forwardedValue = forwardMapping.get(baseRow.get(i));
       if (forwardedValue >= 0) {
-        result[resultSize++] = forwardedValue;
+        row.setValue(resultSize, forwardedValue);
+        resultSize++;
       }
     }
-    return ArrayBasedIndexedInts.of(result, resultSize);
+    row.setSize(resultSize);
+    return row;
   }
 
   @Override

processing/src/main/java/io/druid/query/dimension/PredicateFilteredDimensionSelector.java

@@ -34,6 +34,7 @@ final class PredicateFilteredDimensionSelector implements DimensionSelector
 {
   private final DimensionSelector selector;
   private final Predicate<String> predicate;
+  private final ArrayBasedIndexedInts row = new ArrayBasedIndexedInts();
 
   PredicateFilteredDimensionSelector(DimensionSelector selector, Predicate<String> predicate)
   {
@@ -46,14 +47,16 @@ public IndexedInts getRow()
   {
     IndexedInts baseRow = selector.getRow();
     int baseRowSize = baseRow.size();
-    int[] result = new int[baseRowSize];
+    row.ensureSize(baseRowSize);
     int resultSize = 0;
     for (int i = 0; i < baseRowSize; i++) {
       if (predicate.apply(selector.lookupName(baseRow.get(i)))) {
-        result[resultSize++] = i;
+        row.setValue(resultSize, i);
+        resultSize++;
       }
     }
-    return ArrayBasedIndexedInts.of(result, resultSize);
+    row.setSize(resultSize);
+    return row;
   }
 
   @Override

processing/src/main/java/io/druid/query/groupby/RowBasedColumnSelectorFactory.java

@@ -203,11 +203,14 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
     } else {
       return new DimensionSelector()
       {
+        private final RangeIndexedInts indexedInts = new RangeIndexedInts();
+
         @Override
         public IndexedInts getRow()
         {
           final List<String> dimensionValues = row.get().getDimension(dimension);
-          return RangeIndexedInts.create(dimensionValues != null ? dimensionValues.size() : 0);
+          indexedInts.setSize(dimensionValues != null ? dimensionValues.size() : 0);
+          return indexedInts;
         }
 
         @Override

processing/src/main/java/io/druid/query/groupby/epinephelinae/column/DictionaryBuildingStringGroupByColumnSelectorStrategy.java

@@ -68,21 +68,26 @@ public void initColumnValues(ColumnValueSelector selector, int columnIndex, Obje
   {
     final DimensionSelector dimSelector = (DimensionSelector) selector;
     final IndexedInts row = dimSelector.getRow();
-    final int[] newIds = new int[row.size()];
-
+    ArrayBasedIndexedInts newRow = (ArrayBasedIndexedInts) valuess[columnIndex];
+    if (newRow == null) {
+      newRow = new ArrayBasedIndexedInts();
+      valuess[columnIndex] = newRow;
+    }
+    int rowSize = row.size();
+    newRow.ensureSize(rowSize);
     for (int i = 0; i < row.size(); i++) {
       final String value = dimSelector.lookupName(row.get(i));
       final int dictId = reverseDictionary.getInt(value);
       if (dictId < 0) {
         dictionary.add(value);
         reverseDictionary.put(value, nextId);
-        newIds[i] = nextId;
+        newRow.setValue(i, nextId);
         nextId++;
       } else {
-        newIds[i] = dictId;
+        newRow.setValue(i, dictId);
       }
     }
-    valuess[columnIndex] = ArrayBasedIndexedInts.of(newIds);
+    newRow.setSize(rowSize);
   }
 
   @Override

processing/src/main/java/io/druid/segment/ColumnValueSelector.java

@@ -22,7 +22,8 @@
 import io.druid.guice.annotations.PublicApi;
 
 /**
- * Base type for interfaces that manage column value selection, e.g. DimensionSelector, LongColumnSelector
+ * Base type for interfaces that manage column value selection, e.g. {@link DimensionSelector}, {@link
+ * LongColumnSelector}.
  *
  * This interface has methods to get the value in all primitive types, that have corresponding basic aggregators in
  * Druid: Sum, Min, Max, etc: {@link #getFloat()}, {@link #getDouble()} and {@link #getLong()} to support "polymorphic"

processing/src/main/java/io/druid/segment/DimensionSelector.java

@@ -36,11 +36,15 @@ public interface DimensionSelector extends ColumnValueSelector, HotLoopCallee
   int CARDINALITY_UNKNOWN = -1;
 
   /**
-   * Gets all values for the row inside of an IntBuffer.  I.e. one possible implementation could be
+   * Returns the indexed values at the current position in this DimensionSelector.
    *
-   * return IntBuffer.wrap(lookupExpansion(get());
-   *
-   * @return all values for the row as an IntBuffer
+   * IMPORTANT. The returned {@link IndexedInts} object could generally be reused inside the implementation of
+   * DimensionSelector, i. e. this method could always return the same object for the same selector. Users
+   * of this API, such as {@link io.druid.query.aggregation.Aggregator#aggregate()}, {@link
+   * io.druid.query.aggregation.BufferAggregator#aggregate}, {@link io.druid.query.aggregation.AggregateCombiner#reset},
+   * {@link io.druid.query.aggregation.AggregateCombiner#fold} should be prepared for that and not storing the object
+   * returned from this method in their state, assuming that the object will remain unchanged even when the position of
+   * the selector changes. This may not be the case.
    */
   @CalledFromHotLoop
   IndexedInts getRow();

processing/src/main/java/io/druid/segment/SingleScanTimeDimSelector.java

@@ -38,6 +38,7 @@ public class SingleScanTimeDimSelector implements DimensionSelector
   private final boolean descending;
 
   private final List<String> timeValues = new ArrayList<>();
+  private final SingleIndexedInt row = new SingleIndexedInt();
   private String currentValue = null;
   private long currentTimestamp = Long.MIN_VALUE;
   private int index = -1;
@@ -61,7 +62,8 @@ public SingleScanTimeDimSelector(BaseLongColumnValueSelector selector, Extractio
   @Override
   public IndexedInts getRow()
   {
-    return SingleIndexedInt.of(getDimensionValueIndex());
+    row.setValue(getDimensionValueIndex());
+    return row;
   }
 
   @Override

processing/src/main/java/io/druid/segment/StringDimensionIndexer.java

@@ -366,6 +366,7 @@ public DimensionSelector makeDimensionSelector(
 
     class IndexerDimensionSelector implements DimensionSelector, IdLookup
     {
+      private final ArrayBasedIndexedInts indexedInts = new ArrayBasedIndexedInts();
       private int[] nullIdIntArray;
 
       @Override
@@ -405,7 +406,8 @@ public IndexedInts getRow()
           rowSize = indices.length;
         }
 
-        return ArrayBasedIndexedInts.of(row, rowSize);
+        indexedInts.setValues(row, rowSize);
+        return indexedInts;
       }
 
       @Override

processing/src/main/java/io/druid/segment/column/SimpleDictionaryEncodedColumn.java

@@ -203,10 +203,13 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
       class SingleValueQueryableDimensionSelector extends QueryableDimensionSelector
           implements SingleValueHistoricalDimensionSelector
       {
+        private final SingleIndexedInt row = new SingleIndexedInt();
+
         @Override
         public IndexedInts getRow()
         {
-          return SingleIndexedInt.of(getRowValue());
+          row.setValue(getRowValue());
+          return row;
         }
 
         public int getRowValue()
@@ -217,7 +220,8 @@ public int getRowValue()
         @Override
         public IndexedInts getRow(int offset)
         {
-          return SingleIndexedInt.of(getRowValue(offset));
+          row.setValue(getRowValue(offset));
+          return row;
         }
 
         @Override

processing/src/main/java/io/druid/segment/data/ArrayBasedIndexedInts.java

@@ -27,36 +27,55 @@
  */
 public final class ArrayBasedIndexedInts implements IndexedInts
 {
-  private static final ArrayBasedIndexedInts EMPTY = new ArrayBasedIndexedInts(IntArrays.EMPTY_ARRAY, 0);
+  private int[] expansion;
+  private int size;
 
-  public static ArrayBasedIndexedInts of(int[] expansion)
+  public ArrayBasedIndexedInts()
   {
-    if (expansion.length == 0) {
-      return EMPTY;
-    }
-    return new ArrayBasedIndexedInts(expansion, expansion.length);
+    expansion = IntArrays.EMPTY_ARRAY;
+    size = 0;
+  }
+
+  public ArrayBasedIndexedInts(int[] expansion)
+  {
+    this.expansion = expansion;
+    this.size = expansion.length;
   }
 
-  public static ArrayBasedIndexedInts of(int[] expansion, int size)
+  public void ensureSize(int size)
   {
-    if (size == 0) {
-      return EMPTY;
+    if (expansion.length < size) {
+      expansion = new int[size];
     }
+  }
+
+  public void setSize(int size)
+  {
     if (size < 0 || size > expansion.length) {
-      throw new IAE("Size[%s] should be between 0 and %s", size, expansion.length);
+      throw new IAE("Size[%d] > expansion.length[%d] or < 0", size, expansion.length);
     }
-    return new ArrayBasedIndexedInts(expansion, size);
+    this.size = size;
   }
 
-  private final int[] expansion;
-  private final int size;
-
-  private ArrayBasedIndexedInts(int[] expansion, int size)
+  /**
+   * Sets the values from the given array. The given values array is not reused and not prone to be mutated later.
+   * Instead, the values from this array are copied into an array which is internal to ArrayBasedIndexedInts.
+   */
+  public void setValues(int[] values, int size)
   {
-    this.expansion = expansion;
+    if (size < 0 || size > values.length) {
+      throw new IAE("Size[%d] should be between 0 and %d", size, values.length);
+    }
+    ensureSize(size);
+    System.arraycopy(values, 0, expansion, 0, size);
     this.size = size;
   }
 
+  public void setValue(int index, int value)
+  {
+    expansion[index] = value;
+  }
+
   @Override
   public int size()
   {
@@ -66,8 +85,8 @@ public int size()
   @Override
   public int get(int index)
   {
-    if (index >= size) {
-      throw new IndexOutOfBoundsException("index: " + index + ", size: " + size);
+    if (index < 0 || index >= size) {
+      throw new IAE("index[%d] >= size[%d] or < 0", index, size);
     }
     return expansion[index];
   }

processing/src/main/java/io/druid/segment/data/CompressedVSizeColumnarMultiIntsSupplier.java

@@ -147,10 +147,13 @@ public static class CompressedVSizeColumnarMultiInts implements ColumnarMultiInt
     private final ColumnarInts offsets;
     private final ColumnarInts values;
 
+    private final SliceIndexedInts rowValues;
+
     CompressedVSizeColumnarMultiInts(ColumnarInts offsets, ColumnarInts values)
     {
       this.offsets = offsets;
       this.values = values;
+      this.rowValues = new SliceIndexedInts(values);
     }
 
     @Override
@@ -177,30 +180,8 @@ public IndexedInts get(int index)
     {
       final int offset = offsets.get(index);
       final int size = offsets.get(index + 1) - offset;
-
-      return new IndexedInts()
-      {
-        @Override
-        public int size()
-        {
-          return size;
-        }
-
-        @Override
-        public int get(int index)
-        {
-          if (index >= size) {
-            throw new IAE("Index[%d] >= size[%d]", index, size);
-          }
-          return values.get(index + offset);
-        }
-
-        @Override
-        public void inspectRuntimeShape(RuntimeShapeInspector inspector)
-        {
-          inspector.visit("values", values);
-        }
-      };
+      rowValues.setValues(offset, size);
+      return rowValues;
     }
 
     @Override

processing/src/main/java/io/druid/segment/data/RangeIndexedInts.java

@@ -19,38 +19,26 @@
 
 package io.druid.segment.data;
 
-import com.google.common.base.Preconditions;
+import io.druid.java.util.common.IAE;
 import io.druid.query.monomorphicprocessing.RuntimeShapeInspector;
 
 /**
- * An IndexedInts that always returns [0, 1, ..., N].
+ * Reusable IndexedInts that returns sequences [0, 1, ..., N].
  */
 public class RangeIndexedInts implements IndexedInts
 {
-  private static final int CACHE_LIMIT = 8;
-  private static final RangeIndexedInts[] CACHE = new RangeIndexedInts[CACHE_LIMIT];
+  private int size;
 
-  static {
-    for (int i = 0; i < CACHE_LIMIT; i++) {
-      CACHE[i] = new RangeIndexedInts(i);
-    }
-  }
-
-  private final int size;
-
-  private RangeIndexedInts(int size)
+  public RangeIndexedInts()
   {
-    this.size = size;
   }
 
-  public static RangeIndexedInts create(final int size)
+  public void setSize(int size)
   {
-    Preconditions.checkArgument(size >= 0, "size >= 0");
-    if (size < CACHE_LIMIT) {
-      return CACHE[size];
-    } else {
-      return new RangeIndexedInts(size);
+    if (size < 0) {
+      throw new IAE("Size[%d] must be non-negative", size);
     }
+    this.size = size;
   }
 
   @Override
@@ -63,7 +51,7 @@ public int size()
   public int get(int index)
   {
     if (index < 0 || index >= size) {
-      throw new IndexOutOfBoundsException("index: " + index);
+      throw new IAE("index[%d] >= size[%d] or < 0", index, size);
     }
     return index;
   }