Terms aggregation should remap global ordinal buckets when a sub-aggregator is used to sort the terms (#24941)

`terms` aggregations at the root level use the `global_ordinals` execution hint by default.
When all sub-aggregators can be run in `breadth_first` mode the collected buckets for these sub-aggs are dense (remapped after the initial pruning).
But if a sub-aggregator is not deferrable and needs to collect all buckets before pruning we don't remap global ords and the aggregator needs to deal with sparse buckets.
Most (if not all) aggregators expect dense buckets and uses this information to allocate memories.
This change forces the remap of the global ordinals but only when there is at least one sub-aggregator that cannot be deferred.

Relates #24788

Author: jimczi

core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/GlobalOrdinalsSignificantTermsAggregator.java

@@ -22,7 +22,6 @@
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.common.lease.Releasables;
-import org.elasticsearch.common.util.LongHash;
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.Aggregator;
 import org.elasticsearch.search.aggregations.AggregatorFactories;
@@ -52,22 +51,28 @@ public class GlobalOrdinalsSignificantTermsAggregator extends GlobalOrdinalsStri
     protected final SignificantTermsAggregatorFactory termsAggFactory;
     private final SignificanceHeuristic significanceHeuristic;
 
-    public GlobalOrdinalsSignificantTermsAggregator(String name, AggregatorFactories factories,
-            ValuesSource.Bytes.WithOrdinals.FieldData valuesSource, DocValueFormat format,
-            BucketCountThresholds bucketCountThresholds, IncludeExclude.OrdinalsFilter includeExclude,
-            SearchContext context, Aggregator parent,
-            SignificanceHeuristic significanceHeuristic, SignificantTermsAggregatorFactory termsAggFactory,
-            List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) throws IOException {
-
+    public GlobalOrdinalsSignificantTermsAggregator(String name,
+                                                    AggregatorFactories factories,
+                                                    ValuesSource.Bytes.WithOrdinals.FieldData valuesSource,
+                                                    DocValueFormat format,
+                                                    BucketCountThresholds bucketCountThresholds,
+                                                    IncludeExclude.OrdinalsFilter includeExclude,
+                                                    SearchContext context,
+                                                    Aggregator parent,
+                                                    boolean forceRemapGlobalOrds,
+                                                    SignificanceHeuristic significanceHeuristic,
+                                                    SignificantTermsAggregatorFactory termsAggFactory,
+                                                    List<PipelineAggregator> pipelineAggregators,
+                                                    Map<String, Object> metaData) throws IOException {
         super(name, factories, valuesSource, null, format, bucketCountThresholds, includeExclude, context, parent,
-                SubAggCollectionMode.DEPTH_FIRST, false, pipelineAggregators, metaData);
+            forceRemapGlobalOrds, SubAggCollectionMode.DEPTH_FIRST, false, pipelineAggregators, metaData);
         this.significanceHeuristic = significanceHeuristic;
         this.termsAggFactory = termsAggFactory;
+        this.numCollectedDocs = 0;
     }
 
     @Override
-    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx,
-            final LeafBucketCollector sub) throws IOException {
+    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException {
         return new LeafBucketCollectorBase(super.getLeafCollector(ctx, sub), null) {
             @Override
             public void collect(int doc, long bucket) throws IOException {
@@ -77,18 +82,17 @@ public void collect(int doc, long bucket) throws IOException {
         };
     }
 
-
     @Override
     public SignificantStringTerms buildAggregation(long owningBucketOrdinal) throws IOException {
         assert owningBucketOrdinal == 0;
-        if (globalOrds == null) { // no context in this reader
+        if (valueCount == 0) { // no context in this reader
             return buildEmptyAggregation();
         }
 
         final int size;
         if (bucketCountThresholds.getMinDocCount() == 0) {
             // if minDocCount == 0 then we can end up with more buckets then maxBucketOrd() returns
-            size = (int) Math.min(globalOrds.getValueCount(), bucketCountThresholds.getShardSize());
+            size = (int) Math.min(valueCount, bucketCountThresholds.getShardSize());
         } else {
             size = (int) Math.min(maxBucketOrd(), bucketCountThresholds.getShardSize());
         }
@@ -97,7 +101,7 @@ public SignificantStringTerms buildAggregation(long owningBucketOrdinal) throws
 
         BucketSignificancePriorityQueue<SignificantStringTerms.Bucket> ordered = new BucketSignificancePriorityQueue<>(size);
         SignificantStringTerms.Bucket spare = null;
-        for (long globalTermOrd = 0; globalTermOrd < globalOrds.getValueCount(); ++globalTermOrd) {
+        for (long globalTermOrd = 0; globalTermOrd < valueCount; ++globalTermOrd) {
             if (includeExclude != null && !acceptedGlobalOrdinals.get(globalTermOrd)) {
                 continue;
             }
@@ -114,7 +118,7 @@ public SignificantStringTerms buildAggregation(long owningBucketOrdinal) throws
                 spare = new SignificantStringTerms.Bucket(new BytesRef(), 0, 0, 0, 0, null, format);
             }
             spare.bucketOrd = bucketOrd;
-            copy(globalOrds.lookupOrd(globalTermOrd), spare.termBytes);
+            copy(lookupGlobalOrd.apply(globalTermOrd), spare.termBytes);
             spare.subsetDf = bucketDocCount;
             spare.subsetSize = subsetSize;
             spare.supersetDf = termsAggFactory.getBackgroundFrequency(spare.termBytes);
@@ -147,62 +151,13 @@ public SignificantStringTerms buildEmptyAggregation() {
         IndexReader topReader = searcher.getIndexReader();
         int supersetSize = topReader.numDocs();
         return new SignificantStringTerms(name, bucketCountThresholds.getRequiredSize(), bucketCountThresholds.getMinDocCount(),
-                pipelineAggregators(), metaData(), format, 0, supersetSize, significanceHeuristic, emptyList());
+                pipelineAggregators(), metaData(), format, numCollectedDocs, supersetSize, significanceHeuristic, emptyList());
     }
 
     @Override
     protected void doClose() {
+        super.doClose();
         Releasables.close(termsAggFactory);
     }
-
-    public static class WithHash extends GlobalOrdinalsSignificantTermsAggregator {
-
-        private final LongHash bucketOrds;
-
-        public WithHash(String name, AggregatorFactories factories, ValuesSource.Bytes.WithOrdinals.FieldData valuesSource,
-                DocValueFormat format, BucketCountThresholds bucketCountThresholds, IncludeExclude.OrdinalsFilter includeExclude,
-                SearchContext context, Aggregator parent, SignificanceHeuristic significanceHeuristic,
-                SignificantTermsAggregatorFactory termsAggFactory, List<PipelineAggregator> pipelineAggregators,
-                Map<String, Object> metaData) throws IOException {
-            super(name, factories, valuesSource, format, bucketCountThresholds, includeExclude, context, parent, significanceHeuristic,
-                    termsAggFactory, pipelineAggregators, metaData);
-            bucketOrds = new LongHash(1, context.bigArrays());
-        }
-
-        @Override
-        public LeafBucketCollector getLeafCollector(LeafReaderContext ctx,
-                final LeafBucketCollector sub) throws IOException {
-            return new LeafBucketCollectorBase(super.getLeafCollector(ctx, sub), null) {
-                @Override
-                public void collect(int doc, long bucket) throws IOException {
-                    assert bucket == 0;
-                    numCollectedDocs++;
-                    globalOrds.setDocument(doc);
-                    final int numOrds = globalOrds.cardinality();
-                    for (int i = 0; i < numOrds; i++) {
-                        final long globalOrd = globalOrds.ordAt(i);
-                        long bucketOrd = bucketOrds.add(globalOrd);
-                        if (bucketOrd < 0) {
-                            bucketOrd = -1 - bucketOrd;
-                            collectExistingBucket(sub, doc, bucketOrd);
-                        } else {
-                            collectBucket(sub, doc, bucketOrd);
-                        }
-                    }
-                }
-            };
-        }
-
-        @Override
-        protected long getBucketOrd(long termOrd) {
-            return bucketOrds.find(termOrd);
-        }
-
-        @Override
-        protected void doClose() {
-            Releasables.close(termsAggFactory, bucketOrds);
-        }
-    }
-
 }
 

core/src/main/java/org/elasticsearch/search/aggregations/bucket/significant/SignificantTermsAggregatorFactory.java

@@ -70,10 +70,17 @@ public class SignificantTermsAggregatorFactory extends ValuesSourceAggregatorFac
     private final TermsAggregator.BucketCountThresholds bucketCountThresholds;
     private final SignificanceHeuristic significanceHeuristic;
 
-    public SignificantTermsAggregatorFactory(String name, ValuesSourceConfig<ValuesSource> config, IncludeExclude includeExclude,
-            String executionHint, QueryBuilder filterBuilder, TermsAggregator.BucketCountThresholds bucketCountThresholds,
-            SignificanceHeuristic significanceHeuristic, SearchContext context, AggregatorFactory<?> parent,
-            AggregatorFactories.Builder subFactoriesBuilder, Map<String, Object> metaData) throws IOException {
+    public SignificantTermsAggregatorFactory(String name,
+                                             ValuesSourceConfig<ValuesSource> config,
+                                             IncludeExclude includeExclude,
+                                             String executionHint,
+                                             QueryBuilder filterBuilder,
+                                             TermsAggregator.BucketCountThresholds bucketCountThresholds,
+                                             SignificanceHeuristic significanceHeuristic,
+                                             SearchContext context,
+                                             AggregatorFactory<?> parent,
+                                             AggregatorFactories.Builder subFactoriesBuilder,
+                                             Map<String, Object> metaData) throws IOException {
         super(name, config, context, parent, subFactoriesBuilder, metaData);
         this.includeExclude = includeExclude;
         this.executionHint = executionHint;
@@ -246,44 +253,71 @@ public enum ExecutionMode {
         MAP(new ParseField("map")) {
 
             @Override
-            Aggregator create(String name, AggregatorFactories factories, ValuesSource valuesSource, DocValueFormat format,
-                    TermsAggregator.BucketCountThresholds bucketCountThresholds, IncludeExclude includeExclude,
-                    SearchContext aggregationContext, Aggregator parent, SignificanceHeuristic significanceHeuristic,
-                    SignificantTermsAggregatorFactory termsAggregatorFactory, List<PipelineAggregator> pipelineAggregators,
-                    Map<String, Object> metaData) throws IOException {
+            Aggregator create(String name,
+                              AggregatorFactories factories,
+                              ValuesSource valuesSource,
+                              DocValueFormat format,
+                              TermsAggregator.BucketCountThresholds bucketCountThresholds,
+                              IncludeExclude includeExclude,
+                              SearchContext aggregationContext,
+                              Aggregator parent,
+                              SignificanceHeuristic significanceHeuristic,
+                              SignificantTermsAggregatorFactory termsAggregatorFactory,
+                              List<PipelineAggregator> pipelineAggregators,
+                              Map<String, Object> metaData) throws IOException {
+
                 final IncludeExclude.StringFilter filter = includeExclude == null ? null : includeExclude.convertToStringFilter(format);
                 return new SignificantStringTermsAggregator(name, factories, valuesSource, format, bucketCountThresholds, filter,
                         aggregationContext, parent, significanceHeuristic, termsAggregatorFactory, pipelineAggregators, metaData);
+
             }
 
         },
         GLOBAL_ORDINALS(new ParseField("global_ordinals")) {
 
             @Override
-            Aggregator create(String name, AggregatorFactories factories, ValuesSource valuesSource, DocValueFormat format,
-                    TermsAggregator.BucketCountThresholds bucketCountThresholds, IncludeExclude includeExclude,
-                    SearchContext aggregationContext, Aggregator parent, SignificanceHeuristic significanceHeuristic,
-                    SignificantTermsAggregatorFactory termsAggregatorFactory, List<PipelineAggregator> pipelineAggregators,
-                    Map<String, Object> metaData) throws IOException {
+            Aggregator create(String name,
+                              AggregatorFactories factories,
+                              ValuesSource valuesSource,
+                              DocValueFormat format,
+                              TermsAggregator.BucketCountThresholds bucketCountThresholds,
+                              IncludeExclude includeExclude,
+                              SearchContext aggregationContext,
+                              Aggregator parent,
+                              SignificanceHeuristic significanceHeuristic,
+                              SignificantTermsAggregatorFactory termsAggregatorFactory,
+                              List<PipelineAggregator> pipelineAggregators,
+                              Map<String, Object> metaData) throws IOException {
+
                 final IncludeExclude.OrdinalsFilter filter = includeExclude == null ? null : includeExclude.convertToOrdinalsFilter(format);
                 return new GlobalOrdinalsSignificantTermsAggregator(name, factories,
                         (ValuesSource.Bytes.WithOrdinals.FieldData) valuesSource, format, bucketCountThresholds, filter,
-                        aggregationContext, parent, significanceHeuristic, termsAggregatorFactory, pipelineAggregators, metaData);
+                        aggregationContext, parent, false, significanceHeuristic, termsAggregatorFactory, pipelineAggregators, metaData);
+
             }
 
         },
         GLOBAL_ORDINALS_HASH(new ParseField("global_ordinals_hash")) {
 
             @Override
-            Aggregator create(String name, AggregatorFactories factories, ValuesSource valuesSource, DocValueFormat format,
-                    TermsAggregator.BucketCountThresholds bucketCountThresholds, IncludeExclude includeExclude,
-                    SearchContext aggregationContext, Aggregator parent, SignificanceHeuristic significanceHeuristic,
-                    SignificantTermsAggregatorFactory termsAggregatorFactory, List<PipelineAggregator> pipelineAggregators,
-                    Map<String, Object> metaData) throws IOException {
+            Aggregator create(String name,
+                              AggregatorFactories factories,
+                              ValuesSource valuesSource,
+                              DocValueFormat format,
+                              TermsAggregator.BucketCountThresholds bucketCountThresholds,
+                              IncludeExclude includeExclude,
+                              SearchContext aggregationContext,
+                              Aggregator parent,
+                              SignificanceHeuristic significanceHeuristic,
+                              SignificantTermsAggregatorFactory termsAggregatorFactory,
+                              List<PipelineAggregator> pipelineAggregators,
+                              Map<String, Object> metaData) throws IOException {
+
                 final IncludeExclude.OrdinalsFilter filter = includeExclude == null ? null : includeExclude.convertToOrdinalsFilter(format);
-                return new GlobalOrdinalsSignificantTermsAggregator.WithHash(name, factories,
-                        (ValuesSource.Bytes.WithOrdinals.FieldData) valuesSource, format, bucketCountThresholds, filter,
-                        aggregationContext, parent, significanceHeuristic, termsAggregatorFactory, pipelineAggregators, metaData);
+                return new GlobalOrdinalsSignificantTermsAggregator(name, factories,
+                    (ValuesSource.Bytes.WithOrdinals.FieldData) valuesSource, format, bucketCountThresholds, filter, aggregationContext, parent,
+                    true, significanceHeuristic, termsAggregatorFactory, pipelineAggregators, metaData);
+
             }
         };
 
@@ -302,11 +336,18 @@ public static ExecutionMode fromString(String value) {
             this.parseField = parseField;
         }
 
-        abstract Aggregator create(String name, AggregatorFactories factories, ValuesSource valuesSource,  DocValueFormat format,
-                TermsAggregator.BucketCountThresholds bucketCountThresholds, IncludeExclude includeExclude,
-                SearchContext aggregationContext, Aggregator parent, SignificanceHeuristic significanceHeuristic,
-                SignificantTermsAggregatorFactory termsAggregatorFactory, List<PipelineAggregator> pipelineAggregators,
-                Map<String, Object> metaData) throws IOException;
+        abstract Aggregator create(String name,
+                                   AggregatorFactories factories,
+                                   ValuesSource valuesSource,
+                                   DocValueFormat format,
+                                   TermsAggregator.BucketCountThresholds bucketCountThresholds,
+                                   IncludeExclude includeExclude,
+                                   SearchContext aggregationContext,
+                                   Aggregator parent,
+                                   SignificanceHeuristic significanceHeuristic,
+                                   SignificantTermsAggregatorFactory termsAggregatorFactory,
+                                   List<PipelineAggregator> pipelineAggregators,
+                                   Map<String, Object> metaData) throws IOException;
 
         @Override
         public String toString() {