Fold InitialSearchPhase into AbstractSearchAsyncAction (#47182)

Historically, we have two base classes for search actions that generally need to fan out to multiple shards and then move on to the following phase: InitialSearchPhase and AbstractSearchAsyncAction that extends it. Practically, every search action extends the latter, and there are no direct subclasses of InitialSearchPhase in our codebase.

This commit folds InitialSearchPhase into AbstractSearchAsyncAction in the attempt of simplifying things and making the search code running on the coordinating node easier to reason about.

server/src/main/java/org/elasticsearch/action/search/ArraySearchPhaseResults.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.search;
+
+import org.elasticsearch.common.util.concurrent.AtomicArray;
+import org.elasticsearch.search.SearchPhaseResult;
+
+import java.util.stream.Stream;
+
+/**
+ * This class acts as a basic result collection that can be extended to do on-the-fly reduction or result processing
+ */
+class ArraySearchPhaseResults<Result extends SearchPhaseResult> extends SearchPhaseResults<Result> {
+    final AtomicArray<Result> results;
+
+    ArraySearchPhaseResults(int size) {
+        super(size);
+        this.results = new AtomicArray<>(size);
+    }
+
+    Stream<Result> getSuccessfulResults() {
+        return results.asList().stream();
+    }
+
+    void consumeResult(Result result) {
+        assert results.get(result.getShardIndex()) == null : "shardIndex: " + result.getShardIndex() + " is already set";
+        results.set(result.getShardIndex(), result);
+    }
+
+    boolean hasResult(int shardIndex) {
+        return results.get(shardIndex) != null;
+    }
+
+    @Override
+    AtomicArray<Result> getAtomicArray() {
+        return results;
+    }
+}

server/src/main/java/org/elasticsearch/action/search/CanMatchPreFilterSearchPhase.java

@@ -97,8 +97,7 @@ private GroupShardsIterator<SearchShardIterator> getIterator(BitSetSearchPhaseRe
         return shardsIts;
     }
 
-    private static final class BitSetSearchPhaseResults extends InitialSearchPhase.
-        SearchPhaseResults<SearchService.CanMatchResponse> {
+    private static final class BitSetSearchPhaseResults extends SearchPhaseResults<SearchService.CanMatchResponse> {
 
         private final FixedBitSet possibleMatches;
         private int numPossibleMatches;

server/src/main/java/org/elasticsearch/action/search/DfsQueryPhase.java

@@ -40,16 +40,16 @@
  * @see CountedCollector#onFailure(int, SearchShardTarget, Exception)
  */
 final class DfsQueryPhase extends SearchPhase {
-    private final InitialSearchPhase.ArraySearchPhaseResults<SearchPhaseResult> queryResult;
+    private final ArraySearchPhaseResults<SearchPhaseResult> queryResult;
     private final SearchPhaseController searchPhaseController;
     private final AtomicArray<DfsSearchResult> dfsSearchResults;
-    private final Function<InitialSearchPhase.ArraySearchPhaseResults<SearchPhaseResult>, SearchPhase> nextPhaseFactory;
+    private final Function<ArraySearchPhaseResults<SearchPhaseResult>, SearchPhase> nextPhaseFactory;
     private final SearchPhaseContext context;
     private final SearchTransportService searchTransportService;
 
     DfsQueryPhase(AtomicArray<DfsSearchResult> dfsSearchResults,
                   SearchPhaseController searchPhaseController,
-                  Function<InitialSearchPhase.ArraySearchPhaseResults<SearchPhaseResult>, SearchPhase> nextPhaseFactory,
+                  Function<ArraySearchPhaseResults<SearchPhaseResult>, SearchPhase> nextPhaseFactory,
                   SearchPhaseContext context) {
         super("dfs_query");
         this.queryResult = searchPhaseController.newSearchPhaseResults(context.getRequest(), context.getNumShards());

server/src/main/java/org/elasticsearch/action/search/FetchSearchPhase.java

@@ -48,17 +48,17 @@ final class FetchSearchPhase extends SearchPhase {
     private final BiFunction<InternalSearchResponse, String, SearchPhase> nextPhaseFactory;
     private final SearchPhaseContext context;
     private final Logger logger;
-    private final InitialSearchPhase.SearchPhaseResults<SearchPhaseResult> resultConsumer;
+    private final SearchPhaseResults<SearchPhaseResult> resultConsumer;
 
-    FetchSearchPhase(InitialSearchPhase.SearchPhaseResults<SearchPhaseResult> resultConsumer,
+    FetchSearchPhase(SearchPhaseResults<SearchPhaseResult> resultConsumer,
                      SearchPhaseController searchPhaseController,
                      SearchPhaseContext context) {
         this(resultConsumer, searchPhaseController, context,
             (response, scrollId) -> new ExpandSearchPhase(context, response, // collapse only happens if the request has inner hits
                 (finalResponse) -> sendResponsePhase(finalResponse, scrollId, context)));
     }
 
-    FetchSearchPhase(InitialSearchPhase.SearchPhaseResults<SearchPhaseResult> resultConsumer,
+    FetchSearchPhase(SearchPhaseResults<SearchPhaseResult> resultConsumer,
                      SearchPhaseController searchPhaseController,
                      SearchPhaseContext context, BiFunction<InternalSearchResponse, String, SearchPhase> nextPhaseFactory) {
         super("fetch");