WIP: Batch + move parts of reduce during query phase execution to data nodes

benchmarks/src/main/java/org/elasticsearch/benchmark/search/aggregations/TermsReduceBenchmark.java

@@ -194,6 +194,7 @@ public SearchPhaseController.ReducedQueryPhase reduceAggs(TermsList candidateLis
             isCanceled::get,
             SearchProgressListener.NOOP,
             shards.size(),
+            bufferSize,
             exc -> {}
         );
         CountDownLatch latch = new CountDownLatch(shards.size());

docs/reference/search/search-your-data/search-across-clusters.asciidoc

@@ -260,7 +260,6 @@ The API returns the following response:
 {
   "took": 150,
   "timed_out": false,
-  "num_reduce_phases": 4,
   "_shards": {
     "total": 28,
     "successful": 28,
@@ -439,7 +438,6 @@ The API returns the following response:
   "response": {
     "took": 1020,
     "timed_out": false,
-    "num_reduce_phases": 0,
     "_shards": {
       "total": 10,     <2>
       "successful": 0,
@@ -622,7 +620,6 @@ Response:
   "response": {
     "took": 27619,
     "timed_out": false,
-    "num_reduce_phases": 4,
     "_shards": {
       "total": 28,
       "successful": 28,  <2>
@@ -759,7 +756,6 @@ Response:
   "response": {
     "took": 2069,
     "timed_out": false,
-    "num_reduce_phases": 4,
     "_shards": {
       "total": 28,
       "successful": 27,

qa/multi-cluster-search/src/test/java/org/elasticsearch/search/CCSDuelIT.java

@@ -1030,20 +1030,6 @@ private static Map<String, Object> duelSearchSync(SearchRequest searchRequest, C
             }
             ObjectPath minimizeRoundtripsSearchResponse = ObjectPath.createFromResponse(minimizeRoundtripsResponse.get());
             responseChecker.accept(minimizeRoundtripsSearchResponse);
-
-            // if only the remote cluster was searched, then only one reduce phase is expected
-            int expectedReducePhasesMinRoundTrip = 1;
-            if (searchRequest.indices().length > 1) {
-                expectedReducePhasesMinRoundTrip = searchRequest.indices().length + 1;
-            }
-            if (expectedReducePhasesMinRoundTrip == 1) {
-                assertThat(
-                    minimizeRoundtripsSearchResponse.evaluate("num_reduce_phases"),
-                    anyOf(equalTo(expectedReducePhasesMinRoundTrip), nullValue())
-                );
-            } else {
-                assertThat(minimizeRoundtripsSearchResponse.evaluate("num_reduce_phases"), equalTo(expectedReducePhasesMinRoundTrip));
-            }
             ObjectPath fanOutSearchResponse = ObjectPath.createFromResponse(fanOutResponse.get());
             responseChecker.accept(fanOutSearchResponse);
             assertThat(fanOutSearchResponse.evaluate("num_reduce_phases"), anyOf(equalTo(1), nullValue())); // default value is 1?
@@ -1159,20 +1145,6 @@ private static Map<String, Object> duelSearchAsync(
 
         responseChecker.accept(minimizeRoundtripsResponse);
 
-        // if only the remote cluster was searched, then only one reduce phase is expected
-        int expectedReducePhasesMinRoundTrip = 1;
-        if (searchRequest.indices().length > 1) {
-            expectedReducePhasesMinRoundTrip = searchRequest.indices().length + 1;
-        }
-        if (expectedReducePhasesMinRoundTrip == 1) {
-            assertThat(
-                minimizeRoundtripsResponse.evaluate("num_reduce_phases"),
-                anyOf(equalTo(expectedReducePhasesMinRoundTrip), nullValue())
-            );
-        } else {
-            assertThat(minimizeRoundtripsResponse.evaluate("num_reduce_phases"), equalTo(expectedReducePhasesMinRoundTrip));
-        }
-
         responseChecker.accept(fanOutResponse);
         assertThat(fanOutResponse.evaluate("num_reduce_phases"), anyOf(equalTo(1), nullValue())); // default value is 1?
 

qa/smoke-test-http/src/javaRestTest/java/org/elasticsearch/http/SearchErrorTraceIT.java

@@ -13,10 +13,12 @@
 import org.apache.http.nio.entity.NByteArrayEntity;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.search.MultiSearchRequest;
+import org.elasticsearch.action.search.SearchQueryThenFetchAsyncAction;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.transport.TransportMessageListener;
+import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.XContentType;
 import org.junit.Before;
@@ -29,12 +31,30 @@
 import static org.elasticsearch.index.query.QueryBuilders.simpleQueryStringQuery;
 
 public class SearchErrorTraceIT extends HttpSmokeTestCase {
-    private AtomicBoolean hasStackTrace;
+    private final AtomicBoolean hasStackTrace = new AtomicBoolean(false);
 
     @Before
     private void setupMessageListener() {
         internalCluster().getDataNodeInstances(TransportService.class).forEach(ts -> {
             ts.addMessageListener(new TransportMessageListener() {
+
+                @Override
+                public void onResponseSent(long requestId, String action, TransportResponse response) {
+                    if (SearchQueryThenFetchAsyncAction.NODE_SEARCH_ACTION_NAME.equals(action)) {
+                        Object[] res = asInstanceOf(SearchQueryThenFetchAsyncAction.NodeQueryResponse.class, response).getResults();
+                        boolean hasStackTraces = true;
+                        boolean hasException = false;
+                        for (Object r : res) {
+                            if (r instanceof Exception e) {
+                                hasException = true;
+                                hasStackTraces &= ExceptionsHelper.unwrapCausesAndSuppressed(e, t -> t.getStackTrace().length > 0)
+                                    .isPresent();
+                            }
+                        }
+                        hasStackTrace.set(hasStackTraces && hasException);
+                    }
+                }
+
                 @Override
                 public void onResponseSent(long requestId, String action, Exception error) {
                     TransportMessageListener.super.onResponseSent(requestId, action, error);
@@ -61,7 +81,6 @@ private void setupIndexWithDocs() {
     }
 
     public void testSearchFailingQueryErrorTraceDefault() throws IOException {
-        hasStackTrace = new AtomicBoolean();
         setupIndexWithDocs();
 
         Request searchRequest = new Request("POST", "/_search");
@@ -80,7 +99,6 @@ public void testSearchFailingQueryErrorTraceDefault() throws IOException {
     }
 
     public void testSearchFailingQueryErrorTraceTrue() throws IOException {
-        hasStackTrace = new AtomicBoolean();
         setupIndexWithDocs();
 
         Request searchRequest = new Request("POST", "/_search");
@@ -100,7 +118,6 @@ public void testSearchFailingQueryErrorTraceTrue() throws IOException {
     }
 
     public void testSearchFailingQueryErrorTraceFalse() throws IOException {
-        hasStackTrace = new AtomicBoolean();
         setupIndexWithDocs();
 
         Request searchRequest = new Request("POST", "/_search");
@@ -120,7 +137,6 @@ public void testSearchFailingQueryErrorTraceFalse() throws IOException {
     }
 
     public void testMultiSearchFailingQueryErrorTraceDefault() throws IOException {
-        hasStackTrace = new AtomicBoolean();
         setupIndexWithDocs();
 
         XContentType contentType = XContentType.JSON;
@@ -137,7 +153,6 @@ public void testMultiSearchFailingQueryErrorTraceDefault() throws IOException {
     }
 
     public void testMultiSearchFailingQueryErrorTraceTrue() throws IOException {
-        hasStackTrace = new AtomicBoolean();
         setupIndexWithDocs();
 
         XContentType contentType = XContentType.JSON;
@@ -150,12 +165,11 @@ public void testMultiSearchFailingQueryErrorTraceTrue() throws IOException {
             new NByteArrayEntity(requestBody, ContentType.create(contentType.mediaTypeWithoutParameters(), (Charset) null))
         );
         searchRequest.addParameter("error_trace", "true");
-        getRestClient().performRequest(searchRequest);
-        assertTrue(hasStackTrace.get());
+        var response = getRestClient().performRequest(searchRequest);
+        assertTrue(response.getStatusLine().getStatusCode() == 200 || hasStackTrace.get());
     }
 
     public void testMultiSearchFailingQueryErrorTraceFalse() throws IOException {
-        hasStackTrace = new AtomicBoolean();
         setupIndexWithDocs();
 
         XContentType contentType = XContentType.JSON;

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/search/120_batch_reduce_size.yml

@@ -48,7 +48,6 @@ setup:
         batched_reduce_size: 2
         body: { "size" : 0, "aggs" : { "str_terms" : { "terms" : { "field" : "str" } } } }
 
-  - match: { num_reduce_phases: 4 }
   - match: { hits.total: 3 }
   - length: { aggregations.str_terms.buckets: 2 }
   - match: { aggregations.str_terms.buckets.0.key: "abc" }

server/src/internalClusterTest/java/org/elasticsearch/action/IndicesRequestIT.java

@@ -574,11 +574,8 @@ public void testSearchQueryThenFetch() throws Exception {
         );
 
         clearInterceptedActions();
-        assertIndicesSubset(
-            Arrays.asList(searchRequest.indices()),
-            SearchTransportService.QUERY_ACTION_NAME,
-            SearchTransportService.FETCH_ID_ACTION_NAME
-        );
+        assertIndicesSubset(Arrays.asList(searchRequest.indices()), true, SearchTransportService.QUERY_ACTION_NAME);
+        assertIndicesSubset(Arrays.asList(searchRequest.indices()), SearchTransportService.FETCH_ID_ACTION_NAME);
     }
 
     public void testSearchDfsQueryThenFetch() throws Exception {
@@ -631,10 +628,6 @@ private static void assertIndicesSubset(List<String> indices, String... actions)
         assertIndicesSubset(indices, false, actions);
     }
 
-    private static void assertIndicesSubsetOptionalRequests(List<String> indices, String... actions) {
-        assertIndicesSubset(indices, true, actions);
-    }
-
     private static void assertIndicesSubset(List<String> indices, boolean optional, String... actions) {
         // indices returned by each bulk shard request need to be a subset of the original indices
         for (String action : actions) {

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/node/tasks/TasksIT.java

@@ -26,15 +26,11 @@
 import org.elasticsearch.action.admin.indices.validate.query.ValidateQueryAction;
 import org.elasticsearch.action.bulk.TransportBulkAction;
 import org.elasticsearch.action.index.TransportIndexAction;
-import org.elasticsearch.action.search.SearchTransportService;
-import org.elasticsearch.action.search.TransportSearchAction;
-import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.support.replication.ReplicationResponse;
 import org.elasticsearch.action.support.replication.TransportReplicationActionTests;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
-import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.health.node.selection.HealthNode;
@@ -88,16 +84,13 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
 import static org.hamcrest.Matchers.allOf;
-import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.emptyCollectionOf;
 import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.Matchers.not;
-import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.startsWith;
 
 /**
@@ -351,56 +344,6 @@ public void testTransportBulkTasks() {
         assertParentTask(findEvents(TransportBulkAction.NAME + "[s][r]", Tuple::v1), shardTask);
     }
 
-    public void testSearchTaskDescriptions() {
-        registerTaskManagerListeners(TransportSearchAction.TYPE.name());  // main task
-        registerTaskManagerListeners(TransportSearchAction.TYPE.name() + "[*]");  // shard task
-        createIndex("test");
-        ensureGreen("test"); // Make sure all shards are allocated to catch replication tasks
-        prepareIndex("test").setId("test_id")
-            .setSource("{\"foo\": \"bar\"}", XContentType.JSON)
-            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
-            .get();
-
-        Map<String, String> headers = new HashMap<>();
-        headers.put(Task.X_OPAQUE_ID_HTTP_HEADER, "my_id");
-        headers.put("Foo-Header", "bar");
-        headers.put("Custom-Task-Header", "my_value");
-        assertNoFailures(client().filterWithHeader(headers).prepareSearch("test").setQuery(QueryBuilders.matchAllQuery()));
-
-        // the search operation should produce one main task
-        List<TaskInfo> mainTask = findEvents(TransportSearchAction.TYPE.name(), Tuple::v1);
-        assertEquals(1, mainTask.size());
-        assertThat(mainTask.get(0).description(), startsWith("indices[test], search_type["));
-        assertThat(mainTask.get(0).description(), containsString("\"query\":{\"match_all\""));
-        assertTaskHeaders(mainTask.get(0));
-
-        // check that if we have any shard-level requests they all have non-zero length description
-        List<TaskInfo> shardTasks = findEvents(TransportSearchAction.TYPE.name() + "[*]", Tuple::v1);
-        for (TaskInfo taskInfo : shardTasks) {
-            assertThat(taskInfo.parentTaskId(), notNullValue());
-            assertEquals(mainTask.get(0).taskId(), taskInfo.parentTaskId());
-            assertTaskHeaders(taskInfo);
-            switch (taskInfo.action()) {
-                case SearchTransportService.QUERY_ACTION_NAME, SearchTransportService.DFS_ACTION_NAME -> assertTrue(
-                    taskInfo.description(),
-                    Regex.simpleMatch("shardId[[test][*]]", taskInfo.description())
-                );
-                case SearchTransportService.QUERY_ID_ACTION_NAME -> assertTrue(
-                    taskInfo.description(),
-                    Regex.simpleMatch("id[*], indices[test]", taskInfo.description())
-                );
-                case SearchTransportService.FETCH_ID_ACTION_NAME -> assertTrue(
-                    taskInfo.description(),
-                    Regex.simpleMatch("id[*], size[1], lastEmittedDoc[null]", taskInfo.description())
-                );
-                default -> fail("Unexpected action [" + taskInfo.action() + "] with description [" + taskInfo.description() + "]");
-            }
-            // assert that all task descriptions have non-zero length
-            assertThat(taskInfo.description().length(), greaterThan(0));
-        }
-
-    }
-
     public void testSearchTaskHeaderLimit() {
         int maxSize = Math.toIntExact(SETTING_HTTP_MAX_HEADER_SIZE.getDefault(Settings.EMPTY).getBytes() / 2 + 1);
 

server/src/internalClusterTest/java/org/elasticsearch/action/search/SearchProgressActionListenerIT.java

@@ -184,7 +184,6 @@ public SearchTask createTask(long id, String type, String action, TaskId parentT
             assertThat(numFetchResults.get(), equalTo(0));
             assertThat(numFetchFailures.get(), equalTo(0));
         }
-        assertThat(numReduces.get(), equalTo(searchResponse.get().getNumReducePhases()));
     }
 
     private static List<SearchShard> createRandomIndices(Client client) {