support rollup function in druid 0.9.2 (#210)

* support rollup function in druid 0.9.2

* give isRollup a default value; use NoopEmitter for this dummy emitter

Author: GavinGavinNo1

build.sbt

@@ -14,7 +14,7 @@ val jacksonOneVersion = "1.9.13"
 // See https://github.com/druid-io/druid/pull/1669, https://github.com/druid-io/tranquility/pull/81 before upgrading Jackson
 val jacksonTwoVersion = "2.4.6"
 val jacksonTwoModuleScalaVersion = "2.4.5"
-val druidVersion = "0.9.1"
+val druidVersion = "0.9.2"
 val guiceVersion = "4.0"
 val flinkVersion = "1.0.3"
 val finagleVersion = "6.31.0"

core/src/main/scala/com/metamx/tranquility/druid/DruidBeamMaker.scala

@@ -88,7 +88,8 @@ class DruidBeamMaker[A](
       "granularitySpec" -> Map(
         "type" -> "uniform",
         "segmentGranularity" -> beamTuning.segmentGranularity,
-        "queryGranularity" -> queryGranularityMap
+        "queryGranularity" -> queryGranularityMap,
+        "rollup" -> rollup.isRollup
       )
     )
     val ioConfigMap = Map(

core/src/main/scala/com/metamx/tranquility/druid/DruidBeams.scala

@@ -87,12 +87,13 @@ import scala.reflect.runtime.universe.typeTag
   * val dataSource = "foo"
   * val dimensions = Seq("bar")
   * val aggregators = Seq(new LongSumAggregatorFactory("baz", "baz"))
+  * val isRollup = true
   * val sender = DruidBeams
   *   .builder[Map[String, Any]](eventMap => new DateTime(eventMap("timestamp")))
   *   .curator(curator)
   *   .discoveryPath("/test/discovery")
   *   .location(DruidLocation(new DruidEnvironment("druid:local:indexer", "druid:local:firehose:%s"), dataSource))
-  *   .rollup(DruidRollup(dimensions, aggregators, QueryGranularities.MINUTE))
+  *   .rollup(DruidRollup(dimensions, aggregators, QueryGranularities.MINUTE, isRollup))
   *   .tuning(new ClusteredBeamTuning(Granularity.HOUR, 10.minutes, 1, 1))
   *   .buildTranquilizer()
   * val future = sender.send(Map("timestamp" -> "2010-01-02T03:04:05.678Z", "bar" -> "hey", "baz" -> 3))
@@ -336,7 +337,8 @@ object DruidBeams
           )
       },
       aggregators = fireDepartment.getDataSchema.getAggregators,
-      indexGranularity = fireDepartment.getDataSchema.getGranularitySpec.getQueryGranularity
+      indexGranularity = fireDepartment.getDataSchema.getGranularitySpec.getQueryGranularity,
+      isRollup = fireDepartment.getDataSchema.getGranularitySpec.isRollup
     )
     builder(inputFnFn(rollup, mkparser, timestampSpec), timestamperFn(timestampSpec))
       .curatorFactory(

core/src/main/scala/com/metamx/tranquility/druid/DruidRollup.scala

@@ -35,7 +35,8 @@ import scala.collection.JavaConverters._
 class DruidRollup(
   val dimensions: DruidDimensions,
   val aggregators: IndexedSeq[AggregatorFactory],
-  val indexGranularity: QueryGranularity
+  val indexGranularity: QueryGranularity,
+  val isRollup: Boolean = true
 )
 {
   private val additionalExclusions: Set[String] = {
@@ -176,10 +177,11 @@ object DruidRollup
   def apply(
     dimensions: DruidDimensions,
     aggregators: Seq[AggregatorFactory],
-    indexGranularity: QueryGranularity
+    indexGranularity: QueryGranularity,
+    isRollup: Boolean
   ) =
   {
-    new DruidRollup(dimensions, aggregators.toIndexedSeq, indexGranularity)
+    new DruidRollup(dimensions, aggregators.toIndexedSeq, indexGranularity, isRollup)
   }
 
   /**
@@ -192,13 +194,15 @@ object DruidRollup
   def create(
     dimensions: DruidDimensions,
     aggregators: java.util.List[AggregatorFactory],
-    indexGranularity: QueryGranularity
+    indexGranularity: QueryGranularity,
+    isRollup: Boolean
   ): DruidRollup =
   {
     new DruidRollup(
       dimensions,
       aggregators.asScala.toIndexedSeq,
-      indexGranularity
+      indexGranularity,
+      isRollup
     )
   }
 
@@ -208,13 +212,15 @@ object DruidRollup
   def create(
     dimensions: java.util.List[String],
     aggregators: java.util.List[AggregatorFactory],
-    indexGranularity: QueryGranularity
+    indexGranularity: QueryGranularity,
+    isRollup: Boolean
   ): DruidRollup =
   {
     new DruidRollup(
       SpecificDruidDimensions(dimensions.asScala, Vector.empty),
       aggregators.asScala.toIndexedSeq,
-      indexGranularity
+      indexGranularity,
+      isRollup
     )
   }
 }

core/src/test/java/com/metamx/tranquility/javatests/JavaApiTest.java

@@ -52,7 +52,8 @@ public void testSpecificDimensionsRollupConfiguration() throws Exception
     final DruidRollup rollup = DruidRollup.create(
         DruidDimensions.specific(dimensions),
         aggregators,
-        QueryGranularities.MINUTE
+        QueryGranularities.MINUTE,
+        true
     );
     Assert.assertTrue(rollup.dimensions() instanceof SpecificDruidDimensions);
     Assert.assertEquals("column", ((SpecificDruidDimensions) rollup.dimensions()).dimensions().iterator().next());
@@ -64,7 +65,8 @@ public void testSchemalessDimensionsRollupConfiguration() throws Exception
     final DruidRollup rollup = DruidRollup.create(
         DruidDimensions.schemaless(),
         aggregators,
-        QueryGranularities.MINUTE
+        QueryGranularities.MINUTE,
+        true
     );
     Assert.assertTrue(rollup.dimensions() instanceof SchemalessDruidDimensions);
     Assert.assertEquals(0, ((SchemalessDruidDimensions) rollup.dimensions()).dimensionExclusions().size());
@@ -76,7 +78,8 @@ public void testSchemalessDimensionsWithExclusionsRollupConfiguration() throws E
     final DruidRollup rollup = DruidRollup.create(
         DruidDimensions.schemalessWithExclusions(dimensions),
         aggregators,
-        QueryGranularities.MINUTE
+        QueryGranularities.MINUTE,
+        true
     );
     Assert.assertTrue(rollup.dimensions() instanceof SchemalessDruidDimensions);
     Assert.assertEquals("column", ((SchemalessDruidDimensions) rollup.dimensions()).dimensionExclusions().iterator().next());
@@ -96,7 +99,8 @@ public void testSchemalessDimensionsWithExclusionsAndSpatialDimensionsRollupConf
                            )
                        ),
         aggregators,
-        QueryGranularities.MINUTE
+        QueryGranularities.MINUTE,
+        true
     );
     Assert.assertTrue(rollup.dimensions() instanceof SchemalessDruidDimensions);
     Assert.assertEquals("column", ((SchemalessDruidDimensions) rollup.dimensions()).dimensionExclusions().iterator().next());

core/src/test/resources/tranquility-core.json

@@ -37,7 +37,8 @@
           "granularitySpec" : {
             "queryGranularity" : "none",
             "type" : "uniform",
-            "segmentGranularity" : "hour"
+            "segmentGranularity" : "hour",
+            "rollup": false
           },
           "dataSource" : "foo"
         },

core/src/test/resources/tranquility-core.yaml

@@ -21,6 +21,7 @@ dataSources:
           type: uniform
           segmentGranularity: hour
           queryGranularity: none
+          rollup: false
 
       ioConfig:
         type: realtime

core/src/test/scala/com/metamx/tranquility/test/DirectDruidTest.scala

@@ -96,7 +96,8 @@ object DirectDruidTest
         Vector(MultipleFieldDruidSpatialDimension("coord.geo", Seq("lat", "lon")))
       ),
       IndexedSeq(new LongSumAggregatorFactory("barr", "bar")),
-      QueryGranularities.MINUTE
+      QueryGranularities.MINUTE,
+      true
     )
     val druidEnvironment = new DruidEnvironment(
       "druid/tranquility/indexer" /* Slashes should be converted to colons */ ,

core/src/test/scala/com/metamx/tranquility/test/DruidBeamTest.scala

@@ -20,14 +20,11 @@
 package com.metamx.tranquility.test
 
 import com.fasterxml.jackson.databind
-import com.fasterxml.jackson.databind.DeserializationContext
-import com.fasterxml.jackson.databind.InjectableValues
-import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.{ObjectReader, DeserializationContext, InjectableValues, ObjectMapper}
 import com.google.inject.Key
 import com.metamx.common.Granularity
 import com.metamx.common.scala.untyped.Dict
-import com.metamx.emitter.core.Emitter
-import com.metamx.emitter.core.Event
+import com.metamx.emitter.core.NoopEmitter
 import com.metamx.emitter.service.ServiceEmitter
 import com.metamx.tranquility.beam.ClusteredBeamTuning
 import com.metamx.tranquility.druid.DruidBeamConfig
@@ -56,6 +53,30 @@ import _root_.scala.collection.JavaConverters._
 
 class DruidBeamTest extends FunSuite with Matchers
 {
+  private def emptyEmitter(): ServiceEmitter = new ServiceEmitter(
+    "service", "host", new NoopEmitter()
+  )
+
+  private def defaultObjectReader(): ObjectReader = DruidGuicer.Default.objectMapper.reader(
+    new InjectableValues
+    {
+      override def findInjectableValue(
+          valueId: Any,
+          ctxt: DeserializationContext,
+          forProperty: databind.BeanProperty,
+          beanInstance: scala.Any
+        ): AnyRef =
+      {
+        valueId match {
+          case k: Key[_] if k.getTypeLiteral.getRawType == classOf[ChatHandlerProvider] => new NoopChatHandlerProvider
+          case k: Key[_] if k.getTypeLiteral.getRawType == classOf[ObjectMapper] => DruidGuicer.Default.objectMapper
+          case k: Key[_] if k.getTypeLiteral.getRawType == classOf[EventReceiverFirehoseRegister] =>
+            new EventReceiverFirehoseRegister
+        }
+      }
+    }
+  ).withType(classOf[Task])
+
   test("GenerateAvailabilityGroup") {
     val dt = new DateTime("2010-02-03T04:34:56.789", DateTimeZone.forID("America/Los_Angeles"))
     assert(DruidBeamMaker.generateAvailabilityGroup("x", dt, 1) === "x-2010-02-03T12:34:56.789Z-0001")
@@ -123,23 +144,13 @@ class DruidBeamTest extends FunSuite with Matchers
       DruidRollup(
         dimensions = SpecificDruidDimensions(Seq("dim1", "dim2"), Seq(DruidSpatialDimension.singleField("spatial1"))),
         aggregators = Seq(new LongSumAggregatorFactory("met1", "met1")),
-        indexGranularity = QueryGranularities.MINUTE
+        indexGranularity = QueryGranularities.MINUTE,
+        true
       ),
       new TimestampSpec("ts", "iso", null),
       null,
       null,
-      new ServiceEmitter(
-        "service", "host", new Emitter
-        {
-          override def flush(): Unit = ???
-
-          override def emit(event: Event): Unit = ???
-
-          override def close(): Unit = ???
-
-          override def start(): Unit = ???
-        }
-      ),
+      emptyEmitter(),
       null,
       DruidGuicer.Default.objectMapper
     )
@@ -151,27 +162,7 @@ class DruidBeamTest extends FunSuite with Matchers
       1,
       2
     )
-    val objectReader = DruidGuicer.Default.objectMapper.reader(
-      new InjectableValues
-      {
-        override def findInjectableValue(
-          valueId: Any,
-          ctxt: DeserializationContext,
-          forProperty: databind.BeanProperty,
-          beanInstance: scala.Any
-        ): AnyRef =
-        {
-          valueId match {
-            case k: Key[_] if k.getTypeLiteral.getRawType == classOf[ChatHandlerProvider] => new NoopChatHandlerProvider
-            case k: Key[_] if k.getTypeLiteral.getRawType == classOf[ObjectMapper] => DruidGuicer.Default.objectMapper
-            case k: Key[_] if k.getTypeLiteral.getRawType == classOf[EventReceiverFirehoseRegister] =>
-              new EventReceiverFirehoseRegister
-          }
-        }
-      }
-    ).withType(classOf[Task])
-
-    val task = objectReader.readValue(taskBytes).asInstanceOf[RealtimeIndexTask]
+    val task = defaultObjectReader().readValue(taskBytes).asInstanceOf[RealtimeIndexTask]
     task.getId should be("index_realtime_mydatasource_2000-01-01T00:00:00.000Z_1_2")
     task.getDataSource should be("mydatasource")
     task.getTaskResource.getAvailabilityGroup should be("mygroup")
@@ -202,4 +193,39 @@ class DruidBeamTest extends FunSuite with Matchers
     parseSpec.getDimensionsSpec.getDimensions.asScala.map(_.getName) should be(Seq("dim1", "dim2", "spatial1"))
     parseSpec.getDimensionsSpec.getSpatialDimensions.asScala.map(_.getDimName) should be(Seq("spatial1"))
   }
+
+  test("Attribute isRollup should be passed to task") {
+    val isRollup = false
+    val druidBeamMaker = new DruidBeamMaker[Dict](
+      DruidBeamConfig(),
+      DruidLocation.create("druid/overlord", "mydatasource"),
+      ClusteredBeamTuning(),
+      DruidTuning().toMap,
+      DruidRollup(
+        dimensions = SpecificDruidDimensions(Seq(), Seq()),
+        aggregators = Seq(),
+        indexGranularity = QueryGranularities.NONE,
+        // isRollup is set for test.
+        isRollup
+      ),
+      new TimestampSpec("ts", "iso", null),
+      null,
+      null,
+      emptyEmitter(),
+      null,
+      DruidGuicer.Default.objectMapper
+    )
+
+    val interval = new Interval("2000/PT1H", ISOChronology.getInstanceUTC)
+    val taskBytes = druidBeamMaker.taskBytes(
+      interval,
+      "mygroup",
+      "myfirehose",
+      1,
+      2
+    )
+    val task = defaultObjectReader().readValue(taskBytes).asInstanceOf[RealtimeIndexTask]
+
+    task.getRealtimeIngestionSchema.getDataSchema.getGranularitySpec.isRollup should be(isRollup)
+  }
 }

core/src/test/scala/com/metamx/tranquility/test/DruidRollupTest.scala

@@ -36,7 +36,8 @@ class DruidRollupTest extends FunSuite with Matchers
     val rollup = DruidRollup(
       SpecificDruidDimensions(Vector("hey", "what"), Vector.empty),
       Seq(new CountAggregatorFactory("heyyo")),
-      QueryGranularities.NONE
+      QueryGranularities.NONE,
+      true
     )
     rollup.validate()
   }
@@ -46,7 +47,8 @@ class DruidRollupTest extends FunSuite with Matchers
       DruidRollup(
         SpecificDruidDimensions(Vector("hey", "what"), Vector.empty),
         Seq(new CountAggregatorFactory("hey")),
-        QueryGranularities.NONE
+        QueryGranularities.NONE,
+        true
       )
     }
     e.getMessage should be("Duplicate columns: hey")
@@ -57,7 +59,8 @@ class DruidRollupTest extends FunSuite with Matchers
       DruidRollup(
         SpecificDruidDimensions(Vector("what"), Vector.empty),
         Seq(new CountAggregatorFactory("hey"), new LongSumAggregatorFactory("hey", "blah")),
-        QueryGranularities.NONE
+        QueryGranularities.NONE,
+        true
       )
     }
     e.getMessage should be("Duplicate columns: hey")
@@ -68,7 +71,8 @@ class DruidRollupTest extends FunSuite with Matchers
       DruidRollup(
         SpecificDruidDimensions(Vector("what", "what"), Vector.empty),
         Seq(new CountAggregatorFactory("hey")),
-        QueryGranularities.NONE
+        QueryGranularities.NONE,
+        true
       )
     }
     e.getMessage should be("Duplicate columns: what")
@@ -78,7 +82,8 @@ class DruidRollupTest extends FunSuite with Matchers
     val rollup = DruidRollup(
       SpecificDruidDimensions(Vector("e", "f", "a", "b", "z", "t"), Vector.empty),
       Seq(new CountAggregatorFactory("hey")),
-      QueryGranularities.NONE
+      QueryGranularities.NONE,
+      true
     )
     rollup.dimensions.specMap.get("dimensions").asInstanceOf[java.util.List[String]].asScala should
       be(Seq("e", "f", "a", "b", "z", "t"))
@@ -88,7 +93,8 @@ class DruidRollupTest extends FunSuite with Matchers
     val rollup = DruidRollup(
       SpecificDruidDimensions(Seq("foo", "bar")),
       Seq(new LongSumAggregatorFactory("hey", "there")),
-      QueryGranularities.NONE
+      QueryGranularities.NONE,
+      true
     )
     val timestampSpec = new TimestampSpec("t", "auto", null)
     rollup.isStringDimension(timestampSpec, "t") should be(false)
@@ -104,7 +110,8 @@ class DruidRollupTest extends FunSuite with Matchers
     val rollup = DruidRollup(
       SchemalessDruidDimensions(Set("qux")),
       Seq(new LongSumAggregatorFactory("hey", "there")),
-      QueryGranularities.NONE
+      QueryGranularities.NONE,
+      true
     )
     val timestampSpec = new TimestampSpec("t", "auto", null)
     rollup.isStringDimension(timestampSpec, "t") should be(false)