Save memory when auto_date_histogram is not on top (#57304)

This builds an `auto_date_histogram` aggregator that natively aggregates
from many buckets and uses it when the `auto_date_histogram` used to use
`asMultiBucketAggregator` which should save a significant amount of
memory in those cases. In particular, this happens when
`auto_date_histogram` is a sub-aggregator of a multi-bucketing aggregator
like `terms` or `histogram` or `filters`. For the most part we preserve
the original implementation when `auto_date_histogram` only collects from
a single bucket.

It isn't possible to "just port the aggregator" without taking a pretty
significant performance hit because we used to rewrite all of the
buckets every time we switched to a coarser and coarser rounding
configuration. Without some major surgery to how to delay sub-aggs
we'd end up rewriting the delay list zillions of time if there are many
buckets.

The multi-bucket version of the aggregator has a "budget" of "wasted"
buckets and only rewrites all of the buckets when we exceed that budget.
Now that we don't rebucket every time we increase the rounding we can no
longer get an accurate count of the number of buckets! So instead the
aggregator uses an estimate of the number of buckets to trigger switching
to a coarser rounding. This estimate is likely to be *terrible* when
buckets are far apart compared to the rounding. So it also uses the
difference between the first and last bucket to trigger switching to a
coarser rounding. Which covers for the shortcomings of the bucket
estimation technique pretty well. It also causes the aggregator to emit
fewer buckets in cases where they'd be reduced together on the
coordinating node. This is wonderful! But probably fairly rare.

All of that does buy us some speed improvements when the aggregator is
a child of multi-bucket aggregator:
Without metrics or time zone: 25% faster
With metrics: 15% faster
With time zone: 22% faster

Relates to #56487

Author: nik9000

rest-api-spec/src/main/resources/rest-api-spec/test/search.aggregation/330_auto_date_histogram.yml

@@ -4,6 +4,7 @@ setup:
           index: test
           body:
             settings:
+              number_of_shards: 1
               number_of_replicas: 0
             mappings:
               properties:
@@ -75,3 +76,25 @@ setup:
   - match: { aggregations.histo.buckets.1.doc_count: 2 }
   - match: { aggregations.histo.buckets.1.v.value: 7 }
   - match: { aggregations.histo_avg_v.value: 5 }
+
+---
+"profile at top level":
+  - skip:
+      version: " - 7.99.99"
+      reason: Debug information added in 8.0.0 (to be backported to 7.9.0)
+
+  - do:
+      search:
+        body:
+          profile: true
+          size: 0
+          aggs:
+            histo:
+              auto_date_histogram:
+                field: date
+                buckets: 2
+
+  - match: { hits.total.value: 4 }
+  - length: { aggregations.histo.buckets: 2 }
+  - match: { profile.shards.0.aggregations.0.type: AutoDateHistogramAggregator.FromSingle }
+  - match: { profile.shards.0.aggregations.0.debug.surviving_buckets: 4 }

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

@@ -30,6 +30,7 @@
 import org.elasticsearch.search.aggregations.InternalAggregation;
 import org.elasticsearch.search.aggregations.InternalAggregations;
 import org.elasticsearch.search.aggregations.LeafBucketCollector;
+import org.elasticsearch.search.aggregations.bucket.global.GlobalAggregator;
 import org.elasticsearch.search.aggregations.bucket.terms.LongKeyedBucketOrds;
 import org.elasticsearch.search.aggregations.support.AggregationPath;
 import org.elasticsearch.search.internal.SearchContext;
@@ -329,7 +330,7 @@ protected final <B> InternalAggregation[] buildAggregationsForVariableBuckets(lo
      * @param bucketOrds hash of values to the bucket ordinal
      */
     protected final <B> InternalAggregation[] buildAggregationsForVariableBuckets(long[] owningBucketOrds, LongKeyedBucketOrds bucketOrds,
-            BucketBuilderForVariable<B> bucketBuilder, Function<List<B>, InternalAggregation> resultBuilder) throws IOException {
+            BucketBuilderForVariable<B> bucketBuilder, ResultBuilderForVariable<B> resultBuilder) throws IOException {
         long totalOrdsToCollect = 0;
         for (int ordIdx = 0; ordIdx < owningBucketOrds.length; ordIdx++) {
             totalOrdsToCollect += bucketOrds.bucketsInOrd(owningBucketOrds[ordIdx]);
@@ -360,14 +361,18 @@ protected final <B> InternalAggregation[] buildAggregationsForVariableBuckets(lo
                 }
                 buckets.add(bucketBuilder.build(ordsEnum.value(), bucketDocCount(ordsEnum.ord()), subAggregationResults[b++]));
             }
-            results[ordIdx] = resultBuilder.apply(buckets);
+            results[ordIdx] = resultBuilder.build(owningBucketOrds[ordIdx], buckets);
         }
         return results;
     }
     @FunctionalInterface
     protected interface BucketBuilderForVariable<B> {
         B build(long bucketValue, int docCount, InternalAggregations subAggregationResults);
     }
+    @FunctionalInterface
+    protected interface ResultBuilderForVariable<B> {
+        InternalAggregation build(long owninigBucketOrd, List<B> buckets);
+    }
 
     /**
      * Utility method to build empty aggregations of the sub aggregators.
@@ -407,4 +412,15 @@ public BucketComparator bucketComparator(String key, SortOrder order) {
                 "Either drop the key (a la \"" + name() + "\") or change it to \"doc_count\" (a la \"" + name() +
                 ".doc_count\") or \"key\".");
     }
+
+    public static boolean descendsFromGlobalAggregator(Aggregator parent) {
+        while (parent != null) {
+            if (parent.getClass() == GlobalAggregator.class) {
+                return true;
+            }
+            parent = parent.parent();
+        }
+        return false;
+    }
+
 }

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

@@ -23,7 +23,6 @@
 import org.elasticsearch.search.aggregations.AggregatorFactories;
 import org.elasticsearch.search.aggregations.BucketCollector;
 import org.elasticsearch.search.aggregations.MultiBucketCollector;
-import org.elasticsearch.search.aggregations.bucket.global.GlobalAggregator;
 import org.elasticsearch.search.internal.SearchContext;
 
 import java.io.IOException;
@@ -70,16 +69,6 @@ protected void doPreCollection() throws IOException {
         collectableSubAggregators = MultiBucketCollector.wrap(collectors);
     }
 
-    public static boolean descendsFromGlobalAggregator(Aggregator parent) {
-        while (parent != null) {
-            if (parent.getClass() == GlobalAggregator.class) {
-                return true;
-            }
-            parent = parent.parent();
-        }
-        return false;
-    }
-
     public DeferringBucketCollector getDeferringCollector() {
         // Default impl is a collector that selects the best buckets
         // but an alternative defer policy may be based on best docs.

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/AbstractHistogramAggregator.java

@@ -90,7 +90,7 @@ public InternalAggregation[] buildAggregations(long[] owningBucketOrds) throws I
                 double roundKey = Double.longBitsToDouble(bucketValue);
                 double key = roundKey * interval + offset;
                 return new InternalHistogram.Bucket(key, docCount, keyed, formatter, subAggregationResults);
-            }, buckets -> {
+            }, (owningBucketOrd, buckets) -> {
                 // the contract of the histogram aggregation is that shards must return buckets ordered by key in ascending order
                 CollectionUtil.introSort(buckets, BucketOrder.key(true).comparator());
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/histogram/AutoDateHistogramAggregationBuilder.java

@@ -290,6 +290,10 @@ public long getRoughEstimateDurationMillis() {
             return roughEstimateDurationMillis;
         }
 
+        public long getMaximumRoughEstimateDurationMillis() {
+            return getRoughEstimateDurationMillis() * getMaximumInnerInterval();
+        }
+
         @Override
         public int hashCode() {
             return Objects.hash(rounding, Arrays.hashCode(innerIntervals), dateTimeUnit);