- fixed terms aggs ordering by sub metric agg - introduced MetricsAggregator to enable access to metrics values without creating buckets

fixes: elastic#4643

Author: uboness

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

@@ -106,6 +106,10 @@ public Aggregator parent() {
         return parent;
     }
 
+    public Aggregator[] subAggregators() {
+        return subAggregators;
+    }
+
     /**
      * @return  The current aggregation context.
      */

src/main/java/org/elasticsearch/search/aggregations/bucket/Bucket.java

@@ -35,13 +35,13 @@ public interface Bucket {
 
     Aggregations getAggregations();
 
-    static class Comparator<B extends Bucket> implements java.util.Comparator<B> {
+    static class SubAggregationComparator<B extends Bucket> implements java.util.Comparator<B> {
 
         private final String aggName;
         private final String valueName;
         private final boolean asc;
 
-        public Comparator(String expression, boolean asc) {
+        public SubAggregationComparator(String expression, boolean asc) {
             this.asc = asc;
             int i = expression.indexOf('.');
             if (i < 0) {
@@ -53,7 +53,7 @@ public Comparator(String expression, boolean asc) {
             }
         }
 
-        public Comparator(String aggName, String valueName, boolean asc) {
+        public SubAggregationComparator(String aggName, String valueName, boolean asc) {
             this.aggName = aggName;
             this.valueName = valueName;
             this.asc = asc;

src/main/java/org/elasticsearch/search/aggregations/bucket/BucketsAggregator.java

@@ -29,7 +29,9 @@
 import org.elasticsearch.search.aggregations.support.AggregationContext;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 
 /**
  *
@@ -38,10 +40,19 @@ public abstract class BucketsAggregator extends Aggregator {
 
     private LongArray docCounts;
 
+    private final Aggregator[] collectableSugAggregators;
+
     public BucketsAggregator(String name, BucketAggregationMode bucketAggregationMode, AggregatorFactories factories,
                              long estimatedBucketsCount, AggregationContext context, Aggregator parent) {
         super(name, bucketAggregationMode, factories, estimatedBucketsCount, context, parent);
         docCounts = BigArrays.newLongArray(estimatedBucketsCount, context.pageCacheRecycler(), true);
+        List<Aggregator> collectables = new ArrayList<Aggregator>(subAggregators.length);
+        for (int i = 0; i < subAggregators.length; i++) {
+            if (subAggregators[i].shouldCollect()) {
+                collectables.add((subAggregators[i]));
+            }
+        }
+        collectableSugAggregators = collectables.toArray(new Aggregator[collectables.size()]);
     }
 
     /**
@@ -50,17 +61,17 @@ public BucketsAggregator(String name, BucketAggregationMode bucketAggregationMod
     protected final void collectBucket(int doc, long bucketOrd) throws IOException {
         docCounts = BigArrays.grow(docCounts, bucketOrd + 1);
         docCounts.increment(bucketOrd, 1);
-        for (int i = 0; i < subAggregators.length; i++) {
-            subAggregators[i].collect(doc, bucketOrd);
+        for (int i = 0; i < collectableSugAggregators.length; i++) {
+            collectableSugAggregators[i].collect(doc, bucketOrd);
         }
     }
 
     /**
      * Utility method to collect the given doc in the given bucket but not to update the doc counts of the bucket
      */
     protected final void collectBucketNoCounts(int doc, long bucketOrd) throws IOException {
-        for (int i = 0; i < subAggregators.length; i++) {
-            subAggregators[i].collect(doc, bucketOrd);
+        for (int i = 0; i < collectableSugAggregators.length; i++) {
+            collectableSugAggregators[i].collect(doc, bucketOrd);
         }
     }
 

src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/InternalOrder.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.search.aggregations.bucket.Bucket;
 
 import java.io.IOException;
 import java.util.Comparator;
@@ -70,11 +71,11 @@ static class Aggregation extends InternalOrder {
         static final byte ID = 0;
 
         Aggregation(String key, boolean asc) {
-            super(ID, key, asc, new HistogramBase.Bucket.Comparator<HistogramBase.Bucket>(key, asc));
+            super(ID, key, asc, new Bucket.SubAggregationComparator<HistogramBase.Bucket>(key, asc));
         }
 
         Aggregation(String aggName, String valueName, boolean asc) {
-            super(ID, key(aggName, valueName), asc, new HistogramBase.Bucket.Comparator<HistogramBase.Bucket>(aggName, valueName, asc));
+            super(ID, key(aggName, valueName), asc, new Bucket.SubAggregationComparator<HistogramBase.Bucket>(aggName, valueName, asc));
         }
 
         private static String key(String aggName, String valueName) {

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

@@ -147,7 +147,7 @@ public InternalTerms reduce(ReduceContext reduceContext) {
 
         // TODO: would it be better to sort the backing array buffer of hppc map directly instead of using a PQ?
         final int size = Math.min(requiredSize, buckets.v().size());
-        BucketPriorityQueue ordered = new BucketPriorityQueue(size, order.comparator());
+        BucketPriorityQueue ordered = new BucketPriorityQueue(size, order.comparator(null));
         boolean[] states = buckets.v().allocated;
         Object[] internalBuckets = buckets.v().values;
         for (int i = 0; i < states.length; i++) {

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

@@ -23,7 +23,6 @@
 import org.elasticsearch.index.fielddata.DoubleValues;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
-import org.elasticsearch.search.aggregations.InternalAggregations;
 import org.elasticsearch.search.aggregations.bucket.BucketsAggregator;
 import org.elasticsearch.search.aggregations.bucket.LongHash;
 import org.elasticsearch.search.aggregations.bucket.terms.support.BucketPriorityQueue;
@@ -49,7 +48,7 @@ public DoubleTermsAggregator(String name, AggregatorFactories factories, Numeric
                                InternalOrder order, int requiredSize, int shardSize, AggregationContext aggregationContext, Aggregator parent) {
         super(name, BucketAggregationMode.PER_BUCKET, factories, estimatedBucketCount, aggregationContext, parent);
         this.valuesSource = valuesSource;
-        this.order = order;
+        this.order = InternalOrder.validate(order, this);
         this.requiredSize = requiredSize;
         this.shardSize = shardSize;
         bucketOrds = new LongHash(estimatedBucketCount, aggregationContext.pageCacheRecycler());
@@ -77,24 +76,13 @@ public void collect(int doc, long owningBucketOrdinal) throws IOException {
         }
     }
 
-    // private impl that stores a bucket ord. This allows for computing the aggregations lazily.
-    static class OrdinalBucket extends DoubleTerms.Bucket {
-
-        long bucketOrd;
-
-        public OrdinalBucket() {
-            super(0, 0, (InternalAggregations) null);
-        }
-
-    }
-
     @Override
     public DoubleTerms buildAggregation(long owningBucketOrdinal) {
         assert owningBucketOrdinal == 0;
         final int size = (int) Math.min(bucketOrds.size(), shardSize);
 
-        BucketPriorityQueue ordered = new BucketPriorityQueue(size, order.comparator());
-        OrdinalBucket spare = null;
+        BucketPriorityQueue ordered = new BucketPriorityQueue(size, order.comparator(this));
+        DoubleTerms.Bucket spare = null;
         for (long i = 0; i < bucketOrds.capacity(); ++i) {
             final long ord = bucketOrds.id(i);
             if (ord < 0) {
@@ -103,17 +91,17 @@ public DoubleTerms buildAggregation(long owningBucketOrdinal) {
             }
 
             if (spare == null) {
-                spare = new OrdinalBucket();
+                spare = new DoubleTerms.Bucket(0, 0, null);
             }
             spare.term = Double.longBitsToDouble(bucketOrds.key(i));
             spare.docCount = bucketDocCount(ord);
             spare.bucketOrd = ord;
-            spare = (OrdinalBucket) ordered.insertWithOverflow(spare);
+            spare = (DoubleTerms.Bucket) ordered.insertWithOverflow(spare);
         }
 
         final InternalTerms.Bucket[] list = new InternalTerms.Bucket[ordered.size()];
         for (int i = ordered.size() - 1; i >= 0; --i) {
-            final OrdinalBucket bucket = (OrdinalBucket) ordered.pop();
+            final DoubleTerms.Bucket bucket = (DoubleTerms.Bucket) ordered.pop();
             bucket.aggregations = bucketAggregations(bucket.bucketOrd);
             list[i] = bucket;
         }