Cardinality aggregation.

This aggregation computes unique term counts using the hyperloglog++ algorithm
which uses linear counting to estimate low cardinalities and hyperloglog on
higher cardinalities.

Since this algorithm works on hashes, it is useful for high-cardinality fields
to store the hash of values directly in the index, which is the purpose of
the new `murmur3` field type. This is less necessary on low-cardinality
string fields because the aggregator is smart enough to only compute the hash
once per unique value per segment thanks to ordinals, or on numeric fields
since hashing them is very fast.

Close #5426

docs/reference/search/aggregations/metrics.asciidoc

@@ -14,4 +14,6 @@ include::metrics/extendedstats-aggregation.asciidoc[]
 
 include::metrics/valuecount-aggregation.asciidoc[]
 
-include::metrics/percentile-aggregation.asciidoc[]
+include::metrics/percentile-aggregation.asciidoc[]
+
+include::metrics/cardinality-aggregation.asciidoc[]

docs/reference/search/aggregations/metrics/cardinality-aggregation.asciidoc

@@ -0,0 +1,159 @@
+[[search-aggregations-metrics-cardinality-aggregation]]
+=== Cardinality
+
+coming[1.1.0]
+
+A `single-value` metrics aggregation that calculates an approximate count of
+distinct values. Values can be extracted either from specific fields in the
+document or generated by a script.
+
+.Experimental!
+[IMPORTANT]
+=====
+This feature is marked as experimental, and may be subject to change in the
+future.  If you use this feature, please let us know your experience with it!
+=====
+
+Assume you are indexing books and would like to count the unique authors that
+match a query:
+
+[source,js]
+--------------------------------------------------
+{
+    "aggs" : {
+        "author_count" : { 
+            "cardinality" : { 
+                "field" : "author"
+            }
+        }
+    }
+}
+--------------------------------------------------
+
+This aggregation also supports the `precision_threshold` and `rehash` options:
+
+[source,js]
+--------------------------------------------------
+{
+    "aggs" : {
+        "author_count" : { 
+            "cardinality" : { 
+                "field" : "author_hash",
+                "precision_threshold": 100, <1>
+                "rehash": false <2>
+            }
+        }
+    }
+}
+--------------------------------------------------
+
+<1> The `precision_threshold` options allows to trade memory for accuracy, and
+defines a unique count below which counts are expected to be close to
+accurate. Above this value, counts might become a bit more fuzzy. The maximum
+supported value is 40000, thresholds above this number will have the same
+effect as a threshold of 40000.
+Default value depends on the number of parent aggregations that multiple
+create buckets (such as terms or histograms).
+<2> If you computed a hash on client-side, stored it into your documents and want
+Elasticsearch to use them to compute counts using this hash function without
+rehashing values, it is possible to specify `rehash: false`. Default value is
+`true`. Please note that the hash must be indexed as a long when `rehash` is
+false.
+
+==== Counts are approximate
+
+Computing exact counts requires loading values into a hash set and returning its
+size. This doesn't scale when working on high-cardinality sets and/or large
+values as the required memory usage and the need to communicate those
+per-shard sets between nodes would utilize too many resources of the cluster.
+
+This `cardinality` aggregation is based on the
+http://static.googleusercontent.com/media/research.google.com/fr//pubs/archive/40671.pdf[HyperLogLog++]
+algorithm, which counts based on the hashes of the values with some interesting
+properties:
+
+ * configurable precision, which decides on how to trade memory for accuracy,
+ * excellent accuracy on low-cardinality sets,
+ * fixed memory usage: no matter if there are tens or billions of unique values,
+   memory usage only depends on the configured precision.
+ 
+For a precision threshold of `c`, the implementation that we are using requires
+about `c * 8` bytes.
+
+The following chart shows how the error varies before and after the threshold:
+
+image:images/cardinality_error.png[]
+
+For all 3 thresholds, counts have been accurate up to the configured threshold
+(although not guaranteed, this is likely to be the case). Please also note that
+even with a threshold as low as 100, the error remains under under 5%, even when
+counting millions of items.
+
+==== Pre-computed hashes
+
+If you don't want Elasticsearch to re-compute hashes on every run of this
+aggregation, it is possible to use pre-computed hashes, either by computing a
+hash on client-side, indexing it and specifying `rehash: false`, or by using
+the special `murmur3` field mapper, typically in the context of a `multi-field`
+in the mapping:
+
+[source,js]
+--------------------------------------------------
+{
+    "author": {
+        "type": "string",
+        "fields": {
+            "hash": {
+                "type": "murmur3"
+            }
+        }
+    }
+}
+--------------------------------------------------
+
+With such a mapping, Elasticsearch is going to compute hashes of the `author`
+field at indexing time and store them in the `author.hash` field. This
+way, unique counts can be computed using the cardinality aggregation by only
+loading the hashes into memory, not the values of the `author` field, and
+without computing hashes on the fly:
+
+[source,js]
+--------------------------------------------------
+{
+    "aggs" : {
+        "author_count" : { 
+            "cardinality" : { 
+                "field" : "author.hash"
+            }
+        }
+    }
+}
+--------------------------------------------------
+
+NOTE: `rehash` is automatically set to `false` when computing unique counts on
+a `murmur3` field.
+
+NOTE: Pre-computing hashes is usually only useful on very large and/or
+high-cardinality fields as it saves CPU and memory. However, on numeric
+fields, hashing is very fast and storing the original values requires as much
+or less memory than storing the hashes. This is also true on low-cardinality
+string fields, especially given that those have an optimization in order to
+make sure that hashes are computed at most once per unique value per segment.
+
+==== Script
+
+The `cardinality` metric supports scripting, with a noticeable performance hit
+however since hashes need to be computed on the fly.
+
+[source,js]
+--------------------------------------------------
+{
+    "aggs" : {
+        "author_count" : { 
+            "cardinality" : { 
+                "script": "doc['author.first_name'].value + ' ' + doc['author.last_name'].value"
+            }
+        }
+    }
+}
+--------------------------------------------------

src/main/java/org/elasticsearch/common/hash/MurmurHash3.java

@@ -0,0 +1,156 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.common.hash;
+
+import org.elasticsearch.common.util.UnsafeUtils;
+
+
+/**
+ * MurmurHash3 hashing functions.
+ */
+public enum MurmurHash3 {
+    ;
+
+    /**
+     * A 128-bits hash.
+     */
+    public static class Hash128 {
+        public long h1, h2;
+    }
+
+    private static long C1 = 0x87c37b91114253d5L;
+    private static long C2 = 0x4cf5ad432745937fL;
+
+    protected static long getblock(byte[] key, int offset, int index) {
+        int i_8 = index << 3;
+        int blockOffset = offset + i_8;
+        return UnsafeUtils.readLongLE(key, blockOffset);
+    }
+
+    protected static long fmix(long k) {
+        k ^= k >>> 33;
+        k *= 0xff51afd7ed558ccdL;
+        k ^= k >>> 33;
+        k *= 0xc4ceb9fe1a85ec53L;
+        k ^= k >>> 33;
+        return k;
+    }
+
+    /**
+     * Compute the hash of the MurmurHash3_x64_128 hashing function.
+     */
+    public static Hash128 hash128(byte[] key, int offset, int length, long seed, Hash128 hash) {
+        long h1 = seed;
+        long h2 = seed;
+
+        if (length >= 16) {
+
+            final int len16 = length & 0xFFFFFFF0; // higher multiple of 16 that is lower than or equal to length
+            final int end = offset + len16;
+            for (int i = offset; i < end; i += 16) {
+                long k1 = UnsafeUtils.readLongLE(key, i);
+                long k2 = UnsafeUtils.readLongLE(key, i + 8);
+
+                k1 *= C1;
+                k1 = Long.rotateLeft(k1, 31);
+                k1 *= C2;
+                h1 ^= k1;
+
+                h1 = Long.rotateLeft(h1, 27);
+                h1 += h2;
+                h1 = h1 * 5 + 0x52dce729;
+
+                k2 *= C2;
+                k2 = Long.rotateLeft(k2, 33);
+                k2 *= C1;
+                h2 ^= k2;
+
+                h2 = Long.rotateLeft(h2, 31);
+                h2 += h1;
+                h2 = h2 * 5 + 0x38495ab5;
+            }
+
+            // Advance offset to the unprocessed tail of the data.
+            offset = end;
+        }
+
+        long k1 = 0;
+        long k2 = 0;
+
+        switch (length & 15) {
+            case 15:
+                k2 ^= (key[offset + 14] & 0xFFL) << 48;
+            case 14:
+                k2 ^= (key[offset + 13] & 0xFFL) << 40;
+            case 13:
+                k2 ^= (key[offset + 12] & 0xFFL) << 32;
+            case 12:
+                k2 ^= (key[offset + 11] & 0xFFL) << 24;
+            case 11:
+                k2 ^= (key[offset + 10] & 0xFFL) << 16;
+            case 10:
+                k2 ^= (key[offset + 9] & 0xFFL) << 8;
+            case 9:
+                k2 ^= (key[offset + 8] & 0xFFL) << 0;
+                k2 *= C2;
+                k2 = Long.rotateLeft(k2, 33);
+                k2 *= C1;
+                h2 ^= k2;
+
+            case 8:
+                k1 ^= (key[offset + 7] & 0xFFL) << 56;
+            case 7:
+                k1 ^= (key[offset + 6] & 0xFFL) << 48;
+            case 6:
+                k1 ^= (key[offset + 5] & 0xFFL) << 40;
+            case 5:
+                k1 ^= (key[offset + 4] & 0xFFL) << 32;
+            case 4:
+                k1 ^= (key[offset + 3] & 0xFFL) << 24;
+            case 3:
+                k1 ^= (key[offset + 2] & 0xFFL) << 16;
+            case 2:
+                k1 ^= (key[offset + 1] & 0xFFL) << 8;
+            case 1:
+                k1 ^= (key[offset] & 0xFFL);
+                k1 *= C1;
+                k1 = Long.rotateLeft(k1, 31);
+                k1 *= C2;
+                h1 ^= k1;
+        }
+
+        h1 ^= length;
+        h2 ^= length;
+
+        h1 += h2;
+        h2 += h1;
+
+        h1 = fmix(h1);
+        h2 = fmix(h2);
+
+        h1 += h2;
+        h2 += h1;
+
+        hash.h1 = h1;
+        hash.h2 = h2;
+        return hash;
+    }
+
+}

src/main/java/org/elasticsearch/common/util/BigArrays.java

@@ -140,6 +140,12 @@ public class BigArrays extends AbstractComponent {
             System.arraycopy(buf, offset, array, (int) index, len);
         }
 
+        @Override
+        public void fill(long fromIndex, long toIndex, byte value) {
+            assert indexIsInt(fromIndex);
+            assert indexIsInt(toIndex);
+            Arrays.fill(array, (int) fromIndex, (int) toIndex, value);
+        }
     }
 
     private static class IntArrayWrapper extends AbstractArrayWrapper implements IntArray {

src/main/java/org/elasticsearch/common/util/BigByteArray.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.common.util;
 
+import com.google.common.base.Preconditions;
 import org.apache.lucene.util.ArrayUtil;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.RamUsageEstimator;
@@ -106,6 +107,22 @@ final class BigByteArray extends AbstractBigArray implements ByteArray {
         }
     }
 
+    @Override
+    public void fill(long fromIndex, long toIndex, byte value) {
+        Preconditions.checkArgument(fromIndex <= toIndex);
+        final int fromPage = pageIndex(fromIndex);
+        final int toPage = pageIndex(toIndex - 1);
+        if (fromPage == toPage) {
+            Arrays.fill(pages[fromPage], indexInPage(fromIndex), indexInPage(toIndex - 1) + 1, value);
+        } else {
+            Arrays.fill(pages[fromPage], indexInPage(fromIndex), pages[fromPage].length, value);
+            for (int i = fromPage + 1; i < toPage; ++i) {
+                Arrays.fill(pages[i], value);
+            }
+            Arrays.fill(pages[toPage], 0, indexInPage(toIndex - 1) + 1, value);
+        }
+    }
+
     @Override
     protected int numBytesPerElement() {
         return RamUsageEstimator.NUM_BYTES_BYTE;

src/main/java/org/elasticsearch/common/util/BloomFilter.java

@@ -355,6 +355,7 @@ public class BloomFilter {
     }
 
     // START : MURMUR 3_128
+    // NOTE: don't replace this code with the o.e.common.hashing.MurmurHash3 method which returns a different hash
 
     protected static long getblock(byte[] key, int offset, int index) {
         int i_8 = index << 3;

src/main/java/org/elasticsearch/common/util/ByteArray.java

@@ -46,4 +46,9 @@ public interface ByteArray extends BigArray {
      */
     public abstract void set(long index, byte[] buf, int offset, int len);
 
+    /**
+     * Fill slots between <code>fromIndex</code> inclusive to <code>toIndex</code> exclusive with <code>value</code>.
+     */
+    public abstract void fill(long fromIndex, long toIndex, byte value);
+
 }

src/main/java/org/elasticsearch/common/util/UnsafeUtils.java

@@ -23,6 +23,7 @@ import org.apache.lucene.util.BytesRef;
 import sun.misc.Unsafe;
 
 import java.lang.reflect.Field;
+import java.nio.ByteOrder;
 
 /** Utility methods that use {@link Unsafe}. */
 public enum UnsafeUtils {
@@ -111,4 +112,26 @@ public enum UnsafeUtils {
         return true;
     }
 
+    /**
+     * Read a long using little endian byte order.
+     */
+    public static long readLongLE(byte[] src, int offset) {
+        long value = readLong(src, offset);
+        if (ByteOrder.nativeOrder() == ByteOrder.BIG_ENDIAN) {
+            value = Long.reverseBytes(value);
+        }
+        return value;
+    }
+
+    /**
+     * Read an int using little endian byte order.
+     */
+    public static int readIntLE(byte[] src, int offset) {
+        int value = readInt(src, offset);
+        if (ByteOrder.nativeOrder() == ByteOrder.BIG_ENDIAN) {
+            value = Integer.reverseBytes(value);
+        }
+        return value;
+    }
+
 }

src/main/java/org/elasticsearch/index/fielddata/MurmurHash3Values.java

@@ -0,0 +1,120 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.index.fielddata;
+
+import com.carrotsearch.hppc.hash.MurmurHash3;
+import org.apache.lucene.util.BytesRef;
+
+/**
+ *
+ */
+public class MurmurHash3Values {
+
+    public static LongValues wrap(DoubleValues values) {
+        return new Double(values);
+    }
+
+    public static LongValues wrap(LongValues values) {
+        return new Long(values);
+    }
+
+    public static LongValues wrap(BytesValues values) {
+        return new Bytes(values);
+    }
+
+    private static class Long extends LongValues {
+
+        private final LongValues values;
+
+        public Long(LongValues values) {
+            super(values.isMultiValued());
+            this.values = values;
+        }
+
+        @Override
+        public int setDocument(int docId) {
+            return values.setDocument(docId);
+        }
+
+        @Override
+        public long nextValue() {
+            return MurmurHash3.hash(values.nextValue());
+        }
+
+        @Override
+        public AtomicFieldData.Order getOrder() {
+            return AtomicFieldData.Order.NONE;
+        }
+    }
+
+    private static class Double extends LongValues {
+
+        private final DoubleValues values;
+
+        public Double(DoubleValues values) {
+            super(values.isMultiValued());
+            this.values = values;
+        }
+
+        @Override
+        public int setDocument(int docId) {
+            return values.setDocument(docId);
+        }
+
+        @Override
+        public long nextValue() {
+            return MurmurHash3.hash(java.lang.Double.doubleToLongBits(values.nextValue()));
+        }
+
+        @Override
+        public AtomicFieldData.Order getOrder() {
+            return AtomicFieldData.Order.NONE;
+        }
+    }
+
+    private static class Bytes extends LongValues {
+
+        private final org.elasticsearch.common.hash.MurmurHash3.Hash128 hash = new org.elasticsearch.common.hash.MurmurHash3.Hash128();
+
+        private final BytesValues values;
+
+        public Bytes(BytesValues values) {
+            super(values.isMultiValued());
+            this.values = values;
+        }
+
+        @Override
+        public int setDocument(int docId) {
+            return values.setDocument(docId);
+        }
+
+        @Override
+        public long nextValue() {
+            final BytesRef next = values.nextValue();
+            org.elasticsearch.common.hash.MurmurHash3.hash128(next.bytes, next.offset, next.length, 0, hash);
+            return hash.h1;
+        }
+
+        @Override
+        public AtomicFieldData.Order getOrder() {
+            return AtomicFieldData.Order.NONE;
+        }
+    }
+}

src/main/java/org/elasticsearch/index/mapper/DocumentMapperParser.java

@@ -95,7 +95,8 @@ public class DocumentMapperParser extends AbstractIndexComponent {
                 .put(ObjectMapper.NESTED_CONTENT_TYPE, new ObjectMapper.TypeParser())
                 .put(TypeParsers.MULTI_FIELD_CONTENT_TYPE, TypeParsers.multiFieldConverterTypeParser)
                 .put(CompletionFieldMapper.CONTENT_TYPE, new CompletionFieldMapper.TypeParser())
-                .put(GeoPointFieldMapper.CONTENT_TYPE, new GeoPointFieldMapper.TypeParser());
+                .put(GeoPointFieldMapper.CONTENT_TYPE, new GeoPointFieldMapper.TypeParser())
+                .put(Murmur3FieldMapper.CONTENT_TYPE, new Murmur3FieldMapper.TypeParser());
 
         if (ShapesAvailability.JTS_AVAILABLE) {
             typeParsersBuilder.put(GeoShapeFieldMapper.CONTENT_TYPE, new GeoShapeFieldMapper.TypeParser());

src/main/java/org/elasticsearch/index/mapper/MapperBuilders.java

@@ -150,6 +150,10 @@ public final class MapperBuilders {
         return new LongFieldMapper.Builder(name);
     }
 
+    public static Murmur3FieldMapper.Builder murmur3Field(String name) {
+        return new Murmur3FieldMapper.Builder(name);
+    }
+
     public static FloatFieldMapper.Builder floatField(String name) {
         return new FloatFieldMapper.Builder(name);
     }

src/main/java/org/elasticsearch/index/mapper/core/Murmur3FieldMapper.java

@@ -0,0 +1,111 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.index.mapper.core;
+
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.FieldType;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.common.Explicit;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.hash.MurmurHash3;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.index.codec.docvaluesformat.DocValuesFormatProvider;
+import org.elasticsearch.index.codec.postingsformat.PostingsFormatProvider;
+import org.elasticsearch.index.mapper.Mapper;
+import org.elasticsearch.index.mapper.MapperParsingException;
+import org.elasticsearch.index.mapper.ParseContext;
+import org.elasticsearch.index.similarity.SimilarityProvider;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.index.mapper.MapperBuilders.murmur3Field;
+import static org.elasticsearch.index.mapper.core.TypeParsers.parseNumberField;
+
+public class Murmur3FieldMapper extends LongFieldMapper {
+
+    public static final String CONTENT_TYPE = "murmur3";
+
+    public static class Defaults extends LongFieldMapper.Defaults {
+    }
+
+    public static class Builder extends NumberFieldMapper.Builder<Builder, Murmur3FieldMapper> {
+
+        public Builder(String name) {
+            super(name, new FieldType(Defaults.FIELD_TYPE));
+            builder = this;
+            builder.precisionStep(Integer.MAX_VALUE);
+        }
+
+        @Override
+        public Murmur3FieldMapper build(BuilderContext context) {
+            fieldType.setOmitNorms(fieldType.omitNorms() && boost == 1.0f);
+            Murmur3FieldMapper fieldMapper = new Murmur3FieldMapper(buildNames(context), precisionStep, boost, fieldType, docValues, ~0L,
+                    ignoreMalformed(context), coerce(context), postingsProvider, docValuesProvider, similarity, normsLoading,
+                    fieldDataSettings, context.indexSettings(), multiFieldsBuilder.build(this, context), copyTo);
+            fieldMapper.includeInAll(includeInAll);
+            return fieldMapper;
+        }
+    }
+
+    public static class TypeParser implements Mapper.TypeParser {
+        @Override
+        @SuppressWarnings("unchecked")
+        public Mapper.Builder parse(String name, Map<String, Object> node, ParserContext parserContext) throws MapperParsingException {
+            Builder builder = murmur3Field(name);
+            parseNumberField(builder, name, node, parserContext);
+            return builder;
+        }
+    }
+
+    protected Murmur3FieldMapper(Names names, int precisionStep, float boost, FieldType fieldType, Boolean docValues,
+            Long nullValue, Explicit<Boolean> ignoreMalformed, Explicit<Boolean> coerce,
+            PostingsFormatProvider postingsProvider, DocValuesFormatProvider docValuesProvider,
+            SimilarityProvider similarity, Loading normsLoading, @Nullable Settings fieldDataSettings,
+            Settings indexSettings, MultiFields multiFields, CopyTo copyTo) {
+        super(names, precisionStep, boost, fieldType, docValues, nullValue, ignoreMalformed, coerce,
+                postingsProvider, docValuesProvider, similarity, normsLoading, fieldDataSettings,
+                indexSettings, multiFields, copyTo);
+    }
+
+    @Override
+    protected String contentType() {
+        return CONTENT_TYPE;
+    }
+
+    @Override
+    protected void innerParseCreateField(ParseContext context, List<Field> fields) throws IOException {
+        final Object value;
+        if (context.externalValueSet()) {
+            value = context.externalValue();
+        } else {
+            value = context.parser().textOrNull();
+        }
+        if (value != null) {
+            final BytesRef bytes = new BytesRef(value.toString());
+            final long hash = MurmurHash3.hash128(bytes.bytes, bytes.offset, bytes.length, 0, new MurmurHash3.Hash128()).h1;
+            context.externalValue(hash);
+            super.innerParseCreateField(context, fields);
+        }
+
+    }
+
+}

src/main/java/org/elasticsearch/search/aggregations/AggregationBuilders.java

@@ -31,6 +31,7 @@ import org.elasticsearch.search.aggregations.bucket.range.geodistance.GeoDistanc
 import org.elasticsearch.search.aggregations.bucket.range.ipv4.IPv4RangeBuilder;
 import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
 import org.elasticsearch.search.aggregations.metrics.avg.AvgBuilder;
+import org.elasticsearch.search.aggregations.metrics.cardinality.CardinalityBuilder;
 import org.elasticsearch.search.aggregations.metrics.max.MaxBuilder;
 import org.elasticsearch.search.aggregations.metrics.min.MinBuilder;
 import org.elasticsearch.search.aggregations.metrics.percentiles.PercentilesBuilder;
@@ -126,4 +127,8 @@ public class AggregationBuilders {
     public static PercentilesBuilder percentiles(String name) {
         return new PercentilesBuilder(name);
     }
+
+    public static CardinalityBuilder cardinality(String name) {
+        return new CardinalityBuilder(name);
+    }
 }

src/main/java/org/elasticsearch/search/aggregations/AggregationModule.java

@@ -34,6 +34,7 @@ import org.elasticsearch.search.aggregations.bucket.range.geodistance.GeoDistanc
 import org.elasticsearch.search.aggregations.bucket.range.ipv4.IpRangeParser;
 import org.elasticsearch.search.aggregations.bucket.terms.TermsParser;
 import org.elasticsearch.search.aggregations.metrics.avg.AvgParser;
+import org.elasticsearch.search.aggregations.metrics.cardinality.CardinalityParser;
 import org.elasticsearch.search.aggregations.metrics.max.MaxParser;
 import org.elasticsearch.search.aggregations.metrics.min.MinParser;
 import org.elasticsearch.search.aggregations.metrics.percentiles.PercentilesParser;
@@ -60,6 +61,7 @@ public class AggregationModule extends AbstractModule {
         parsers.add(ExtendedStatsParser.class);
         parsers.add(ValueCountParser.class);
         parsers.add(PercentilesParser.class);
+        parsers.add(CardinalityParser.class);
 
         parsers.add(GlobalParser.class);
         parsers.add(MissingParser.class);

src/main/java/org/elasticsearch/search/aggregations/TransportAggregationModule.java

@@ -35,6 +35,7 @@ import org.elasticsearch.search.aggregations.bucket.terms.LongTerms;
 import org.elasticsearch.search.aggregations.bucket.terms.StringTerms;
 import org.elasticsearch.search.aggregations.bucket.terms.UnmappedTerms;
 import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
+import org.elasticsearch.search.aggregations.metrics.cardinality.InternalCardinality;
 import org.elasticsearch.search.aggregations.metrics.max.InternalMax;
 import org.elasticsearch.search.aggregations.metrics.min.InternalMin;
 import org.elasticsearch.search.aggregations.metrics.percentiles.InternalPercentiles;
@@ -60,6 +61,7 @@ public class TransportAggregationModule extends AbstractModule {
         InternalExtendedStats.registerStreams();
         InternalValueCount.registerStreams();
         InternalPercentiles.registerStreams();
+        InternalCardinality.registerStreams();
 
         // buckets
         InternalGlobal.registerStreams();

src/main/java/org/elasticsearch/search/aggregations/metrics/cardinality/Cardinality.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics.cardinality;
+
+import org.elasticsearch.search.aggregations.Aggregation;
+
+/**
+ * A number of unique terms.
+ */
+public interface Cardinality extends Aggregation {
+
+    public long getValue();
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/cardinality/CardinalityAggregator.java

@@ -0,0 +1,271 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics.cardinality;
+
+import com.google.common.base.Preconditions;
+import org.apache.lucene.index.AtomicReaderContext;
+import org.apache.lucene.util.BytesRef;
+import org.apache.lucene.util.FixedBitSet;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.lease.Releasables;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.util.LongArray;
+import org.elasticsearch.common.util.ObjectArray;
+import org.elasticsearch.index.fielddata.BytesValues;
+import org.elasticsearch.index.fielddata.LongValues;
+import org.elasticsearch.index.fielddata.MurmurHash3Values;
+import org.elasticsearch.index.fielddata.ordinals.Ordinals;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.metrics.MetricsAggregator;
+import org.elasticsearch.search.aggregations.support.AggregationContext;
+import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.numeric.NumericValuesSource;
+
+import java.io.IOException;
+
+/**
+ * An aggregator that computes approximate counts of unique values.
+ */
+public class CardinalityAggregator extends MetricsAggregator.SingleValue {
+
+    private final int precision;
+    private final boolean rehash;
+    private final ValuesSource valuesSource;
+
+    // Expensive to initialize, so we only initialize it when we have an actual value source
+    @Nullable
+    private HyperLogLogPlusPlus counts;
+
+    private Collector collector;
+
+    public CardinalityAggregator(String name, long estimatedBucketsCount, ValuesSource valuesSource, boolean rehash,
+                                 int precision, AggregationContext context, Aggregator parent) {
+        super(name, estimatedBucketsCount, context, parent);
+        this.valuesSource = valuesSource;
+        this.rehash = rehash;
+        this.precision = precision;
+        this.counts = valuesSource == null ? null : new HyperLogLogPlusPlus(precision, bigArrays, estimatedBucketsCount);
+    }
+
+    @Override
+    public void setNextReader(AtomicReaderContext reader) {
+        postCollectLastCollector();
+        collector = createCollector(reader);
+    }
+
+    private Collector createCollector(AtomicReaderContext reader) {
+
+        // if rehash is false then the value source is either already hashed, or the user explicitly
+        // requested not to hash the values (perhaps they already hashed the values themselves before indexing the doc)
+        // so we can just work with the original value source as is
+        if (!rehash) {
+            LongValues hashValues = ((NumericValuesSource) valuesSource).longValues();
+            return new DirectCollector(counts, hashValues);
+        }
+
+        if (valuesSource instanceof NumericValuesSource) {
+            NumericValuesSource source = (NumericValuesSource) valuesSource;
+            LongValues hashValues = source.isFloatingPoint() ? MurmurHash3Values.wrap(source.doubleValues()) : MurmurHash3Values.wrap(source.longValues());
+            return new DirectCollector(counts, hashValues);
+        }
+
+        final BytesValues bytesValues = valuesSource.bytesValues();
+        if (bytesValues instanceof BytesValues.WithOrdinals) {
+            BytesValues.WithOrdinals values = (BytesValues.WithOrdinals) bytesValues;
+            final long maxOrd = values.ordinals().getMaxOrd();
+            if (maxOrd <= reader.reader().maxDoc()) {
+                return new OrdinalsCollector(counts, values, bigArrays);
+            }
+        }
+
+        return new DirectCollector(counts, MurmurHash3Values.wrap(bytesValues));
+    }
+
+
+    @Override
+    public boolean shouldCollect() {
+        return valuesSource != null;
+    }
+
+    @Override
+    public void collect(int doc, long owningBucketOrdinal) throws IOException {
+        collector.collect(doc, owningBucketOrdinal);
+    }
+
+    private void postCollectLastCollector() {
+        if (collector != null) {
+            try {
+                collector.postCollect();
+                collector.release();
+            } finally {
+                collector = null;
+            }
+        }
+    }
+
+    @Override
+    protected void doPostCollection() {
+        postCollectLastCollector();
+    }
+
+    @Override
+    public double metric(long owningBucketOrd) {
+        return counts == null ? 0 : counts.cardinality(owningBucketOrd);
+    }
+
+    @Override
+    public InternalAggregation buildAggregation(long owningBucketOrdinal) {
+        if (counts == null || owningBucketOrdinal >= counts.maxBucket() || counts.cardinality(owningBucketOrdinal) == 0) {
+            return buildEmptyAggregation();
+        }
+        // We need to build a copy because the returned Aggregation needs remain usable after
+        // this Aggregator (and its HLL++ counters) is released.
+        HyperLogLogPlusPlus copy = new HyperLogLogPlusPlus(precision, BigArrays.NON_RECYCLING_INSTANCE, 1);
+        copy.merge(0, counts, owningBucketOrdinal);
+        return new InternalCardinality(name, copy);
+    }
+
+    @Override
+    public InternalAggregation buildEmptyAggregation() {
+        return new InternalCardinality(name, null);
+    }
+
+    @Override
+    protected void doRelease() {
+        Releasables.release(counts, collector);
+    }
+
+    private static interface Collector extends Releasable {
+
+        void collect(int doc, long bucketOrd);
+
+        void postCollect();
+
+    }
+
+    private static class DirectCollector implements Collector {
+
+        private final LongValues hashes;
+        private final HyperLogLogPlusPlus counts;
+
+        DirectCollector(HyperLogLogPlusPlus counts, LongValues values) {
+            this.counts = counts;
+            this.hashes = values;
+        }
+
+        @Override
+        public void collect(int doc, long bucketOrd) {
+            final int valueCount = hashes.setDocument(doc);
+            for (int i = 0; i < valueCount; ++i) {
+                counts.collect(bucketOrd, hashes.nextValue());
+            }
+        }
+
+        @Override
+        public void postCollect() {
+            // no-op
+        }
+
+        @Override
+        public boolean release() throws ElasticsearchException {
+            return true;
+        }
+
+    }
+
+    private static class OrdinalsCollector implements Collector {
+
+        private final BigArrays bigArrays;
+        private final BytesValues.WithOrdinals values;
+        private final Ordinals.Docs ordinals;
+        private final int maxOrd;
+        private final HyperLogLogPlusPlus counts;
+        private ObjectArray<FixedBitSet> visitedOrds;
+
+        OrdinalsCollector(HyperLogLogPlusPlus counts, BytesValues.WithOrdinals values, BigArrays bigArrays) {
+            ordinals = values.ordinals();
+            Preconditions.checkArgument(ordinals.getMaxOrd() <= Integer.MAX_VALUE);
+            maxOrd = (int) ordinals.getMaxOrd();
+            this.bigArrays = bigArrays;
+            this.counts = counts;
+            this.values = values;
+            visitedOrds = bigArrays.newObjectArray(1);
+        }
+
+        @Override
+        public void collect(int doc, long bucketOrd) {
+            visitedOrds = bigArrays.grow(visitedOrds, bucketOrd + 1);
+            FixedBitSet bits = visitedOrds.get(bucketOrd);
+            if (bits == null) {
+                bits = new FixedBitSet(maxOrd);
+                visitedOrds.set(bucketOrd, bits);
+            }
+            final int valueCount = ordinals.setDocument(doc);
+            for (int i = 0; i < valueCount; ++i) {
+                bits.set((int) ordinals.nextOrd());
+            }
+        }
+
+        @Override
+        public void postCollect() {
+            final FixedBitSet allVisitedOrds = new FixedBitSet(maxOrd);
+            for (long bucket = visitedOrds.size() - 1; bucket >= 0; --bucket) {
+                final FixedBitSet bits = visitedOrds.get(bucket);
+                if (bits != null) {
+                    allVisitedOrds.or(bits);
+                }
+            }
+
+            final org.elasticsearch.common.hash.MurmurHash3.Hash128 hash = new org.elasticsearch.common.hash.MurmurHash3.Hash128();
+            final LongArray hashes = bigArrays.newLongArray(maxOrd, false);
+            boolean success = false;
+            try {
+                for (int ord = allVisitedOrds.nextSetBit(0); ord != -1; ord = ord + 1 < maxOrd ? allVisitedOrds.nextSetBit(ord + 1) : -1) {
+                    final BytesRef value = values.getValueByOrd(ord);
+                    org.elasticsearch.common.hash.MurmurHash3.hash128(value.bytes, value.offset, value.length, 0, hash);
+                    hashes.set(ord, hash.h1);
+                }
+
+                for (long bucket = visitedOrds.size() - 1; bucket >= 0; --bucket) {
+                    final FixedBitSet bits = visitedOrds.get(bucket);
+                    if (bits != null) {
+                        for (int ord = bits.nextSetBit(0); ord != -1; ord = ord + 1 < maxOrd ? bits.nextSetBit(ord + 1) : -1) {
+                            counts.collect(bucket, hashes.get(ord));
+                        }
+                    }
+                }
+                success = true;
+            } finally {
+                Releasables.release(success, hashes);
+            }
+        }
+
+        @Override
+        public boolean release() throws ElasticsearchException {
+            Releasables.release(visitedOrds);
+            return true;
+        }
+
+    }
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/cardinality/CardinalityAggregatorFactory.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics.cardinality;
+
+import org.elasticsearch.search.aggregations.AggregationExecutionException;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.Aggregator.BucketAggregationMode;
+import org.elasticsearch.search.aggregations.support.AggregationContext;
+import org.elasticsearch.search.aggregations.support.ValueSourceAggregatorFactory;
+import org.elasticsearch.search.aggregations.support.ValuesSource;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.aggregations.support.numeric.NumericValuesSource;
+
+final class CardinalityAggregatorFactory extends ValueSourceAggregatorFactory<ValuesSource> {
+
+    private final long precisionThreshold;
+    private final boolean rehash;
+
+    CardinalityAggregatorFactory(String name, ValuesSourceConfig valuesSourceConfig, long precisionThreshold, boolean rehash) {
+        super(name, InternalCardinality.TYPE.name(), valuesSourceConfig);
+        this.precisionThreshold = precisionThreshold;
+        this.rehash = rehash;
+    }
+
+    private int precision(Aggregator parent) {
+        return precisionThreshold < 0 ? defaultPrecision(parent) : HyperLogLogPlusPlus.precisionFromThreshold(precisionThreshold);
+    }
+
+    @Override
+    protected Aggregator createUnmapped(AggregationContext context, Aggregator parent) {
+        return new CardinalityAggregator(name, parent == null ? 1 : parent.estimatedBucketCount(), null, true, precision(parent), context, parent);
+    }
+
+    @Override
+    protected Aggregator create(ValuesSource valuesSource, long expectedBucketsCount, AggregationContext context, Aggregator parent) {
+        if (!(valuesSource instanceof NumericValuesSource) && !rehash) {
+            throw new AggregationExecutionException("Turning off rehashing for cardinality aggregation [" + name + "] on non-numeric values in not allowed");
+        }
+        return new CardinalityAggregator(name, parent == null ? 1 : parent.estimatedBucketCount(), valuesSource, rehash, precision(parent), context, parent);
+    }
+
+    /*
+     * If one of the parent aggregators is a MULTI_BUCKET one, we might want to lower the precision
+     * because otherwise it might be memory-intensive. On the other hand, for top-level aggregators
+     * we try to focus on accuracy.
+     */
+    private int defaultPrecision(Aggregator parent) {
+        int precision = HyperLogLogPlusPlus.DEFAULT_PRECISION;
+        while (parent != null) {
+            if (parent.bucketAggregationMode() == BucketAggregationMode.MULTI_BUCKETS) {
+                // if the parent is a multi-bucket aggregator, we substract 5 to the precision,
+                // which will effectively divide the memory usage of each counter by 32
+                precision -= 5;
+            }
+            parent = parent.parent();
+        }
+
+        return Math.max(precision, HyperLogLogPlusPlus.MIN_PRECISION);
+    }
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/cardinality/CardinalityBuilder.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics.cardinality;
+
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.aggregations.metrics.ValuesSourceMetricsAggregationBuilder;
+
+import java.io.IOException;
+
+public class CardinalityBuilder extends ValuesSourceMetricsAggregationBuilder<CardinalityBuilder> {
+
+    private Long precisionThreshold;
+    private Boolean rehash;
+
+    public CardinalityBuilder(String name) {
+        super(name, InternalCardinality.TYPE.name());
+    }
+
+    public CardinalityBuilder precisionThreshold(long precisionThreshold) {
+        this.precisionThreshold = precisionThreshold;
+        return this;
+    }
+
+    public CardinalityBuilder rehash(boolean rehash) {
+        this.rehash = rehash;
+        return this;
+    }
+
+    @Override
+    protected void internalXContent(XContentBuilder builder, Params params) throws IOException {
+        super.internalXContent(builder, params);
+        if (precisionThreshold != null) {
+            builder.field("precision_threshold", precisionThreshold);
+        }
+        if (rehash != null) {
+            builder.field("rehash", rehash);
+        }
+    }
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/cardinality/CardinalityParser.java

@@ -0,0 +1,134 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics.cardinality;
+
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.index.fielddata.IndexFieldData;
+import org.elasticsearch.index.mapper.FieldMapper;
+import org.elasticsearch.index.mapper.core.Murmur3FieldMapper;
+import org.elasticsearch.index.mapper.core.NumberFieldMapper;
+import org.elasticsearch.search.SearchParseException;
+import org.elasticsearch.search.aggregations.Aggregator;
+import org.elasticsearch.search.aggregations.AggregatorFactory;
+import org.elasticsearch.search.aggregations.support.FieldContext;
+import org.elasticsearch.search.aggregations.support.ValuesSourceConfig;
+import org.elasticsearch.search.aggregations.support.bytes.BytesValuesSource;
+import org.elasticsearch.search.aggregations.support.numeric.NumericValuesSource;
+import org.elasticsearch.search.internal.SearchContext;
+
+import java.io.IOException;
+import java.util.Map;
+
+
+public class CardinalityParser implements Aggregator.Parser {
+
+    private static final ParseField PRECISION_THRESHOLD = new ParseField("precision_threshold");
+
+    @Override
+    public String type() {
+        return InternalCardinality.TYPE.name();
+    }
+
+    @Override
+    public AggregatorFactory parse(String name, XContentParser parser, SearchContext context) throws IOException {
+        long precisionThreshold = -1;
+        Boolean rehash = null;
+        String field = null;
+        String script = null;
+        String scriptLang = null;
+        Map<String, Object> scriptParams = null;
+
+        XContentParser.Token token;
+        String currentFieldName = null;
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                currentFieldName = parser.currentName();
+            } else if (token == XContentParser.Token.VALUE_STRING) {
+                if ("field".equals(currentFieldName)) {
+                    field = parser.text();
+                } else if ("script".equals(currentFieldName)) {
+                    script = parser.text();
+                } else if ("lang".equals(currentFieldName)) {
+                    scriptLang = parser.text();
+                } else {
+                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + name + "]: [" + currentFieldName + "].");
+                }
+            } else if (token == XContentParser.Token.START_OBJECT) {
+                if ("params".equals(currentFieldName)) {
+                    scriptParams = parser.map();
+                } else {
+                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + name + "]: [" + currentFieldName + "].");
+                }
+            } else if (token == XContentParser.Token.VALUE_BOOLEAN) {
+                if ("rehash".equals(currentFieldName)) {
+                    rehash = parser.booleanValue();
+                } else {
+                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + name + "]: [" + currentFieldName + "].");
+                }
+            } else if (token == XContentParser.Token.VALUE_NUMBER) {
+                if (PRECISION_THRESHOLD.match(currentFieldName)) {
+                    precisionThreshold = parser.longValue();
+                } else {
+                    throw new SearchParseException(context, "Unknown key for a " + token + " in [" + name + "]: [" + currentFieldName + "].");
+                }
+            } else {
+                throw new SearchParseException(context, "Unexpected token " + token + " in [" + name + "].");
+            }
+        }
+
+        ValuesSourceConfig<?> config = null;
+
+        if (script != null) {
+            config = new ValuesSourceConfig<BytesValuesSource>(BytesValuesSource.class);
+            config.script(context.scriptService().search(context.lookup(), scriptLang, script, scriptParams));
+        }
+
+        if (field != null) {
+            FieldMapper<?> mapper = context.smartNameFieldMapper(field);
+            if (config == null) {
+                if (mapper instanceof NumberFieldMapper) {
+                    config = new ValuesSourceConfig<NumericValuesSource>(NumericValuesSource.class);
+                } else {
+                    config = new ValuesSourceConfig<BytesValuesSource>(BytesValuesSource.class);
+                    if (mapper == null) {
+                        config.unmapped(true);
+                    }
+                }
+                if (rehash == null && mapper instanceof Murmur3FieldMapper) {
+                    rehash = false;
+                }
+            }
+            if (mapper != null) {
+                IndexFieldData<?> indexFieldData = context.fieldData().getForField(mapper);
+                config.fieldContext(new FieldContext(field, indexFieldData));
+            }
+        } else if (config == null) {
+            config = new ValuesSourceConfig<BytesValuesSource>(BytesValuesSource.class);
+        }
+
+        if (rehash == null) {
+            rehash = true;
+        }
+
+        return new CardinalityAggregatorFactory(name, config, precisionThreshold, rehash);
+    }
+
+}

src/main/java/org/elasticsearch/search/aggregations/metrics/cardinality/InternalCardinality.java

@@ -0,0 +1,138 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics.cardinality;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.aggregations.AggregationStreams;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.metrics.MetricsAggregation;
+import org.elasticsearch.search.aggregations.support.numeric.ValueFormatterStreams;
+
+import java.io.IOException;
+import java.util.List;
+
+public final class InternalCardinality extends MetricsAggregation.SingleValue implements Cardinality {
+
+    public final static Type TYPE = new Type("cardinality");
+
+    public final static AggregationStreams.Stream STREAM = new AggregationStreams.Stream() {
+        @Override
+        public InternalCardinality readResult(StreamInput in) throws IOException {
+            InternalCardinality result = new InternalCardinality();
+            result.readFrom(in);
+            return result;
+        }
+    };
+
+    public static void registerStreams() {
+        AggregationStreams.registerStream(STREAM, TYPE.stream());
+    }
+
+    private HyperLogLogPlusPlus counts;
+
+    InternalCardinality(String name, HyperLogLogPlusPlus counts) {
+        super(name);
+        this.counts = counts;
+    }
+
+    private InternalCardinality() {
+    }
+
+    @Override
+    public double value() {
+        return getValue();
+    }
+
+    @Override
+    public long getValue() {
+        return counts == null ? 0 : counts.cardinality(0);
+    }
+
+    @Override
+    public Type type() {
+        return TYPE;
+    }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        name = in.readString();
+        valueFormatter = ValueFormatterStreams.readOptional(in);
+        if (in.readBoolean()) {
+            counts = HyperLogLogPlusPlus.readFrom(in, BigArrays.NON_RECYCLING_INSTANCE);
+        } else {
+            counts = null;
+        }
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(name);
+        ValueFormatterStreams.writeOptional(valueFormatter, out);
+        if (counts != null) {
+            out.writeBoolean(true);
+            counts.writeTo(0, out);
+        } else {
+            out.writeBoolean(false);
+        }
+    }
+
+    @Override
+    public InternalAggregation reduce(ReduceContext reduceContext) {
+        List<InternalAggregation> aggregations = reduceContext.aggregations();
+        InternalCardinality reduced = null;
+        for (InternalAggregation aggregation : aggregations) {
+            final InternalCardinality cardinality = (InternalCardinality) aggregation;
+            if (cardinality.counts != null) {
+                if (reduced == null) {
+                    reduced = cardinality;
+                } else {
+                    reduced.merge(cardinality);
+                }
+            }
+        }
+
+        if (reduced == null) { // all empty
+            reduced = (InternalCardinality) aggregations.get(0);
+        }
+
+        return reduced;
+    }
+
+    public void merge(InternalCardinality other) {
+        assert counts != null && other != null;
+        counts.merge(0, other.counts, 0);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject(name);
+        final long cardinality = getValue();
+        builder.field(CommonFields.VALUE, cardinality);
+        if (valueFormatter != null) {
+            builder.field(CommonFields.VALUE_AS_STRING, valueFormatter.format(cardinality));
+        }
+        builder.endObject();
+        return builder;
+    }
+
+}

src/test/java/org/elasticsearch/benchmark/search/aggregations/CardinalityAggregationSearchBenchmark.java

@@ -0,0 +1,162 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.benchmark.search.aggregations;
+
+import com.carrotsearch.randomizedtesting.generators.RandomInts;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.StopWatch;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.search.aggregations.metrics.cardinality.Cardinality;
+
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
+import static org.elasticsearch.client.Requests.createIndexRequest;
+import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS;
+import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
+import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
+import static org.elasticsearch.search.aggregations.AggregationBuilders.cardinality;
+
+public class CardinalityAggregationSearchBenchmark {
+
+    private static final Random R = new Random();
+    private static final String CLUSTER_NAME = CardinalityAggregationSearchBenchmark.class.getSimpleName();
+    private static final int NUM_DOCS = 10000000;
+    private static final int LOW_CARD = 1000;
+    private static final int HIGH_CARD = 1000000;
+    private static final int BATCH = 100;
+    private static final int WARM = 5;
+    private static final int RUNS = 10;
+    private static final int ITERS = 5;
+
+    public static void main(String[] args) {
+        Settings settings = settingsBuilder()
+                .put("index.refresh_interval", "-1")
+                .put(SETTING_NUMBER_OF_SHARDS, 5)
+                .put(SETTING_NUMBER_OF_REPLICAS, 0)
+                .build();
+
+        Node[] nodes = new Node[1];
+        for (int i = 0; i < nodes.length; i++) {
+            nodes[i] = nodeBuilder().clusterName(CLUSTER_NAME)
+                    .settings(settingsBuilder().put(settings).put("name", "node" + i))
+                    .node();
+        }
+
+        Node clientNode = nodeBuilder()
+                .clusterName(CLUSTER_NAME)
+                .settings(settingsBuilder().put(settings).put("name", "client")).client(true).node();
+
+        Client client = clientNode.client();
+
+        try {
+            client.admin().indices().create(createIndexRequest("index").settings(settings).mapping("type",
+                    jsonBuilder().startObject().startObject("type").startObject("properties")
+                        .startObject("low_card_str_value")
+                            .field("type", "multi_field")
+                            .startObject("fields")
+                                .startObject("low_card_str_value")
+                                    .field("type", "string")
+                                .endObject()
+                                .startObject("hash")
+                                    .field("type", "murmur3")
+                                .endObject()
+                            .endObject()
+                        .endObject()
+                        .startObject("high_card_str_value")
+                            .field("type", "multi_field")
+                            .startObject("fields")
+                                .startObject("high_card_str_value")
+                                    .field("type", "string")
+                                .endObject()
+                                .startObject("hash")
+                                    .field("type", "murmur3")
+                                .endObject()
+                            .endObject()
+                        .endObject()
+                        .startObject("low_card_num_value")
+                            .field("type", "long")
+                        .endObject()
+                        .startObject("high_card_num_value")
+                            .field("type", "long")
+                        .endObject()
+                    .endObject().endObject().endObject())).actionGet();
+
+            System.out.println("Indexing " + NUM_DOCS + " documents");
+
+            StopWatch stopWatch = new StopWatch().start();
+            for (int i = 0; i < NUM_DOCS; ) {
+                BulkRequestBuilder request = client.prepareBulk();
+                for (int j = 0; j < BATCH && i < NUM_DOCS; ++j) {
+                    final int lowCard = RandomInts.randomInt(R, LOW_CARD);
+                    final int highCard = RandomInts.randomInt(R, HIGH_CARD);
+                    request.add(client.prepareIndex("index", "type", Integer.toString(i)).setSource("low_card_str_value", "str" + lowCard, "high_card_str_value", "str" + highCard, "low_card_num_value", lowCard , "high_card_num_value", highCard));
+                    ++i;
+                }
+                BulkResponse response = request.execute().actionGet();
+                if (response.hasFailures()) {
+                    System.err.println("--> failures...");
+                    System.err.println(response.buildFailureMessage());
+                }
+                if ((i % 100000) == 0) {
+                    System.out.println("--> Indexed " + i + " took " + stopWatch.stop().lastTaskTime());
+                    stopWatch.start();
+                }
+            }
+
+            client.admin().indices().prepareRefresh("index").execute().actionGet();
+        } catch (Exception e) {
+            System.out.println("Index already exists, skipping index creation");
+        }
+
+        ClusterHealthResponse clusterHealthResponse = client.admin().cluster().prepareHealth().setWaitForGreenStatus().setTimeout("10m").execute().actionGet();
+        if (clusterHealthResponse.isTimedOut()) {
+            System.err.println("--> Timed out waiting for cluster health");
+        }
+
+        for (int i = 0; i < WARM + RUNS; ++i) {
+            if (i >= WARM) {
+                System.out.println("RUN " + (i - WARM));
+            }
+            for (String field : new String[] {"low_card_str_value", "low_card_str_value.hash", "high_card_str_value", "high_card_str_value.hash", "low_card_num_value", "high_card_num_value"}) {
+                long start = System.nanoTime();
+                SearchResponse resp = null;
+                for (int j = 0; j < ITERS; ++j) {
+                    resp = client.prepareSearch("index").setSearchType(SearchType.COUNT).addAggregation(cardinality("cardinality").field(field)).execute().actionGet();
+                }
+                long end = System.nanoTime();
+                final long cardinality = ((Cardinality) resp.getAggregations().get("cardinality")).getValue();
+                if (i >= WARM) {
+                    System.out.println(field + "\t" + new TimeValue((end - start) / ITERS, TimeUnit.NANOSECONDS) + "\tcardinality=" + cardinality);
+                }
+            }
+        }
+    }
+
+}

src/test/java/org/elasticsearch/common/hashing/MurmurHash3Tests.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.common.hashing;
+
+import com.google.common.hash.HashCode;
+import com.google.common.hash.Hashing;
+import org.elasticsearch.common.hash.MurmurHash3;
+import org.elasticsearch.test.ElasticsearchTestCase;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.LongBuffer;
+
+public class MurmurHash3Tests extends ElasticsearchTestCase {
+
+    public void testHash128() {
+        final int iters = atLeast(100);
+        for (int i = 0; i < iters; ++i) {
+            final int seed = randomInt();
+            final int offset = randomInt(20);
+            final int len = randomInt(randomBoolean() ? 20 : 200);
+            final byte[] bytes = new byte[len + offset + randomInt(3)];
+            getRandom().nextBytes(bytes);
+            HashCode h1 = Hashing.murmur3_128(seed).hashBytes(bytes, offset, len);;
+            MurmurHash3.Hash128 h2 = MurmurHash3.hash128(bytes, offset, len, seed, new MurmurHash3.Hash128());
+            assertEquals(h1, h2);
+        }
+    }
+
+    private void assertEquals(HashCode h1, MurmurHash3.Hash128 h2) {
+        final LongBuffer longs = ByteBuffer.wrap(h1.asBytes()).order(ByteOrder.LITTLE_ENDIAN).asLongBuffer();
+        assertEquals(2, longs.limit());
+        assertEquals(h1.asLong(), h2.h1);
+        assertEquals(longs.get(), h2.h1);
+        assertEquals(longs.get(), h2.h2);
+    }
+
+}

src/test/java/org/elasticsearch/common/util/BigArraysTests.java

@@ -144,6 +144,27 @@ public class BigArraysTests extends ElasticsearchTestCase {
         array.release();
     }
 
+    public void testByteArrayFill() {
+        final int len = randomIntBetween(1, 100000);
+        final int fromIndex = randomIntBetween(0, len - 1);
+        final int toIndex = randomBoolean()
+            ? Math.min(fromIndex + randomInt(100), len) // single page
+            : randomIntBetween(fromIndex, len); // likely multiple pages
+        final ByteArray array2 = bigArrays.newByteArray(len, randomBoolean());
+        final byte[] array1 = new byte[len];
+        for (int i = 0; i < len; ++i) {
+            array1[i] = randomByte();
+            array2.set(i, array1[i]);
+        }
+        final byte rand = randomByte();
+        Arrays.fill(array1, fromIndex, toIndex, rand);
+        array2.fill(fromIndex, toIndex, rand);
+        for (int i = 0; i < len; ++i) {
+            assertEquals(array1[i], array2.get(i), 0.001d);
+        }
+        array2.release();
+    }
+
     public void testFloatArrayFill() {
         final int len = randomIntBetween(1, 100000);
         final int fromIndex = randomIntBetween(0, len - 1);

src/test/java/org/elasticsearch/search/aggregations/metrics/CardinalityTests.java

@@ -0,0 +1,431 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics;
+
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.search.aggregations.bucket.terms.Terms;
+import org.elasticsearch.search.aggregations.metrics.cardinality.Cardinality;
+import org.elasticsearch.test.ElasticsearchIntegrationTest;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.search.aggregations.AggregationBuilders.cardinality;
+import static org.elasticsearch.search.aggregations.AggregationBuilders.terms;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertSearchResponse;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.core.IsNull.notNullValue;
+
+public class CardinalityTests extends ElasticsearchIntegrationTest {
+
+    @Override
+    public Settings indexSettings() {
+        return ImmutableSettings.builder()
+                .put("index.number_of_shards", between(1, 5))
+                .put("index.number_of_replicas", between(0, 1))
+                .build();
+    }
+
+    long numDocs;
+    long precisionThreshold;
+
+    @Before
+    public void init() throws Exception {
+
+        prepareCreate("idx").addMapping("type",
+                jsonBuilder().startObject().startObject("type").startObject("properties")
+                    .startObject("str_value")
+                        .field("type", "string")
+                        .startObject("fields")
+                            .startObject("hash")
+                                .field("type", "murmur3")
+                            .endObject()
+                        .endObject()
+                    .endObject()
+                    .startObject("str_values")
+                        .field("type", "string")
+                        .startObject("fields")
+                            .startObject("hash")
+                                .field("type", "murmur3")
+                            .endObject()
+                        .endObject()
+                    .endObject()
+                    .startObject("l_value")
+                        .field("type", "long")
+                        .startObject("fields")
+                            .startObject("hash")
+                                .field("type", "murmur3")
+                            .endObject()
+                        .endObject()
+                    .endObject()
+                    .startObject("l_values")
+                        .field("type", "long")
+                        .startObject("fields")
+                            .startObject("hash")
+                                .field("type", "murmur3")
+                            .endObject()
+                        .endObject()
+                    .endObject()
+                        .startObject("d_value")
+                            .field("type", "double")
+                            .startObject("fields")
+                                .startObject("hash")
+                                    .field("type", "murmur3")
+                                .endObject()
+                            .endObject()
+                        .endObject()
+                        .startObject("d_values")
+                            .field("type", "double")
+                                .startObject("fields")
+                                    .startObject("hash")
+                                        .field("type", "murmur3")
+                                    .endObject()
+                                .endObject()
+                            .endObject()
+                        .endObject()
+                        .endObject().endObject()).execute().actionGet();
+
+        numDocs = randomIntBetween(2, 100);
+        precisionThreshold = randomIntBetween(0, 1 << randomInt(20));
+        IndexRequestBuilder[] builders = new IndexRequestBuilder[(int) numDocs];
+        for (int i = 0; i < numDocs; ++i) {
+            builders[i] = client().prepareIndex("idx", "type").setSource(jsonBuilder()
+                    .startObject()
+                        .field("str_value", "s" + i)
+                        .field("str_values", new String[]{"s" + (i * 2), "s" + (i * 2 + 1)})
+                        .field("l_value", i)
+                        .field("l_values", new int[] {i * 2, i * 2 + 1})
+                        .field("d_value", i)
+                        .field("d_values", new double[]{i * 2, i * 2 + 1})
+                    .endObject());
+        }
+        indexRandom(true, builders);
+        createIndex("idx_unmapped");
+        ensureSearchable();
+    }
+
+    private void assertCount(Cardinality count, long value) {
+        if (value <= precisionThreshold) {
+            // linear counting should be picked, and should be accurate
+            assertEquals(value, count.getValue());
+        } else {
+            // error is not bound, so let's just make sure it is > 0
+            assertThat(count.getValue(), greaterThan(0L));
+        }
+    }
+     private String singleNumericField(boolean hash) {
+        return (randomBoolean() ? "l_value" : "d_value") + (hash ? ".hash" : "");
+    }
+
+    private String multiNumericField(boolean hash) {
+        return (randomBoolean() ? "l_values" : "d_values") + (hash ? ".hash" : "");
+    }
+
+    @Test
+    public void unmapped() throws Exception {
+        SearchResponse response = client().prepareSearch("idx_unmapped").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, 0);
+    }
+
+    @Test
+    public void partiallyUnmapped() throws Exception {
+        SearchResponse response = client().prepareSearch("idx", "idx_unmapped").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void singleValuedString() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void singleValuedStringHashed() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_value.hash"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void singleValuedNumeric() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field(singleNumericField(false)))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void singleValuedNumericHashed() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field(singleNumericField(true)))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void multiValuedString() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_values"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void multiValuedStringHashed() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_values.hash"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void multiValuedNumeric() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field(multiNumericField(false)))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void multiValuedNumericHashed() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field(multiNumericField(true)))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void singleValuedStringScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).script("doc['str_value'].value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void multiValuedStringScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).script("doc['str_values'].values"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void singleValuedNumericScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).script("doc['" + singleNumericField(false) + "'].value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void multiValuedNumericScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).script("doc['" + multiNumericField(false) + "'].values"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void singleValuedStringValueScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_value").script("_value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void multiValuedStringValueScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_values").script("_value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void singleValuedNumericValueScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field(singleNumericField(false)).script("_value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs);
+    }
+
+    @Test
+    public void multiValuedNumericValueScript() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field(multiNumericField(false)).script("_value"))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Cardinality count = response.getAggregations().get("cardinality");
+        assertThat(count, notNullValue());
+        assertThat(count.getName(), equalTo("cardinality"));
+        assertCount(count, numDocs * 2);
+    }
+
+    @Test
+    public void asSubAgg() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(terms("terms").field("str_value").subAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_values")))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Terms terms = response.getAggregations().get("terms");
+        for (Terms.Bucket bucket : terms.getBuckets()) {
+            Cardinality count = bucket.getAggregations().get("cardinality");
+            assertThat(count, notNullValue());
+            assertThat(count.getName(), equalTo("cardinality"));
+            assertCount(count, 2);
+        }
+    }
+
+    @Test
+    public void asSubAggHashed() throws Exception {
+        SearchResponse response = client().prepareSearch("idx").setTypes("type")
+                .addAggregation(terms("terms").field("str_value").subAggregation(cardinality("cardinality").precisionThreshold(precisionThreshold).field("str_values.hash")))
+                .execute().actionGet();
+
+        assertSearchResponse(response);
+
+        Terms terms = response.getAggregations().get("terms");
+        for (Terms.Bucket bucket : terms.getBuckets()) {
+            Cardinality count = bucket.getAggregations().get("cardinality");
+            assertThat(count, notNullValue());
+            assertThat(count.getName(), equalTo("cardinality"));
+            assertCount(count, 2);
+        }
+    }
+
+}

src/test/java/org/elasticsearch/search/aggregations/metrics/cardinality/HyperLogLogPlusPlusTests.java

@@ -0,0 +1,137 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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 org.elasticsearch.search.aggregations.metrics.cardinality;
+
+import com.carrotsearch.hppc.IntOpenHashSet;
+import com.carrotsearch.hppc.hash.MurmurHash3;
+import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.test.ElasticsearchTestCase;
+import org.junit.Test;
+
+import static org.elasticsearch.search.aggregations.metrics.cardinality.HyperLogLogPlusPlus.MAX_PRECISION;
+import static org.elasticsearch.search.aggregations.metrics.cardinality.HyperLogLogPlusPlus.MIN_PRECISION;
+import static org.hamcrest.Matchers.closeTo;
+
+public class HyperLogLogPlusPlusTests extends ElasticsearchTestCase {
+
+    @Test
+    public void encodeDecode() {
+        final int iters = atLeast(100000);
+        // random hashes
+        for (int i = 0; i < iters; ++i) {
+            final int p1 = randomIntBetween(4, 24);
+            final long hash = randomLong();
+            testEncodeDecode(p1, hash);
+        }
+        // special cases
+        for (int p1 = MIN_PRECISION; p1 <= MAX_PRECISION; ++p1) {
+            testEncodeDecode(p1, 0);
+            testEncodeDecode(p1, 1);
+            testEncodeDecode(p1, ~0L);
+        }
+    }
+
+    private void testEncodeDecode(int p1, long hash) {
+        final long index = HyperLogLogPlusPlus.index(hash, p1);
+        final int runLen = HyperLogLogPlusPlus.runLen(hash, p1);
+        final int encoded = HyperLogLogPlusPlus.encodeHash(hash, p1);
+        assertEquals(index, HyperLogLogPlusPlus.decodeIndex(encoded, p1));
+        assertEquals(runLen, HyperLogLogPlusPlus.decodeRunLen(encoded, p1));
+    }
+
+    @Test
+    public void accuracy() {
+        final long bucket = randomInt(20);
+        final int numValues = randomIntBetween(1, 1000000);
+        final int maxValue = randomIntBetween(1, randomBoolean() ? 1000: 1000000);
+        final int p = randomIntBetween(14, MAX_PRECISION);
+        IntOpenHashSet set = new IntOpenHashSet();
+        HyperLogLogPlusPlus e = new HyperLogLogPlusPlus(p, BigArrays.NON_RECYCLING_INSTANCE, 1);
+        for (int i = 0; i < numValues; ++i) {
+            final int n = randomInt(maxValue);
+            set.add(n);
+            final long hash = MurmurHash3.hash((long) n);
+            e.collect(bucket, hash);
+            if (randomInt(100) == 0) {
+                //System.out.println(e.cardinality(bucket) + " <> " + set.size());
+                assertThat((double) e.cardinality(bucket), closeTo(set.size(), 0.1 * set.size()));
+            }
+        }
+        assertThat((double) e.cardinality(bucket), closeTo(set.size(), 0.1 * set.size()));
+    }
+
+    @Test
+    public void merge() {
+        final int p = randomIntBetween(MIN_PRECISION, MAX_PRECISION);
+        final HyperLogLogPlusPlus single = new HyperLogLogPlusPlus(p, BigArrays.NON_RECYCLING_INSTANCE, 0);
+        final HyperLogLogPlusPlus[] multi = new HyperLogLogPlusPlus[randomIntBetween(2, 100)];
+        final long[] bucketOrds = new long[multi.length];
+        for (int i = 0; i < multi.length; ++i) {
+            bucketOrds[i] = randomInt(20);
+            multi[i] = new HyperLogLogPlusPlus(p, BigArrays.NON_RECYCLING_INSTANCE, 5);
+        }
+        final int numValues = randomIntBetween(1, 100000);
+        final int maxValue = randomIntBetween(1, randomBoolean() ? 1000: 1000000);
+        for (int i = 0; i < numValues; ++i) {
+            final int n = randomInt(maxValue);
+            final long hash = MurmurHash3.hash((long) n);
+            single.collect(0, hash);
+            // use a gaussian so that all instances don't collect as many hashes
+            final int index = (int) (Math.pow(randomDouble(), 2));
+            multi[index].collect(bucketOrds[index], hash);
+            if (randomInt(100) == 0) {
+                HyperLogLogPlusPlus merged = new HyperLogLogPlusPlus(p, BigArrays.NON_RECYCLING_INSTANCE, 0);
+                for (int j = 0; j < multi.length; ++j) {
+                    merged.merge(0, multi[j], bucketOrds[j]);
+                }
+                assertEquals(single.cardinality(0), merged.cardinality(0));
+            }
+        }
+    }
+
+    @Test
+    public void fakeHashes() {
+        // hashes with lots of leading zeros trigger different paths in the code that we try to go through here
+        final int p = randomIntBetween(MIN_PRECISION, MAX_PRECISION);
+        final HyperLogLogPlusPlus counts = new HyperLogLogPlusPlus(p, BigArrays.NON_RECYCLING_INSTANCE, 0);
+
+        counts.collect(0, 0);
+        assertEquals(1, counts.cardinality(0));
+        if (randomBoolean()) {
+            counts.collect(0, 1);
+            assertEquals(2, counts.cardinality(0));
+        }
+        counts.upgradeToHll(0);
+        // all hashes felt into the same bucket so hll would expect a count of 1
+        assertEquals(1, counts.cardinality(0));
+    }
+
+    @Test
+    public void precisionFromThreshold() {
+        assertEquals(4, HyperLogLogPlusPlus.precisionFromThreshold(0));
+        assertEquals(6, HyperLogLogPlusPlus.precisionFromThreshold(10));
+        assertEquals(10, HyperLogLogPlusPlus.precisionFromThreshold(100));
+        assertEquals(13, HyperLogLogPlusPlus.precisionFromThreshold(1000));
+        assertEquals(16, HyperLogLogPlusPlus.precisionFromThreshold(10000));
+        assertEquals(18, HyperLogLogPlusPlus.precisionFromThreshold(100000));
+        assertEquals(18, HyperLogLogPlusPlus.precisionFromThreshold(1000000));
+    }
+
+}

src/test/java/org/elasticsearch/test/cache/recycler/MockBigArrays.java

@@ -298,6 +298,11 @@ public class MockBigArrays extends BigArrays {
             in.set(index, buf, offset, len);
         }
 
+        @Override
+        public void fill(long fromIndex, long toIndex, byte value) {
+            in.fill(fromIndex, toIndex, value);
+        }
+
     }
 
     private class IntArrayWrapper extends AbstractArrayWrapper implements IntArray {