fix bug with expressions on sparse string realtime columns without explicit null valued rows

benchmarks/src/test/java/org/apache/druid/benchmark/indexing/StringDimensionIndexerBenchmark.java

@@ -59,7 +59,7 @@ public class StringDimensionIndexerBenchmark
   @Setup
   public void setup()
   {
-    indexer = new StringDimensionIndexer(DimensionSchema.MultiValueHandling.ofDefault(), true);
+    indexer = new StringDimensionIndexer(DimensionSchema.MultiValueHandling.ofDefault(), true, false);
 
     for (int i = 0; i < cardinality; i++) {
       indexer.processRowValsToUnsortedEncodedKeyComponent("abcd-" + i, true);

indexing-hadoop/src/main/java/org/apache/druid/indexer/IndexGeneratorJob.java

@@ -47,7 +47,7 @@
 import org.apache.druid.segment.BaseProgressIndicator;
 import org.apache.druid.segment.ProgressIndicator;
 import org.apache.druid.segment.QueryableIndex;
-import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ColumnCapabilities;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexSchema;
 import org.apache.druid.segment.indexing.TuningConfigs;
@@ -289,7 +289,7 @@ private static IncrementalIndex makeIncrementalIndex(
       AggregatorFactory[] aggs,
       HadoopDruidIndexerConfig config,
       Iterable<String> oldDimOrder,
-      Map<String, ColumnCapabilitiesImpl> oldCapabilities
+      Map<String, ColumnCapabilities> oldCapabilities
   )
   {
     final HadoopTuningConfig tuningConfig = config.getSchema().getTuningConfig();

processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/GroupByQueryEngineV2.java

@@ -340,7 +340,7 @@ public static boolean isAllSingleValueDims(
 
               // Now check column capabilities.
               final ColumnCapabilities columnCapabilities = capabilitiesFunction.apply(dimension.getDimension());
-              return (columnCapabilities != null && !columnCapabilities.hasMultipleValues().isMaybeTrue()) ||
+              return (columnCapabilities != null && columnCapabilities.hasMultipleValues().isFalse()) ||
                      (missingMeansNonExistent && columnCapabilities == null);
             });
   }

processing/src/main/java/org/apache/druid/query/metadata/SegmentAnalyzer.java

@@ -227,7 +227,7 @@ private ColumnAnalysis analyzeStringColumn(
         min = NullHandling.nullToEmptyIfNeeded(bitmapIndex.getValue(0));
         max = NullHandling.nullToEmptyIfNeeded(bitmapIndex.getValue(cardinality - 1));
       }
-    } else if (capabilities.isDictionaryEncoded()) {
+    } else if (capabilities.isDictionaryEncoded().isTrue()) {
       // fallback if no bitmap index
       DictionaryEncodedColumn<String> theColumn = (DictionaryEncodedColumn<String>) columnHolder.getColumn();
       cardinality = theColumn.getCardinality();

processing/src/main/java/org/apache/druid/query/topn/TopNQueryEngine.java

@@ -194,7 +194,7 @@ private static boolean canUsePooledAlgorithm(
     }
     if (capabilities != null && capabilities.getType() == ValueType.STRING) {
       // string columns must use the on heap algorithm unless they have the following capabilites
-      return capabilities.isDictionaryEncoded() && capabilities.areDictionaryValuesUnique().isTrue();
+      return capabilities.isDictionaryEncoded().isTrue() && capabilities.areDictionaryValuesUnique().isTrue();
     } else {
       // non-strings are not eligible to use the pooled algorithm, and should use a heap algorithm
       return false;

processing/src/main/java/org/apache/druid/segment/DimensionHandlerUtils.java

@@ -73,16 +73,16 @@ private DimensionHandlerUtils()
   )
   {
     if (capabilities == null) {
-      return new StringDimensionHandler(dimensionName, multiValueHandling, true);
+      return new StringDimensionHandler(dimensionName, multiValueHandling, true, false);
     }
 
     multiValueHandling = multiValueHandling == null ? MultiValueHandling.ofDefault() : multiValueHandling;
 
     if (capabilities.getType() == ValueType.STRING) {
-      if (!capabilities.isDictionaryEncoded()) {
+      if (!capabilities.isDictionaryEncoded().isTrue()) {
         throw new IAE("String column must have dictionary encoding.");
       }
-      return new StringDimensionHandler(dimensionName, multiValueHandling, capabilities.hasBitmapIndexes());
+      return new StringDimensionHandler(dimensionName, multiValueHandling, capabilities.hasBitmapIndexes(), capabilities.hasSpatialIndexes());
     }
 
     if (capabilities.getType() == ValueType.LONG) {
@@ -98,7 +98,7 @@ private DimensionHandlerUtils()
     }
 
     // Return a StringDimensionHandler by default (null columns will be treated as String typed)
-    return new StringDimensionHandler(dimensionName, multiValueHandling, true);
+    return new StringDimensionHandler(dimensionName, multiValueHandling, true, false);
   }
 
   public static List<ValueType> getValueTypesFromDimensionSpecs(List<DimensionSpec> dimSpecs)
@@ -226,11 +226,11 @@ private static ColumnCapabilities getEffectiveCapabilities(
       capabilities = ColumnCapabilitiesImpl.copyOf(capabilities)
                                            .setType(ValueType.STRING)
                                            .setDictionaryValuesUnique(
-                                               capabilities.isDictionaryEncoded() &&
+                                               capabilities.isDictionaryEncoded().isTrue() &&
                                                fn.getExtractionType() == ExtractionFn.ExtractionType.ONE_TO_ONE
                                            )
                                            .setDictionaryValuesSorted(
-                                               capabilities.isDictionaryEncoded() && fn.preservesOrdering()
+                                               capabilities.isDictionaryEncoded().isTrue() && fn.preservesOrdering()
                                            );
     }
 

processing/src/main/java/org/apache/druid/segment/DimensionIndexer.java

@@ -22,6 +22,7 @@
 import org.apache.druid.collections.bitmap.BitmapFactory;
 import org.apache.druid.collections.bitmap.MutableBitmap;
 import org.apache.druid.query.dimension.DimensionSpec;
+import org.apache.druid.segment.column.ColumnCapabilities;
 import org.apache.druid.segment.data.CloseableIndexed;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexRowHolder;
@@ -236,6 +237,7 @@ ColumnValueSelector<?> makeColumnValueSelector(
       IncrementalIndex.DimensionDesc desc
   );
 
+  ColumnCapabilities getColumnCapabilities();
   /**
    * Compares the row values for this DimensionIndexer's dimension from a Row key.
    *

processing/src/main/java/org/apache/druid/segment/DoubleDimensionIndexer.java

@@ -25,6 +25,9 @@
 import org.apache.druid.java.util.common.guava.Comparators;
 import org.apache.druid.query.dimension.DimensionSpec;
 import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.segment.data.CloseableIndexed;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexRowHolder;
@@ -38,6 +41,9 @@ public class DoubleDimensionIndexer implements DimensionIndexer<Double, Double,
 {
   public static final Comparator<Double> DOUBLE_COMPARATOR = Comparators.naturalNullsFirst();
 
+  private final ColumnCapabilitiesImpl capabilities =
+      ColumnCapabilitiesImpl.createSimpleNumericColumnCapabilities(ValueType.DOUBLE);
+
   @Override
   public Double processRowValsToUnsortedEncodedKeyComponent(@Nullable Object dimValues, boolean reportParseExceptions)
   {
@@ -89,6 +95,12 @@ public int getCardinality()
     return DimensionDictionarySelector.CARDINALITY_UNKNOWN;
   }
 
+  @Override
+  public ColumnCapabilities getColumnCapabilities()
+  {
+    return capabilities;
+  }
+
   @Override
   public DimensionSelector makeDimensionSelector(
       DimensionSpec spec,

processing/src/main/java/org/apache/druid/segment/FloatDimensionIndexer.java

@@ -25,6 +25,9 @@
 import org.apache.druid.java.util.common.guava.Comparators;
 import org.apache.druid.query.dimension.DimensionSpec;
 import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.segment.data.CloseableIndexed;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexRowHolder;
@@ -38,6 +41,9 @@ public class FloatDimensionIndexer implements DimensionIndexer<Float, Float, Flo
 {
   public static final Comparator<Float> FLOAT_COMPARATOR = Comparators.naturalNullsFirst();
 
+  private final ColumnCapabilitiesImpl capabilities =
+      ColumnCapabilitiesImpl.createSimpleNumericColumnCapabilities(ValueType.FLOAT);
+
   @Override
   public Float processRowValsToUnsortedEncodedKeyComponent(@Nullable Object dimValues, boolean reportParseExceptions)
   {
@@ -90,6 +96,12 @@ public int getCardinality()
     return DimensionDictionarySelector.CARDINALITY_UNKNOWN;
   }
 
+  @Override
+  public ColumnCapabilities getColumnCapabilities()
+  {
+    return capabilities;
+  }
+
   @Override
   public DimensionSelector makeDimensionSelector(
       DimensionSpec spec,

processing/src/main/java/org/apache/druid/segment/IndexMergerV9.java

@@ -88,6 +88,63 @@ public class IndexMergerV9 implements IndexMerger
 {
   private static final Logger log = new Logger(IndexMergerV9.class);
 
+  // merge logic for the state capabilities will be in after incremental index is persisted
+  public static final ColumnCapabilities.CoercionLogic DIMENSION_CAPABILITY_MERGE_LOGIC =
+      new ColumnCapabilities.CoercionLogic()
+      {
+        @Override
+        public boolean dictionaryEncoded()
+        {
+          return true;
+        }
+
+        @Override
+        public boolean dictionaryValuesSorted()
+        {
+          return true;
+        }
+
+        @Override
+        public boolean dictionaryValuesUnique()
+        {
+          return true;
+        }
+
+        @Override
+        public boolean multipleValues()
+        {
+          return false;
+        }
+      };
+
+  public static final ColumnCapabilities.CoercionLogic METRIC_CAPABILITY_MERGE_LOGIC =
+      new ColumnCapabilities.CoercionLogic()
+      {
+        @Override
+        public boolean dictionaryEncoded()
+        {
+          return false;
+        }
+
+        @Override
+        public boolean dictionaryValuesSorted()
+        {
+          return false;
+        }
+
+        @Override
+        public boolean dictionaryValuesUnique()
+        {
+          return false;
+        }
+
+        @Override
+        public boolean multipleValues()
+        {
+          return false;
+        }
+      };
+
   private final ObjectMapper mapper;
   private final IndexIO indexIO;
   private final SegmentWriteOutMediumFactory defaultSegmentWriteOutMediumFactory;
@@ -724,14 +781,14 @@ private void mergeCapabilities(
       for (String dimension : adapter.getDimensionNames()) {
         ColumnCapabilities capabilities = adapter.getCapabilities(dimension);
         capabilitiesMap.compute(dimension, (d, existingCapabilities) ->
-            ColumnCapabilitiesImpl.snapshot(capabilities)
-                                  .merge(ColumnCapabilitiesImpl.snapshot(existingCapabilities)));
+            ColumnCapabilitiesImpl.merge(capabilities, existingCapabilities, DIMENSION_CAPABILITY_MERGE_LOGIC)
+        );
       }
       for (String metric : adapter.getMetricNames()) {
         ColumnCapabilities capabilities = adapter.getCapabilities(metric);
         capabilitiesMap.compute(metric, (m, existingCapabilities) ->
-            ColumnCapabilitiesImpl.snapshot(capabilities)
-                                  .merge(ColumnCapabilitiesImpl.snapshot(existingCapabilities)));
+            ColumnCapabilitiesImpl.merge(capabilities, existingCapabilities, METRIC_CAPABILITY_MERGE_LOGIC)
+        );
         metricsValueTypes.put(metric, capabilities.getType());
         metricTypeNames.put(metric, adapter.getMetricType(metric));
       }
@@ -1011,7 +1068,10 @@ private Map<String, DimensionHandler> makeDimensionHandlers(
   {
     Map<String, DimensionHandler> handlers = new LinkedHashMap<>();
     for (int i = 0; i < mergedDimensions.size(); i++) {
-      ColumnCapabilities capabilities = dimCapabilities.get(i);
+      ColumnCapabilities capabilities = ColumnCapabilitiesImpl.snapshot(
+          dimCapabilities.get(i),
+          DIMENSION_CAPABILITY_MERGE_LOGIC
+      );
       String dimName = mergedDimensions.get(i);
       DimensionHandler handler = DimensionHandlerUtils.getHandlerFromCapabilities(dimName, capabilities, null);
       handlers.put(dimName, handler);

processing/src/main/java/org/apache/druid/segment/LongDimensionIndexer.java

@@ -25,6 +25,9 @@
 import org.apache.druid.java.util.common.guava.Comparators;
 import org.apache.druid.query.dimension.DimensionSpec;
 import org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.segment.data.CloseableIndexed;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexRowHolder;
@@ -38,6 +41,10 @@ public class LongDimensionIndexer implements DimensionIndexer<Long, Long, Long>
 {
   public static final Comparator LONG_COMPARATOR = Comparators.<Long>naturalNullsFirst();
 
+  private final ColumnCapabilitiesImpl capabilities =
+      ColumnCapabilitiesImpl.createSimpleNumericColumnCapabilities(ValueType.LONG);
+
+
   @Override
   public Long processRowValsToUnsortedEncodedKeyComponent(@Nullable Object dimValues, boolean reportParseExceptions)
   {
@@ -90,6 +97,12 @@ public int getCardinality()
     return DimensionDictionarySelector.CARDINALITY_UNKNOWN;
   }
 
+  @Override
+  public ColumnCapabilities getColumnCapabilities()
+  {
+    return capabilities;
+  }
+
   @Override
   public DimensionSelector makeDimensionSelector(
       DimensionSpec spec,

processing/src/main/java/org/apache/druid/segment/StringDimensionHandler.java

@@ -98,12 +98,14 @@ private static IndexedInts getRow(ColumnValueSelector s)
   private final String dimensionName;
   private final MultiValueHandling multiValueHandling;
   private final boolean hasBitmapIndexes;
+  private final boolean hasSpatialIndexes;
 
-  public StringDimensionHandler(String dimensionName, MultiValueHandling multiValueHandling, boolean hasBitmapIndexes)
+  public StringDimensionHandler(String dimensionName, MultiValueHandling multiValueHandling, boolean hasBitmapIndexes, boolean hasSpatialIndexes)
   {
     this.dimensionName = dimensionName;
     this.multiValueHandling = multiValueHandling;
     this.hasBitmapIndexes = hasBitmapIndexes;
+    this.hasSpatialIndexes = hasSpatialIndexes;
   }
 
   @Override
@@ -139,7 +141,7 @@ public SettableColumnValueSelector makeNewSettableEncodedValueSelector()
   @Override
   public DimensionIndexer<Integer, int[], String> makeIndexer()
   {
-    return new StringDimensionIndexer(multiValueHandling, hasBitmapIndexes);
+    return new StringDimensionIndexer(multiValueHandling, hasBitmapIndexes, hasSpatialIndexes);
   }
 
   @Override