Revert "Updated the topN selection logic for reduce"

This reverts commit a711100.

Author: vinaykpud

server/src/main/java/org/opensearch/search/aggregations/bucket/terms/InternalTerms.java

@@ -458,15 +458,21 @@ For backward compatibility, we disable the merge sort and use ({@link InternalTe
         }
         final B[] list;
         if (reduceContext.isFinalReduce() || reduceContext.isSliceLevel()) {
-            B[] reducedBucketsArr = createBucketsArray(reducedBuckets.size());
-            for (int i = 0; i < reducedBuckets.size(); i++) {
-                reducedBucketsArr[i] = reducedBuckets.get(i);
-            }
             final int size = Math.min(localBucketCountThresholds.getRequiredSize(), reducedBuckets.size());
-            final Comparator<MultiBucketsAggregation.Bucket> cmp = order.comparator();
             if (size < reducedBuckets.size()) {
-                ArrayUtil.select(reducedBucketsArr, 0, reducedBuckets.size(), size, cmp);
-                int selectedSize = 0;
+                Comparator<MultiBucketsAggregation.Bucket> cmp = order.comparator();
+                B[] reducedBucketsArr  = createBucketsArray(reducedBuckets.size());;
+                for (int i = 0; i < reducedBuckets.size(); i++) {
+                    reducedBucketsArr[i] = reducedBuckets.get(i);
+                }
+                ArrayUtil.select(
+                    reducedBucketsArr,
+                    0,
+                    reducedBuckets.size(),
+                    size,
+                    cmp
+                );
+                int sz = 0;
                 for (B bucket : reducedBucketsArr) {
                     if (sumDocCountError == -1) {
                         bucket.setDocCountError(-1);
@@ -475,38 +481,47 @@ For backward compatibility, we disable the merge sort and use ({@link InternalTe
                         bucket.setDocCountError(docCountError -> docCountError + finalSumDocCountError);
                     }
                     if (bucket.getDocCount() >= localBucketCountThresholds.getMinDocCount()) {
-                        B removed = ((selectedSize == size) ? bucket : null);
+                        B removed = ((sz == size) ? bucket : null);
                         if (removed != null) {
                             otherDocCount += removed.getDocCount();
                             reduceContext.consumeBucketsAndMaybeBreak(-countInnerBucket(removed));
                         } else {
-                            selectedSize++;
+                            sz++;
                             reduceContext.consumeBucketsAndMaybeBreak(1);
                         }
                     } else {
                         reduceContext.consumeBucketsAndMaybeBreak(-countInnerBucket(bucket));
                     }
                 }
-                list = createBucketsArray(selectedSize);
-                System.arraycopy(reducedBucketsArr, 0, list, 0, selectedSize);
+                list = createBucketsArray(sz);
+                if (sz >= 0) System.arraycopy(reducedBucketsArr, 0, list, 0, sz);
+                Arrays.sort(list, cmp);
             } else {
-                // since only else case possible is size == reducedBuckets.size() we can use the entire list of reduced buckets
-                list = reducedBucketsArr;
-                for (B bucket : reducedBucketsArr) {
+                final BucketPriorityQueue<B> ordered = new BucketPriorityQueue<>(size, order.comparator());
+                for (B bucket : reducedBuckets) {
                     if (sumDocCountError == -1) {
                         bucket.setDocCountError(-1);
                     } else {
                         final long finalSumDocCountError = sumDocCountError;
                         bucket.setDocCountError(docCountError -> docCountError + finalSumDocCountError);
                     }
                     if (bucket.getDocCount() >= localBucketCountThresholds.getMinDocCount()) {
-                        reduceContext.consumeBucketsAndMaybeBreak(1);
+                        B removed = ordered.insertWithOverflow(bucket);
+                        if (removed != null) {
+                            otherDocCount += removed.getDocCount();
+                            reduceContext.consumeBucketsAndMaybeBreak(-countInnerBucket(removed));
+                        } else {
+                            reduceContext.consumeBucketsAndMaybeBreak(1);
+                        }
                     } else {
                         reduceContext.consumeBucketsAndMaybeBreak(-countInnerBucket(bucket));
                     }
                 }
+                list = createBucketsArray(ordered.size());
+                for (int i = ordered.size() - 1; i >= 0; i--) {
+                    list[i] = ordered.pop();
+                }
             }
-            Arrays.sort(list, cmp);
         } else {
             // we can prune the list on partial reduce if the aggregation is ordered by key
             // and not filtered (minDocCount == 0)