Fix cardinality agg pruning optimization by self collecting

CHANGELOG.md

@@ -78,6 +78,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix IntervalQuery flaky test ([#19332](https://github.com/opensearch-project/OpenSearch/pull/19332))
 - Fix ingestion state xcontent serialization in IndexMetadata and fail fast on mapping errors([#19320](https://github.com/opensearch-project/OpenSearch/pull/19320))
 - Fix updated keyword field params leading to stale responses from request cache ([#19385](https://github.com/opensearch-project/OpenSearch/pull/19385))
+- Fix cardinality agg pruning optimization by self collecting ([#19473](https://github.com/opensearch-project/OpenSearch/pull/19473))
 - Implement SslHandler retrieval logic for transport-reactor-netty4 plugin ([#19458](https://github.com/opensearch-project/OpenSearch/pull/19458))
 
 ### Dependencies

server/src/main/java/org/opensearch/search/aggregations/metrics/CardinalityAggregator.java

@@ -40,7 +40,6 @@
 import org.apache.lucene.index.Term;
 import org.apache.lucene.index.Terms;
 import org.apache.lucene.index.TermsEnum;
-import org.apache.lucene.search.BulkScorer;
 import org.apache.lucene.search.CollectionTerminatedException;
 import org.apache.lucene.search.DisiPriorityQueue;
 import org.apache.lucene.search.DisiWrapper;
@@ -227,12 +226,14 @@ private Collector tryWrapWithPruningCollector(Collector collector, Terms terms,
     private boolean tryScoreWithPruningCollector(LeafReaderContext ctx, Collector pruningCollector) throws IOException {
         try {
             Weight weight = context.query().rewrite(context.searcher()).createWeight(context.searcher(), ScoreMode.TOP_DOCS, 1f);
-            BulkScorer scorer = weight.bulkScorer(ctx);
+            Scorer scorer = weight.scorer(ctx);
             if (scorer == null) {
                 return false;
             }
-            Bits liveDocs = ctx.reader().getLiveDocs();
-            scorer.score(pruningCollector, liveDocs, 0, DocIdSetIterator.NO_MORE_DOCS);
+            pruningCollector.setScorer(scorer);
+            DocIdSetIterator iterator = scorer.iterator();
+            bulkCollect(ctx.reader().getLiveDocs(), iterator, pruningCollector);
+
             pruningCollector.postCollect();
             Releasables.close(pruningCollector);
         } catch (Exception e) {
@@ -247,6 +248,26 @@ private boolean tryScoreWithPruningCollector(LeafReaderContext ctx, Collector pr
         return true;
     }
 
+    private void bulkCollect(Bits acceptDocs, DocIdSetIterator iterator, Collector pruningCollector) throws IOException {
+        DocIdSetIterator competitiveIterator = pruningCollector.competitiveIterator();
+
+        int doc = iterator.nextDoc();
+        while (doc < DocIdSetIterator.NO_MORE_DOCS) {
+            assert competitiveIterator.docID() <= doc; // invariant
+            if (competitiveIterator.docID() < doc) {
+                int competitiveNext = competitiveIterator.advance(doc);
+                if (competitiveNext != doc) {
+                    doc = iterator.advance(competitiveNext);
+                    continue;
+                }
+            }
+            if ((acceptDocs == null || acceptDocs.get(doc))) {
+                pruningCollector.collect(doc);
+            }
+            doc = iterator.nextDoc();
+        }
+    }
+
     private Collector getNoOpCollector() {
         return new Collector() {
             @Override