Add support for querying HLLSketch types (Netflix-Skunkworks#454)

These represent approximate distinct counts. We can treat them as
longsum/counters. Usage will be nuanced, we'll probably need to
ensure callers are performing per-step queries for the results to make
sense.

atlas-druid/src/main/scala/com/netflix/atlas/druid/DruidClient.scala

@@ -251,7 +251,12 @@ object DruidClient {
 
  case class Metric(name: String, dataType: String = "DOUBLE") {
 
- def isCounter: Boolean = dataType == "DOUBLE" || dataType == "FLOAT" || dataType == "LONG"
+ def isSketch: Boolean = {
+ dataType == "HLLSketch"
+ }
+
+ def isCounter: Boolean =
+ dataType == "DOUBLE" || dataType == "FLOAT" || dataType == "LONG" || isSketch
 
  def isTimer: Boolean = {
  dataType == "spectatorHistogramTimer"
@@ -512,6 +517,7 @@ object DruidClient {
  def sum(fieldName: String): Aggregation = Aggregation("doubleSum", fieldName)
  def min(fieldName: String): Aggregation = Aggregation("doubleMin", fieldName)
  def max(fieldName: String): Aggregation = Aggregation("doubleMax", fieldName)
+ def distinct(fieldName: String): Aggregation = Aggregation("HLLSketchMerge", fieldName)
  def timer(fieldName: String): Aggregation = Aggregation("timer", fieldName)
  def distSummary(fieldName: String): Aggregation = Aggregation("dist-summary", fieldName)
  }

atlas-druid/src/main/scala/com/netflix/atlas/druid/DruidDatabaseActor.scala

@@ -439,6 +439,7 @@ object DruidDatabaseActor {
  case Histogram(_) if metric.isDistSummary => Aggregation.distSummary(metric.name)
  case DataExpr.GroupBy(e, _) => toAggregation(metric, e)
  case DataExpr.Consolidation(e, _) => toAggregation(metric, e)
+ case _ if metric.isSketch => Aggregation.distinct(metric.name)
  case _: DataExpr.Sum => Aggregation.sum(metric.name)
  case _: DataExpr.Max => Aggregation.max(metric.name)
  case _: DataExpr.Min => Aggregation.min(metric.name)
@@ -563,7 +564,8 @@ object DruidDatabaseActor {
  // performance. This filter cannot be used with the spectatorHistogram type in druid.
  val metricValueFilter = Query.Not(Query.Equal(name, "0")).and(Query.HasKey(name))
  val finalQueryWithFilter =
- if (m.metric.isHistogram) finalQuery else finalQuery.and(metricValueFilter)
+ if (m.metric.isHistogram || m.metric.isSketch) finalQuery
+ else finalQuery.and(metricValueFilter)
 
  val groupByQuery = GroupByQuery(
  dataSource = datasource,

atlas-druid/src/test/resources/segmentMetadataResponse.json

@@ -65,6 +65,17 @@
  "minValue": null,
  "maxValue": null,
  "errorMessage": null
+ },
+ "test.metric.hllsketch": {
+ "typeSignature": "COMPLEX<HLLSketch>",
+ "type": "HLLSketch",
+ "hasMultipleValues": false,
+ "hasNulls": true,
+ "size": 0,
+ "cardinality": null,
+ "minValue": null,
+ "maxValue": null,
+ "errorMessage": null
  }
  },
  "size": 158732338,

atlas-druid/src/test/scala/com/netflix/atlas/druid/DruidClientSuite.scala

@@ -153,7 +153,8 @@ class DruidClientSuite extends FunSuite {
  "test.dim.2",
  "test.metric.histogram.dist.1",
  "test.metric.histogram.dist.2",
- "test.metric.histogram.timer"
+ "test.metric.histogram.timer",
+ "test.metric.hllsketch"
  )
  assertEquals(columns.keySet, expected)
  }
@@ -165,6 +166,7 @@ class DruidClientSuite extends FunSuite {
  assertEquals(columns("test.metric.histogram.dist.1").`type`, "spectatorHistogram")
  assertEquals(columns("test.metric.histogram.dist.2").`type`, "spectatorHistogramDistribution")
  assertEquals(columns("test.metric.histogram.timer").`type`, "spectatorHistogramTimer")
+ assertEquals(columns("test.metric.hllsketch").`type`, "HLLSketch")
  assertEquals(columns("test.dim.1").`type`, "STRING")
  assertEquals(columns("test.dim.1").`type`, "STRING")
  }
@@ -177,6 +179,7 @@ class DruidClientSuite extends FunSuite {
  case "test.metric.histogram.dist.1" => assert(m.isDistSummary)
  case "test.metric.histogram.dist.2" => assert(m.isDistSummary)
  case "test.metric.histogram.timer" => assert(m.isTimer)
+ case "test.metric.hllsketch" => assert(m.isSketch)
  case name => throw new MatchError(name)
  }
  }
@@ -243,6 +246,13 @@ class DruidClientSuite extends FunSuite {
  assert(!json.contains("aggrType"))
  }
 
+ test("aggregation encode, distinct type") {
+ val aggr = Aggregation.distinct("foo")
+ val json = Json.encode(aggr)
+ assert(json.contains("HLLSketchMerge"))
+ assert(!json.contains("aggrType"))
+ }
+
  test("aggregation encode, doubleSum type") {
  val aggr = Aggregation.sum("foo")
  val json = Json.encode(aggr)