Make global ords terms simpler to understand (backport of #57241) (#57311)

When the `terms` enum operates on non-numeric data it can collect it via
global ordinals. It actually has two separate collection strategies for,
one "dense" and one "remapping". Each of *those* strategies has two
"iteration" strategies that it uses to build buckets, depending on
whether or not we need buckets with `0` docs in them. Previously this
was done with several `null` checks and never really explained. This
change replaces those checks with two `CollectionStrategy` classes which
have good stuff like documentation.

Author: nik9000

rest-api-spec/src/main/resources/rest-api-spec/test/search.aggregation/20_terms.yml

@@ -822,6 +822,7 @@ setup:
   - match: { profile.shards.0.aggregations.0.description: str_terms }
   - match: { profile.shards.0.aggregations.0.breakdown.collect_count: 4 }
   - match: { profile.shards.0.aggregations.0.debug.deferred_aggregators: [ max_number ] }
+  - match: { profile.shards.0.aggregations.0.debug.collection_strategy: dense }
   - match: { profile.shards.0.aggregations.0.children.0.type: MaxAggregator }
   - match: { profile.shards.0.aggregations.0.children.0.description: max_number }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/GlobalOrdinalsSignificantTermsAggregator.java

@@ -96,50 +96,33 @@ public InternalAggregation[] buildAggregations(long[] owningBucketOrds) throws I
         long subsetSize = numCollectedDocs;
 
         BucketSignificancePriorityQueue<SignificantStringTerms.Bucket> ordered = new BucketSignificancePriorityQueue<>(size);
-        SignificantStringTerms.Bucket spare = null;
-        final boolean needsFullScan = bucketOrds == null || bucketCountThresholds.getMinDocCount() == 0;
-        final long maxId = needsFullScan ? valueCount : bucketOrds.size();
-        for (long ord = 0; ord < maxId; ord++) {
-            final long globalOrd;
-            final long bucketOrd;
-            if (needsFullScan) {
-                bucketOrd = bucketOrds == null ? ord : bucketOrds.find(ord);
-                globalOrd = ord;
-            } else {
-                assert bucketOrds != null;
-                bucketOrd = ord;
-                globalOrd = bucketOrds.get(ord);
-            }
-            if (includeExclude != null && !acceptedGlobalOrdinals.get(globalOrd)) {
-                continue;
-            }
-            final int bucketDocCount = bucketOrd < 0 ? 0 : bucketDocCount(bucketOrd);
-            if (bucketCountThresholds.getMinDocCount() > 0 && bucketDocCount == 0) {
-                continue;
-            }
-            if (bucketDocCount < bucketCountThresholds.getShardMinDocCount()) {
-                continue;
-            }
+        collectionStrategy.forEach(new BucketInfoConsumer() {
+            SignificantStringTerms.Bucket spare = null;
 
-            if (spare == null) {
-                spare = new SignificantStringTerms.Bucket(new BytesRef(), 0, 0, 0, 0, null, format, 0);
-            }
-            spare.bucketOrd = bucketOrd;
-            copy(lookupGlobalOrd.apply(globalOrd), spare.termBytes);
-            spare.subsetDf = bucketDocCount;
-            spare.subsetSize = subsetSize;
-            spare.supersetDf = termsAggFactory.getBackgroundFrequency(spare.termBytes);
-            spare.supersetSize = supersetSize;
-            // During shard-local down-selection we use subset/superset stats
-            // that are for this shard only
-            // Back at the central reducer these properties will be updated with
-            // global stats
-            spare.updateScore(significanceHeuristic);
-            spare = ordered.insertWithOverflow(spare);
-            if (spare == null) {
-                consumeBucketsAndMaybeBreak(1);
+            @Override
+            public void accept(long globalOrd, long bucketOrd, long docCount) throws IOException {
+                if (docCount >= bucketCountThresholds.getShardMinDocCount()) {
+                    if (spare == null) {
+                        spare = new SignificantStringTerms.Bucket(new BytesRef(), 0, 0, 0, 0, null, format, 0);
+                    }
+                    spare.bucketOrd = bucketOrd;
+                    copy(lookupGlobalOrd.apply(globalOrd), spare.termBytes);
+                    spare.subsetDf = docCount;
+                    spare.subsetSize = subsetSize;
+                    spare.supersetDf = termsAggFactory.getBackgroundFrequency(spare.termBytes);
+                    spare.supersetSize = supersetSize;
+                    // During shard-local down-selection we use subset/superset stats
+                    // that are for this shard only
+                    // Back at the central reducer these properties will be updated with
+                    // global stats
+                    spare.updateScore(significanceHeuristic);
+                    spare = ordered.insertWithOverflow(spare);
+                    if (spare == null) {
+                        consumeBucketsAndMaybeBreak(1);
+                    }
+                }
             }
-        }
+        });
 
         final SignificantStringTerms.Bucket[] list = new SignificantStringTerms.Bucket[ordered.size()];
         for (int i = ordered.size() - 1; i >= 0; i--) {