Only call reduce on a single InternalAggregation when needed (#62525)

Adds a new abstract method in InternalAggregation that flags the framework if it needs to reduce on a single InternalAggregation.

Author: iverase

modules/aggs-matrix-stats/src/main/java/org/elasticsearch/search/aggregations/matrix/stats/InternalMatrixStats.java

@@ -254,6 +254,11 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Reduce
         return new InternalMatrixStats(name, runningStats.docCount, runningStats, null, getMetadata());
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return true;
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(super.hashCode(), stats, results);

server/src/main/java/org/elasticsearch/search/aggregations/InternalAggregation.java

@@ -223,9 +223,17 @@ public InternalAggregation reducePipelines(
      * aggregations are of the same type (the same type as this aggregation). For best efficiency, when implementing,
      * try reusing an existing instance (typically the first in the given list) to save on redundant object
      * construction.
+     *
+     * @see #mustReduceOnSingleInternalAgg()
      */
     public abstract InternalAggregation reduce(List<InternalAggregation> aggregations, ReduceContext reduceContext);
 
+    /**
+     * Signal the framework if the {@linkplain InternalAggregation#reduce(List, ReduceContext)} phase needs to be called
+     * when there is only one {@linkplain InternalAggregation}.
+     */
+    protected abstract boolean mustReduceOnSingleInternalAgg();
+
     /**
      * Return true if this aggregation is mapped, and can lead a reduction.  If this agg returns
      * false, it should return itself if asked to lead a reduction

server/src/main/java/org/elasticsearch/search/aggregations/InternalAggregations.java

@@ -161,7 +161,12 @@ public static InternalAggregations reduce(List<InternalAggregations> aggregation
             // If all aggs are unmapped, the agg that leads the reduction will just return itself
             aggregations.sort(INTERNAL_AGG_COMPARATOR);
             InternalAggregation first = aggregations.get(0); // the list can't be empty as it's created on demand
-            reducedAggregations.add(first.reduce(aggregations, context));
+            if (first.mustReduceOnSingleInternalAgg() || aggregations.size() > 1) {
+                reducedAggregations.add(first.reduce(aggregations, context));
+            } else {
+                // no need for reduce phase
+                reducedAggregations.add(first);
+            }
         }
 
         return from(reducedAggregations);

server/src/main/java/org/elasticsearch/search/aggregations/InternalMultiBucketAggregation.java

@@ -177,6 +177,11 @@ public InternalAggregation copyWithRewritenBuckets(Function<InternalAggregations
         return modified ? create(newBuckets) : this;
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return true;
+    }
+
     @Override
     public void forEachBucket(Consumer<InternalAggregations> consumer) {
         for (B bucket : getBuckets()) {

server/src/main/java/org/elasticsearch/search/aggregations/bucket/InternalSingleBucketAggregation.java

@@ -176,6 +176,11 @@ public final double sortValue(AggregationPath.PathElement head, Iterator<Aggrega
         return aggregations.sortValue(head, tail);
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return true;
+    }
+
     @Override
     public InternalAggregation copyWithRewritenBuckets(Function<InternalAggregations, InternalAggregations> rewriter) {
         InternalAggregations rewritten = rewriter.apply(aggregations);

server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalGeoBounds.java

@@ -116,6 +116,11 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Reduce
         return new InternalGeoBounds(name, top, bottom, posLeft, posRight, negLeft, negRight, wrapLongitude, getMetadata());
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return false;
+    }
+
     @Override
     public Object getProperty(List<String> path) {
         if (path.isEmpty()) {

server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalGeoCentroid.java

@@ -133,6 +133,11 @@ public InternalGeoCentroid reduce(List<InternalAggregation> aggregations, Reduce
         return new InternalGeoCentroid(name, result, totalCount, getMetadata());
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return false;
+    }
+
     @Override
     public Object getProperty(List<String> path) {
         if (path.isEmpty()) {

server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalNumericMetricsAggregation.java

@@ -128,6 +128,11 @@ public final double sortValue(AggregationPath.PathElement head, Iterator<Aggrega
         throw new IllegalArgumentException("Metrics aggregations cannot have sub-aggregations (at [>" + head + "]");
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return false;
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(super.hashCode(), format);

server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalScriptedMetric.java

@@ -74,7 +74,7 @@ protected void doWriteTo(StreamOutput out) throws IOException {
                 /*
                  * I *believe* that this situation can only happen in cross
                  * cluster search right now. Thus the message. But computers
-                 * are hard. 
+                 * are hard.
                  */
                 throw new IllegalArgumentException("scripted_metric doesn't support cross cluster search until 7.8.0");
             }
@@ -134,6 +134,11 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Reduce
         return new InternalScriptedMetric(firstAggregation.getName(), aggregation, firstAggregation.reduceScript, getMetadata());
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return true;
+    }
+
     @Override
     public Object getProperty(List<String> path) {
         if (path.isEmpty()) {

server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalTopHits.java

@@ -41,7 +41,8 @@
 /**
  * Results of the {@link TopHitsAggregator}.
  */
-public class InternalTopHits extends InternalAggregation implements TopHits {
+public class
+InternalTopHits extends InternalAggregation implements TopHits {
     private int from;
     private int size;
     private TopDocsAndMaxScore topDocs;
@@ -162,6 +163,11 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Reduce
             new SearchHits(hits, reducedTopDocs.totalHits, maxScore), getMetadata());
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return true;
+    }
+
     @Override
     public Object getProperty(List<String> path) {
         if (path.isEmpty()) {