Introduce ability to minimize round-trips in CCS

client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java

@@ -398,6 +398,9 @@ private static void addSearchRequestParams(Params params, SearchRequest searchRe
         params.withPreference(searchRequest.preference());
         params.withIndicesOptions(searchRequest.indicesOptions());
         params.putParam("search_type", searchRequest.searchType().name().toLowerCase(Locale.ROOT));
+        if (searchRequest.getCCSReduceMode() != null) {
+            params.putParam("ccs_reduce_mode", searchRequest.getCCSReduceMode().toString());
+        }
         if (searchRequest.requestCache() != null) {
             params.putParam("request_cache", Boolean.toString(searchRequest.requestCache()));
         }

client/rest-high-level/src/test/java/org/elasticsearch/client/RequestConvertersTests.java

@@ -41,6 +41,7 @@
 import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.get.MultiGetRequest;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.CCSReduceMode;
 import org.elasticsearch.action.search.ClearScrollRequest;
 import org.elasticsearch.action.search.MultiSearchRequest;
 import org.elasticsearch.action.search.SearchRequest;
@@ -1239,7 +1240,7 @@ public void testMultiSearch() throws IOException {
             requests.add(searchRequest);
         };
         MultiSearchRequest.readMultiLineFormat(new BytesArray(EntityUtils.toByteArray(request.getEntity())),
-                REQUEST_BODY_CONTENT_TYPE.xContent(), consumer, null, multiSearchRequest.indicesOptions(), null, null, null,
+                REQUEST_BODY_CONTENT_TYPE.xContent(), consumer, null, multiSearchRequest.indicesOptions(), null, null, null, null,
                 xContentRegistry(), true);
         assertEquals(requests, multiSearchRequest.requests());
     }
@@ -1862,6 +1863,10 @@ private static void setRandomSearchParams(SearchRequest searchRequest,
             searchRequest.scroll(randomTimeValue());
             expectedParams.put("scroll", searchRequest.scroll().keepAlive().getStringRep());
         }
+        if (randomBoolean()) {
+            searchRequest.setCCSReduceMode(randomFrom(CCSReduceMode.values()));
+            expectedParams.put("ccs_reduce_mode", searchRequest.getCCSReduceMode().toString());
+        }
     }
 
     static void setRandomIndicesOptions(Consumer<IndicesOptions> setter, Supplier<IndicesOptions> getter,

docs/reference/modules/cross-cluster-search.asciidoc

@@ -65,13 +65,15 @@ GET /cluster_one:twitter/_search
 {
   "took": 150,
   "timed_out": false,
+  "num_reduce_phases": 2,
   "_shards": {
     "total": 1,
     "successful": 1,
     "failed": 0,
     "skipped": 0
   },
   "_clusters": {
+    "ccs_reduce_mode": "remote",
     "total": 1,
     "successful": 1,
     "skipped": 0
@@ -130,13 +132,15 @@ will be prefixed with their remote cluster name:
 {
   "took": 150,
   "timed_out": false,
+  "num_reduce_phases": 3,
   "_shards": {
     "total": 2,
     "successful": 2,
     "failed": 0,
     "skipped": 0
   },
   "_clusters": {
+    "ccs_reduce_mode": "remote",
     "total": 2,
     "successful": 2,
     "skipped": 0
@@ -222,13 +226,15 @@ GET /cluster_one:twitter,cluster_two:twitter,twitter/_search <1>
 {
   "took": 150,
   "timed_out": false,
+  "num_reduce_phases": 3,
   "_shards": {
     "total": 2,
     "successful": 2,
     "failed": 0,
     "skipped": 0
   },
   "_clusters": { <1>
+    "ccs_reduce_mode": "remote",
     "total": 3,
     "successful": 2,
     "skipped": 1
@@ -273,3 +279,42 @@ GET /cluster_one:twitter,cluster_two:twitter,twitter/_search <1>
 // TESTRESPONSE[s/"_score": 1/"_score": "$body.hits.hits.0._score"/]
 // TESTRESPONSE[s/"_score": 2/"_score": "$body.hits.hits.1._score"/]
 <1> The `clusters` section indicates that one cluster was unavailable and got skipped
+
+[float]
+[[ccs-reduce-mode]]
+=== Reduce mode
+
+When searching across remote clusters, there are two possible reduce modes:
+
+- `remote`: the coordinating node sends one search request to each cluster.
+Each cluster performs the search independently, reducing and fetching results
+which are then returned to the caller. Once the CCS node has received all the
+responses, it performs another reduction and returns the relevant results back
+to the user. The `remote` reduce mode is beneficial when there is network
+latency between the coordinating node and the remote clusters involved, which
+is typically the case. A single request is sent to each remote cluster, at the
+cost of retrieving `from` + `size` already fetched results.
+
+- `local`: the coordinating node sends a <<search-shards,search shards>>
+request to each remote cluster, in order to collect information about their
+corresponding remote indices involved in the search request and the shards
+where their data is located. Once each cluster has responded to such request,
+the search executes as if all shards were part of the same cluster. The
+coordinating node sends one request to each shard involved, each shard
+executes the query and returns its own results which are then reduced (and
+fetched, depending on the <<search-request-search-type, search type>>) by the
+coordinating node. The `local` reduce mode is beneficial whenever there is very
+ low network latency between the coordinating node and the remote clusters
+ involved, as it treats all shards the same, at the cost of sending many
+ requests to each remote cluster.
+
+By default `remote` is used whenever possible. In case a scroll is provided,
+or inner hits are requested as part of field collapsing, `local` is used
+instead. The <<search-request-body, search API>> supports the `ccs_reduce_mode`
+parameter which allows to select the reduce mode if needed. The reduce mode
+that was used for a cross-cluster search request is returned as part of the
+`_clusters` section.
+
+Note that all the communication between the nodes, regardless of which cluster
+they belong to and the selected reduce mode, happens through the
+<<modules-transport,transport layer>>.

docs/reference/search/request-body.asciidoc

@@ -113,6 +113,10 @@ And here is a sample response:
     reduce the memory overhead per search request if the potential  number of
     shards in the request can be large.
 
+`ccs_reduce_mode`::
+
+    The cross-cluster search reduce mode. Can be `local` or `remote`.
+    See <<ccs-reduce-mode>> for more.
 
 
 Out of the above, the `search_type`, `request_cache` and the `allow_partial_search_results` 

modules/lang-mustache/src/main/java/org/elasticsearch/script/mustache/MultiSearchTemplateRequest.java

@@ -65,7 +65,6 @@ public MultiSearchTemplateRequest add(SearchTemplateRequest request) {
         return this;
     }
 
-
     /**
      * Returns the amount of search requests specified in this multi search requests are allowed to be ran concurrently.
      */

modules/lang-mustache/src/test/java/org/elasticsearch/script/mustache/MultiSearchTemplateResponseTests.java

@@ -19,6 +19,7 @@
 package org.elasticsearch.script.mustache;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.search.CCSReduceMode;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.common.Strings;
@@ -50,7 +51,7 @@ protected MultiSearchTemplateResponse createTestInstance() {
             int successfulShards = randomIntBetween(0, totalShards);
             int skippedShards = totalShards - successfulShards;
             InternalSearchResponse internalSearchResponse = InternalSearchResponse.empty();
-            SearchResponse.Clusters clusters = new SearchResponse.Clusters(totalShards, successfulShards, skippedShards);
+            SearchResponse.Clusters clusters = randomClusters();
             SearchTemplateResponse searchTemplateResponse = new SearchTemplateResponse();
             SearchResponse searchResponse = new SearchResponse(internalSearchResponse, null, totalShards,
                     successfulShards, skippedShards, tookInMillis, ShardSearchFailure.EMPTY_ARRAY, clusters);
@@ -59,22 +60,28 @@ protected MultiSearchTemplateResponse createTestInstance() {
         }
         return new MultiSearchTemplateResponse(items, overallTookInMillis);
     }
-
+
+    private static SearchResponse.Clusters randomClusters() {
+        int totalClusters = randomIntBetween(0, 10);
+        int successfulClusters = randomIntBetween(0, totalClusters);
+        int skippedClusters = totalClusters - successfulClusters;
+        CCSReduceMode executionMode = randomFrom(CCSReduceMode.values());
+        return new SearchResponse.Clusters(totalClusters, successfulClusters, skippedClusters, executionMode);
+    }
 
     private static  MultiSearchTemplateResponse createTestInstanceWithFailures() {
         int numItems = randomIntBetween(0, 128);
         long overallTookInMillis = randomNonNegativeLong();
         MultiSearchTemplateResponse.Item[] items = new MultiSearchTemplateResponse.Item[numItems];
         for (int i = 0; i < numItems; i++) {
             if (randomBoolean()) {
-                // Creating a minimal response is OK, because SearchResponse self
-                // is tested elsewhere.
+                // Creating a minimal response is OK, because SearchResponse is tested elsewhere.
                 long tookInMillis = randomNonNegativeLong();
                 int totalShards = randomIntBetween(1, Integer.MAX_VALUE);
                 int successfulShards = randomIntBetween(0, totalShards);
                 int skippedShards = totalShards - successfulShards;
                 InternalSearchResponse internalSearchResponse = InternalSearchResponse.empty();
-                SearchResponse.Clusters clusters = new SearchResponse.Clusters(totalShards, successfulShards, skippedShards);
+                SearchResponse.Clusters clusters = randomClusters();
                 SearchTemplateResponse searchTemplateResponse = new SearchTemplateResponse();
                 SearchResponse searchResponse = new SearchResponse(internalSearchResponse, null, totalShards,
                         successfulShards, skippedShards, tookInMillis, ShardSearchFailure.EMPTY_ARRAY, clusters);
@@ -133,6 +140,5 @@ public void testFromXContentWithFailures() throws IOException {
         AbstractXContentTestCase.testFromXContent(NUMBER_OF_TEST_RUNS, instanceSupplier, supportsUnknownFields, Strings.EMPTY_ARRAY,
                 getRandomFieldsExcludeFilterWhenResultHasErrors(), this::createParser, this::doParseInstance,
                 this::assertEqualInstances, assertToXContentEquivalence, ToXContent.EMPTY_PARAMS);
-    }    
-
+    }
 }

qa/ccs-unavailable-clusters/src/test/java/org/elasticsearch/search/CrossClusterSearchUnavailableClusterIT.java

@@ -22,6 +22,7 @@
 import org.apache.http.HttpEntity;
 import org.apache.http.entity.ContentType;
 import org.apache.http.nio.entity.NStringEntity;
+import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsAction;
@@ -32,9 +33,12 @@
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.CCSReduceMode;
+import org.elasticsearch.action.search.SearchAction;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchScrollRequest;
+import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
@@ -49,6 +53,8 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
+import org.elasticsearch.search.aggregations.InternalAggregations;
+import org.elasticsearch.search.internal.InternalSearchResponse;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.test.transport.MockTransportService;
 import org.elasticsearch.threadpool.TestThreadPool;
@@ -107,6 +113,14 @@ private static MockTransportService startTransport(
                         channel.sendResponse(new ClusterSearchShardsResponse(new ClusterSearchShardsGroup[0],
                                 knownNodes.toArray(new DiscoveryNode[0]), Collections.emptyMap()));
                     });
+            newService.registerRequestHandler(SearchAction.NAME, ThreadPool.Names.SAME, SearchRequest::new,
+                (request, channel, task) -> {
+                    InternalSearchResponse response = new InternalSearchResponse(new SearchHits(new SearchHit[0],
+                        new TotalHits(0, TotalHits.Relation.EQUAL_TO), Float.NaN), InternalAggregations.EMPTY, null, null, false, null, 1);
+                    SearchResponse searchResponse = new SearchResponse(response, null, 1, 1, 0, 100, ShardSearchFailure.EMPTY_ARRAY,
+                        SearchResponse.Clusters.EMPTY);
+                    channel.sendResponse(searchResponse);
+                });
             newService.registerRequestHandler(ClusterStateAction.NAME, ThreadPool.Names.SAME, ClusterStateRequest::new,
                 (request, channel, task) -> {
                         DiscoveryNodes.Builder builder = DiscoveryNodes.builder();
@@ -151,6 +165,7 @@ public void testSearchSkipUnavailable() throws IOException {
                 assertEquals(2, response.getClusters().getTotal());
                 assertEquals(2, response.getClusters().getSuccessful());
                 assertEquals(0, response.getClusters().getSkipped());
+                assertEquals(CCSReduceMode.REMOTE, response.getClusters().getCCSReduceMode());
                 assertEquals(10, response.getHits().getTotalHits().value);
                 assertEquals(10, response.getHits().getHits().length);
             }
@@ -159,6 +174,7 @@ public void testSearchSkipUnavailable() throws IOException {
                 assertEquals(1, response.getClusters().getTotal());
                 assertEquals(1, response.getClusters().getSuccessful());
                 assertEquals(0, response.getClusters().getSkipped());
+                assertEquals(CCSReduceMode.REMOTE, response.getClusters().getCCSReduceMode());
                 assertEquals(0, response.getHits().getTotalHits().value);
             }
 
@@ -168,6 +184,7 @@ public void testSearchSkipUnavailable() throws IOException {
                 assertEquals(2, response.getClusters().getTotal());
                 assertEquals(2, response.getClusters().getSuccessful());
                 assertEquals(0, response.getClusters().getSkipped());
+                assertEquals(CCSReduceMode.LOCAL, response.getClusters().getCCSReduceMode());
                 assertEquals(10, response.getHits().getTotalHits().value);
                 assertEquals(10, response.getHits().getHits().length);
                 String scrollId = response.getScrollId();
@@ -186,6 +203,7 @@ public void testSearchSkipUnavailable() throws IOException {
                 assertEquals(2, response.getClusters().getTotal());
                 assertEquals(1, response.getClusters().getSuccessful());
                 assertEquals(1, response.getClusters().getSkipped());
+                assertEquals(CCSReduceMode.REMOTE, response.getClusters().getCCSReduceMode());
                 assertEquals(10, response.getHits().getTotalHits().value);
                 assertEquals(10, response.getHits().getHits().length);
             }
@@ -194,6 +212,7 @@ public void testSearchSkipUnavailable() throws IOException {
                 assertEquals(1, response.getClusters().getTotal());
                 assertEquals(0, response.getClusters().getSuccessful());
                 assertEquals(1, response.getClusters().getSkipped());
+                assertEquals(CCSReduceMode.REMOTE, response.getClusters().getCCSReduceMode());
                 assertEquals(0, response.getHits().getTotalHits().value);
             }
 
@@ -203,6 +222,7 @@ public void testSearchSkipUnavailable() throws IOException {
                 assertEquals(2, response.getClusters().getTotal());
                 assertEquals(1, response.getClusters().getSuccessful());
                 assertEquals(1, response.getClusters().getSkipped());
+                assertEquals(CCSReduceMode.LOCAL, response.getClusters().getCCSReduceMode());
                 assertEquals(10, response.getHits().getTotalHits().value);
                 assertEquals(10, response.getHits().getHits().length);
                 String scrollId = response.getScrollId();

qa/multi-cluster-search/src/test/resources/rest-api-spec/test/multi_cluster/10_basic.yml

@@ -36,6 +36,11 @@
               terms:
                 field: f1.keyword
 
+  - match: {_clusters.total: 2}
+  - match: {_clusters.successful: 2}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
+  - match: {num}
   - match: { _shards.total: 5 }
   - match: { hits.total: 11 }
   - gte: { hits.hits.0._seq_no: 0 }
@@ -59,6 +64,10 @@
               terms:
                 field: f1.keyword
 
+  - match: {_clusters.total: 2}
+  - match: {_clusters.successful: 2}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
   - match: { _shards.total: 5 }
   - match: { hits.total: 6}
   - match: { hits.hits.0._index: "my_remote_cluster:test_index"}
@@ -76,6 +85,10 @@
               terms:
                 field: f1.keyword
 
+  - match: {_clusters.total: 1}
+  - match: {_clusters.successful: 1}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
   - match: { _shards.total: 3 }
   - match: { hits.total: 6}
   - match: { hits.hits.0._index: "my_remote_cluster:test_index"}
@@ -93,6 +106,7 @@
               terms:
                 field: f1.keyword
 
+  - is_false: _clusters
   - match: { _shards.total: 2 }
   - match: { hits.total: 5}
   - match: { hits.hits.0._index: "test_index"}
@@ -122,6 +136,10 @@
         rest_total_hits_as_int: true
         index: test_remote_cluster:test_index
 
+  - match: {_clusters.total: 1}
+  - match: {_clusters.successful: 1}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
   - match: { _shards.total: 3 }
   - match: { hits.total: 6 }
   - match: { hits.hits.0._index: "test_remote_cluster:test_index" }
@@ -148,6 +166,10 @@
         rest_total_hits_as_int: true
         index: "*:test_index"
 
+  - match: {_clusters.total: 2}
+  - match: {_clusters.successful: 2}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
   - match: { _shards.total: 6 }
   - match: { hits.total: 12 }
 
@@ -159,6 +181,10 @@
         rest_total_hits_as_int: true
         index: my_remote_cluster:aliased_test_index
 
+  - match: {_clusters.total: 1}
+  - match: {_clusters.successful: 1}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
   - match: { _shards.total: 3 }
   - match: { hits.total: 2 }
   - match: { hits.hits.0._source.filter_field: 1 }
@@ -172,6 +198,10 @@
         rest_total_hits_as_int: true
         index: my_remote_cluster:aliased_test_index,my_remote_cluster:field_caps_index_1
 
+  - match: {_clusters.total: 1}
+  - match: {_clusters.successful: 1}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
   - match: { _shards.total: 4 }
   - match: { hits.total: 2 }
   - match: { hits.hits.0._source.filter_field: 1 }
@@ -185,6 +215,10 @@
         rest_total_hits_as_int: true
         index: "my_remote_cluster:single_doc_index"
 
+  - match: {_clusters.total: 1}
+  - match: {_clusters.successful: 1}
+  - match: {_clusters.skipped: 0}
+  - match: {_clusters.ccs_reduce_mode: "remote"}
   - match: { _shards.total: 1 }
   - match: { hits.total: 1 }
   - match: { hits.hits.0._index: "my_remote_cluster:single_doc_index"}