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,6 +21,7 @@
 
 import org.apache.datasketches.hll.HllSketch;
 import org.apache.druid.java.util.common.StringEncoding;
+import org.apache.druid.java.util.common.UOE;
 import org.apache.druid.segment.BaseDoubleColumnValueSelector;
 import org.apache.druid.segment.BaseFloatColumnValueSelector;
 import org.apache.druid.segment.BaseLongColumnValueSelector;
@@ -96,6 +97,13 @@ public Consumer<Supplier<HllSketch>> makeLongProcessor(BaseLongColumnValueSelect
     };
   }
 
+  @Override
+  public Consumer<Supplier<HllSketch>> makeArrayProcessor(BaseObjectColumnValueSelector<?> selector)
+  {
+    // todo (clint): pass in type info so we can convert these arrays to byte arrays
+    throw new UOE("HLL sketch does not support ARRAY inputs");
+  }
+
   @Override
   public Consumer<Supplier<HllSketch>> makeComplexProcessor(BaseObjectColumnValueSelector<?> selector)
   {

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

@@ -83,6 +83,15 @@ public HllSketchBuildVectorProcessor makeLongProcessor(ColumnCapabilities capabi
     return new LongHllSketchBuildVectorProcessor(helper, selector);
   }
 
+  @Override
+  public HllSketchBuildVectorProcessor makeArrayProcessor(
+      ColumnCapabilities capabilities,
+      VectorObjectSelector selector
+  )
+  {
+    return null;
+  }
+
   @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,19 @@ public VectorAggregator makeLongProcessor(ColumnCapabilities capabilities, Vecto
             return new KllDoublesSketchBuildVectorAggregator(selector, getK(), getMaxIntermediateSizeWithNulls());
           }
 
+          @Override
+          public VectorAggregator makeArrayProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
+          {
+            // todo (clint): y tho? shouldn't this (and string inputs) be an error?
+            /*
+            throw new UOE(
+                "KLL Doubles sketch does not support[%s] inputs",
+                capabilities.toColumnType()
+            );
+             */
+            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,20 @@ public VectorAggregator makeLongProcessor(ColumnCapabilities capabilities, Vecto
             return new KllFloatsSketchBuildVectorAggregator(selector, getK(), getMaxIntermediateSizeWithNulls());
           }
 
+          @Override
+          public VectorAggregator makeArrayProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
+          {
+
+            // todo (clint): y tho? shouldn't this (and string inputs) be an error?
+            /*
+            throw new UOE(
+                "KLL Floats sketch does not support[%s] inputs",
+                capabilities.toColumnType()
+            );
+             */
+            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,19 @@ public VectorAggregator makeLongProcessor(ColumnCapabilities capabilities, Vecto
             return new DoublesSketchBuildVectorAggregator(selector, k, getMaxIntermediateSizeWithNulls());
           }
 
+          @Override
+          public VectorAggregator makeArrayProcessor(ColumnCapabilities capabilities, VectorObjectSelector selector)
+          {
+            /*
+            throw new UOE(
+                "Doubles sketch does not support[%s] inputs",
+                capabilities.toColumnType()
+            );
+             */
+            // todo (clint): y tho? shouldn't this (and string inputs) be an error?
+            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", "", null))
                           ),
                           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()
         ),
@@ -895,7 +898,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 +914,7 @@ public void testGroupByAggregatorDefaultValues()
                                         null,
                                         true
                                     ),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new HllSketchBuildAggregatorFactory(
@@ -923,7 +926,7 @@ public void testGroupByAggregatorDefaultValues()
                                         false,
                                         true
                                     ),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 )
                             )
                         )
@@ -954,19 +957,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)
                                 )
                             )
                         )

extensions-core/datasketches/src/test/java/org/apache/druid/query/aggregation/datasketches/quantiles/sql/DoublesSketchSqlAggregatorTest.java

@@ -815,27 +815,27 @@ 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 DoublesSketchAggregatorFactory("a0:agg", "m1", null),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new DoublesSketchAggregatorFactory("a1:agg", "qsketch_m1", null),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new DoublesSketchAggregatorFactory("a2:agg", "m1", null, null, false),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new DoublesSketchAggregatorFactory("a3:agg", "qsketch_m1", null, null, false),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 )
                             )
                         )
@@ -886,27 +886,27 @@ public void testGroupByAggregatorDefaultValuesWithFinalizeSketches()
             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 DoublesSketchAggregatorFactory("a0:agg", "m1", null),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new DoublesSketchAggregatorFactory("a1:agg", "qsketch_m1", null),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new DoublesSketchAggregatorFactory("a2:agg", "m1", null, null, true),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 ),
                                 new FilteredAggregatorFactory(
                                     new DoublesSketchAggregatorFactory("a3:agg", "qsketch_m1", null, null, true),
-                                    selector("dim1", "nonexistent", null)
+                                    equality("dim1", "nonexistent", ColumnType.STRING)
                                 )
                             )
                         )