build v9 directly

docs/content/ingestion/batch-ingestion.md

@@ -104,7 +104,8 @@ The spec\_file is a path to a file that contains JSON and an example looks like:
     "ignoreInvalidRows" : false,
     "jobProperties" : { },
     "combineText" : false,        
-    "rowFlushBoundary" : 300000
+    "rowFlushBoundary" : 300000,
+    "buildV9Directly" : false
   }
 }
 ```
@@ -205,6 +206,7 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |ignoreInvalidRows|Boolean|Ignore rows found to have problems.|no (default == false)|
 |useCombiner|Boolean|Use hadoop combiner to merge rows at mapper if possible.|no (default == false)|
 |jobProperties|Object|a map of properties to add to the Hadoop job configuration.|no (default == null)|
+|buildV9Directly|Boolean|Whether to build v9 index directly instead of building v8 index and convert it to v9 format|no (default = false)|
 
 ### Partitioning specification
 

docs/content/ingestion/realtime-ingestion.md

@@ -142,6 +142,7 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
 |rejectionPolicy|Object|Controls how data sets the data acceptance policy for creating and handing off segments. More on this below.|no (default=='serverTime')|
 |maxPendingPersists|Integer|Maximum number of persists that can be pending, but not started. If this limit would be exceeded by a new intermediate persist, ingestion will block until the currently-running persist finishes. Maximum heap memory usage for indexing scales with maxRowsInMemory * (2 + maxPendingPersists).|no (default == 0; meaning one persist can be running concurrently with ingestion, and none can be queued up)|
 |shardSpec|Object|This describes the shard that is represented by this server. This must be specified properly in order to have multiple realtime nodes indexing the same data stream in a [sharded fashion](#sharding).|no (default == 'NoneShardSpec'|
+|buildV9Directly|Boolean|Whether to build v9 index directly instead of building v8 index and convert it to v9 format|no (default = false)|
 
 #### Rejection Policy
 

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

@@ -25,8 +25,6 @@
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.data.GenericIndexed;
 import io.druid.segment.data.ObjectStrategy;
-import io.druid.segment.serde.ColumnPartSerde;
-import io.druid.segment.serde.ComplexColumnPartSerde;
 import io.druid.segment.serde.ComplexColumnPartSupplier;
 import io.druid.segment.serde.ComplexMetricExtractor;
 import io.druid.segment.serde.ComplexMetricSerde;
@@ -67,11 +65,10 @@ public Object extractValue(InputRow inputRow, String metricName)
   }
 
   @Override
-  public ColumnPartSerde deserializeColumn(ByteBuffer buffer, ColumnBuilder builder)
+  public void deserializeColumn(ByteBuffer buffer, ColumnBuilder builder)
   {
     GenericIndexed<Sketch> ge = GenericIndexed.read(buffer, strategy);
     builder.setComplexColumn(new ComplexColumnPartSupplier(getTypeName(), ge));
-    return new ComplexColumnPartSerde(ge, getTypeName());
   }
 
   @Override

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

@@ -24,8 +24,6 @@
 import io.druid.segment.column.ColumnBuilder;
 import io.druid.segment.data.GenericIndexed;
 import io.druid.segment.data.ObjectStrategy;
-import io.druid.segment.serde.ColumnPartSerde;
-import io.druid.segment.serde.ComplexColumnPartSerde;
 import io.druid.segment.serde.ComplexColumnPartSupplier;
 import io.druid.segment.serde.ComplexMetricExtractor;
 import io.druid.segment.serde.ComplexMetricSerde;
@@ -92,15 +90,12 @@ public ApproximateHistogram extractValue(InputRow inputRow, String metricName)
   }
 
   @Override
-  public ColumnPartSerde deserializeColumn(
+  public void deserializeColumn(
       ByteBuffer byteBuffer, ColumnBuilder columnBuilder
   )
   {
     final GenericIndexed column = GenericIndexed.read(byteBuffer, getObjectStrategy());
-
     columnBuilder.setComplexColumn(new ComplexColumnPartSupplier(getTypeName(), column));
-
-    return new ComplexColumnPartSerde(column, getTypeName());
   }
 
   public ObjectStrategy getObjectStrategy()

indexing-hadoop/src/main/java/io/druid/indexer/HadoopDruidIndexerConfig.java

@@ -51,6 +51,7 @@
 import io.druid.initialization.Initialization;
 import io.druid.segment.IndexIO;
 import io.druid.segment.IndexMerger;
+import io.druid.segment.IndexMergerV9;
 import io.druid.segment.IndexSpec;
 import io.druid.segment.indexing.granularity.GranularitySpec;
 import io.druid.server.DruidNode;
@@ -89,6 +90,7 @@ public class HadoopDruidIndexerConfig
   public static final ObjectMapper JSON_MAPPER;
   public static final IndexIO INDEX_IO;
   public static final IndexMerger INDEX_MERGER;
+  public static final IndexMergerV9 INDEX_MERGER_V9;
 
   private static final String DEFAULT_WORKING_PATH = "/tmp/druid-indexing";
 
@@ -112,6 +114,7 @@ public void configure(Binder binder)
     JSON_MAPPER = injector.getInstance(ObjectMapper.class);
     INDEX_IO = injector.getInstance(IndexIO.class);
     INDEX_MERGER = injector.getInstance(IndexMerger.class);
+    INDEX_MERGER_V9 = injector.getInstance(IndexMergerV9.class);
   }
 
   public static enum IndexJobCounters
@@ -351,6 +354,11 @@ public HadoopyShardSpec getShardSpec(Bucket bucket)
     return schema.getTuningConfig().getShardSpecs().get(bucket.time).get(bucket.partitionNum);
   }
 
+  public boolean isBuildV9Directly()
+  {
+    return schema.getTuningConfig().getBuildV9Directly();
+  }
+
   /**
    * Job instance should have Configuration set (by calling {@link #addJobProperties(Job)}
    * or via injected system properties) before this method is called.  The {@link PathSpec} may

indexing-hadoop/src/main/java/io/druid/indexer/HadoopTuningConfig.java

@@ -42,6 +42,7 @@ public class HadoopTuningConfig implements TuningConfig
   private static final IndexSpec DEFAULT_INDEX_SPEC = new IndexSpec();
   private static final int DEFAULT_ROW_FLUSH_BOUNDARY = 80000;
   private static final boolean DEFAULT_USE_COMBINER = false;
+  private static final Boolean DEFAULT_BUILD_V9_DIRECTLY = Boolean.FALSE;
 
   public static HadoopTuningConfig makeDefaultTuningConfig()
   {
@@ -59,7 +60,8 @@ public static HadoopTuningConfig makeDefaultTuningConfig()
         null,
         false,
         false,
-        null
+        null,
+        DEFAULT_BUILD_V9_DIRECTLY
     );
   }
 
@@ -76,6 +78,7 @@ public static HadoopTuningConfig makeDefaultTuningConfig()
   private final Map<String, String> jobProperties;
   private final boolean combineText;
   private final boolean useCombiner;
+  private final Boolean buildV9Directly;
 
   @JsonCreator
   public HadoopTuningConfig(
@@ -93,7 +96,8 @@ public HadoopTuningConfig(
       final @JsonProperty("combineText") boolean combineText,
       final @JsonProperty("useCombiner") Boolean useCombiner,
       // See https://github.com/druid-io/druid/pull/1922
-      final @JsonProperty("rowFlushBoundary") Integer maxRowsInMemoryCOMPAT
+      final @JsonProperty("rowFlushBoundary") Integer maxRowsInMemoryCOMPAT,
+      final @JsonProperty("buildV9Directly") Boolean buildV9Directly
   )
   {
     this.workingPath = workingPath;
@@ -111,6 +115,7 @@ public HadoopTuningConfig(
                           : ImmutableMap.copyOf(jobProperties));
     this.combineText = combineText;
     this.useCombiner = useCombiner == null ? DEFAULT_USE_COMBINER : useCombiner.booleanValue();
+    this.buildV9Directly = buildV9Directly == null ? DEFAULT_BUILD_V9_DIRECTLY : buildV9Directly;
   }
 
   @JsonProperty
@@ -191,6 +196,11 @@ public boolean getUseCombiner()
     return useCombiner;
   }
 
+  @JsonProperty
+  public Boolean getBuildV9Directly() {
+    return buildV9Directly;
+  }
+
   public HadoopTuningConfig withWorkingPath(String path)
   {
     return new HadoopTuningConfig(
@@ -207,7 +217,8 @@ public HadoopTuningConfig withWorkingPath(String path)
         jobProperties,
         combineText,
         useCombiner,
-        null
+        null,
+        buildV9Directly
     );
   }
 
@@ -227,7 +238,8 @@ public HadoopTuningConfig withVersion(String ver)
         jobProperties,
         combineText,
         useCombiner,
-        null
+        null,
+        buildV9Directly
     );
   }
 
@@ -247,7 +259,8 @@ public HadoopTuningConfig withShardSpecs(Map<DateTime, List<HadoopyShardSpec>> s
         jobProperties,
         combineText,
         useCombiner,
-        null
+        null,
+        buildV9Directly
     );
   }
 }

indexing-hadoop/src/main/java/io/druid/indexer/IndexGeneratorJob.java

@@ -454,6 +454,7 @@ protected ProgressIndicator makeProgressIndicator(final Context context)
         @Override
         public void progress()
         {
+          super.progress();
           context.progress();
         }
       };
@@ -466,9 +467,15 @@ private File persist(
         final ProgressIndicator progressIndicator
     ) throws IOException
     {
-      return HadoopDruidIndexerConfig.INDEX_MERGER.persist(
-          index, interval, file, null, config.getIndexSpec(), progressIndicator
-      );
+      if (config.isBuildV9Directly()) {
+        return HadoopDruidIndexerConfig.INDEX_MERGER_V9.persist(
+            index, interval, file, null, config.getIndexSpec(), progressIndicator
+        );
+      } else {
+        return HadoopDruidIndexerConfig.INDEX_MERGER.persist(
+            index, interval, file, null, config.getIndexSpec(), progressIndicator
+        );
+      }
     }
 
     protected File mergeQueryableIndex(
@@ -478,9 +485,15 @@ protected File mergeQueryableIndex(
         ProgressIndicator progressIndicator
     ) throws IOException
     {
-      return HadoopDruidIndexerConfig.INDEX_MERGER.mergeQueryableIndex(
-          indexes, aggs, file, config.getIndexSpec(), progressIndicator
-      );
+      if (config.isBuildV9Directly()) {
+        return HadoopDruidIndexerConfig.INDEX_MERGER_V9.mergeQueryableIndex(
+            indexes, aggs, file, config.getIndexSpec(), progressIndicator
+        );
+      } else {
+        return HadoopDruidIndexerConfig.INDEX_MERGER.mergeQueryableIndex(
+            indexes, aggs, file, config.getIndexSpec(), progressIndicator
+        );
+      }
     }
 
     @Override
@@ -586,7 +599,7 @@ protected void reduce(
             indexes.add(HadoopDruidIndexerConfig.INDEX_IO.loadIndex(file));
           }
           mergedBase = mergeQueryableIndex(
-                        indexes, aggregators, new File(baseFlushFile, "merged"), progressIndicator
+              indexes, aggregators, new File(baseFlushFile, "merged"), progressIndicator
           );
         }
         final FileSystem outputFS = new Path(config.getSchema().getIOConfig().getSegmentOutputPath())

indexing-hadoop/src/test/java/io/druid/indexer/BatchDeltaIngestionTest.java

@@ -381,6 +381,7 @@ private HadoopDruidIndexerConfig makeHadoopDruidIndexerConfig(Map<String, Object
                 null,
                 false,
                 false,
+                null,
                 null
             )
         )

indexing-hadoop/src/test/java/io/druid/indexer/DetermineHashedPartitionsJobTest.java

@@ -160,6 +160,7 @@ public DetermineHashedPartitionsJobTest(String dataFilePath, long targetPartitio
             null,
             false,
             false,
+            null,
             null
         )
     );

indexing-hadoop/src/test/java/io/druid/indexer/DeterminePartitionsJobTest.java

@@ -264,6 +264,7 @@ public DeterminePartitionsJobTest(
                 null,
                 false,
                 false,
+                null,
                 null
             )
         )

indexing-hadoop/src/test/java/io/druid/indexer/HadoopDruidIndexerConfigTest.java

@@ -207,6 +207,7 @@ public void testHashedBucketSelection()
             null,
             false,
             false,
+            null,
             null
         )
     );

indexing-hadoop/src/test/java/io/druid/indexer/HadoopTuningConfigTest.java

@@ -53,6 +53,7 @@ public void testSerde() throws Exception
         null,
         true,
         true,
+        null,
         null
     );
 

indexing-hadoop/src/test/java/io/druid/indexer/IndexGeneratorJobTest.java

@@ -395,6 +395,7 @@ public void setUp() throws Exception
                 ImmutableMap.of(JobContext.NUM_REDUCES, "0"), //verifies that set num reducers is ignored
                 false,
                 useCombiner,
+                null,
                 null
             )
         )

indexing-hadoop/src/test/java/io/druid/indexer/JobHelperTest.java

@@ -115,6 +115,7 @@ public void setup() throws Exception
                 ),
                 false,
                 false,
+                null,
                 null
             )
         )

indexing-hadoop/src/test/java/io/druid/indexer/path/GranularityPathSpecTest.java

@@ -120,7 +120,7 @@ public void testAddInputPath() throws Exception
             jsonMapper
         ),
         new HadoopIOConfig(null, null, null),
-        new HadoopTuningConfig(null, null, null, null, null, null, false, false, false, false, null, false, false, null)
+        new HadoopTuningConfig(null, null, null, null, null, null, false, false, false, false, null, false, false, null, null)
     );
 
     granularityPathSpec.setDataGranularity(Granularity.HOUR);

indexing-hadoop/src/test/java/io/druid/indexer/updater/HadoopConverterJobTest.java

@@ -201,6 +201,7 @@ public InputStream openStream() throws IOException
                 null,
                 false,
                 false,
+                null,
                 null
             )
         )

indexing-service/src/main/java/io/druid/indexing/common/TaskToolbox.java

@@ -37,6 +37,7 @@
 import io.druid.query.QueryRunnerFactoryConglomerate;
 import io.druid.segment.IndexIO;
 import io.druid.segment.IndexMerger;
+import io.druid.segment.IndexMergerV9;
 import io.druid.segment.loading.DataSegmentArchiver;
 import io.druid.segment.loading.DataSegmentKiller;
 import io.druid.segment.loading.DataSegmentMover;
@@ -80,7 +81,7 @@ public class TaskToolbox
   private final IndexIO indexIO;
   private final Cache cache;
   private final CacheConfig cacheConfig;
-
+  private final IndexMergerV9 indexMergerV9;
 
   public TaskToolbox(
       TaskConfig config,
@@ -102,7 +103,8 @@ public TaskToolbox(
       IndexMerger indexMerger,
       IndexIO indexIO,
       Cache cache,
-      CacheConfig cacheConfig
+      CacheConfig cacheConfig,
+      IndexMergerV9 indexMergerV9
   )
   {
     this.config = config;
@@ -125,6 +127,7 @@ public TaskToolbox(
     this.indexIO = Preconditions.checkNotNull(indexIO, "Null IndexIO");
     this.cache = cache;
     this.cacheConfig = cacheConfig;
+    this.indexMergerV9 = Preconditions.checkNotNull(indexMergerV9, "Null IndexMergerV9");
   }
 
   public TaskConfig getConfig()
@@ -247,4 +250,8 @@ public CacheConfig getCacheConfig()
   {
     return cacheConfig;
   }
+
+  public IndexMergerV9 getIndexMergerV9() {
+    return indexMergerV9;
+  }
 }