remix nested columns

changes: * introduce ColumnFormat to separate physical storage format from logical type. ColumnFormat is now used instead of ColumnCapabilities to get column handlers for segment creation * introduce new 'standard type' indexer, merger, and family of serializers, which is the next logical iteration of the nested column stuff. Essentially this is an automatic type column indexer that produces the most appropriate column for the given inputs, making either STRING, ARRAY<STRING>, LONG, ARRAY<LONG>, DOUBLE, ARRAY<DOUBLE>, or COMPLEX<json>. * revert NestedDataColumnIndexer, NestedDataColumnMerger, NestedDataColumnSerializer to their version pre #13803 behavior (v4) for backwards compatibility * fix a bug in RoaringBitmapSerdeFactory if anything actually ever wrote out an empty bitmap using toBytes and then later tried to read it (the nerve!)
apache · Apr 2, 2023 · 6561769 · 6561769
1 parent 518698a
commit 6561769
Show file tree

Hide file tree

Showing 123 changed files with 8,890 additions and 1,383 deletions.
diff --git a/...xtensions/src/test/java/org/apache/druid/data/input/protobuf/ProtobufInputFormatTest.java b/...xtensions/src/test/java/org/apache/druid/data/input/protobuf/ProtobufInputFormatTest.java
@@ -42,7 +42,7 @@
 import org.apache.druid.java.util.common.parsers.JSONPathSpec;
 import org.apache.druid.math.expr.ExpressionProcessing;
 import org.apache.druid.query.expression.TestExprMacroTable;
-import org.apache.druid.segment.NestedDataDimensionSchema;
+import org.apache.druid.segment.StandardTypeColumnSchema;
 import org.apache.druid.segment.transform.ExpressionTransform;
 import org.apache.druid.segment.transform.TransformSpec;
 import org.apache.druid.segment.transform.TransformingInputEntityReader;
@@ -269,14 +269,14 @@ public void testParseNestedData() throws Exception
             timestampSpec,
             new DimensionsSpec(
                 Lists.newArrayList(
-                    new StringDimensionSchema("event"),
-                    new StringDimensionSchema("id"),
-                    new StringDimensionSchema("someOtherId"),
-                    new StringDimensionSchema("isValid"),
-                    new StringDimensionSchema("eventType"),
-                    new NestedDataDimensionSchema("foo"),
-                    new NestedDataDimensionSchema("bar"),
-                    new StringDimensionSchema("someBytesColumn")
+                    new StandardTypeColumnSchema("event"),
+                    new StandardTypeColumnSchema("id"),
+                    new StandardTypeColumnSchema("someOtherId"),
+                    new StandardTypeColumnSchema("isValid"),
+                    new StandardTypeColumnSchema("eventType"),
+                    new StandardTypeColumnSchema("foo"),
+                    new StandardTypeColumnSchema("bar"),
+                    new StandardTypeColumnSchema("someBytesColumn")
                 )
             ),
             null

diff --git a/indexing-hadoop/src/main/java/org/apache/druid/indexer/IndexGeneratorJob.java b/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.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnFormat;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexSchema;
 import org.apache.druid.timeline.DataSegment;
@@ -292,7 +292,7 @@ private static IncrementalIndex makeIncrementalIndex(
       AggregatorFactory[] aggs,
       HadoopDruidIndexerConfig config,
       Iterable<String> oldDimOrder,
-      Map<String, ColumnCapabilities> oldCapabilities
+      Map<String, ColumnFormat> oldCapabilities
   )
   {
     final HadoopTuningConfig tuningConfig = config.getSchema().getTuningConfig();
@@ -456,7 +456,7 @@ protected void reduce(final BytesWritable key, Iterable<BytesWritable> values, f
             dimOrder.addAll(index.getDimensionOrder());
             log.info("current index full due to [%s]. creating new index.", index.getOutOfRowsReason());
             flushIndexToContextAndClose(key, index, context);
-            index = makeIncrementalIndex(bucket, combiningAggs, config, dimOrder, index.getColumnHandlerCapabilities());
+            index = makeIncrementalIndex(bucket, combiningAggs, config, dimOrder, index.getColumnFormats());
           }
 
           index.add(value);
@@ -752,7 +752,7 @@ public void doRun()
                 combiningAggs,
                 config,
                 allDimensionNames,
-                persistIndex.getColumnHandlerCapabilities()
+                persistIndex.getColumnFormats()
             );
             startTime = System.currentTimeMillis();
             ++indexCount;

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java
@@ -1109,7 +1109,7 @@ private void processDimensionsSpec(final QueryableIndex index)
         );
 
         if (!uniqueDims.containsKey(dimension)) {
-          final DimensionHandler dimensionHandler = Preconditions.checkNotNull(
+          Preconditions.checkNotNull(
               dimensionHandlerMap.get(dimension),
               "Cannot find dimensionHandler for dimension[%s]",
               dimension
@@ -1118,11 +1118,7 @@ private void processDimensionsSpec(final QueryableIndex index)
           uniqueDims.put(dimension, uniqueDims.size());
           dimensionSchemaMap.put(
               dimension,
-              createDimensionSchema(
-                  dimension,
-                  columnHolder.getHandlerCapabilities(),
-                  dimensionHandler.getMultivalueHandling()
-              )
+              columnHolder.getColumnFormat().getColumnSchema(dimension)
           );
         }
       }

diff --git a/...-service/src/main/java/org/apache/druid/indexing/overlord/sampler/InputSourceSampler.java b/...-service/src/main/java/org/apache/druid/indexing/overlord/sampler/InputSourceSampler.java
@@ -247,11 +247,12 @@ public SamplerResponse sample(
           if (!SamplerInputRow.SAMPLER_ORDERING_COLUMN.equals(dimensionDesc.getName())) {
             final ColumnType columnType = dimensionDesc.getCapabilities().toColumnType();
             signatureBuilder.add(dimensionDesc.getName(), columnType);
+            // for now, use legacy types instead of standard type
             logicalDimensionSchemas.add(
                 DimensionSchema.getDefaultSchemaForBuiltInType(dimensionDesc.getName(), dimensionDesc.getCapabilities())
             );
             physicalDimensionSchemas.add(
-                dimensionDesc.getHandler().getDimensionSchema(dimensionDesc.getCapabilities())
+                dimensionDesc.getIndexer().getFormat().getColumnSchema(dimensionDesc.getName())
             );
           }
         }

diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/DimensionSchema.java b/processing/src/main/java/org/apache/druid/data/input/impl/DimensionSchema.java
@@ -29,10 +29,14 @@
 import org.apache.druid.guice.annotations.PublicApi;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.segment.DimensionHandler;
+import org.apache.druid.segment.DimensionHandlerUtils;
 import org.apache.druid.segment.NestedDataDimensionSchema;
+import org.apache.druid.segment.StandardTypeColumnSchema;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.TypeSignature;
 import org.apache.druid.segment.column.ValueType;
+import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.nested.NestedDataComplexTypeSerde;
 
 import java.util.Objects;
@@ -47,7 +51,9 @@
     @JsonSubTypes.Type(name = DimensionSchema.FLOAT_TYPE_NAME, value = FloatDimensionSchema.class),
     @JsonSubTypes.Type(name = DimensionSchema.DOUBLE_TYPE_NAME, value = DoubleDimensionSchema.class),
     @JsonSubTypes.Type(name = DimensionSchema.SPATIAL_TYPE_NAME, value = NewSpatialDimensionSchema.class),
-    @JsonSubTypes.Type(name = NestedDataComplexTypeSerde.TYPE_NAME, value = NestedDataDimensionSchema.class)
+    @JsonSubTypes.Type(name = NestedDataComplexTypeSerde.TYPE_NAME, value = NestedDataDimensionSchema.class),
+    @JsonSubTypes.Type(name = StandardTypeColumnSchema.TYPE, value = StandardTypeColumnSchema.class),
+    @JsonSubTypes.Type(name = "auto", value = StandardTypeColumnSchema.class)
 })
 public abstract class DimensionSchema
 {
@@ -150,6 +156,17 @@ public boolean hasBitmapIndex()
   @JsonIgnore
   public abstract ColumnType getColumnType();
 
+  @JsonIgnore
+  public DimensionHandler getDimensionHandler()
+  {
+    // default implementation for backwards compatibility
+    return DimensionHandlerUtils.getHandlerFromCapabilities(
+        name,
+        IncrementalIndex.makeDefaultCapabilitiesFromValueType(getColumnType()),
+        multiValueHandling
+    );
+  }
+
   @Override
   public boolean equals(final Object o)
   {

diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/DoubleDimensionSchema.java b/processing/src/main/java/org/apache/druid/data/input/impl/DoubleDimensionSchema.java
@@ -21,6 +21,8 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.segment.DimensionHandler;
+import org.apache.druid.segment.DoubleDimensionHandler;
 import org.apache.druid.segment.column.ColumnType;
 
 public class DoubleDimensionSchema extends DimensionSchema
@@ -42,4 +44,10 @@ public ColumnType getColumnType()
   {
     return ColumnType.DOUBLE;
   }
+
+  @Override
+  public DimensionHandler getDimensionHandler()
+  {
+    return new DoubleDimensionHandler(getName());
+  }
 }
diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/FloatDimensionSchema.java b/processing/src/main/java/org/apache/druid/data/input/impl/FloatDimensionSchema.java
@@ -22,6 +22,8 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.segment.DimensionHandler;
+import org.apache.druid.segment.FloatDimensionHandler;
 import org.apache.druid.segment.column.ColumnType;
 
 public class FloatDimensionSchema extends DimensionSchema
@@ -46,4 +48,10 @@ public ColumnType getColumnType()
   {
     return ColumnType.FLOAT;
   }
+
+  @Override
+  public DimensionHandler getDimensionHandler()
+  {
+    return new FloatDimensionHandler(getName());
+  }
 }
diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/LongDimensionSchema.java b/processing/src/main/java/org/apache/druid/data/input/impl/LongDimensionSchema.java
@@ -22,6 +22,8 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.segment.DimensionHandler;
+import org.apache.druid.segment.LongDimensionHandler;
 import org.apache.druid.segment.column.ColumnType;
 
 
@@ -47,4 +49,10 @@ public ColumnType getColumnType()
   {
     return ColumnType.LONG;
   }
+
+  @Override
+  public DimensionHandler getDimensionHandler()
+  {
+    return new LongDimensionHandler(getName());
+  }
 }
diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/NewSpatialDimensionSchema.java b/processing/src/main/java/org/apache/druid/data/input/impl/NewSpatialDimensionSchema.java
@@ -22,6 +22,8 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.segment.DimensionHandler;
+import org.apache.druid.segment.StringDimensionHandler;
 import org.apache.druid.segment.column.ColumnType;
 
 import java.util.List;
@@ -67,6 +69,12 @@ public ColumnType getColumnType()
     return ColumnType.STRING;
   }
 
+  @Override
+  public DimensionHandler getDimensionHandler()
+  {
+    return new StringDimensionHandler(getName(), getMultiValueHandling(), hasBitmapIndex(), true);
+  }
+
   @Override
   public boolean equals(Object o)
   {

diff --git a/processing/src/main/java/org/apache/druid/data/input/impl/StringDimensionSchema.java b/processing/src/main/java/org/apache/druid/data/input/impl/StringDimensionSchema.java
@@ -22,6 +22,8 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.segment.DimensionHandler;
+import org.apache.druid.segment.StringDimensionHandler;
 import org.apache.druid.segment.column.ColumnType;
 
 public class StringDimensionSchema extends DimensionSchema
@@ -61,4 +63,10 @@ public ColumnType getColumnType()
   {
     return ColumnType.STRING;
   }
+
+  @Override
+  public DimensionHandler getDimensionHandler()
+  {
+    return new StringDimensionHandler(getName(), getMultiValueHandling(), hasBitmapIndex(), false);
+  }
 }
diff --git a/processing/src/main/java/org/apache/druid/query/expression/NestedDataExpressions.java b/processing/src/main/java/org/apache/druid/query/expression/NestedDataExpressions.java
@@ -30,7 +30,7 @@
 import org.apache.druid.math.expr.ExprType;
 import org.apache.druid.math.expr.ExpressionType;
 import org.apache.druid.math.expr.NamedFunction;
-import org.apache.druid.segment.nested.NestedDataComplexTypeSerde;
+import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.nested.NestedPathFinder;
 import org.apache.druid.segment.nested.NestedPathPart;
 import org.apache.druid.segment.nested.StructuredData;
@@ -47,7 +47,7 @@
 public class NestedDataExpressions
 {
   public static final ExpressionType TYPE = Preconditions.checkNotNull(
-      ExpressionType.fromColumnType(NestedDataComplexTypeSerde.TYPE)
+      ExpressionType.fromColumnType(ColumnType.NESTED_DATA)
   );
 
   public static class JsonObjectExprMacro implements ExprMacroTable.ExprMacro

diff --git a/processing/src/main/java/org/apache/druid/segment/DimensionIndexer.java b/processing/src/main/java/org/apache/druid/segment/DimensionIndexer.java
@@ -22,12 +22,19 @@
 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.CapabilitiesBasedFormat;
 import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.column.ColumnCapabilitiesImpl;
+import org.apache.druid.segment.column.ColumnFormat;
 import org.apache.druid.segment.data.CloseableIndexed;
 import org.apache.druid.segment.incremental.IncrementalIndex;
 import org.apache.druid.segment.incremental.IncrementalIndexRowHolder;
+import org.apache.druid.segment.nested.FieldTypeInfo;
+import org.apache.druid.segment.nested.NestedPathFinder;
+import org.apache.druid.segment.nested.SortedValueDictionary;
 
 import javax.annotation.Nullable;
+import java.util.SortedMap;
 
 /**
  * Processing related interface
@@ -175,6 +182,18 @@ EncodedKeyComponent<EncodedKeyComponentType> processRowValsToUnsortedEncodedKeyC
   CloseableIndexed<ActualType> getSortedIndexedValues();
 
 
+  default SortedValueDictionary getSortedValueLookups()
+  {
+    throw new UnsupportedOperationException("Column does not support value dictionaries.");
+  }
+
+  default void mergeNestedFields(SortedMap<String, FieldTypeInfo.MutableTypeSet> mergedFields)
+  {
+    mergedFields.put(
+        NestedPathFinder.JSON_PATH_ROOT,
+        new FieldTypeInfo.MutableTypeSet().add(getColumnCapabilities().toColumnType())
+    );
+  }
   /**
    * Get the minimum dimension value seen by this indexer.
    *
@@ -237,9 +256,14 @@ ColumnValueSelector<?> makeColumnValueSelector(
 
   ColumnCapabilities getColumnCapabilities();
 
-  default ColumnCapabilities getHandlerCapabilities()
+  default ColumnFormat getFormat()
   {
-    return getColumnCapabilities();
+    return new CapabilitiesBasedFormat(
+        ColumnCapabilitiesImpl.snapshot(
+            getColumnCapabilities(),
+            CapabilitiesBasedFormat.DIMENSION_CAPABILITY_MERGE_LOGIC
+        )
+    );
   }
 
   /**

diff --git a/processing/src/main/java/org/apache/druid/segment/IndexIO.java b/processing/src/main/java/org/apache/druid/segment/IndexIO.java
@@ -788,7 +788,7 @@ private void registerColumnHolder(
      */
     @VisibleForTesting
     ColumnHolder deserializeColumn(
-        String columnName, // columnName is not used in this method, but used in tests.
+        String columnName,
         ObjectMapper mapper,
         ByteBuffer byteBuffer,
         SmooshedFileMapper smooshedFiles
@@ -797,7 +797,7 @@ ColumnHolder deserializeColumn(
       ColumnDescriptor serde = mapper.readValue(
           SERIALIZER_UTILS.readString(byteBuffer), ColumnDescriptor.class
       );
-      return serde.read(byteBuffer, columnConfig, smooshedFiles);
+      return serde.read(columnName, byteBuffer, columnConfig, smooshedFiles);
     }
   }