Node level can match action

docs/reference/modules/threadpool.asciidoc

@@ -21,6 +21,11 @@ There are several thread pools, but the important ones include:
     For count/search/suggest/get operations on `search_throttled indices`.
     Thread pool type is `fixed` with a size of `1`, and queue_size of `100`.
 
+`search_coordination`::
+    For lightweight search-related coordination operations. Thread pool type is
+    `fixed` with a size of a max of `min(5, (`<<node.processors,
+`# of allocated processors`>>`) / 2)`, and queue_size of `1000`.
+
 `get`::
     For get operations. Thread pool type is `fixed`
     with a size of <<node.processors, `# of allocated processors`>>,

qa/ccs-rolling-upgrade-remote-cluster/src/test/java/org/elasticsearch/upgrades/SearchStatesIT.java

@@ -198,7 +198,7 @@ static int indexDocs(RestHighLevelClient client, String index, int numDocs) thro
         return numDocs;
     }
 
-    void verifySearch(String localIndex, int localNumDocs, String remoteIndex, int remoteNumDocs) {
+    void verifySearch(String localIndex, int localNumDocs, String remoteIndex, int remoteNumDocs, Integer preFilterShardSize) {
         try (RestHighLevelClient localClient = newLocalClient()) {
             Request request = new Request("POST", "/_search");
             final int expectedDocs;
@@ -212,6 +212,12 @@ void verifySearch(String localIndex, int localNumDocs, String remoteIndex, int r
             if (UPGRADE_FROM_VERSION.onOrAfter(Version.V_7_0_0)) {
                 request.addParameter("ccs_minimize_roundtrips", Boolean.toString(randomBoolean()));
             }
+            if (preFilterShardSize == null && randomBoolean()) {
+                preFilterShardSize = randomIntBetween(1, 100);
+            }
+            if (preFilterShardSize != null) {
+                request.addParameter("pre_filter_shard_size", Integer.toString(preFilterShardSize));
+            }
             int size = between(1, 100);
             request.setJsonEntity("{\"sort\": \"f\", \"size\": " + size + "}");
             Response response = localClient.getLowLevelClient().performRequest(request);
@@ -245,7 +251,32 @@ public void testBWCSearchStates() throws Exception {
             configureRemoteClusters(getNodes(remoteClient.getLowLevelClient()));
             int iterations = between(1, 20);
             for (int i = 0; i < iterations; i++) {
-                verifySearch(localIndex, localNumDocs, CLUSTER_ALIAS + ":" + remoteIndex, remoteNumDocs);
+                verifySearch(localIndex, localNumDocs, CLUSTER_ALIAS + ":" + remoteIndex, remoteNumDocs, null);
+            }
+            localClient.indices().delete(new DeleteIndexRequest(localIndex), RequestOptions.DEFAULT);
+            remoteClient.indices().delete(new DeleteIndexRequest(remoteIndex), RequestOptions.DEFAULT);
+        }
+    }
+
+    public void testCanMatch() throws Exception {
+        String localIndex = "test_can_match_local_index";
+        String remoteIndex = "test_can_match_remote_index";
+        try (RestHighLevelClient localClient = newLocalClient();
+             RestHighLevelClient remoteClient = newRemoteClient()) {
+            localClient.indices().create(new CreateIndexRequest(localIndex)
+                    .settings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, between(5, 20))),
+                RequestOptions.DEFAULT);
+            int localNumDocs = indexDocs(localClient, localIndex, between(10, 100));
+
+            remoteClient.indices().create(new CreateIndexRequest(remoteIndex)
+                    .settings(Settings.builder().put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, between(5, 20))),
+                RequestOptions.DEFAULT);
+            int remoteNumDocs = indexDocs(remoteClient, remoteIndex, between(10, 100));
+
+            configureRemoteClusters(getNodes(remoteClient.getLowLevelClient()));
+            int iterations = between(1, 10);
+            for (int i = 0; i < iterations; i++) {
+                verifySearch(localIndex, localNumDocs, CLUSTER_ALIAS + ":" + remoteIndex, remoteNumDocs, between(1, 10));
             }
             localClient.indices().delete(new DeleteIndexRequest(localIndex), RequestOptions.DEFAULT);
             remoteClient.indices().delete(new DeleteIndexRequest(remoteIndex), RequestOptions.DEFAULT);

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

@@ -724,7 +724,11 @@ private static void assumeMultiClusterSetup() {
     private static SearchRequest initSearchRequest() {
         List<String> indices = Arrays.asList(INDEX_NAME, "my_remote_cluster:" + INDEX_NAME);
         Collections.shuffle(indices, random());
-        return new SearchRequest(indices.toArray(new String[0]));
+        final SearchRequest request = new SearchRequest(indices.toArray(new String[0]));
+        if (randomBoolean()) {
+            request.setPreFilterShardSize(between(1, 20));
+        }
+        return request;
     }
 
     private static void duelSearch(SearchRequest searchRequest, Consumer<SearchResponse> responseChecker) throws Exception {

server/src/main/java/org/elasticsearch/action/OriginalIndices.java

@@ -14,6 +14,7 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Objects;
 
 /**
  * Used to keep track of original indices within internal (e.g. shard level) requests
@@ -67,4 +68,19 @@ public String toString() {
             ", indicesOptions=" + indicesOptions +
             '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        OriginalIndices that = (OriginalIndices) o;
+        return Arrays.equals(indices, that.indices) && Objects.equals(indicesOptions, that.indicesOptions);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = Objects.hash(indicesOptions);
+        result = 31 * result + Arrays.hashCode(indices);
+        return result;
+    }
 }