Aggregations: Return the sum of the doc counts of other buckets.

This commit adds a new field to the response of the terms aggregation called
`sum_other_doc_count` which is equal to the sum of the doc counts of the buckets
that did not make it to the list of top buckets. It is typically useful to have
a sector called eg. `other` when using terms aggregations to build pie charts.

Example query and response:

```json
GET test/_search?search_type=count
{
  "aggs": {
    "colors": {
      "terms": {
        "field": "color",
        "size": 3
      }
    }
  }
}
```

```json
{
   [...],
   "aggregations": {
      "colors": {
         "doc_count_error_upper_bound": 0,
         "sum_other_doc_count": 4,
         "buckets": [
            {
               "key": "blue",
               "doc_count": 65
            },
            {
               "key": "red",
               "doc_count": 14
            },
            {
               "key": "brown",
               "doc_count": 3
            }
         ]
      }
   }
}
```

Close elastic#8213

Author: jpountz

docs/reference/search/aggregations/bucket/terms-aggregation.asciidoc

@@ -25,7 +25,9 @@ Response:
 
     "aggregations" : {
         "genders" : {
-            "buckets" : [
+            "doc_count_error_upper_bound": 0, <1>
+            "sum_other_doc_count": 0, <2>
+            "buckets" : [ <3>
                 {
                     "key" : "male",
                     "doc_count" : 10
@@ -40,6 +42,10 @@ Response:
 }
 --------------------------------------------------
 
+<1> an upper bound of the error on the document counts for each term, see <<search-aggregations-bucket-terms-aggregation-approximate-counts,below>>
+<2> when there are lots of unique terms, elasticsearch only returns the top terms; this number is the sum of the document counts for all buckets that are not part of the response
+<3> the list of the top buckets, the meaning of `top` being defined by the <<search-aggregations-bucket-terms-aggregation-order,order>>
+
 By default, the `terms` aggregation will return the buckets for the top ten terms ordered by the `doc_count`. One can
 change this default behaviour by setting the `size` parameter.
 
@@ -52,6 +58,7 @@ This means that if the number of unique terms is greater than `size`, the return
 (it could be that the term counts are slightly off and it could even be that a term that should have been in the top
 size buckets was not returned). If set to `0`, the `size` will be set to `Integer.MAX_VALUE`.
 
+[[search-aggregations-bucket-terms-aggregation-approximate-counts]]
 ==== Document counts are approximate
 
 As described above, the document counts (and the results of any sub aggregations) in the terms aggregation are not always
@@ -226,6 +233,7 @@ does not return a particular term which appears in the results from another shar
 aggregation is either sorted by a sub aggregation or in order of ascending document count, the error in the document counts cannot be
 determined and is given a value of -1 to indicate this.
 
+[[search-aggregations-bucket-terms-aggregation-order]]
 ==== Order
 
 The order of the buckets can be customized by setting the `order` parameter. By default, the buckets are ordered by

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/AbstractStringTermsAggregator.java

@@ -45,7 +45,7 @@ public boolean shouldCollect() {
 
     @Override
     public InternalAggregation buildEmptyAggregation() {
-        return new StringTerms(name, order, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Collections.<InternalTerms.Bucket>emptyList(), showTermDocCountError, 0);
+        return new StringTerms(name, order, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Collections.<InternalTerms.Bucket>emptyList(), showTermDocCountError, 0, 0);
     }
 
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/DoubleTerms.java

@@ -98,8 +98,8 @@ Bucket newBucket(long docCount, InternalAggregations aggs, long docCountError) {
 
     DoubleTerms() {} // for serialization
 
-    public DoubleTerms(String name, InternalOrder order, @Nullable ValueFormatter formatter, int requiredSize, int shardSize, long minDocCount, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError) {
-        super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError);
+    public DoubleTerms(String name, InternalOrder order, @Nullable ValueFormatter formatter, int requiredSize, int shardSize, long minDocCount, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount) {
+        super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount);
         this.formatter = formatter;
     }
 
@@ -109,8 +109,8 @@ public Type type() {
     }
 
     @Override
-    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError) {
-        return new DoubleTerms(name, order, formatter, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError);
+    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount) {
+        return new DoubleTerms(name, order, formatter, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount);
     }
 
     @Override
@@ -132,6 +132,9 @@ public void readFrom(StreamInput in) throws IOException {
             this.showTermDocCountError = false;
         }
         this.minDocCount = in.readVLong();
+        if (in.getVersion().onOrAfter(Version.V_1_4_0)) {
+            this.otherDocCount = in.readVLong();
+        }
         int size = in.readVInt();
         List<InternalTerms.Bucket> buckets = new ArrayList<>(size);
         for (int i = 0; i < size; i++) {
@@ -162,6 +165,9 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(showTermDocCountError);
         }
         out.writeVLong(minDocCount);
+        if (out.getVersion().onOrAfter(Version.V_1_4_0)) {
+            out.writeVLong(otherDocCount);
+        }
         out.writeVInt(buckets.size());
         for (InternalTerms.Bucket bucket : buckets) {
             out.writeDouble(((Bucket) bucket).term);
@@ -176,6 +182,7 @@ public void writeTo(StreamOutput out) throws IOException {
     @Override
     public XContentBuilder doXContentBody(XContentBuilder builder, Params params) throws IOException {
         builder.field(InternalTerms.DOC_COUNT_ERROR_UPPER_BOUND_FIELD_NAME, docCountError);
+        builder.field(SUM_OF_OTHER_DOC_COUNTS, otherDocCount);
         builder.startArray(CommonFields.BUCKETS);
         for (InternalTerms.Bucket bucket : buckets) {
             builder.startObject();

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/DoubleTermsAggregator.java

@@ -69,7 +69,7 @@ private static DoubleTerms convertToDouble(LongTerms terms) {
         for (int i = 0; i < buckets.length; ++i) {
             buckets[i] = convertToDouble(buckets[i]);
         }
-        return new DoubleTerms(terms.getName(), terms.order, terms.formatter, terms.requiredSize, terms.shardSize, terms.minDocCount, Arrays.asList(buckets), terms.showTermDocCountError, terms.docCountError);
+        return new DoubleTerms(terms.getName(), terms.order, terms.formatter, terms.requiredSize, terms.shardSize, terms.minDocCount, Arrays.asList(buckets), terms.showTermDocCountError, terms.docCountError, terms.otherDocCount);
     }
 
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/GlobalOrdinalsStringTermsAggregator.java

@@ -150,6 +150,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
         } else {
             size = (int) Math.min(maxBucketOrd(), bucketCountThresholds.getShardSize());
         }
+        long otherDocCount = 0;
         BucketPriorityQueue ordered = new BucketPriorityQueue(size, order.comparator(this));
         OrdBucket spare = new OrdBucket(-1, 0, null, showTermDocCountError, 0);
         for (long globalTermOrd = 0; globalTermOrd < globalOrds.getValueCount(); ++globalTermOrd) {
@@ -161,6 +162,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
             if (bucketCountThresholds.getMinDocCount() > 0 && bucketDocCount == 0) {
                 continue;
             }
+            otherDocCount += bucketDocCount;
             spare.globalOrd = globalTermOrd;
             spare.bucketOrd = bucketOrd;
             spare.docCount = bucketDocCount;
@@ -182,6 +184,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
             copy(globalOrds.lookupOrd(bucket.globalOrd), scratch);
             list[i] = new StringTerms.Bucket(scratch, bucket.docCount, null, showTermDocCountError, 0);
             list[i].bucketOrd = bucket.bucketOrd;
+            otherDocCount -= list[i].docCount;
         }
         //replay any deferred collections
         runDeferredCollections(survivingBucketOrds);
@@ -193,7 +196,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
           bucket.docCountError = 0;
         }
 
-        return new StringTerms(name, order, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Arrays.asList(list), showTermDocCountError, 0);
+        return new StringTerms(name, order, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Arrays.asList(list), showTermDocCountError, 0, otherDocCount);
     }
 
     /** This is used internally only, just for compare using global ordinal instead of term bytes in the PQ */

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalTerms.java

@@ -37,6 +37,7 @@
 public abstract class InternalTerms extends InternalAggregation implements Terms, ToXContent, Streamable {
 
     protected static final String DOC_COUNT_ERROR_UPPER_BOUND_FIELD_NAME = "doc_count_error_upper_bound";
+    protected static final String SUM_OF_OTHER_DOC_COUNTS = "sum_other_doc_count";
 
     public static abstract class Bucket extends Terms.Bucket {
 
@@ -104,10 +105,11 @@ public Bucket reduce(List<? extends Bucket> buckets, ReduceContext context) {
     protected Map<String, Bucket> bucketMap;
     protected long docCountError;
     protected boolean showTermDocCountError;
+    protected long otherDocCount;
 
     protected InternalTerms() {} // for serialization
 
-    protected InternalTerms(String name, InternalOrder order, int requiredSize, int shardSize, long minDocCount, List<Bucket> buckets, boolean showTermDocCountError, long docCountError) {
+    protected InternalTerms(String name, InternalOrder order, int requiredSize, int shardSize, long minDocCount, List<Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount) {
         super(name);
         this.order = order;
         this.requiredSize = requiredSize;
@@ -116,6 +118,7 @@ protected InternalTerms(String name, InternalOrder order, int requiredSize, int
         this.buckets = buckets;
         this.showTermDocCountError = showTermDocCountError;
         this.docCountError = docCountError;
+        this.otherDocCount = otherDocCount;
     }
 
     @Override
@@ -139,14 +142,21 @@ public long getDocCountError() {
         return docCountError;
     }
 
+    @Override
+    public long getSumOfOtherDocCounts() {
+        return otherDocCount;
+    }
+
     @Override
     public InternalAggregation reduce(ReduceContext reduceContext) {
         List<InternalAggregation> aggregations = reduceContext.aggregations();
 
         Multimap<Object, InternalTerms.Bucket> buckets = ArrayListMultimap.create();
         long sumDocCountError = 0;
+        long otherDocCount = 0;
         for (InternalAggregation aggregation : aggregations) {
             InternalTerms terms = (InternalTerms) aggregation;
+            otherDocCount += terms.getSumOfOtherDocCounts();
             final long thisAggDocCountError;
             if (terms.buckets.size() < this.shardSize || this.order == InternalOrder.TERM_ASC || this.order == InternalOrder.TERM_DESC) {
                 thisAggDocCountError = 0;
@@ -182,7 +192,10 @@ public InternalAggregation reduce(ReduceContext reduceContext) {
                 }
             }
             if (b.docCount >= minDocCount) {
-                ordered.insertWithOverflow(b);
+                Terms.Bucket removed = ordered.insertWithOverflow(b);
+                if (removed != null) {
+                    otherDocCount += removed.getDocCount();
+                }
             }
         }
         Bucket[] list = new Bucket[ordered.size()];
@@ -195,9 +208,9 @@ public InternalAggregation reduce(ReduceContext reduceContext) {
         } else {
             docCountError = aggregations.size() == 1 ? 0 : sumDocCountError;
         }
-        return newAggregation(name, Arrays.asList(list), showTermDocCountError, docCountError);
+        return newAggregation(name, Arrays.asList(list), showTermDocCountError, docCountError, otherDocCount);
     }
 
-    protected abstract InternalTerms newAggregation(String name, List<Bucket> buckets, boolean showTermDocCountError, long docCountError);
+    protected abstract InternalTerms newAggregation(String name, List<Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount);
 
 }

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/LongTerms.java

@@ -99,8 +99,8 @@ Bucket newBucket(long docCount, InternalAggregations aggs, long docCountError) {
 
     LongTerms() {} // for serialization
 
-    public LongTerms(String name, InternalOrder order, @Nullable ValueFormatter formatter, int requiredSize, int shardSize, long minDocCount, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError) {
-        super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError);
+    public LongTerms(String name, InternalOrder order, @Nullable ValueFormatter formatter, int requiredSize, int shardSize, long minDocCount, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount) {
+        super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount);
         this.formatter = formatter;
     }
 
@@ -110,8 +110,8 @@ public Type type() {
     }
 
     @Override
-    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError) {
-        return new LongTerms(name, order, formatter, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError);
+    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount) {
+        return new LongTerms(name, order, formatter, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount);
     }
 
     @Override
@@ -133,6 +133,9 @@ public void readFrom(StreamInput in) throws IOException {
             this.showTermDocCountError = false;
         }
         this.minDocCount = in.readVLong();
+        if (in.getVersion().onOrAfter(Version.V_1_4_0)) {
+            this.otherDocCount = in.readVLong();
+        }
         int size = in.readVInt();
         List<InternalTerms.Bucket> buckets = new ArrayList<>(size);
         for (int i = 0; i < size; i++) {
@@ -163,6 +166,9 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(showTermDocCountError);
         }
         out.writeVLong(minDocCount);
+        if (out.getVersion().onOrAfter(Version.V_1_4_0)) {
+            out.writeVLong(otherDocCount);
+        }
         out.writeVInt(buckets.size());
         for (InternalTerms.Bucket bucket : buckets) {
             out.writeLong(((Bucket) bucket).term);
@@ -177,6 +183,7 @@ public void writeTo(StreamOutput out) throws IOException {
     @Override
     public XContentBuilder doXContentBody(XContentBuilder builder, Params params) throws IOException {
         builder.field(InternalTerms.DOC_COUNT_ERROR_UPPER_BOUND_FIELD_NAME, docCountError);
+        builder.field(SUM_OF_OTHER_DOC_COUNTS, otherDocCount);
         builder.startArray(CommonFields.BUCKETS);
         for (InternalTerms.Bucket bucket : buckets) {
             builder.startObject();

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/LongTermsAggregator.java

@@ -115,6 +115,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
 
         final int size = (int) Math.min(bucketOrds.size(), bucketCountThresholds.getShardSize());
 
+        long otherDocCount = 0;
         BucketPriorityQueue ordered = new BucketPriorityQueue(size, order.comparator(this));
         LongTerms.Bucket spare = null;
         for (long i = 0; i < bucketOrds.size(); i++) {
@@ -123,6 +124,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
             }
             spare.term = bucketOrds.get(i);
             spare.docCount = bucketDocCount(i);
+            otherDocCount += spare.docCount;
             spare.bucketOrd = i;
             if (bucketCountThresholds.getShardMinDocCount() <= spare.docCount) {
                 spare = (LongTerms.Bucket) ordered.insertWithOverflow(spare);
@@ -136,6 +138,7 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
             final LongTerms.Bucket bucket = (LongTerms.Bucket) ordered.pop();
             survivingBucketOrds[i] = bucket.bucketOrd;
             list[i] = bucket;
+            otherDocCount -= bucket.docCount;
         }
 
         runDeferredCollections(survivingBucketOrds);
@@ -146,13 +149,13 @@ public InternalAggregation buildAggregation(long owningBucketOrdinal) {
           list[i].docCountError = 0;
         }
 
-        return new LongTerms(name, order, formatter, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Arrays.asList(list), showTermDocCountError, 0);
+        return new LongTerms(name, order, formatter, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Arrays.asList(list), showTermDocCountError, 0, otherDocCount);
     }
 
 
     @Override
     public InternalAggregation buildEmptyAggregation() {
-        return new LongTerms(name, order, formatter, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Collections.<InternalTerms.Bucket>emptyList(), showTermDocCountError, 0);
+        return new LongTerms(name, order, formatter, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getShardSize(), bucketCountThresholds.getMinDocCount(), Collections.<InternalTerms.Bucket>emptyList(), showTermDocCountError, 0, 0);
     }
 
     @Override

src/main/java/org/elasticsearch/search/aggregations/bucket/terms/StringTerms.java

@@ -98,8 +98,8 @@ Bucket newBucket(long docCount, InternalAggregations aggs, long docCountError) {
 
     StringTerms() {} // for serialization
 
-    public StringTerms(String name, InternalOrder order, int requiredSize, int shardSize, long minDocCount, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError) {
-        super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError);
+    public StringTerms(String name, InternalOrder order, int requiredSize, int shardSize, long minDocCount, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount) {
+        super(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount);
     }
 
     @Override
@@ -108,8 +108,8 @@ public Type type() {
     }
 
     @Override
-    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError) {
-        return new StringTerms(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError);
+    protected InternalTerms newAggregation(String name, List<InternalTerms.Bucket> buckets, boolean showTermDocCountError, long docCountError, long otherDocCount) {
+        return new StringTerms(name, order, requiredSize, shardSize, minDocCount, buckets, showTermDocCountError, docCountError, otherDocCount);
     }
 
     @Override
@@ -130,6 +130,9 @@ public void readFrom(StreamInput in) throws IOException {
             this.showTermDocCountError = false;
         }
         this.minDocCount = in.readVLong();
+        if (in.getVersion().onOrAfter(Version.V_1_4_0)) {
+            this.otherDocCount = in.readVLong();
+        }
         int size = in.readVInt();
         List<InternalTerms.Bucket> buckets = new ArrayList<>(size);
         for (int i = 0; i < size; i++) {
@@ -159,6 +162,9 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(showTermDocCountError);
         }
         out.writeVLong(minDocCount);
+        if (out.getVersion().onOrAfter(Version.V_1_4_0)) {
+            out.writeVLong(otherDocCount);
+        }
         out.writeVInt(buckets.size());
         for (InternalTerms.Bucket bucket : buckets) {
             out.writeBytesRef(((Bucket) bucket).termBytes);
@@ -173,6 +179,7 @@ public void writeTo(StreamOutput out) throws IOException {
     @Override
     public XContentBuilder doXContentBody(XContentBuilder builder, Params params) throws IOException {
         builder.field(InternalTerms.DOC_COUNT_ERROR_UPPER_BOUND_FIELD_NAME, docCountError);
+        builder.field(SUM_OF_OTHER_DOC_COUNTS, otherDocCount);
         builder.startArray(CommonFields.BUCKETS);
         for (InternalTerms.Bucket bucket : buckets) {
             builder.startObject();