Enable GenericIndexed V2 for built-in(druid-io managed) complex columns. (#3987)

* Enable GenericIndexed V2 for complex columns.

* SerializerBuilder to use  GenericColumnSerializer.

extensions-core/datasketches/src/main/java/io/druid/query/aggregation/datasketches/theta/SketchMergeComplexMetricSerde.java

@@ -21,12 +21,15 @@ package io.druid.query.aggregation.datasketches.theta;
 
 import com.yahoo.sketches.theta.Sketch;
 import io.druid.data.input.InputRow;
+import io.druid.segment.GenericColumnSerializer;
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.data.GenericIndexed;
+import io.druid.segment.data.IOPeon;
 import io.druid.segment.data.ObjectStrategy;
 import io.druid.segment.serde.ComplexColumnPartSupplier;
 import io.druid.segment.serde.ComplexMetricExtractor;
 import io.druid.segment.serde.ComplexMetricSerde;
+import io.druid.segment.serde.LargeColumnSupportedComplexColumnSerializer;
 
 import java.nio.ByteBuffer;
 
@@ -76,4 +79,10 @@ public class SketchMergeComplexMetricSerde extends ComplexMetricSerde
     return strategy;
   }
 
+  @Override
+  public GenericColumnSerializer getSerializer(IOPeon peon, String column)
+  {
+    return LargeColumnSupportedComplexColumnSerializer.create(peon, column, this.getObjectStrategy());
+  }
+
 }

extensions-core/histogram/src/main/java/io/druid/query/aggregation/histogram/ApproximateHistogramFoldingSerde.java

@@ -21,12 +21,15 @@ package io.druid.query.aggregation.histogram;
 
 import com.google.common.collect.Ordering;
 import io.druid.data.input.InputRow;
+import io.druid.segment.GenericColumnSerializer;
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.data.GenericIndexed;
+import io.druid.segment.data.IOPeon;
 import io.druid.segment.data.ObjectStrategy;
 import io.druid.segment.serde.ComplexColumnPartSupplier;
 import io.druid.segment.serde.ComplexMetricExtractor;
 import io.druid.segment.serde.ComplexMetricSerde;
+import io.druid.segment.serde.LargeColumnSupportedComplexColumnSerializer;
 
 import java.nio.ByteBuffer;
 import java.util.Iterator;
@@ -98,6 +101,12 @@ public class ApproximateHistogramFoldingSerde extends ComplexMetricSerde
     columnBuilder.setComplexColumn(new ComplexColumnPartSupplier(getTypeName(), column));
   }
 
+  @Override
+  public GenericColumnSerializer getSerializer(IOPeon peon, String column)
+  {
+    return LargeColumnSupportedComplexColumnSerializer.create(peon, column, this.getObjectStrategy());
+  }
+
   public ObjectStrategy getObjectStrategy()
   {
     return new ObjectStrategy<ApproximateHistogram>()

extensions-core/stats/src/main/java/io/druid/query/aggregation/variance/VarianceSerde.java

@@ -21,12 +21,15 @@ package io.druid.query.aggregation.variance;
 
 import com.google.common.collect.Ordering;
 import io.druid.data.input.InputRow;
+import io.druid.segment.GenericColumnSerializer;
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.data.GenericIndexed;
+import io.druid.segment.data.IOPeon;
 import io.druid.segment.data.ObjectStrategy;
 import io.druid.segment.serde.ComplexColumnPartSupplier;
 import io.druid.segment.serde.ComplexMetricExtractor;
 import io.druid.segment.serde.ComplexMetricSerde;
+import io.druid.segment.serde.LargeColumnSupportedComplexColumnSerializer;
 
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -118,4 +121,11 @@ public class VarianceSerde extends ComplexMetricSerde
       }
     };
   }
+
+  @Override
+  public GenericColumnSerializer getSerializer(IOPeon peon, String column)
+  {
+    return LargeColumnSupportedComplexColumnSerializer.create(peon, column, this.getObjectStrategy());
+  }
+
 }

processing/src/main/java/io/druid/query/aggregation/hyperloglog/HyperUniquesSerde.java

@@ -21,14 +21,17 @@ package io.druid.query.aggregation.hyperloglog;
 
 import com.google.common.collect.Ordering;
 import io.druid.data.input.InputRow;
-import io.druid.hll.HyperLogLogHash;
 import io.druid.hll.HyperLogLogCollector;
+import io.druid.hll.HyperLogLogHash;
+import io.druid.segment.GenericColumnSerializer;
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.data.GenericIndexed;
+import io.druid.segment.data.IOPeon;
 import io.druid.segment.data.ObjectStrategy;
 import io.druid.segment.serde.ComplexColumnPartSupplier;
 import io.druid.segment.serde.ComplexMetricExtractor;
 import io.druid.segment.serde.ComplexMetricSerde;
+import io.druid.segment.serde.LargeColumnSupportedComplexColumnSerializer;
 
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -141,4 +144,11 @@ public class HyperUniquesSerde extends ComplexMetricSerde
       }
     };
   }
+
+  @Override
+  public GenericColumnSerializer getSerializer(IOPeon peon, String column)
+  {
+    return LargeColumnSupportedComplexColumnSerializer.create(peon, column, this.getObjectStrategy());
+  }
+
 }

processing/src/main/java/io/druid/segment/IndexMergerV9.java

@@ -48,7 +48,6 @@ import io.druid.segment.data.IOPeon;
 import io.druid.segment.data.TmpFileIOPeon;
 import io.druid.segment.loading.MMappedQueryableSegmentizerFactory;
 import io.druid.segment.serde.ComplexColumnPartSerde;
-import io.druid.segment.serde.ComplexColumnSerializer;
 import io.druid.segment.serde.ComplexMetricSerde;
 import io.druid.segment.serde.ComplexMetrics;
 import io.druid.segment.serde.FloatGenericColumnPartSerde;
@@ -367,7 +366,7 @@ public class IndexMergerV9 extends IndexMerger
           builder.setValueType(ValueType.COMPLEX);
           builder.addSerde(
               ComplexColumnPartSerde.serializerBuilder().withTypeName(typeName)
-                                    .withDelegate((ComplexColumnSerializer) writer)
+                                    .withDelegate(writer)
                                     .build()
           );
           break;

processing/src/main/java/io/druid/segment/serde/ComplexColumnPartSerde.java

@@ -22,6 +22,7 @@ package io.druid.segment.serde;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.druid.java.util.common.io.smoosh.FileSmoosher;
+import io.druid.segment.GenericColumnSerializer;
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.column.ColumnConfig;
 import io.druid.segment.data.GenericIndexed;
@@ -92,7 +93,7 @@ public class ComplexColumnPartSerde implements ColumnPartSerde
   public static class SerializerBuilder
   {
     private String typeName = null;
-    private ComplexColumnSerializer delegate = null;
+    private GenericColumnSerializer delegate = null;
 
     public SerializerBuilder withTypeName(final String typeName)
     {
@@ -100,7 +101,7 @@ public class ComplexColumnPartSerde implements ColumnPartSerde
       return this;
     }
 
-    public SerializerBuilder withDelegate(final ComplexColumnSerializer delegate)
+    public SerializerBuilder withDelegate(final GenericColumnSerializer delegate)
     {
       this.delegate = delegate;
       return this;