Revert "#17593 Add Extension Points For Pre and Post Collection of Scores in QueryPhase"

CHANGELOG.md

@@ -27,7 +27,6 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Workload Management] Modify logging message in WorkloadGroupService ([#18712](https://github.com/opensearch-project/OpenSearch/pull/18712))
 - Add BooleanQuery rewrite moving constant-scoring must clauses to filter clauses ([#18510](https://github.com/opensearch-project/OpenSearch/issues/18510))
 - Add functionality for plugins to inject QueryCollectorContext during QueryPhase ([#18637](https://github.com/opensearch-project/OpenSearch/pull/18637))
-- Add QueryPhaseListener interface for pre/post collection hooks ([#17593](https://github.com/opensearch-project/OpenSearch/issues/17593))
 - Add support for non-timing info in profiler ([#18460](https://github.com/opensearch-project/OpenSearch/issues/18460))
 - [Rule-based auto tagging] Bug fix and improvements ([#18726](https://github.com/opensearch-project/OpenSearch/pull/18726))
 - Extend Approximation Framework to other numeric types ([#18530](https://github.com/opensearch-project/OpenSearch/issues/18530))

server/src/main/java/org/opensearch/search/query/ConcurrentQueryPhaseSearcher.java

@@ -19,6 +19,7 @@
 import org.opensearch.search.internal.ContextIndexSearcher;
 import org.opensearch.search.internal.SearchContext;
 import org.opensearch.search.profile.query.ProfileCollectorManager;
+import org.opensearch.search.query.QueryPhase.DefaultQueryPhaseSearcher;
 
 import java.io.IOException;
 import java.util.LinkedList;
@@ -29,7 +30,7 @@
  * The implementation of the {@link QueryPhaseSearcher} which attempts to use concurrent
  * search of Apache Lucene segments if it has been enabled.
  */
-public class ConcurrentQueryPhaseSearcher extends AbstractQueryPhaseSearcher {
+public class ConcurrentQueryPhaseSearcher extends DefaultQueryPhaseSearcher {
     private static final Logger LOGGER = LogManager.getLogger(ConcurrentQueryPhaseSearcher.class);
     private final AggregationProcessor aggregationProcessor = new ConcurrentAggregationProcessor();
 
@@ -39,15 +40,15 @@ public class ConcurrentQueryPhaseSearcher extends AbstractQueryPhaseSearcher {
     public ConcurrentQueryPhaseSearcher() {}
 
     @Override
-    protected boolean doSearchWith(
+    protected boolean searchWithCollector(
         SearchContext searchContext,
         ContextIndexSearcher searcher,
         Query query,
         LinkedList<QueryCollectorContext> collectors,
+        QueryCollectorContext queryCollectorContext,
         boolean hasFilterCollector,
         boolean hasTimeout
     ) throws IOException {
-        QueryCollectorContext queryCollectorContext = getQueryCollectorContext(searchContext, hasFilterCollector);
         return searchWithCollectorManager(
             searchContext,
             searcher,

server/src/main/java/org/opensearch/search/query/QueryPhase.java

@@ -76,13 +76,15 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.stream.Collectors;
 
 import static org.opensearch.search.query.QueryCollectorContext.createEarlyTerminationCollectorContext;
 import static org.opensearch.search.query.QueryCollectorContext.createFilteredCollectorContext;
 import static org.opensearch.search.query.QueryCollectorContext.createMinScoreCollectorContext;
 import static org.opensearch.search.query.QueryCollectorContext.createMultiCollectorContext;
+import static org.opensearch.search.query.TopDocsCollectorContext.createTopDocsCollectorContext;
 
 /**
  * Query phase of a search request, used to run the query and get back from each shard information about the matching documents
@@ -409,7 +411,7 @@ public static class TimeExceededException extends RuntimeException {
      *
      * @opensearch.internal
      */
-    public static class DefaultQueryPhaseSearcher extends AbstractQueryPhaseSearcher {
+    public static class DefaultQueryPhaseSearcher implements QueryPhaseSearcher {
         private final AggregationProcessor aggregationProcessor;
 
         /**
@@ -420,7 +422,7 @@ protected DefaultQueryPhaseSearcher() {
         }
 
         @Override
-        protected boolean doSearchWith(
+        public boolean searchWith(
             SearchContext searchContext,
             ContextIndexSearcher searcher,
             Query query,
@@ -445,6 +447,47 @@ protected boolean searchWithCollector(
             boolean hasTimeout
         ) throws IOException {
             QueryCollectorContext queryCollectorContext = getQueryCollectorContext(searchContext, hasFilterCollector);
+            return searchWithCollector(searchContext, searcher, query, collectors, queryCollectorContext, hasFilterCollector, hasTimeout);
+        }
+
+        private QueryCollectorContext getQueryCollectorContext(SearchContext searchContext, boolean hasFilterCollector) throws IOException {
+            // create the top docs collector last when the other collectors are known
+            final Optional<QueryCollectorContext> queryCollectorContextOpt = QueryCollectorContextSpecRegistry.getQueryCollectorContextSpec(
+                searchContext,
+                new QueryCollectorArguments.Builder().hasFilterCollector(hasFilterCollector).build()
+            ).map(queryCollectorContextSpec -> new QueryCollectorContext(queryCollectorContextSpec.getContextName()) {
+                @Override
+                Collector create(Collector in) throws IOException {
+                    return queryCollectorContextSpec.create(in);
+                }
+
+                @Override
+                CollectorManager<?, ReduceableSearchResult> createManager(CollectorManager<?, ReduceableSearchResult> in)
+                    throws IOException {
+                    return queryCollectorContextSpec.createManager(in);
+                }
+
+                @Override
+                void postProcess(QuerySearchResult result) throws IOException {
+                    queryCollectorContextSpec.postProcess(result);
+                }
+            });
+            if (queryCollectorContextOpt.isPresent()) {
+                return queryCollectorContextOpt.get();
+            } else {
+                return createTopDocsCollectorContext(searchContext, hasFilterCollector);
+            }
+        }
+
+        protected boolean searchWithCollector(
+            SearchContext searchContext,
+            ContextIndexSearcher searcher,
+            Query query,
+            LinkedList<QueryCollectorContext> collectors,
+            QueryCollectorContext queryCollectorContext,
+            boolean hasFilterCollector,
+            boolean hasTimeout
+        ) throws IOException {
             return QueryPhase.searchWithCollector(
                 searchContext,
                 searcher,
@@ -455,6 +498,5 @@ protected boolean searchWithCollector(
                 hasTimeout
             );
         }
-
     }
 }

server/src/main/java/org/opensearch/search/query/QueryPhaseSearcher.java

@@ -17,9 +17,7 @@
 import org.opensearch.search.internal.SearchContext;
 
 import java.io.IOException;
-import java.util.Collections;
 import java.util.LinkedList;
-import java.util.List;
 
 /**
  * The extension point which allows to plug in custom search implementation to be
@@ -55,12 +53,4 @@ boolean searchWith(
     default AggregationProcessor aggregationProcessor(SearchContext searchContext) {
         return new DefaultAggregationProcessor();
     }
-
-    /**
-     * Get the list of query phase listeners that should be executed before and after score collection.
-     * @return list of query phase listeners, empty list if none
-     */
-    default List<QueryPhaseListener> queryPhaseListeners() {
-        return Collections.emptyList();
-    }
 }