lower segment heap footprint and fix bug with expression type coercion

indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java

@@ -2129,13 +2129,12 @@ void removeMetadata(File file)
             file,
             new SimpleQueryableIndex(
                 index.getDataInterval(),
-                index.getColumnNames(),
                 index.getAvailableDimensions(),
                 index.getBitmapFactoryForDimensions(),
                 index.getColumns(),
                 index.getFileMapper(),
                 null,
-                () -> index.getDimensionHandlers()
+                false
             )
         );
       }

processing/src/main/java/org/apache/druid/java/util/common/io/smoosh/SmooshedFileMapper.java

@@ -48,7 +48,11 @@
  */
 public class SmooshedFileMapper implements Closeable
 {
-  private static final Interner<String> STRING_INTERNER = Interners.newWeakInterner();
+  /**
+   * Interner for smoosh internal files, which includes all column names since every column has an internal file
+   * associated with it
+   */
+  public static final Interner<String> STRING_INTERNER = Interners.newWeakInterner();
 
   public static SmooshedFileMapper load(File baseDir) throws IOException
   {

processing/src/main/java/org/apache/druid/math/expr/ExpressionTypeConversion.java

@@ -88,10 +88,10 @@ public static ExpressionType operator(@Nullable ExpressionType type, @Nullable E
       return type;
     }
     if (type.is(ExprType.COMPLEX) || other.is(ExprType.COMPLEX)) {
-      if (type.getElementType() == null) {
+      if (type.getComplexTypeName() == null) {
         return other;
       }
-      if (other.getElementType() == null) {
+      if (other.getComplexTypeName() == null) {
         return type;
       }
       if (!Objects.equals(type, other)) {

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

@@ -60,7 +60,9 @@
 import org.apache.druid.segment.data.CompressedColumnarLongsSupplier;
 import org.apache.druid.segment.data.GenericIndexed;
 import org.apache.druid.segment.data.ImmutableRTreeObjectStrategy;
+import org.apache.druid.segment.data.Indexed;
 import org.apache.druid.segment.data.IndexedIterable;
+import org.apache.druid.segment.data.ListIndexed;
 import org.apache.druid.segment.data.VSizeColumnarMultiInts;
 import org.apache.druid.segment.serde.ComplexColumnPartSupplier;
 import org.apache.druid.segment.serde.DictionaryEncodedColumnSupplier;
@@ -639,14 +641,12 @@ public QueryableIndex load(File inDir, ObjectMapper mapper, boolean lazy, Segmen
           loadFailed
       );
 
-      final GenericIndexed<String> finalCols, finalDims;
+      final Indexed<String> finalCols, finalDims;
 
       if (allCols != null) {
         // To restore original column order, we merge allCols/allDims and nonNullCols/nonNullDims, respectively.
-        final List<String> mergedCols = restoreColumns(nonNullCols, allCols);
-        final List<String> mergedDims = restoreColumns(nonNullDims, allDims);
-        finalCols = GenericIndexed.fromIterable(mergedCols, GenericIndexed.STRING_STRATEGY);
-        finalDims = GenericIndexed.fromIterable(mergedDims, GenericIndexed.STRING_STRATEGY);
+        finalCols = new ListIndexed<>(restoreColumns(nonNullCols, allCols));
+        finalDims = new ListIndexed<>(restoreColumns(nonNullDims, allDims));
       } else {
         finalCols = nonNullCols;
         finalDims = nonNullDims;
@@ -701,9 +701,9 @@ private List<String> restoreColumns(GenericIndexed<String> nonNullCols, GenericI
               + "while allColsIterator expects one. This is likely a potential bug in creating this segment. "
               + "Try reingesting your data with storeEmptyColumns setting to false in task context."
           );
-          mergedCols.add(nonNullColsIterator.next());
+          mergedCols.add(SmooshedFileMapper.STRING_INTERNER.intern(nonNullColsIterator.next()));
         } else {
-          mergedCols.add(next);
+          mergedCols.add(SmooshedFileMapper.STRING_INTERNER.intern(next));
         }
       }
 
@@ -712,7 +712,7 @@ private List<String> restoreColumns(GenericIndexed<String> nonNullCols, GenericI
 
     private void registerColumnHolders(
         File inDir,
-        GenericIndexed<String> cols,
+        Indexed<String> cols,
         boolean lazy,
         Map<String, Supplier<ColumnHolder>> columns,
         ObjectMapper mapper,
@@ -726,7 +726,7 @@ private void registerColumnHolders(
           continue;
         }
 
-        ByteBuffer colBuffer = smooshedFiles.mapFile(columnName);
+        final ByteBuffer colBuffer = smooshedFiles.mapFile(columnName);
         registerColumnHolder(
             lazy,
             columns,
@@ -749,12 +749,16 @@ private void registerColumnHolder(
         SegmentLazyLoadFailCallback loadFailed
     ) throws IOException
     {
+
+      // we use the interner here too even though it might have already been added by restoreColumns(..) because that
+      // only happens if there are some null columns
+      final String internedColumnName = SmooshedFileMapper.STRING_INTERNER.intern(columnName);
       if (lazy) {
-        columns.put(columnName, Suppliers.memoize(
+        columns.put(internedColumnName, Suppliers.memoize(
             () -> {
               try {
                 return deserializeColumn(
-                    columnName,
+                    internedColumnName,
                     mapper,
                     colBuffer,
                     smooshedFiles
@@ -768,13 +772,13 @@ private void registerColumnHolder(
             }
         ));
       } else {
-        ColumnHolder columnHolder = deserializeColumn(
-            columnName,
+        final ColumnHolder columnHolder = deserializeColumn(
+            internedColumnName,
             mapper,
             colBuffer,
             smooshedFiles
         );
-        columns.put(columnName, () -> columnHolder);
+        columns.put(internedColumnName, () -> columnHolder);
       }
     }
 

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

@@ -83,40 +83,6 @@ public SimpleQueryableIndex(
     }
   }
 
-  private Map<String, DimensionHandler> initDimensionHandlers(Indexed<String> availableDimensions)
-  {
-    Map<String, DimensionHandler> dimensionHandlerMap = Maps.newLinkedHashMap();
-    for (String dim : availableDimensions) {
-      final ColumnHolder columnHolder = getColumnHolder(dim);
-      ColumnCapabilities capabilities = columnHolder.getHandlerCapabilities();
-      DimensionHandler handler = DimensionHandlerUtils.getHandlerFromCapabilities(dim, capabilities, null);
-      dimensionHandlerMap.put(dim, handler);
-    }
-    return dimensionHandlerMap;
-  }
-
-  @VisibleForTesting
-  public SimpleQueryableIndex(
-      Interval interval,
-      List<String> columnNames,
-      Indexed<String> availableDimensions,
-      BitmapFactory bitmapFactory,
-      Map<String, Supplier<ColumnHolder>> columns,
-      SmooshedFileMapper fileMapper,
-      @Nullable Metadata metadata,
-      Supplier<Map<String, DimensionHandler>> dimensionHandlers
-  )
-  {
-    this.dataInterval = interval;
-    this.columnNames = columnNames;
-    this.availableDimensions = availableDimensions;
-    this.bitmapFactory = bitmapFactory;
-    this.columns = columns;
-    this.fileMapper = fileMapper;
-    this.metadata = metadata;
-    this.dimensionHandlers = dimensionHandlers;
-  }
-
   @Override
   public Interval getDataInterval()
   {
@@ -193,4 +159,15 @@ public Map<String, DimensionHandler> getDimensionHandlers()
     return dimensionHandlers.get();
   }
 
+  private Map<String, DimensionHandler> initDimensionHandlers(Indexed<String> availableDimensions)
+  {
+    Map<String, DimensionHandler> dimensionHandlerMap = Maps.newLinkedHashMap();
+    for (String dim : availableDimensions) {
+      final ColumnHolder columnHolder = getColumnHolder(dim);
+      ColumnCapabilities capabilities = columnHolder.getHandlerCapabilities();
+      DimensionHandler handler = DimensionHandlerUtils.getHandlerFromCapabilities(dim, capabilities, null);
+      dimensionHandlerMap.put(dim, handler);
+    }
+    return dimensionHandlerMap;
+  }
 }

processing/src/test/java/org/apache/druid/math/expr/OutputTypeTest.java

@@ -21,6 +21,7 @@
 
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.segment.nested.NestedDataComplexTypeSerde;
 import org.apache.druid.testing.InitializedNullHandlingTest;
 import org.junit.Assert;
 import org.junit.Rule;
@@ -537,6 +538,20 @@ public void testOperatorAutoConversion()
         ExpressionType.STRING_ARRAY,
         ExpressionTypeConversion.operator(ExpressionType.STRING_ARRAY, ExpressionType.STRING_ARRAY)
     );
+
+    ExpressionType nested = ExpressionType.fromColumnType(NestedDataComplexTypeSerde.TYPE);
+    Assert.assertEquals(
+        nested,
+        ExpressionTypeConversion.operator(nested, nested)
+    );
+    Assert.assertEquals(
+        nested,
+        ExpressionTypeConversion.operator(nested, ExpressionType.UNKNOWN_COMPLEX)
+    );
+    Assert.assertEquals(
+        nested,
+        ExpressionTypeConversion.operator(ExpressionType.UNKNOWN_COMPLEX, nested)
+    );
   }
 
   @Test
@@ -601,6 +616,19 @@ public void testFunctionAutoConversion()
         ExpressionType.STRING_ARRAY,
         ExpressionTypeConversion.function(ExpressionType.STRING_ARRAY, ExpressionType.STRING_ARRAY)
     );
+    ExpressionType nested = ExpressionType.fromColumnType(NestedDataComplexTypeSerde.TYPE);
+    Assert.assertEquals(
+        nested,
+        ExpressionTypeConversion.function(nested, nested)
+    );
+    Assert.assertEquals(
+        nested,
+        ExpressionTypeConversion.function(nested, ExpressionType.UNKNOWN_COMPLEX)
+    );
+    Assert.assertEquals(
+        nested,
+        ExpressionTypeConversion.function(ExpressionType.UNKNOWN_COMPLEX, nested)
+    );
   }
 
   @Test