implement special distinctcount

docs/content/development/extensions-contrib/distinctcount.md

@@ -0,0 +1,76 @@
+---
+layout: doc_page
+---
+
+# DistinctCount aggregator
+
+(1) First use single dimension hash-based partitioning to partition data by a dimension for example visitor_id, this to make sure all rows with a particular value for that dimension will go into the same segment or this might over count.
+(2) Second use distinctCount to calculate exact distinct count, make sure queryGranularity is divide exactly by segmentGranularity or else the result will be wrong.
+There is some limitations, when use with groupBy, the groupBy keys' numbers should not exceed maxIntermediateRows in every segment, if exceed the result will wrong. And when use with topN, numValuesPerPass should not too big, if too big the distinctCount will use many memory and cause the JVM out of service.
+
+This has been used in production.
+
+Example:
+# Timeseries Query
+
+```json
+{
+  "queryType": "timeseries",
+  "dataSource": "sample_datasource",
+  "granularity": "day",
+  "aggregations": [
+    {
+      "type": "distinctCount",
+      "name": "uv",
+      "fieldName": "visitor_id"
+    }
+  ],
+  "intervals": [
+    "2016-03-01T00:00:00.000/2013-03-20T00:00:00.000"
+  ]
+}
+```
+
+# TopN Query
+
+```json
+{
+  "queryType": "topN",
+  "dataSource": "sample_datasource",
+  "dimension": "sample_dim",
+  "threshold": 5,
+  "metric": "uv",
+  "granularity": "all",
+  "aggregations": [
+    {
+      "type": "distinctCount",
+      "name": "uv",
+      "fieldName": "visitor_id"
+    }
+  ],
+  "intervals": [
+    "2016-03-06T00:00:00/2016-03-06T23:59:59"
+  ]
+}
+```
+
+# GroupBy Query
+
+```json
+{
+  "queryType": "groupBy",
+  "dataSource": "sample_datasource",
+  "dimensions": "[sample_dim]",
+  "granularity": "all",
+  "aggregations": [
+    {
+      "type": "distinctCount",
+      "name": "uv",
+      "fieldName": "visitor_id"
+    }
+  ],
+  "intervals": [
+    "2016-03-06T00:00:00/2016-03-06T23:59:59"
+  ]
+}
+```

docs/content/development/extensions.md

@@ -47,6 +47,7 @@ If you'd like to take on maintenance for a community extension, please post on [
 |druid-rabbitmq|RabbitMQ firehose.|[link](../development/extensions-contrib/rabbitmq.html)|
 |druid-rocketmq|RocketMQ firehose.|[link](../development/extensions-contrib/rocketmq.html)|
 |graphite-emitter|Graphite metrics emitter|[link](../development/extensions-contrib/graphite.html)|
+|druid-distinctcount|DistinctCount aggregator|[link](../development/extensions-contrib/distinctcount.html)|
 
 ## Promoting Community Extension to Core Extension
 

extensions-contrib/distinctcount/pom.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. Metamarkets licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>io.druid.extensions.contrib</groupId>
+    <artifactId>druid-distinctcount</artifactId>
+    <name>druid-distinctcount</name>
+    <description>druid-distinctcount</description>
+
+    <parent>
+        <groupId>io.druid</groupId>
+        <artifactId>druid</artifactId>
+        <version>0.9.1-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-processing</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <!-- Tests -->
+        <dependency>
+            <groupId>io.druid</groupId>
+            <artifactId>druid-processing</artifactId>
+            <version>${project.parent.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

extensions-contrib/distinctcount/src/main/java/io/druid/query/aggregation/distinctcount/BitMapFactory.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.query.aggregation.distinctcount;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.metamx.collections.bitmap.MutableBitmap;
+
+/**
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = RoaringBitMapFactory.class)
+@JsonSubTypes(value = {
+    @JsonSubTypes.Type(name = "java", value = JavaBitMapFactory.class),
+    @JsonSubTypes.Type(name = "concise", value = ConciseBitMapFactory.class),
+    @JsonSubTypes.Type(name = "roaring", value = RoaringBitMapFactory.class)
+})
+public interface BitMapFactory
+{
+  /**
+   * Create a new empty bitmap
+   *
+   * @return the new bitmap
+   */
+  public MutableBitmap makeEmptyMutableBitmap();
+}

extensions-contrib/distinctcount/src/main/java/io/druid/query/aggregation/distinctcount/ConciseBitMapFactory.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.query.aggregation.distinctcount;
+
+import com.metamx.collections.bitmap.BitmapFactory;
+import com.metamx.collections.bitmap.ConciseBitmapFactory;
+import com.metamx.collections.bitmap.MutableBitmap;
+
+public class ConciseBitMapFactory implements BitMapFactory
+{
+  private static final BitmapFactory bitmapFactory = new ConciseBitmapFactory();
+
+  public ConciseBitMapFactory() {}
+
+  @Override
+  public MutableBitmap makeEmptyMutableBitmap()
+  {
+    return bitmapFactory.makeEmptyMutableBitmap();
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ConciseBitMapFactory";
+  }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    return this == o || o instanceof ConciseBitMapFactory;
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return 0;
+  }
+}

extensions-contrib/distinctcount/src/main/java/io/druid/query/aggregation/distinctcount/DistinctCountAggregator.java

@@ -0,0 +1,87 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.druid.query.aggregation.distinctcount;
+
+import com.metamx.collections.bitmap.MutableBitmap;
+import io.druid.query.aggregation.Aggregator;
+import io.druid.segment.DimensionSelector;
+
+public class DistinctCountAggregator implements Aggregator
+{
+
+  private final String name;
+  private final DimensionSelector selector;
+  private final MutableBitmap mutableBitmap;
+
+  public DistinctCountAggregator(
+      String name,
+      DimensionSelector selector,
+      MutableBitmap mutableBitmap
+  )
+  {
+    this.name = name;
+    this.selector = selector;
+    this.mutableBitmap = mutableBitmap;
+  }
+
+  @Override
+  public void aggregate()
+  {
+    for (final Integer index : selector.getRow()) {
+      mutableBitmap.add(index);
+    }
+  }
+
+  @Override
+  public void reset()
+  {
+    mutableBitmap.clear();
+  }
+
+  @Override
+  public Object get()
+  {
+    return mutableBitmap.size();
+  }
+
+  @Override
+  public float getFloat()
+  {
+    return (float) mutableBitmap.size();
+  }
+
+  @Override
+  public String getName()
+  {
+    return name;
+  }
+
+  @Override
+  public void close()
+  {
+    mutableBitmap.clear();
+  }
+
+  @Override
+  public long getLong()
+  {
+    return (long) mutableBitmap.size();
+  }
+}