add equality, null, and range filter

benchmarks/src/test/java/org/apache/druid/benchmark/DictionaryEncodedStringIndexSupplierBenchmark.java

@@ -26,12 +26,12 @@
 import org.apache.druid.collections.bitmap.RoaringBitmapFactory;
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.segment.column.BitmapColumnIndex;
-import org.apache.druid.segment.column.IndexedUtf8ValueSetIndex;
-import org.apache.druid.segment.column.StringValueSetIndex;
 import org.apache.druid.segment.data.BitmapSerdeFactory;
 import org.apache.druid.segment.data.GenericIndexed;
 import org.apache.druid.segment.data.RoaringBitmapSerdeFactory;
+import org.apache.druid.segment.index.BitmapColumnIndex;
+import org.apache.druid.segment.index.IndexedUtf8ValueSetIndex;
+import org.apache.druid.segment.index.StringValueSetIndex;
 import org.apache.druid.segment.serde.StringUtf8ColumnIndexSupplier;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;

extensions-contrib/tdigestsketch/src/test/java/org/apache/druid/query/aggregation/tdigestsketch/sql/TDigestSketchSqlAggregatorTest.java

@@ -38,7 +38,6 @@
 import org.apache.druid.query.aggregation.tdigestsketch.TDigestSketchToQuantilePostAggregator;
 import org.apache.druid.query.dimension.DefaultDimensionSpec;
 import org.apache.druid.query.groupby.GroupByQuery;
-import org.apache.druid.query.ordering.StringComparators;
 import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
 import org.apache.druid.segment.IndexBuilder;
 import org.apache.druid.segment.QueryableIndex;
@@ -444,7 +443,7 @@ public void testEmptyTimeseriesResults()
             Druids.newTimeseriesQueryBuilder()
                   .dataSource(CalciteTests.DATASOURCE1)
                   .intervals(new MultipleIntervalSegmentSpec(ImmutableList.of(Filtration.eternity())))
-                  .filters(bound("dim2", "0", "0", false, false, null, StringComparators.NUMERIC))
+                  .filters(equality("dim2", 0L, ColumnType.LONG))
                   .granularity(Granularities.ALL)
                   .aggregators(ImmutableList.of(
                       new TDigestSketchAggregatorFactory("a0:agg", "m1", TDigestSketchAggregatorFactory.DEFAULT_COMPRESSION),
@@ -476,19 +475,19 @@ public void testGroupByAggregatorDefaultValues()
             GroupByQuery.builder()
                         .setDataSource(CalciteTests.DATASOURCE1)
                         .setInterval(querySegmentSpec(Filtration.eternity()))
-                        .setDimFilter(selector("dim2", "a", null))
+                        .setDimFilter(equality("dim2", "a", ColumnType.STRING))
                         .setGranularity(Granularities.ALL)
                         .setVirtualColumns(expressionVirtualColumn("v0", "'a'", ColumnType.STRING))
                         .setDimensions(new DefaultDimensionSpec("v0", "d0", ColumnType.STRING))
                         .setAggregatorSpecs(
                             aggregators(
                                 new FilteredAggregatorFactory(
                                     new TDigestSketchAggregatorFactory("a0:agg", "m1", TDigestSketchAggregatorFactory.DEFAULT_COMPRESSION),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new TDigestSketchAggregatorFactory("a1:agg", "qsketch_m1", 100),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 )
                             )
                         )

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/hll/HllSketchBuildColumnProcessorFactory.java

@@ -21,13 +21,17 @@
 
 import org.apache.datasketches.hll.HllSketch;
 import org.apache.druid.java.util.common.StringEncoding;
+import org.apache.druid.math.expr.ExprEval;
+import org.apache.druid.math.expr.ExpressionType;
 import org.apache.druid.segment.BaseDoubleColumnValueSelector;
 import org.apache.druid.segment.BaseFloatColumnValueSelector;
 import org.apache.druid.segment.BaseLongColumnValueSelector;
 import org.apache.druid.segment.BaseObjectColumnValueSelector;
 import org.apache.druid.segment.ColumnProcessorFactory;
 import org.apache.druid.segment.DimensionSelector;
+import org.apache.druid.segment.column.ColumnCapabilities;
 import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.NullableTypeStrategy;
 import org.apache.druid.segment.data.IndexedInts;
 
 import java.util.function.Consumer;
@@ -96,6 +100,25 @@ public Consumer<Supplier<HllSketch>> makeLongProcessor(BaseLongColumnValueSelect
     };
   }
 
+  @Override
+  public Consumer<Supplier<HllSketch>> makeArrayProcessor(
+      BaseObjectColumnValueSelector<?> selector,
+      ColumnCapabilities columnCapabilities
+  )
+  {
+    final ExpressionType expressionType = ExpressionType.fromColumnType(columnCapabilities);
+    final NullableTypeStrategy<Object> strategy = expressionType.getNullableStrategy();
+    return sketch -> {
+      final Object o = selector.getObject();
+      if (o != null) {
+        byte[] bytes = ExprEval.toBytes(expressionType, strategy, o);
+        sketch.get().update(bytes);
+      }
+    };
+  }
+
+
+
   @Override
   public Consumer<Supplier<HllSketch>> makeComplexProcessor(BaseObjectColumnValueSelector<?> selector)
   {

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/hll/vector/HllSketchBuildVectorProcessorFactory.java

@@ -19,15 +19,22 @@
 
 package org.apache.druid.query.aggregation.datasketches.hll.vector;
 
+import org.apache.datasketches.hll.HllSketch;
 import org.apache.druid.java.util.common.StringEncoding;
+import org.apache.druid.math.expr.ExprEval;
+import org.apache.druid.math.expr.ExpressionType;
 import org.apache.druid.query.aggregation.datasketches.hll.HllSketchBuildBufferAggregatorHelper;
 import org.apache.druid.segment.VectorColumnProcessorFactory;
 import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.NullableTypeStrategy;
 import org.apache.druid.segment.vector.MultiValueDimensionVectorSelector;
 import org.apache.druid.segment.vector.SingleValueDimensionVectorSelector;
 import org.apache.druid.segment.vector.VectorObjectSelector;
 import org.apache.druid.segment.vector.VectorValueSelector;
 
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+
 public class HllSketchBuildVectorProcessorFactory implements VectorColumnProcessorFactory<HllSketchBuildVectorProcessor>
 {
   private final HllSketchBuildBufferAggregatorHelper helper;
@@ -83,6 +90,49 @@ public HllSketchBuildVectorProcessor makeLongProcessor(ColumnCapabilities capabi
     return new LongHllSketchBuildVectorProcessor(helper, selector);
   }
 
+  @Override
+  public HllSketchBuildVectorProcessor makeArrayProcessor(
+      ColumnCapabilities capabilities,
+      VectorObjectSelector selector
+  )
+  {
+    final ExpressionType expressionType = ExpressionType.fromColumnType(capabilities);
+    final NullableTypeStrategy<Object> typeStrategy = expressionType.getNullableStrategy();
+    return new HllSketchBuildVectorProcessor()
+    {
+      @Override
+      public void aggregate(ByteBuffer buf, int position, int startRow, int endRow)
+      {
+        final Object[] vector = selector.getObjectVector();
+        final HllSketch sketch = helper.getSketchAtPosition(buf, position);
+
+        for (int i = startRow; i < endRow; i++) {
+          if (vector[i] != null) {
+            byte[] bytes = ExprEval.toBytes(expressionType, typeStrategy, vector[i]);
+            sketch.update(bytes);
+          }
+        }
+      }
+
+      @Override
+      public void aggregate(ByteBuffer buf, int numRows, int[] positions, @Nullable int[] rows, int positionOffset)
+      {
+        final Object[] vector = selector.getObjectVector();
+
+        for (int i = 0; i < numRows; i++) {
+          final int idx = rows != null ? rows[i] : i;
+          final int position = positions[i] + positionOffset;
+          final HllSketch sketch = helper.getSketchAtPosition(buf, position);
+
+          if (vector[idx] != null) {
+            byte[] bytes = ExprEval.toBytes(expressionType, typeStrategy, vector[idx]);
+            sketch.update(bytes);
+          }
+        }
+      }
+    };
+  }
+
   @Override
   public HllSketchBuildVectorProcessor makeObjectProcessor(
       ColumnCapabilities capabilities,

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/kll/KllDoublesSketchAggregatorFactory.java

@@ -179,7 +179,7 @@ public VectorAggregator makeSingleValueDimensionProcessor(
               SingleValueDimensionVectorSelector selector
           )
           {
-            return new KllSketchNoOpBufferAggregator<KllDoublesSketch>(getEmptySketch());
+            return new KllSketchNoOpBufferAggregator<>(getEmptySketch());
           }
 
           @Override
@@ -188,7 +188,7 @@ public VectorAggregator makeMultiValueDimensionProcessor(
               MultiValueDimensionVectorSelector selector
           )
           {
-            return new KllSketchNoOpBufferAggregator<KllDoublesSketch>(getEmptySketch());
+            return new KllSketchNoOpBufferAggregator<>(getEmptySketch());
           }
 
           @Override
@@ -209,6 +209,12 @@ public VectorAggregator makeLongProcessor(ColumnCapabilities capabilities, Vecto
             return new KllDoublesSketchBuildVectorAggregator(selector, getK(), getMaxIntermediateSizeWithNulls());
           }
 
+          @Override
+          public VectorAggregator makeArrayProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
+          {
+            return new KllSketchNoOpBufferAggregator<>(getEmptySketch());
+          }
+
           @Override
           public VectorAggregator makeObjectProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
           {

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/kll/KllFloatsSketchAggregatorFactory.java

@@ -179,7 +179,7 @@ public VectorAggregator makeSingleValueDimensionProcessor(
               SingleValueDimensionVectorSelector selector
           )
           {
-            return new KllSketchNoOpBufferAggregator<KllFloatsSketch>(getEmptySketch());
+            return new KllSketchNoOpBufferAggregator<>(getEmptySketch());
           }
 
           @Override
@@ -188,7 +188,7 @@ public VectorAggregator makeMultiValueDimensionProcessor(
               MultiValueDimensionVectorSelector selector
           )
           {
-            return new KllSketchNoOpBufferAggregator<KllFloatsSketch>(getEmptySketch());
+            return new KllSketchNoOpBufferAggregator<>(getEmptySketch());
           }
 
           @Override
@@ -209,6 +209,12 @@ public VectorAggregator makeLongProcessor(ColumnCapabilities capabilities, Vecto
             return new KllFloatsSketchBuildVectorAggregator(selector, getK(), getMaxIntermediateSizeWithNulls());
           }
 
+          @Override
+          public VectorAggregator makeArrayProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
+          {
+            return new KllSketchNoOpBufferAggregator<>(getEmptySketch());
+          }
+
           @Override
           public VectorAggregator makeObjectProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
           {

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/quantiles/DoublesSketchAggregatorFactory.java

@@ -208,6 +208,12 @@ public VectorAggregator makeLongProcessor(ColumnCapabilities capabilities, Vecto
             return new DoublesSketchBuildVectorAggregator(selector, k, getMaxIntermediateSizeWithNulls());
           }
 
+          @Override
+          public VectorAggregator makeArrayProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
+          {
+            return new NoopDoublesSketchBufferAggregator();
+          }
+
           @Override
           public VectorAggregator makeObjectProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
           {

extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/util/ToObjectVectorColumnProcessorFactory.java

@@ -133,6 +133,12 @@ public Supplier<Object[]> makeLongProcessor(ColumnCapabilities capabilities, Vec
     };
   }
 
+  @Override
+  public Supplier<Object[]> makeArrayProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
+  {
+    return selector::getObjectVector;
+  }
+
   @Override
   public Supplier<Object[]> makeObjectProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
   {

extensions-core/datasketches/src/test/java/org/apache/druid/query/aggregation/datasketches/hll/sql/HllSketchSqlAggregatorTest.java

@@ -175,7 +175,10 @@ public class HllSketchSqlAggregatorTest extends BaseCalciteQueryTest
   private static final List<AggregatorFactory> EXPECTED_FILTERED_AGGREGATORS =
       EXPECTED_PA_AGGREGATORS.stream()
                              .limit(5)
-                             .map(factory -> new FilteredAggregatorFactory(factory, selector("dim2", "a", null)))
+                             .map(factory -> new FilteredAggregatorFactory(
+                                 factory,
+                                 equality("dim2", "a", ColumnType.STRING)
+                             ))
                              .collect(Collectors.toList());
 
   /**
@@ -344,7 +347,7 @@ public void testApproxCountDistinctHllSketch()
                           new HllSketchBuildAggregatorFactory("a1", "dim2", null, null, null, null, ROUND),
                           new FilteredAggregatorFactory(
                               new HllSketchBuildAggregatorFactory("a2", "dim2", null, null, null, null, ROUND),
-                              BaseCalciteQueryTest.not(BaseCalciteQueryTest.selector("dim2", "", null))
+                              not(equality("dim2", "", ColumnType.STRING))
                           ),
                           new HllSketchBuildAggregatorFactory("a3", "v0", null, null, null, null, ROUND),
                           new HllSketchBuildAggregatorFactory("a4", "v1", null, null, null, null, ROUND),
@@ -436,7 +439,7 @@ public void testAvgDailyCountDistinctHllSketch()
                                 new LongSumAggregatorFactory("_a0:sum", "a0"),
                                 new FilteredAggregatorFactory(
                                     new CountAggregatorFactory("_a0:count"),
-                                    BaseCalciteQueryTest.not(BaseCalciteQueryTest.selector("a0", null, null))
+                                    notNull("a0")
                                 )
                             )
                         )
@@ -480,7 +483,7 @@ public void testApproxCountDistinctHllSketchIsRounded()
                                 new HllSketchBuildAggregatorFactory("a0", "m1", null, null, null, true, true)
                             )
                         )
-                        .setHavingSpec(having(selector("a0", "2", null)))
+                        .setHavingSpec(having(equality("a0", 2L, ColumnType.LONG)))
                         .setContext(QUERY_CONTEXT_DEFAULT)
                         .build()
         ),
@@ -852,7 +855,11 @@ public void testEmptyTimeseriesResults()
         ImmutableList.of(Druids.newTimeseriesQueryBuilder()
                                .dataSource(CalciteTests.DATASOURCE1)
                                .intervals(querySegmentSpec(Filtration.eternity()))
-                               .filters(bound("dim2", "0", "0", false, false, null, StringComparators.NUMERIC))
+                               .filters(
+                                   NullHandling.replaceWithDefault()
+                                   ? bound("dim2", "0", "0", false, false, null, StringComparators.NUMERIC)
+                                   : equality("dim2", 0L, ColumnType.LONG)
+                               )
                                .granularity(Granularities.ALL)
                                .aggregators(
                                    aggregators(
@@ -895,7 +902,7 @@ public void testGroupByAggregatorDefaultValues()
             GroupByQuery.builder()
                         .setDataSource(CalciteTests.DATASOURCE1)
                         .setInterval(querySegmentSpec(Filtration.eternity()))
-                        .setDimFilter(selector("dim2", "a", null))
+                        .setDimFilter(equality("dim2", "a", ColumnType.STRING))
                         .setGranularity(Granularities.ALL)
                         .setVirtualColumns(expressionVirtualColumn("v0", "'a'", ColumnType.STRING))
                         .setDimensions(new DefaultDimensionSpec("v0", "d0", ColumnType.STRING))
@@ -911,7 +918,7 @@ public void testGroupByAggregatorDefaultValues()
                                         null,
                                         true
                                     ),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new HllSketchBuildAggregatorFactory(
@@ -923,7 +930,7 @@ public void testGroupByAggregatorDefaultValues()
                                         false,
                                         true
                                     ),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 )
                             )
                         )
@@ -954,19 +961,19 @@ public void testGroupByAggregatorDefaultValuesFinalizeOuterSketches()
             GroupByQuery.builder()
                         .setDataSource(CalciteTests.DATASOURCE1)
                         .setInterval(querySegmentSpec(Filtration.eternity()))
-                        .setDimFilter(selector("dim2", "a", null))
+                        .setDimFilter(equality("dim2", "a", ColumnType.STRING))
                         .setGranularity(Granularities.ALL)
                         .setVirtualColumns(expressionVirtualColumn("v0", "'a'", ColumnType.STRING))
                         .setDimensions(new DefaultDimensionSpec("v0", "d0", ColumnType.STRING))
                         .setAggregatorSpecs(
                             aggregators(
                                 new FilteredAggregatorFactory(
                                     new HllSketchBuildAggregatorFactory("a0", "v0", null, null, null, null, true),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new HllSketchBuildAggregatorFactory("a1", "v0", null, null, null, null, true),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 )
                             )
                         )