fix some issues with filters on numeric columns with nulls (#9251)

* fix issue with long column predicate filters and nulls

* dang

* uncomment a thing

* styles

* oops

* allcaps

* review stuff

processing/src/main/java/org/apache/druid/query/filter/JavaScriptDimFilter.java

@@ -254,21 +254,60 @@ public class JavaScriptDimFilter implements DimFilter
     public DruidLongPredicate makeLongPredicate()
     {
       // Can't avoid boxing here because the Mozilla JS Function.call() only accepts Object[]
-      return input -> applyObject(input);
+      return new DruidLongPredicate()
+      {
+        @Override
+        public boolean applyLong(long input)
+        {
+          return JavaScriptPredicateFactory.this.applyObject(input);
+        }
+
+        @Override
+        public boolean applyNull()
+        {
+          return JavaScriptPredicateFactory.this.applyObject(null);
+        }
+      };
     }
 
     @Override
     public DruidFloatPredicate makeFloatPredicate()
     {
       // Can't avoid boxing here because the Mozilla JS Function.call() only accepts Object[]
-      return input -> applyObject(input);
+      return new DruidFloatPredicate()
+      {
+        @Override
+        public boolean applyFloat(float input)
+        {
+          return JavaScriptPredicateFactory.this.applyObject(input);
+        }
+
+        @Override
+        public boolean applyNull()
+        {
+          return JavaScriptPredicateFactory.this.applyObject(null);
+        }
+      };
     }
 
     @Override
     public DruidDoublePredicate makeDoublePredicate()
     {
       // Can't avoid boxing here because the Mozilla JS Function.call() only accepts Object[]
-      return input -> applyObject(input);
+      return new DruidDoublePredicate()
+      {
+        @Override
+        public boolean applyDouble(double input)
+        {
+          return JavaScriptPredicateFactory.this.applyObject(input);
+        }
+
+        @Override
+        public boolean applyNull()
+        {
+          return JavaScriptPredicateFactory.this.applyObject(null);
+        }
+      };
     }
 
     public boolean applyObject(final Object input)

processing/src/main/java/org/apache/druid/query/filter/vector/DoubleVectorValueMatcher.java

@@ -61,11 +61,15 @@ public class DoubleVectorValueMatcher implements VectorValueMatcherFactory
       {
         final double[] vector = selector.getDoubleVector();
         final int[] selection = match.getSelection();
-
+        final boolean[] nulls = selector.getNullVector();
+        final boolean hasNulls = canHaveNulls && nulls != null;
         int numRows = 0;
 
         for (int i = 0; i < mask.getSelectionSize(); i++) {
           final int rowNum = mask.getSelection()[i];
+          if (hasNulls && nulls[rowNum]) {
+            continue;
+          }
           if (vector[rowNum] == matchValDouble) {
             selection[numRows++] = rowNum;
           }
@@ -92,12 +96,18 @@ public class DoubleVectorValueMatcher implements VectorValueMatcherFactory
       {
         final double[] vector = selector.getDoubleVector();
         final int[] selection = match.getSelection();
+        final boolean[] nulls = selector.getNullVector();
+        final boolean hasNulls = canHaveNulls && nulls != null;
 
         int numRows = 0;
 
         for (int i = 0; i < mask.getSelectionSize(); i++) {
           final int rowNum = mask.getSelection()[i];
-          if (predicate.applyDouble(vector[rowNum])) {
+          if (hasNulls && nulls[rowNum]) {
+            if (predicate.applyNull()) {
+              selection[numRows++] = rowNum;
+            }
+          } else if (predicate.applyDouble(vector[rowNum])) {
             selection[numRows++] = rowNum;
           }
         }

processing/src/main/java/org/apache/druid/query/filter/vector/FloatVectorValueMatcher.java

@@ -61,11 +61,16 @@ public class FloatVectorValueMatcher implements VectorValueMatcherFactory
       {
         final float[] vector = selector.getFloatVector();
         final int[] selection = match.getSelection();
+        final boolean[] nulls = selector.getNullVector();
+        final boolean hasNulls = canHaveNulls && nulls != null;
 
         int numRows = 0;
 
         for (int i = 0; i < mask.getSelectionSize(); i++) {
           final int rowNum = mask.getSelection()[i];
+          if (hasNulls && nulls[rowNum]) {
+            continue;
+          }
           if (vector[rowNum] == matchValFloat) {
             selection[numRows++] = rowNum;
           }
@@ -92,12 +97,18 @@ public class FloatVectorValueMatcher implements VectorValueMatcherFactory
       {
         final float[] vector = selector.getFloatVector();
         final int[] selection = match.getSelection();
+        final boolean[] nulls = selector.getNullVector();
+        final boolean hasNulls = canHaveNulls && nulls != null;
 
         int numRows = 0;
 
         for (int i = 0; i < mask.getSelectionSize(); i++) {
           final int rowNum = mask.getSelection()[i];
-          if (predicate.applyFloat(vector[rowNum])) {
+          if (hasNulls && nulls[rowNum]) {
+            if (predicate.applyNull()) {
+              selection[numRows++] = rowNum;
+            }
+          } else if (predicate.applyFloat(vector[rowNum])) {
             selection[numRows++] = rowNum;
           }
         }

processing/src/main/java/org/apache/druid/query/filter/vector/LongVectorValueMatcher.java

@@ -61,11 +61,16 @@ public class LongVectorValueMatcher implements VectorValueMatcherFactory
       {
         final long[] vector = selector.getLongVector();
         final int[] selection = match.getSelection();
+        final boolean[] nulls = selector.getNullVector();
+        final boolean hasNulls = canHaveNulls && nulls != null;
 
         int numRows = 0;
 
         for (int i = 0; i < mask.getSelectionSize(); i++) {
           final int rowNum = mask.getSelection()[i];
+          if (hasNulls && nulls[rowNum]) {
+            continue;
+          }
           if (vector[rowNum] == matchValLong) {
             selection[numRows++] = rowNum;
           }
@@ -92,12 +97,18 @@ public class LongVectorValueMatcher implements VectorValueMatcherFactory
       {
         final long[] vector = selector.getLongVector();
         final int[] selection = match.getSelection();
+        final boolean[] nulls = selector.getNullVector();
+        final boolean hasNulls = canHaveNulls && nulls != null;
 
         int numRows = 0;
 
         for (int i = 0; i < mask.getSelectionSize(); i++) {
           final int rowNum = mask.getSelection()[i];
-          if (predicate.applyLong(vector[rowNum])) {
+          if (hasNulls && nulls[rowNum]) {
+            if (predicate.applyNull()) {
+              selection[numRows++] = rowNum;
+            }
+          } else if (predicate.applyLong(vector[rowNum])) {
             selection[numRows++] = rowNum;
           }
         }

processing/src/main/java/org/apache/druid/segment/filter/DimensionPredicateFilter.java

@@ -90,19 +90,58 @@ public class DimensionPredicateFilter implements Filter
         @Override
         public DruidLongPredicate makeLongPredicate()
         {
-          return input -> baseStringPredicate.apply(extractionFn.apply(input));
+          return new DruidLongPredicate()
+          {
+            @Override
+            public boolean applyLong(long input)
+            {
+              return baseStringPredicate.apply(extractionFn.apply(input));
+            }
+
+            @Override
+            public boolean applyNull()
+            {
+              return baseStringPredicate.apply(extractionFn.apply(null));
+            }
+          };
         }
 
         @Override
         public DruidFloatPredicate makeFloatPredicate()
         {
-          return input -> baseStringPredicate.apply(extractionFn.apply(input));
+          return new DruidFloatPredicate()
+          {
+            @Override
+            public boolean applyFloat(float input)
+            {
+              return baseStringPredicate.apply(extractionFn.apply(input));
+            }
+
+            @Override
+            public boolean applyNull()
+            {
+              return baseStringPredicate.apply(extractionFn.apply(null));
+            }
+          };
         }
 
         @Override
         public DruidDoublePredicate makeDoublePredicate()
         {
-          return input -> baseStringPredicate.apply(extractionFn.apply(input));
+          return new DruidDoublePredicate()
+          {
+            @Override
+            public boolean applyDouble(double input)
+            {
+              return baseStringPredicate.apply(extractionFn.apply(input));
+            }
+
+            @Override
+            public boolean applyNull()
+            {
+              return baseStringPredicate.apply(extractionFn.apply(null));
+            }
+          };
         }
       };
     }

processing/src/main/java/org/apache/druid/segment/filter/Filters.java

@@ -36,23 +36,18 @@ import org.apache.druid.query.dimension.DefaultDimensionSpec;
 import org.apache.druid.query.filter.BitmapIndexSelector;
 import org.apache.druid.query.filter.BooleanFilter;
 import org.apache.druid.query.filter.DimFilter;
-import org.apache.druid.query.filter.DruidLongPredicate;
 import org.apache.druid.query.filter.DruidPredicateFactory;
 import org.apache.druid.query.filter.Filter;
 import org.apache.druid.query.filter.FilterTuning;
 import org.apache.druid.query.filter.ValueMatcher;
 import org.apache.druid.query.filter.ValueMatcherColumnSelectorStrategy;
 import org.apache.druid.query.filter.ValueMatcherColumnSelectorStrategyFactory;
-import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
-import org.apache.druid.segment.BaseLongColumnValueSelector;
 import org.apache.druid.segment.ColumnSelector;
 import org.apache.druid.segment.ColumnSelectorFactory;
 import org.apache.druid.segment.DimensionHandlerUtils;
 import org.apache.druid.segment.IntIteratorUtils;
 import org.apache.druid.segment.column.BitmapIndex;
-import org.apache.druid.segment.column.ColumnCapabilities;
 import org.apache.druid.segment.column.ColumnHolder;
-import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.segment.data.CloseableIndexed;
 import org.apache.druid.segment.data.Indexed;
 
@@ -156,16 +151,6 @@ public class Filters
       final DruidPredicateFactory predicateFactory
   )
   {
-    final ColumnCapabilities capabilities = columnSelectorFactory.getColumnCapabilities(columnName);
-
-    // This should be folded into the ValueMatcherColumnSelectorStrategy once that can handle LONG typed columns.
-    if (capabilities != null && capabilities.getType() == ValueType.LONG) {
-      return getLongPredicateMatcher(
-          columnSelectorFactory.makeColumnValueSelector(columnName),
-          predicateFactory.makeLongPredicate()
-      );
-    }
-
     final ColumnSelectorPlus<ValueMatcherColumnSelectorStrategy> selector =
         DimensionHandlerUtils.createColumnSelectorPlus(
             ValueMatcherColumnSelectorStrategyFactory.instance(),
@@ -454,28 +439,6 @@ public class Filters
     return false;
   }
 
-  public static ValueMatcher getLongPredicateMatcher(
-      final BaseLongColumnValueSelector longSelector,
-      final DruidLongPredicate predicate
-  )
-  {
-    return new ValueMatcher()
-    {
-      @Override
-      public boolean matches()
-      {
-        return predicate.applyLong(longSelector.getLong());
-      }
-
-      @Override
-      public void inspectRuntimeShape(RuntimeShapeInspector inspector)
-      {
-        inspector.visit("longSelector", longSelector);
-        inspector.visit("predicate", predicate);
-      }
-    };
-  }
-
   @Nullable
   public static Filter convertToCNFFromQueryContext(Query query, @Nullable Filter filter)
   {

processing/src/main/java/org/apache/druid/segment/vector/VectorSelectorUtils.java

@@ -22,6 +22,7 @@ package org.apache.druid.segment.vector;
 import org.roaringbitmap.PeekableIntIterator;
 
 import javax.annotation.Nullable;
+import java.util.Arrays;
 
 public class VectorSelectorUtils
 {
@@ -57,6 +58,7 @@ public class VectorSelectorUtils
         final int row = i + startOffset;
         nullIterator.advanceIfNeeded(row);
         if (!nullIterator.hasNext()) {
+          Arrays.fill(retVal, i, offset.getCurrentVectorSize(), false);
           break;
         }
         retVal[i] = row == nullIterator.peekNext();
@@ -71,6 +73,7 @@ public class VectorSelectorUtils
         final int row = currentOffsets[i];
         nullIterator.advanceIfNeeded(row);
         if (!nullIterator.hasNext()) {
+          Arrays.fill(retVal, i, offset.getCurrentVectorSize(), false);
           break;
         }
         retVal[i] = row == nullIterator.peekNext();

processing/src/test/java/org/apache/druid/segment/filter/BaseFilterTest.java

@@ -24,8 +24,19 @@ import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Iterables;
+import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.common.guava.SettableSupplier;
 import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.DoubleDimensionSchema;
+import org.apache.druid.data.input.impl.FloatDimensionSchema;
+import org.apache.druid.data.input.impl.InputRowParser;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.java.util.common.Pair;
@@ -62,6 +73,7 @@ import org.apache.druid.segment.data.ConciseBitmapSerdeFactory;
 import org.apache.druid.segment.data.IndexedInts;
 import org.apache.druid.segment.data.RoaringBitmapSerdeFactory;
 import org.apache.druid.segment.incremental.IncrementalIndex;
+import org.apache.druid.segment.incremental.IncrementalIndexSchema;
 import org.apache.druid.segment.incremental.IncrementalIndexStorageAdapter;
 import org.apache.druid.segment.vector.SingleValueDimensionVectorSelector;
 import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
@@ -77,6 +89,7 @@ import org.junit.Rule;
 import org.junit.rules.TemporaryFolder;
 import org.junit.runners.Parameterized;
 
+import javax.annotation.Nullable;
 import java.io.Closeable;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -90,6 +103,8 @@ import java.util.Set;
 
 public abstract class BaseFilterTest extends InitializedNullHandlingTest
 {
+  static final String TIMESTAMP_COLUMN = "timestamp";
+
   static final VirtualColumns VIRTUAL_COLUMNS = VirtualColumns.create(
       ImmutableList.of(
           new ExpressionVirtualColumn("expr", "1.0 + 0.1", ValueType.FLOAT, TestExprMacroTable.INSTANCE),
@@ -98,6 +113,61 @@ public abstract class BaseFilterTest extends InitializedNullHandlingTest
       )
   );
 
+  static final TimestampSpec DEFAULT_TIMESTAMP_SPEC = new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000"));
+  static final DimensionsSpec DEFAULT_DIM_SPEC = new DimensionsSpec(
+      ImmutableList.<DimensionSchema>builder()
+          .addAll(DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3", "timeDim")))
+          .add(new DoubleDimensionSchema("d0"))
+          .add(new FloatDimensionSchema("f0"))
+          .add(new LongDimensionSchema("l0"))
+          .build(),
+      null,
+      null
+  );
+
+  static final InputRowParser<Map<String, Object>> DEFAULT_PARSER = new MapInputRowParser(
+      new TimeAndDimsParseSpec(
+          DEFAULT_TIMESTAMP_SPEC,
+          DEFAULT_DIM_SPEC
+      )
+  );
+
+  static final List<InputRow> DEFAULT_ROWS = ImmutableList.of(
+      makeDefaultSchemaRow("0", "", ImmutableList.of("a", "b"), "2017-07-25", 0.0, 0.0f, 0L),
+      makeDefaultSchemaRow("1", "10", ImmutableList.of(), "2017-07-25", 10.1, 10.1f, 100L),
+      makeDefaultSchemaRow("2", "2", ImmutableList.of(""), "2017-05-25", null, 5.5f, 40L),
+      makeDefaultSchemaRow("3", "1", ImmutableList.of("a"), "2020-01-25", 120.0245, 110.0f, null),
+      makeDefaultSchemaRow("4", "abdef", ImmutableList.of("c"), null, 60.0, null, 9001L),
+      makeDefaultSchemaRow("5", "abc", null, "2020-01-25", 765.432, 123.45f, 12345L)
+  );
+
+  static final IncrementalIndexSchema DEFAULT_INDEX_SCHEMA = new IncrementalIndexSchema.Builder()
+      .withDimensionsSpec(DEFAULT_DIM_SPEC)
+      .withMetrics(new CountAggregatorFactory("count"))
+      .build();
+
+  static InputRow makeDefaultSchemaRow(
+      @Nullable String dim0,
+      @Nullable String dim1,
+      @Nullable List<String> dim2,
+      @Nullable String timeDim,
+      @Nullable Double d0,
+      @Nullable Float f0,
+      @Nullable Long l0)
+  {
+    // for row selector to work correctly as part of the test matrix, default value coercion needs to happen to columns
+    Map<String, Object> mapRow = new HashMap<>(6);
+    mapRow.put("dim0", NullHandling.nullToEmptyIfNeeded(dim0));
+    mapRow.put("dim1", NullHandling.nullToEmptyIfNeeded(dim1));
+    mapRow.put("dim2", dim2 != null ? dim2 : NullHandling.defaultStringValue());
+    mapRow.put("timeDim", NullHandling.nullToEmptyIfNeeded(timeDim));
+    mapRow.put("d0", d0 != null ? d0 : NullHandling.defaultDoubleValue());
+    mapRow.put("f0", f0 != null ? f0 : NullHandling.defaultFloatValue());
+    mapRow.put("l0", l0 != null ? l0 : NullHandling.defaultLongValue());
+    return DEFAULT_PARSER.parseBatch(mapRow).get(0);
+  }
+
+
   @Rule
   public TemporaryFolder temporaryFolder = new TemporaryFolder();
 
@@ -225,9 +295,9 @@ public abstract class BaseFilterTest extends InitializedNullHandlingTest
               );
               final IndexBuilder indexBuilder = IndexBuilder
                   .create()
+                  .schema(DEFAULT_INDEX_SCHEMA)
                   .indexSpec(new IndexSpec(bitmapSerdeFactoryEntry.getValue(), null, null, null))
                   .segmentWriteOutMediumFactory(segmentWriteOutMediumFactoryEntry.getValue());
-
               constructors.add(new Object[]{testName, indexBuilder, finisherEntry.getValue(), cnf, optimize});
             }
           }

processing/src/test/java/org/apache/druid/segment/filter/BoundFilterTest.java

@@ -21,15 +21,8 @@ package org.apache.druid.segment.filter;
 
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.data.input.InputRow;
-import org.apache.druid.data.input.impl.DimensionsSpec;
-import org.apache.druid.data.input.impl.InputRowParser;
-import org.apache.druid.data.input.impl.MapInputRowParser;
-import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
-import org.apache.druid.data.input.impl.TimestampSpec;
-import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.js.JavaScriptConfig;
 import org.apache.druid.query.extraction.ExtractionFn;
@@ -45,30 +38,15 @@ import org.junit.runners.Parameterized;
 
 import java.io.Closeable;
 import java.util.List;
-import java.util.Map;
 
 @RunWith(Parameterized.class)
 public class BoundFilterTest extends BaseFilterTest
 {
-  private static final String TIMESTAMP_COLUMN = "timestamp";
+  private static final List<InputRow> ROWS = ImmutableList.<InputRow>builder()
-
+      .addAll(DEFAULT_ROWS)
-  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
+      .add(makeDefaultSchemaRow("6", "-1000", ImmutableList.of("a"), null, 6.6, null, 10L))
-      new TimeAndDimsParseSpec(
+      .add(makeDefaultSchemaRow("7", "-10.012", ImmutableList.of("d"), null, null, 3.0f, null))
-          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
+      .build();
-          new DimensionsSpec(null, null, null)
-      )
-  );
-
-  private static final List<InputRow> ROWS = ImmutableList.of(
-      PARSER.parseBatch(ImmutableMap.of("dim0", "0", "dim1", "", "dim2", ImmutableList.of("a", "b"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.<String>of())).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "def", "dim2", ImmutableList.of("c"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "6", "dim1", "-1000", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "7", "dim1", "-10.012", "dim2", ImmutableList.of("d"))).get(0)
-  );
 
   public BoundFilterTest(
       String testName,
@@ -524,7 +502,16 @@ public class BoundFilterTest extends BaseFilterTest
     }
 
     assertFilterMatches(
-        new BoundDimFilter("dim1", "super-ab", "super-abd", true, true, false, superFn, StringComparators.LEXICOGRAPHIC),
+        new BoundDimFilter(
+            "dim1",
+            "super-ab",
+            "super-abd",
+            true,
+            true,
+            false,
+            superFn,
+            StringComparators.LEXICOGRAPHIC
+        ),
         ImmutableList.of("5")
     );
 
@@ -534,7 +521,16 @@ public class BoundFilterTest extends BaseFilterTest
     );
 
     assertFilterMatches(
-        new BoundDimFilter("dim2", "super-", "super-zzzzzz", false, false, false, superFn, StringComparators.LEXICOGRAPHIC),
+        new BoundDimFilter(
+            "dim2",
+            "super-",
+            "super-zzzzzz",
+            false,
+            false,
+            false,
+            superFn,
+            StringComparators.LEXICOGRAPHIC
+        ),
         ImmutableList.of("0", "1", "2", "3", "4", "5", "6", "7")
     );
 
@@ -603,12 +599,30 @@ public class BoundFilterTest extends BaseFilterTest
     }
 
     assertFilterMatches(
-        new BoundDimFilter("dim3", "super-null", "super-null", false, false, false, superFn, StringComparators.LEXICOGRAPHIC),
+        new BoundDimFilter(
+            "dim3",
+            "super-null",
+            "super-null",
+            false,
+            false,
+            false,
+            superFn,
+            StringComparators.LEXICOGRAPHIC
+        ),
         ImmutableList.of("0", "1", "2", "3", "4", "5", "6", "7")
     );
 
     assertFilterMatches(
-        new BoundDimFilter("dim4", "super-null", "super-null", false, false, false, superFn, StringComparators.LEXICOGRAPHIC),
+        new BoundDimFilter(
+            "dim4",
+            "super-null",
+            "super-null",
+            false,
+            false,
+            false,
+            superFn,
+            StringComparators.LEXICOGRAPHIC
+        ),
         ImmutableList.of("0", "1", "2", "3", "4", "5", "6", "7")
     );
 
@@ -617,4 +631,94 @@ public class BoundFilterTest extends BaseFilterTest
         ImmutableList.of("0", "1", "2", "3", "4", "5", "6", "7")
     );
   }
+
+  @Test
+  public void testNumericNullsAndZeros()
+  {
+    assertFilterMatches(
+        new BoundDimFilter(
+            "d0",
+            "0.0",
+            "1.0",
+            false,
+            false,
+            false,
+            null,
+            StringComparators.NUMERIC
+        ),
+        NullHandling.replaceWithDefault() ? ImmutableList.of("0", "2", "7") : ImmutableList.of("0")
+    );
+
+    assertFilterMatches(
+        new BoundDimFilter(
+            "f0",
+            "0.0",
+            "1.0",
+            false,
+            false,
+            false,
+            null,
+            StringComparators.NUMERIC
+        ),
+        NullHandling.replaceWithDefault() ? ImmutableList.of("0", "4", "6") : ImmutableList.of("0")
+    );
+
+    assertFilterMatches(
+        new BoundDimFilter(
+            "l0",
+            "0.0",
+            "1.0",
+            false,
+            false,
+            false,
+            null,
+            StringComparators.NUMERIC
+        ),
+        NullHandling.replaceWithDefault() ? ImmutableList.of("0", "3", "7") : ImmutableList.of("0")
+    );
+  }
+
+  @Test
+  public void testNumericNulls()
+  {
+    assertFilterMatches(
+        new BoundDimFilter(
+            "f0",
+            "1.0",
+            null,
+            false,
+            false,
+            false,
+            null,
+            StringComparators.NUMERIC
+        ),
+        ImmutableList.of("1", "2", "3", "5", "7")
+    );
+    assertFilterMatches(
+        new BoundDimFilter(
+            "d0",
+            "1",
+            null,
+            false,
+            false,
+            false,
+            null,
+            StringComparators.NUMERIC
+        ),
+        ImmutableList.of("1", "3", "4", "5", "6")
+    );
+    assertFilterMatches(
+        new BoundDimFilter(
+            "l0",
+            "1",
+            null,
+            false,
+            false,
+            false,
+            null,
+            StringComparators.NUMERIC
+        ),
+        ImmutableList.of("1", "2", "4", "5", "6")
+    );
+  }
 }

processing/src/test/java/org/apache/druid/segment/filter/FilterPartitionTest.java

@@ -22,15 +22,8 @@ package org.apache.druid.segment.filter;
 import com.google.common.base.Function;
 import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import org.apache.druid.common.config.NullHandling;
 import org.apache.druid.data.input.InputRow;
-import org.apache.druid.data.input.impl.DimensionsSpec;
-import org.apache.druid.data.input.impl.InputRowParser;
-import org.apache.druid.data.input.impl.MapInputRowParser;
-import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
-import org.apache.druid.data.input.impl.TimestampSpec;
-import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.js.JavaScriptConfig;
 import org.apache.druid.query.extraction.ExtractionFn;
@@ -59,7 +52,6 @@ import org.junit.runners.Parameterized;
 import java.io.Closeable;
 import java.util.Arrays;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 
 @RunWith(Parameterized.class)
@@ -159,34 +151,17 @@ public class FilterPartitionTest extends BaseFilterTest
     }
   }
 
-  private static String JS_FN = "function(str) { return 'super-' + str; }";
+  private static final String JS_FN = "function(str) { return 'super-' + str; }";
-  private static ExtractionFn JS_EXTRACTION_FN = new JavaScriptExtractionFn(JS_FN, false, JavaScriptConfig.getEnabledInstance());
+  private static final ExtractionFn JS_EXTRACTION_FN =
+      new JavaScriptExtractionFn(JS_FN, false, JavaScriptConfig.getEnabledInstance());
 
-  private static final String TIMESTAMP_COLUMN = "timestamp";
+  private static final List<InputRow> ROWS = ImmutableList.<InputRow>builder()
-
+      .addAll(DEFAULT_ROWS)
-  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
+      .add(makeDefaultSchemaRow("6", "B453B411", ImmutableList.of("c", "d", "e"), null, null, null, null))
-      new TimeAndDimsParseSpec(
+      .add(makeDefaultSchemaRow("7", "HELLO", ImmutableList.of("foo"), null, null, null, null))
-          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
+      .add(makeDefaultSchemaRow("8", "abc", ImmutableList.of("bar"), null, null, null, null))
-          new DimensionsSpec(
+      .add(makeDefaultSchemaRow("9", "1", ImmutableList.of("foo", "bar"), null, null, null, null))
-              DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3")),
+      .build();
-              null,
-              null
-          )
-      )
-  );
-
-  private static final List<InputRow> ROWS = ImmutableList.of(
-      PARSER.parseBatch(ImmutableMap.of("dim0", "0", "dim1", "", "dim2", ImmutableList.of("a", "b"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.of())).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "def", "dim2", ImmutableList.of("c"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "6", "dim1", "B453B411", "dim2", ImmutableList.of("c", "d", "e"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "7", "dim1", "HELLO", "dim2", ImmutableList.of("foo"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "8", "dim1", "abc", "dim2", ImmutableList.of("bar"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "9", "dim1", "1", "dim2", ImmutableList.of("foo", "bar"))).get(0)
-  );
 
   public FilterPartitionTest(
       String testName,
@@ -217,7 +192,7 @@ public class FilterPartitionTest extends BaseFilterTest
     assertFilterMatches(new SelectorDimFilter("dim1", "10", null), ImmutableList.of("1"));
     assertFilterMatches(new SelectorDimFilter("dim1", "2", null), ImmutableList.of("2"));
     assertFilterMatches(new SelectorDimFilter("dim1", "1", null), ImmutableList.of("3", "9"));
-    assertFilterMatches(new SelectorDimFilter("dim1", "def", null), ImmutableList.of("4"));
+    assertFilterMatches(new SelectorDimFilter("dim1", "abdef", null), ImmutableList.of("4"));
     assertFilterMatches(new SelectorDimFilter("dim1", "abc", null), ImmutableList.of("5", "8"));
     assertFilterMatches(new SelectorDimFilter("dim1", "ab", null), ImmutableList.of());
   }
@@ -234,7 +209,7 @@ public class FilterPartitionTest extends BaseFilterTest
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "10", null), ImmutableList.of("1"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "2", null), ImmutableList.of("2"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "1", null), ImmutableList.of("3", "9"));
-    assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "def", null), ImmutableList.of("4"));
+    assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "abdef", null), ImmutableList.of("4"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "abc", null), ImmutableList.of("5", "8"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "ab", null), ImmutableList.of());
 
@@ -246,7 +221,7 @@ public class FilterPartitionTest extends BaseFilterTest
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "super-10", JS_EXTRACTION_FN), ImmutableList.of("1"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "super-2", JS_EXTRACTION_FN), ImmutableList.of("2"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "super-1", JS_EXTRACTION_FN), ImmutableList.of("3", "9"));
-    assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "super-def", JS_EXTRACTION_FN), ImmutableList.of("4"));
+    assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "super-abdef", JS_EXTRACTION_FN), ImmutableList.of("4"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "super-abc", JS_EXTRACTION_FN), ImmutableList.of("5", "8"));
     assertFilterMatches(new NoBitmapSelectorDimFilter("dim1", "super-ab", JS_EXTRACTION_FN), ImmutableList.of());
   }
@@ -639,7 +614,7 @@ public class FilterPartitionTest extends BaseFilterTest
     DimFilter dimFilter1 = new OrDimFilter(Arrays.asList(
         new SelectorDimFilter("dim0", "6", null),
         new AndDimFilter(Arrays.asList(
-            new NoBitmapSelectorDimFilter("dim1", "def", null),
+            new NoBitmapSelectorDimFilter("dim1", "abdef", null),
             new SelectorDimFilter("dim2", "c", null)
         )
         ))
@@ -693,7 +668,7 @@ public class FilterPartitionTest extends BaseFilterTest
     DimFilter dimFilter1 = new OrDimFilter(Arrays.asList(
         new SelectorDimFilter("dim0", "super-6", JS_EXTRACTION_FN),
         new AndDimFilter(Arrays.asList(
-            new NoBitmapSelectorDimFilter("dim1", "super-def", JS_EXTRACTION_FN),
+            new NoBitmapSelectorDimFilter("dim1", "super-abdef", JS_EXTRACTION_FN),
             new SelectorDimFilter("dim2", "super-c", JS_EXTRACTION_FN)
         )
         ))

processing/src/test/java/org/apache/druid/segment/filter/JavaScriptFilterTest.java

@@ -23,13 +23,6 @@ import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.common.config.NullHandling;
-import org.apache.druid.data.input.InputRow;
-import org.apache.druid.data.input.impl.DimensionsSpec;
-import org.apache.druid.data.input.impl.InputRowParser;
-import org.apache.druid.data.input.impl.MapInputRowParser;
-import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
-import org.apache.druid.data.input.impl.TimestampSpec;
-import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.js.JavaScriptConfig;
 import org.apache.druid.query.extraction.ExtractionFn;
@@ -45,34 +38,11 @@ import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
 import java.io.Closeable;
-import java.util.List;
 import java.util.Map;
 
 @RunWith(Parameterized.class)
 public class JavaScriptFilterTest extends BaseFilterTest
 {
-  private static final String TIMESTAMP_COLUMN = "timestamp";
-
-  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
-      new TimeAndDimsParseSpec(
-          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
-          new DimensionsSpec(
-              DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3")),
-              null,
-              null
-          )
-      )
-  );
-
-  private static final List<InputRow> ROWS = ImmutableList.of(
-      PARSER.parseBatch(ImmutableMap.of("dim0", "0", "dim1", "", "dim2", ImmutableList.of("a", "b"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.of())).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "def", "dim2", ImmutableList.of("c"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0)
-  );
-
   public JavaScriptFilterTest(
       String testName,
       IndexBuilder indexBuilder,
@@ -81,7 +51,7 @@ public class JavaScriptFilterTest extends BaseFilterTest
       boolean optimize
   )
   {
-    super(testName, ROWS, indexBuilder, finisher, cnf, optimize);
+    super(testName, DEFAULT_ROWS, indexBuilder, finisher, cnf, optimize);
   }
 
   @AfterClass
@@ -90,11 +60,17 @@ public class JavaScriptFilterTest extends BaseFilterTest
     BaseFilterTest.tearDown(JavaScriptFilterTest.class.getName());
   }
 
-  private final String jsNullFilter = "function(x) { return(x === null) }";
+  private final String jsNullFilter = "function(x) { return x === null }";
 
   private String jsValueFilter(String value)
   {
-    String jsFn = "function(x) { return(x === '" + value + "') }";
+    String jsFn = "function(x) { return x === '" + value + "' }";
+    return jsFn;
+  }
+
+  private String jsNumericValueFilter(String value)
+  {
+    String jsFn = "function(x) { return x === " + value + " }";
     return jsFn;
   }
 
@@ -119,7 +95,7 @@ public class JavaScriptFilterTest extends BaseFilterTest
     assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("dim1", jsValueFilter("10"), null), ImmutableList.of("1"));
     assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("dim1", jsValueFilter("2"), null), ImmutableList.of("2"));
     assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("dim1", jsValueFilter("1"), null), ImmutableList.of("3"));
-    assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("dim1", jsValueFilter("def"), null), ImmutableList.of("4"));
+    assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("dim1", jsValueFilter("abdef"), null), ImmutableList.of("4"));
     assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("dim1", jsValueFilter("abc"), null), ImmutableList.of("5"));
     assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("dim1", jsValueFilter("ab"), null), ImmutableList.of());
   }
@@ -176,7 +152,7 @@ public class JavaScriptFilterTest extends BaseFilterTest
     final Map<String, String> stringMap = ImmutableMap.of(
         "1", "HELLO",
         "a", "HELLO",
-        "def", "HELLO",
+        "abdef", "HELLO",
         "abc", "UNKNOWN"
     );
     LookupExtractor mapExtractor = new MapLookupExtractor(stringMap, false);
@@ -228,6 +204,23 @@ public class JavaScriptFilterTest extends BaseFilterTest
     );
   }
 
+  @Test
+  public void testNumericNull()
+  {
+    if (NullHandling.replaceWithDefault()) {
+      assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("f0", jsNullFilter, null), ImmutableList.of());
+      assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("d0", jsNullFilter, null), ImmutableList.of());
+      assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("l0", jsNullFilter, null), ImmutableList.of());
+    } else {
+      assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("f0", jsNullFilter, null), ImmutableList.of("4"));
+      assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("d0", jsNullFilter, null), ImmutableList.of("2"));
+      assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("l0", jsNullFilter, null), ImmutableList.of("3"));
+    }
+    assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("f0", jsNumericValueFilter("5.5"), null), ImmutableList.of("2"));
+    assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("d0", jsNumericValueFilter("120.0245"), null), ImmutableList.of("3"));
+    assertFilterMatchesSkipVectorize(newJavaScriptDimFilter("l0", jsNumericValueFilter("9001"), null), ImmutableList.of("4"));
+  }
+
   private JavaScriptDimFilter newJavaScriptDimFilter(
       final String dimension,
       final String function,

processing/src/test/java/org/apache/druid/segment/filter/RegexFilterTest.java

@@ -21,15 +21,7 @@ package org.apache.druid.segment.filter;
 
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import org.apache.druid.common.config.NullHandling;
-import org.apache.druid.data.input.InputRow;
-import org.apache.druid.data.input.impl.DimensionsSpec;
-import org.apache.druid.data.input.impl.InputRowParser;
-import org.apache.druid.data.input.impl.MapInputRowParser;
-import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
-import org.apache.druid.data.input.impl.TimestampSpec;
-import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.js.JavaScriptConfig;
 import org.apache.druid.query.extraction.ExtractionFn;
@@ -43,34 +35,10 @@ import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
 import java.io.Closeable;
-import java.util.List;
-import java.util.Map;
 
 @RunWith(Parameterized.class)
 public class RegexFilterTest extends BaseFilterTest
 {
-  private static final String TIMESTAMP_COLUMN = "timestamp";
-
-  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
-      new TimeAndDimsParseSpec(
-          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
-          new DimensionsSpec(
-              DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3")),
-              null,
-              null
-          )
-      )
-  );
-
-  private static final List<InputRow> ROWS = ImmutableList.of(
-      PARSER.parseBatch(ImmutableMap.of("dim0", "0", "dim1", "", "dim2", ImmutableList.of("a", "b"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.of())).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "abdef", "dim2", ImmutableList.of("c"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0)
-  );
-
   public RegexFilterTest(
       String testName,
       IndexBuilder indexBuilder,
@@ -79,7 +47,7 @@ public class RegexFilterTest extends BaseFilterTest
       boolean optimize
   )
   {
-    super(testName, ROWS, indexBuilder, finisher, cnf, optimize);
+    super(testName, DEFAULT_ROWS, indexBuilder, finisher, cnf, optimize);
   }
 
   @AfterClass

processing/src/test/java/org/apache/druid/segment/filter/SearchQueryFilterTest.java

@@ -21,15 +21,7 @@ package org.apache.druid.segment.filter;
 
 import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import org.apache.druid.common.config.NullHandling;
-import org.apache.druid.data.input.InputRow;
-import org.apache.druid.data.input.impl.DimensionsSpec;
-import org.apache.druid.data.input.impl.InputRowParser;
-import org.apache.druid.data.input.impl.MapInputRowParser;
-import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
-import org.apache.druid.data.input.impl.TimestampSpec;
-import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.js.JavaScriptConfig;
 import org.apache.druid.query.extraction.ExtractionFn;
@@ -45,34 +37,10 @@ import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
 import java.io.Closeable;
-import java.util.List;
-import java.util.Map;
 
 @RunWith(Parameterized.class)
 public class SearchQueryFilterTest extends BaseFilterTest
 {
-  private static final String TIMESTAMP_COLUMN = "timestamp";
-
-  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
-      new TimeAndDimsParseSpec(
-          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
-          new DimensionsSpec(
-              DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3")),
-              null,
-              null
-          )
-      )
-  );
-
-  private static final List<InputRow> ROWS = ImmutableList.of(
-      PARSER.parseBatch(ImmutableMap.of("dim0", "0", "dim1", "", "dim2", ImmutableList.of("a", "b"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.of())).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "abdef", "dim2", ImmutableList.of("c"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0)
-  );
-
   public SearchQueryFilterTest(
       String testName,
       IndexBuilder indexBuilder,
@@ -81,7 +49,7 @@ public class SearchQueryFilterTest extends BaseFilterTest
       boolean optimize
   )
   {
-    super(testName, ROWS, indexBuilder, finisher, cnf, optimize);
+    super(testName, DEFAULT_ROWS, indexBuilder, finisher, cnf, optimize);
   }
 
   @AfterClass

processing/src/test/java/org/apache/druid/segment/filter/SelectorFilterTest.java

@@ -23,13 +23,6 @@ import com.google.common.base.Function;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.common.config.NullHandling;
-import org.apache.druid.data.input.InputRow;
-import org.apache.druid.data.input.impl.DimensionsSpec;
-import org.apache.druid.data.input.impl.InputRowParser;
-import org.apache.druid.data.input.impl.MapInputRowParser;
-import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
-import org.apache.druid.data.input.impl.TimestampSpec;
-import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.query.extraction.MapLookupExtractor;
 import org.apache.druid.query.extraction.TimeDimExtractionFn;
@@ -40,7 +33,6 @@ import org.apache.druid.query.lookup.LookupExtractionFn;
 import org.apache.druid.query.lookup.LookupExtractor;
 import org.apache.druid.segment.IndexBuilder;
 import org.apache.druid.segment.StorageAdapter;
-import org.apache.druid.segment.incremental.IncrementalIndexSchema;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Test;
@@ -49,45 +41,11 @@ import org.junit.runners.Parameterized;
 
 import java.io.Closeable;
 import java.util.Arrays;
-import java.util.List;
 import java.util.Map;
 
 @RunWith(Parameterized.class)
 public class SelectorFilterTest extends BaseFilterTest
 {
-  private static final String TIMESTAMP_COLUMN = "timestamp";
-
-  private static final InputRowParser<Map<String, Object>> PARSER = new MapInputRowParser(
-      new TimeAndDimsParseSpec(
-          new TimestampSpec(TIMESTAMP_COLUMN, "iso", DateTimes.of("2000")),
-          new DimensionsSpec(
-              DimensionsSpec.getDefaultSchemas(ImmutableList.of("dim0", "dim1", "dim2", "dim3", "dim6")),
-              null,
-              null
-          )
-      )
-  );
-
-  private static final List<InputRow> ROWS = ImmutableList.of(
-      PARSER.parseBatch(ImmutableMap.of(
-          "dim0",
-          "0",
-          "dim1",
-          "",
-          "dim2",
-          ImmutableList.of("a", "b"),
-          "dim6",
-          "2017-07-25"
-      )).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "1", "dim1", "10", "dim2", ImmutableList.of(), "dim6", "2017-07-25"))
-            .get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "2", "dim1", "2", "dim2", ImmutableList.of(""), "dim6", "2017-05-25"))
-            .get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "3", "dim1", "1", "dim2", ImmutableList.of("a"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "4", "dim1", "def", "dim2", ImmutableList.of("c"))).get(0),
-      PARSER.parseBatch(ImmutableMap.of("dim0", "5", "dim1", "abc")).get(0)
-  );
-
   public SelectorFilterTest(
       String testName,
       IndexBuilder indexBuilder,
@@ -96,17 +54,7 @@ public class SelectorFilterTest extends BaseFilterTest
       boolean optimize
   )
   {
-    super(
+    super(testName, DEFAULT_ROWS, indexBuilder, finisher, cnf, optimize);
-        testName,
-        ROWS,
-        indexBuilder.schema(
-            new IncrementalIndexSchema.Builder()
-                .withDimensionsSpec(PARSER.getParseSpec().getDimensionsSpec()).build()
-        ),
-        finisher,
-        cnf,
-        optimize
-    );
   }
 
   @AfterClass
@@ -123,19 +71,25 @@ public class SelectorFilterTest extends BaseFilterTest
         ImmutableList.of()
     );
     assertFilterMatches(
-        new SelectorDimFilter("dim6", null, new TimeDimExtractionFn("yyyy-MM-dd", "yyyy-MM", true)),
+        new SelectorDimFilter("timeDim", null, new TimeDimExtractionFn("yyyy-MM-dd", "yyyy-MM", true)),
-        ImmutableList.of("3", "4", "5")
+        ImmutableList.of("4")
     );
     assertFilterMatches(new SelectorDimFilter(
-        "dim6",
+        "timeDim",
         "2017-07",
         new TimeDimExtractionFn("yyyy-MM-dd", "yyyy-MM", true)
     ), ImmutableList.of("0", "1"));
     assertFilterMatches(new SelectorDimFilter(
-        "dim6",
+        "timeDim",
         "2017-05",
         new TimeDimExtractionFn("yyyy-MM-dd", "yyyy-MM", true)
     ), ImmutableList.of("2"));
+
+    assertFilterMatches(new SelectorDimFilter(
+        "timeDim",
+        "2020-01",
+        new TimeDimExtractionFn("yyyy-MM-dd", "yyyy-MM", true)
+    ), ImmutableList.of("3", "5"));
   }
 
   @Test
@@ -160,7 +114,7 @@ public class SelectorFilterTest extends BaseFilterTest
     assertFilterMatches(new SelectorDimFilter("dim1", "10", null), ImmutableList.of("1"));
     assertFilterMatches(new SelectorDimFilter("dim1", "2", null), ImmutableList.of("2"));
     assertFilterMatches(new SelectorDimFilter("dim1", "1", null), ImmutableList.of("3"));
-    assertFilterMatches(new SelectorDimFilter("dim1", "def", null), ImmutableList.of("4"));
+    assertFilterMatches(new SelectorDimFilter("dim1", "abdef", null), ImmutableList.of("4"));
     assertFilterMatches(new SelectorDimFilter("dim1", "abc", null), ImmutableList.of("5"));
     assertFilterMatches(new SelectorDimFilter("dim1", "ab", null), ImmutableList.of());
   }
@@ -225,7 +179,7 @@ public class SelectorFilterTest extends BaseFilterTest
     final Map<String, String> stringMap = ImmutableMap.of(
         "1", "HELLO",
         "a", "HELLO",
-        "def", "HELLO",
+        "abdef", "HELLO",
         "abc", "UNKNOWN"
     );
     LookupExtractor mapExtractor = new MapLookupExtractor(stringMap, false);
@@ -354,4 +308,24 @@ public class SelectorFilterTest extends BaseFilterTest
       );
     }
   }
+
+  @Test
+  public void testNumericColumnNullsAndDefaults()
+  {
+    if (NullHandling.replaceWithDefault()) {
+      assertFilterMatches(new SelectorDimFilter("f0", "0", null), ImmutableList.of("0", "4"));
+      assertFilterMatches(new SelectorDimFilter("d0", "0", null), ImmutableList.of("0", "2"));
+      assertFilterMatches(new SelectorDimFilter("l0", "0", null), ImmutableList.of("0", "3"));
+      assertFilterMatches(new SelectorDimFilter("f0", null, null), ImmutableList.of());
+      assertFilterMatches(new SelectorDimFilter("d0", null, null), ImmutableList.of());
+      assertFilterMatches(new SelectorDimFilter("l0", null, null), ImmutableList.of());
+    } else {
+      assertFilterMatches(new SelectorDimFilter("f0", "0", null), ImmutableList.of("0"));
+      assertFilterMatches(new SelectorDimFilter("d0", "0", null), ImmutableList.of("0"));
+      assertFilterMatches(new SelectorDimFilter("l0", "0", null), ImmutableList.of("0"));
+      assertFilterMatches(new SelectorDimFilter("f0", null, null), ImmutableList.of("4"));
+      assertFilterMatches(new SelectorDimFilter("d0", null, null), ImmutableList.of("2"));
+      assertFilterMatches(new SelectorDimFilter("l0", null, null), ImmutableList.of("3"));
+    }
+  }
 }

processing/src/test/java/org/apache/druid/segment/vector/VectorSelectorUtilsTest.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.segment.vector;
 
+import com.google.common.collect.Sets;
 import org.apache.druid.collections.IntSetTestUtility;
 import org.apache.druid.collections.bitmap.ImmutableBitmap;
 import org.apache.druid.collections.bitmap.MutableBitmap;
@@ -31,64 +32,106 @@ import org.junit.Assert;
 import org.junit.Test;
 import org.roaringbitmap.PeekableIntIterator;
 
+import java.util.ArrayList;
 import java.util.Set;
 
 public class VectorSelectorUtilsTest
 {
+  private static final Set<Integer> NULLS = IntSetTestUtility.getSetBits();
+  private static final Set<Integer> NULLS_PATTERN = alternatngPattern(10, 12);
+
   @Test
   public void testBitSetNullVector()
   {
     final WrappedBitSetBitmap bitmap = new WrappedBitSetBitmap();
-    populate(bitmap);
+    populate(bitmap, NULLS);
-    assertNullVector(bitmap);
+    assertNullVector(bitmap, NULLS);
+
+    final WrappedBitSetBitmap bitmap2 = new WrappedBitSetBitmap();
+    populate(bitmap2, NULLS_PATTERN);
+    assertNullVector(bitmap2, NULLS_PATTERN);
   }
 
   @Test
   public void testConciseMutableNullVector()
   {
     final WrappedConciseBitmap bitmap = new WrappedConciseBitmap();
-    populate(bitmap);
+    populate(bitmap, NULLS);
-    assertNullVector(bitmap);
+    assertNullVector(bitmap, NULLS);
+
+    final WrappedConciseBitmap bitmap2 = new WrappedConciseBitmap();
+    populate(bitmap2, NULLS_PATTERN);
+    assertNullVector(bitmap2, NULLS_PATTERN);
   }
 
   @Test
   public void testConciseImmutableNullVector()
   {
     final WrappedConciseBitmap bitmap = new WrappedConciseBitmap();
-    populate(bitmap);
+    populate(bitmap, NULLS);
     final ImmutableBitmap immutable = new WrappedImmutableConciseBitmap(
         ImmutableConciseSet.newImmutableFromMutable(bitmap.getBitmap())
     );
-    assertNullVector(immutable);
+    assertNullVector(immutable, NULLS);
+
+    final WrappedConciseBitmap bitmap2 = new WrappedConciseBitmap();
+    populate(bitmap2, NULLS_PATTERN);
+    final ImmutableBitmap immutable2 = new WrappedImmutableConciseBitmap(
+        ImmutableConciseSet.newImmutableFromMutable(bitmap2.getBitmap())
+    );
+    assertNullVector(immutable2, NULLS_PATTERN);
   }
 
   @Test
   public void testRoaringMutableNullVector()
   {
     WrappedRoaringBitmap bitmap = new WrappedRoaringBitmap();
-    populate(bitmap);
+    populate(bitmap, NULLS);
-    assertNullVector(bitmap);
+    assertNullVector(bitmap, NULLS);
+
+    WrappedRoaringBitmap bitmap2 = new WrappedRoaringBitmap();
+    populate(bitmap2, NULLS_PATTERN);
+    assertNullVector(bitmap2, NULLS_PATTERN);
   }
 
   @Test
   public void testRoaringImmutableNullVector()
   {
     WrappedRoaringBitmap bitmap = new WrappedRoaringBitmap();
-    populate(bitmap);
+    populate(bitmap, NULLS);
-    assertNullVector(bitmap.toImmutableBitmap());
+    assertNullVector(bitmap.toImmutableBitmap(), NULLS);
+
+    WrappedRoaringBitmap bitmap2 = new WrappedRoaringBitmap();
+    populate(bitmap2, NULLS_PATTERN);
+    assertNullVector(bitmap2.toImmutableBitmap(), NULLS_PATTERN);
   }
 
-  public static void populate(MutableBitmap bitmap)
+  public static void populate(MutableBitmap bitmap, Set<Integer> setBits)
   {
-    for (int i : IntSetTestUtility.getSetBits()) {
+    for (int i : setBits) {
       bitmap.add(i);
     }
   }
 
-  private void assertNullVector(ImmutableBitmap bitmap)
+  private static Set<Integer> alternatngPattern(int smallSize, int rowCount)
   {
+    ArrayList<Integer> bits = new ArrayList<>();
+    boolean flipped = true;
+    for (int i = 0; i < rowCount; i++) {
+      if (i > 0 && i % smallSize == 0) {
+        flipped = !flipped;
+      }
+      if (flipped) {
+        bits.add(i);
+      }
+    }
+    return Sets.newTreeSet(bits);
+  }
+
+  private void assertNullVector(ImmutableBitmap bitmap, Set<Integer> nulls)
+  {
+    // test entire set in one vector
     PeekableIntIterator iterator = bitmap.peekableIterator();
-    Set<Integer> nulls = IntSetTestUtility.getSetBits();
     final int vectorSize = 32;
     final boolean[] nullVector = new boolean[vectorSize];
     ReadableVectorOffset someOffset = new NoFilterVectorOffset(vectorSize, 0, vectorSize);
@@ -98,6 +141,7 @@ public class VectorSelectorUtilsTest
       Assert.assertEquals(nulls.contains(i), nullVector[i]);
     }
 
+    // test entire set split into 4 chunks with smaller vectors
     iterator = bitmap.peekableIterator();
     final int smallerVectorSize = 8;
     boolean[] smallVector = null;
@@ -111,11 +155,11 @@ public class VectorSelectorUtilsTest
           Assert.assertEquals(nulls.contains(offset + i), smallVector[i]);
         }
       }
-      smallVector = null;
     }
 
+    // a magical vector perfectly sized to the number of nulls with a bitmap vector offset of just the nulls
     iterator = bitmap.peekableIterator();
-    ReadableVectorOffset allTheNulls = new BitmapVectorOffset(8, bitmap, 0, 22);
+    ReadableVectorOffset allTheNulls = new BitmapVectorOffset(nulls.size(), bitmap, 0, 32);
     smallVector = VectorSelectorUtils.populateNullVector(smallVector, allTheNulls, iterator);
     for (int i = 0; i < nulls.size(); i++) {
       Assert.assertTrue(smallVector[i]);

sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java

@@ -3198,6 +3198,69 @@ public class CalciteQueryTest extends BaseCalciteQueryTest
     );
   }
 
+  @Test
+  public void testLongPredicateFilterNulls() throws Exception
+  {
+    testQuery(
+        "SELECT COUNT(*)\n"
+        + "FROM druid.numfoo\n"
+        + "WHERE l1 > 3",
+        ImmutableList.of(
+            Druids.newTimeseriesQueryBuilder()
+                  .dataSource(CalciteTests.DATASOURCE3)
+                  .intervals(querySegmentSpec(Filtration.eternity()))
+                  .granularity(Granularities.ALL)
+                  .filters(bound("l1", "3", null, true, false, null, StringComparators.NUMERIC))
+                  .aggregators(aggregators(new CountAggregatorFactory("a0")))
+                  .context(TIMESERIES_CONTEXT_DEFAULT)
+                  .build()
+        ),
+        ImmutableList.of(new Object[]{2L})
+    );
+  }
+
+  @Test
+  public void testDoublePredicateFilterNulls() throws Exception
+  {
+    testQuery(
+        "SELECT COUNT(*)\n"
+        + "FROM druid.numfoo\n"
+        + "WHERE d1 > 0",
+        ImmutableList.of(
+            Druids.newTimeseriesQueryBuilder()
+                  .dataSource(CalciteTests.DATASOURCE3)
+                  .intervals(querySegmentSpec(Filtration.eternity()))
+                  .granularity(Granularities.ALL)
+                  .filters(bound("d1", "0", null, true, false, null, StringComparators.NUMERIC))
+                  .aggregators(aggregators(new CountAggregatorFactory("a0")))
+                  .context(TIMESERIES_CONTEXT_DEFAULT)
+                  .build()
+        ),
+        ImmutableList.of(new Object[]{2L})
+    );
+  }
+
+  @Test
+  public void testFloatPredicateFilterNulls() throws Exception
+  {
+    testQuery(
+        "SELECT COUNT(*)\n"
+        + "FROM druid.numfoo\n"
+        + "WHERE f1 > 0",
+        ImmutableList.of(
+            Druids.newTimeseriesQueryBuilder()
+                  .dataSource(CalciteTests.DATASOURCE3)
+                  .intervals(querySegmentSpec(Filtration.eternity()))
+                  .granularity(Granularities.ALL)
+                  .filters(bound("f1", "0", null, true, false, null, StringComparators.NUMERIC))
+                  .aggregators(aggregators(new CountAggregatorFactory("a0")))
+                  .context(TIMESERIES_CONTEXT_DEFAULT)
+                  .build()
+        ),
+        ImmutableList.of(new Object[]{2L})
+    );
+  }
+
   @Test
   public void testEmptyStringEquality() throws Exception
   {