[Backport] reduce heap footprint of ingesting auto typed columns by pushing compression and index generation into writeTo

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

@@ -99,11 +99,11 @@ public EqualityFilter(
       throw InvalidInput.exception("Invalid equality filter on column [%s], matchValueType cannot be null", column);
     }
     this.matchValueType = matchValueType;
-    if (matchValue == null) {
-      throw InvalidInput.exception("Invalid equality filter on column [%s], matchValue cannot be null", column);
-    }
     this.matchValue = matchValue;
     this.matchValueEval = ExprEval.ofType(ExpressionType.fromColumnTypeStrict(matchValueType), matchValue);
+    if (matchValueEval.value() == null) {
+      throw InvalidInput.exception("Invalid equality filter on column [%s], matchValue cannot be null", column);
+    }
     this.filterTuning = filterTuning;
     this.predicateFactory = new EqualityPredicateFactory(matchValueEval);
   }
@@ -239,6 +239,8 @@ public BitmapColumnIndex getBitmapColumnIndex(ColumnIndexSelector selector)
 
     final ValueIndexes valueIndexes = indexSupplier.as(ValueIndexes.class);
     if (valueIndexes != null) {
+      // matchValueEval.value() cannot be null here due to check in the constructor
+      //noinspection DataFlowIssue
       return valueIndexes.forValue(matchValueEval.value(), matchValueType);
     }
 

processing/src/main/java/org/apache/druid/segment/data/FixedIndexedWriter.java

@@ -24,7 +24,6 @@
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.io.smoosh.FileSmoosher;
 import org.apache.druid.segment.column.TypeStrategy;
-import org.apache.druid.segment.serde.Serializer;
 import org.apache.druid.segment.writeout.SegmentWriteOutMedium;
 import org.apache.druid.segment.writeout.WriteOutBytes;
 
@@ -39,7 +38,7 @@
 /**
  * Writer for a {@link FixedIndexed}
  */
-public class FixedIndexedWriter<T> implements Serializer
+public class FixedIndexedWriter<T> implements DictionaryWriter<T>
 {
   private static final int PAGE_SIZE = 4096;
   private final SegmentWriteOutMedium segmentWriteOutMedium;
@@ -73,11 +72,19 @@ public FixedIndexedWriter(
     this.isSorted = isSorted;
   }
 
+  @Override
+  public boolean isSorted()
+  {
+    return isSorted;
+  }
+
+  @Override
   public void open() throws IOException
   {
     this.valuesOut = segmentWriteOutMedium.makeWriteOutBytes();
   }
 
+  @Override
   public int getCardinality()
   {
     return hasNulls ? numWritten + 1 : numWritten;
@@ -89,6 +96,7 @@ public long getSerializedSize()
     return Byte.BYTES + Byte.BYTES + Integer.BYTES + valuesOut.size();
   }
 
+  @Override
   public void write(@Nullable T objectToWrite) throws IOException
   {
     if (prevObject != null && isSorted && comparator.compare(prevObject, objectToWrite) >= 0) {
@@ -140,6 +148,7 @@ public void writeTo(
   }
 
   @SuppressWarnings("unused")
+  @Override
   @Nullable
   public T get(int index) throws IOException
   {

processing/src/main/java/org/apache/druid/segment/index/AllFalseBitmapColumnIndex.java

@@ -19,18 +19,24 @@
 
 package org.apache.druid.segment.index;
 
+import org.apache.druid.collections.bitmap.BitmapFactory;
 import org.apache.druid.query.BitmapResultFactory;
 import org.apache.druid.query.filter.ColumnIndexSelector;
 import org.apache.druid.segment.column.ColumnIndexCapabilities;
 import org.apache.druid.segment.column.SimpleColumnIndexCapabilities;
 
 public class AllFalseBitmapColumnIndex implements BitmapColumnIndex
 {
-  private final ColumnIndexSelector selector;
+  private final BitmapFactory bitmapFactory;
 
   public AllFalseBitmapColumnIndex(ColumnIndexSelector indexSelector)
   {
-    this.selector = indexSelector;
+    this(indexSelector.getBitmapFactory());
+  }
+
+  public AllFalseBitmapColumnIndex(BitmapFactory bitmapFactory)
+  {
+    this.bitmapFactory = bitmapFactory;
   }
 
   @Override
@@ -48,6 +54,6 @@ public double estimateSelectivity(int totalRows)
   @Override
   public <T> T computeBitmapResult(BitmapResultFactory<T> bitmapResultFactory)
   {
-    return bitmapResultFactory.wrapAllFalse(selector.getBitmapFactory().makeEmptyImmutableBitmap());
+    return bitmapResultFactory.wrapAllFalse(bitmapFactory.makeEmptyImmutableBitmap());
   }
 }

processing/src/main/java/org/apache/druid/segment/index/IndexedUtf8ValueIndexes.java

@@ -37,6 +37,7 @@
 import org.apache.druid.segment.index.semantic.Utf8ValueSetIndexes;
 import org.apache.druid.segment.index.semantic.ValueIndexes;
 
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import java.nio.ByteBuffer;
 import java.util.Comparator;
@@ -101,7 +102,7 @@ private ImmutableBitmap getBitmapForValue()
 
   @Nullable
   @Override
-  public BitmapColumnIndex forValue(Object value, TypeSignature<ValueType> valueType)
+  public BitmapColumnIndex forValue(@Nonnull Object value, TypeSignature<ValueType> valueType)
   {
     if (valueType.isPrimitive()) {
       return forValue(

processing/src/main/java/org/apache/druid/segment/index/semantic/ArrayElementIndexes.java

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.segment.index.semantic;
+
+import org.apache.druid.collections.bitmap.ImmutableBitmap;
+import org.apache.druid.segment.column.TypeSignature;
+import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.segment.index.BitmapColumnIndex;
+
+import javax.annotation.Nullable;
+
+public interface ArrayElementIndexes
+{
+  /**
+   * Get the {@link ImmutableBitmap} corresponding to rows with array elements matching the supplied value.  Generates
+   * an empty bitmap when passed a value that doesn't exist in any array. May return null if a value index cannot be
+   * computed for the supplied value type.
+   *
+   * @param value       value to match against any array element in a row
+   * @param valueType   type of the value to match, used to assist conversion from the match value type to the column
+   *                    value type
+   * @return            {@link ImmutableBitmap} corresponding to the rows with array elements which match the value, or
+   *                    null if an index connot be computed for the supplied value type
+   */
+  @Nullable
+  BitmapColumnIndex containsValue(@Nullable Object value, TypeSignature<ValueType> valueType);
+}

processing/src/main/java/org/apache/druid/segment/index/semantic/ValueIndexes.java

@@ -24,14 +24,17 @@
 import org.apache.druid.segment.column.ValueType;
 import org.apache.druid.segment.index.BitmapColumnIndex;
 
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 public interface ValueIndexes
 {
 
   /**
-   * Get the {@link ImmutableBitmap} corresponding to the supplied value.  Generates an empty bitmap when passed a
-   * value that doesn't exist. May return null if a value index cannot be computed for the supplied value type.
+   * Get the {@link ImmutableBitmap} corresponding to rows matching the supplied value.  Generates an empty bitmap when
+   * passed a value that doesn't exist. May return null if a value index cannot be computed for the supplied value type.
+   *
+   * Does not match null, use {@link NullValueIndex} for matching nulls.
    *
    * @param value       value to match
    * @param valueType   type of the value to match, used to assist conversion from the match value type to the column
@@ -40,5 +43,5 @@ public interface ValueIndexes
    *                    connot be computed for the supplied value type
    */
   @Nullable
-  BitmapColumnIndex forValue(Object value, TypeSignature<ValueType> valueType);
+  BitmapColumnIndex forValue(@Nonnull Object value, TypeSignature<ValueType> valueType);
 }

processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleColumnAndIndexSupplier.java

@@ -65,6 +65,7 @@
 import org.apache.druid.segment.index.semantic.ValueIndexes;
 import org.apache.druid.segment.serde.NestedCommonFormatColumnPartSerde;
 
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -228,11 +229,12 @@ private class DoubleValueIndexes implements ValueIndexes
   {
     @Nullable
     @Override
-    public BitmapColumnIndex forValue(Object value, TypeSignature<ValueType> valueType)
+    public BitmapColumnIndex forValue(@Nonnull Object value, TypeSignature<ValueType> valueType)
     {
       final ExprEval<?> eval = ExprEval.ofType(ExpressionType.fromColumnTypeStrict(valueType), value)
                                        .castTo(ExpressionType.DOUBLE);
       if (eval.isNumericNull()) {
+        // value wasn't null, but not a number?
         return null;
       }
       final double doubleValue = eval.asDouble();