One granularity (apache#3850)

* Refactor Segment Granularity * Beginning of one granularity * Copy the fix for custom periods in segment-grunalrity over here. * Remove the custom serialization for now. * Compilation cleanup * Reformat code * Fixing unit tests * Unify to use a single iterable * Backward compatibility for rolling upgrade * Minor check style. Cosmetic changes. * Rename length and millis to duration * CR feedback * Minor changes.
Gamehu · Feb 25, 2017 · c3bf401 · c3bf401
1 parent 58b704c
commit c3bf401
Show file tree

Hide file tree

Showing 214 changed files with 2,597 additions and 3,294 deletions.
diff --git a/api/src/main/java/io/druid/timeline/DataSegment.java b/api/src/main/java/io/druid/timeline/DataSegment.java
@@ -34,7 +34,7 @@
 
 import io.druid.jackson.CommaListJoinDeserializer;
 import io.druid.jackson.CommaListJoinSerializer;
-import io.druid.java.util.common.Granularity;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.query.SegmentDescriptor;
 import io.druid.timeline.partition.NoneShardSpec;
 import io.druid.timeline.partition.ShardSpec;
@@ -288,8 +288,8 @@ public int compare(DataSegment lhs, DataSegment rhs)
       {
         int retVal;
 
-        DateTime lhsMonth = Granularity.MONTH.truncate(lhs.getInterval().getStart());
-        DateTime rhsMonth = Granularity.MONTH.truncate(rhs.getInterval().getStart());
+        DateTime lhsMonth = Granularity.MONTH.bucketStart(lhs.getInterval().getStart());
+        DateTime rhsMonth = Granularity.MONTH.bucketStart(rhs.getInterval().getStart());
 
         retVal = lhsMonth.compareTo(rhsMonth);
 

diff --git a/benchmarks/src/main/java/io/druid/benchmark/FilterPartitionBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/FilterPartitionBenchmark.java
@@ -30,9 +30,9 @@
 import io.druid.benchmark.datagen.BenchmarkSchemas;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.jackson.DefaultObjectMapper;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.guava.Sequence;
 import io.druid.java.util.common.guava.Sequences;
 import io.druid.java.util.common.logger.Logger;
@@ -230,7 +230,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .build(),
@@ -508,7 +508,7 @@ public void readComplexOrFilterCNF(Blackhole blackhole) throws Exception
 
   private Sequence<Cursor> makeCursors(StorageAdapter sa, Filter filter)
   {
-    return sa.makeCursors(filter, schemaInfo.getDataInterval(), VirtualColumns.EMPTY, QueryGranularities.ALL, false);
+    return sa.makeCursors(filter, schemaInfo.getDataInterval(), VirtualColumns.EMPTY, Granularity.ALL, false);
   }
 
   private Sequence<List<String>> readCursors(Sequence<Cursor> cursors, final Blackhole blackhole)

diff --git a/benchmarks/src/main/java/io/druid/benchmark/FilteredAggregatorBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/FilteredAggregatorBenchmark.java
@@ -29,9 +29,9 @@
 import io.druid.benchmark.query.QueryBenchmarkUtil;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.jackson.DefaultObjectMapper;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.guava.Sequence;
 import io.druid.java.util.common.guava.Sequences;
 import io.druid.java.util.common.logger.Logger;
@@ -214,7 +214,7 @@ public void setup() throws IOException
 
     query = Druids.newTimeseriesQueryBuilder()
                   .dataSource("blah")
-                  .granularity(QueryGranularities.ALL)
+                  .granularity(Granularity.ALL)
                   .intervals(intervalSpec)
                   .aggregators(queryAggs)
                   .descending(false)
@@ -231,7 +231,7 @@ private IncrementalIndex makeIncIndex(AggregatorFactory[] metrics)
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(metrics)
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .build(),

diff --git a/benchmarks/src/main/java/io/druid/benchmark/GroupByTypeInterfaceBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/GroupByTypeInterfaceBenchmark.java
@@ -37,10 +37,9 @@
 import io.druid.data.input.InputRow;
 import io.druid.data.input.Row;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
-import io.druid.granularity.QueryGranularity;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.jackson.DefaultObjectMapper;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.guava.Sequence;
 import io.druid.java.util.common.guava.Sequences;
 import io.druid.java.util.common.logger.Logger;
@@ -190,7 +189,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       GroupByQuery queryLongFloat = GroupByQuery
@@ -204,7 +203,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       GroupByQuery queryLong = GroupByQuery
@@ -217,7 +216,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       GroupByQuery queryFloat = GroupByQuery
@@ -230,7 +229,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       basicQueries.put("string", queryString);
@@ -258,7 +257,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularities.DAY)
+          .setGranularity(Granularity.DAY)
           .build();
 
       GroupByQuery queryA = GroupByQuery
@@ -271,7 +270,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularities.WEEK)
+          .setGranularity(Granularity.WEEK)
           .build();
 
       basicQueries.put("nested", queryA);
@@ -435,7 +434,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .build(),

diff --git a/benchmarks/src/main/java/io/druid/benchmark/IncrementalIndexRowTypeBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/IncrementalIndexRowTypeBenchmark.java
@@ -22,8 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.MapBasedInputRow;
-import io.druid.granularity.QueryGranularities;
-
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.query.aggregation.AggregatorFactory;
 import io.druid.query.aggregation.CountAggregatorFactory;
 import io.druid.query.aggregation.DoubleSumAggregatorFactory;
@@ -123,7 +122,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         0,
-        QueryGranularities.NONE,
+        Granularity.NONE,
         aggs,
         false,
         false,

diff --git a/benchmarks/src/main/java/io/druid/benchmark/TopNTypeInterfaceBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/TopNTypeInterfaceBenchmark.java
@@ -31,9 +31,9 @@
 import io.druid.concurrent.Execs;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.jackson.DefaultObjectMapper;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.guava.Sequence;
 import io.druid.java.util.common.guava.Sequences;
 import io.druid.java.util.common.logger.Logger;
@@ -171,7 +171,7 @@ private void setupQueries()
       // Use an IdentityExtractionFn to force usage of DimExtractionTopNAlgorithm
       TopNQueryBuilder queryBuilderString = new TopNQueryBuilder()
           .dataSource("blah")
-          .granularity(QueryGranularities.ALL)
+          .granularity(Granularity.ALL)
           .dimension(new ExtractionDimensionSpec("dimSequential", "dimSequential", IdentityExtractionFn.getInstance()))
           .metric("sumFloatNormal")
           .intervals(intervalSpec)
@@ -180,15 +180,15 @@ private void setupQueries()
       // DimExtractionTopNAlgorithm is always used for numeric columns
       TopNQueryBuilder queryBuilderLong = new TopNQueryBuilder()
           .dataSource("blah")
-          .granularity(QueryGranularities.ALL)
+          .granularity(Granularity.ALL)
           .dimension("metLongUniform")
           .metric("sumFloatNormal")
           .intervals(intervalSpec)
           .aggregators(queryAggs);
 
       TopNQueryBuilder queryBuilderFloat = new TopNQueryBuilder()
           .dataSource("blah")
-          .granularity(QueryGranularities.ALL)
+          .granularity(Granularity.ALL)
           .dimension("metFloatNormal")
           .metric("sumFloatNormal")
           .intervals(intervalSpec)
@@ -206,7 +206,7 @@ private void setupQueries()
 
       TopNQueryBuilder queryBuilderA = new TopNQueryBuilder()
           .dataSource("blah")
-          .granularity(QueryGranularities.ALL)
+          .granularity(Granularity.ALL)
           .dimension("dimUniform")
           .metric(new DimensionTopNMetricSpec(null, StringComparators.NUMERIC))
           .intervals(intervalSpec)
@@ -222,7 +222,7 @@ private void setupQueries()
 
       TopNQueryBuilder queryBuilderA = new TopNQueryBuilder()
           .dataSource("blah")
-          .granularity(QueryGranularities.ALL)
+          .granularity(Granularity.ALL)
           .dimension("dimUniform")
           .metric(new DimensionTopNMetricSpec(null, StringComparators.ALPHANUMERIC))
           .intervals(intervalSpec)
@@ -316,7 +316,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .build(),

diff --git a/benchmarks/src/main/java/io/druid/benchmark/indexing/IncrementalIndexReadBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/indexing/IncrementalIndexReadBenchmark.java
@@ -25,8 +25,8 @@
 import io.druid.benchmark.datagen.BenchmarkSchemas;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
 import io.druid.hll.HyperLogLogHash;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.guava.Sequence;
 import io.druid.java.util.common.guava.Sequences;
 import io.druid.java.util.common.logger.Logger;
@@ -125,7 +125,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .withRollup(rollup)
@@ -203,7 +203,7 @@ private Sequence<Cursor> makeCursors(IncrementalIndexStorageAdapter sa, DimFilte
         filter.toFilter(),
         schemaInfo.getDataInterval(),
         VirtualColumns.EMPTY,
-        QueryGranularities.ALL,
+        Granularity.ALL,
         false
     );
   }

diff --git a/benchmarks/src/main/java/io/druid/benchmark/indexing/IndexIngestionBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/indexing/IndexIngestionBenchmark.java
@@ -24,7 +24,7 @@
 import io.druid.benchmark.datagen.BenchmarkSchemas;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.query.aggregation.hyperloglog.HyperUniquesSerde;
@@ -106,7 +106,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .withRollup(rollup)

diff --git a/benchmarks/src/main/java/io/druid/benchmark/indexing/IndexMergeBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/indexing/IndexMergeBenchmark.java
@@ -26,9 +26,9 @@
 import io.druid.benchmark.datagen.BenchmarkSchemas;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.jackson.DefaultObjectMapper;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.query.aggregation.hyperloglog.HyperUniquesSerde;
 import io.druid.segment.IndexIO;
@@ -163,7 +163,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .withRollup(rollup)

diff --git a/benchmarks/src/main/java/io/druid/benchmark/indexing/IndexPersistBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/indexing/IndexPersistBenchmark.java
@@ -26,9 +26,9 @@
 import io.druid.benchmark.datagen.BenchmarkSchemas;
 import io.druid.data.input.InputRow;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.jackson.DefaultObjectMapper;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.logger.Logger;
 import io.druid.query.aggregation.hyperloglog.HyperUniquesSerde;
 import io.druid.segment.IndexIO;
@@ -157,7 +157,7 @@ private IncrementalIndex makeIncIndex()
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .withRollup(rollup)

diff --git a/benchmarks/src/main/java/io/druid/benchmark/query/GroupByBenchmark.java b/benchmarks/src/main/java/io/druid/benchmark/query/GroupByBenchmark.java
@@ -37,10 +37,9 @@
 import io.druid.data.input.InputRow;
 import io.druid.data.input.Row;
 import io.druid.data.input.impl.DimensionsSpec;
-import io.druid.granularity.QueryGranularities;
-import io.druid.granularity.QueryGranularity;
 import io.druid.hll.HyperLogLogHash;
 import io.druid.jackson.DefaultObjectMapper;
+import io.druid.java.util.common.granularity.Granularity;
 import io.druid.java.util.common.guava.Sequence;
 import io.druid.java.util.common.guava.Sequences;
 import io.druid.java.util.common.logger.Logger;
@@ -195,7 +194,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       basicQueries.put("A", queryA);
@@ -220,7 +219,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularities.DAY)
+          .setGranularity(Granularity.DAY)
           .build();
 
       GroupByQuery queryA = GroupByQuery
@@ -233,7 +232,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularities.WEEK)
+          .setGranularity(Granularity.WEEK)
           .build();
 
       basicQueries.put("nested", queryA);
@@ -262,7 +261,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       simpleQueries.put("A", queryA);
@@ -289,7 +288,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       simpleLongQueries.put("A", queryA);
@@ -316,7 +315,7 @@ private void setupQueries()
           .setAggregatorSpecs(
               queryAggs
           )
-          .setGranularity(QueryGranularity.fromString(queryGranularity))
+          .setGranularity(Granularity.fromString(queryGranularity))
           .build();
 
       simpleFloatQueries.put("A", queryA);
@@ -478,7 +477,7 @@ private IncrementalIndex makeIncIndex(boolean withRollup)
   {
     return new OnheapIncrementalIndex(
         new IncrementalIndexSchema.Builder()
-            .withQueryGranularity(QueryGranularities.NONE)
+            .withQueryGranularity(Granularity.NONE)
             .withMetrics(schemaInfo.getAggsArray())
             .withDimensionsSpec(new DimensionsSpec(null, null, null))
             .withRollup(withRollup)