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`>>,

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;
+    }
 }

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

@@ -0,0 +1,201 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.action.search;
+
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.action.OriginalIndices;
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.search.Scroll;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.search.internal.AliasFilter;
+import org.elasticsearch.search.internal.ShardSearchContextId;
+import org.elasticsearch.search.internal.ShardSearchRequest;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.transport.TransportRequest;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Node-level request used during can-match phase
+ */
+public class CanMatchRequest extends TransportRequest implements IndicesRequest {
+
+    private final OriginalIndices originalIndices;
+    private final SearchSourceBuilder source;
+    private final List<Shard> shards;
+    private final SearchType searchType;
+    private final Boolean requestCache;
+    private final boolean allowPartialSearchResults;
+    private final Scroll scroll;
+    private final int numberOfShards;
+    private final long nowInMillis;
+    @Nullable
+    private final String clusterAlias;
+
+    public static class Shard implements Writeable {
+        private final ShardId shardId;
+        private final int shardRequestIndex;
+        private final AliasFilter aliasFilter;
+        private final float indexBoost;
+        private final ShardSearchContextId readerId;
+        private final TimeValue keepAlive;
+
+        public Shard(ShardId shardId,
+                     int shardRequestIndex,
+                     AliasFilter aliasFilter,
+                     float indexBoost,
+                     ShardSearchContextId readerId,
+                     TimeValue keepAlive) {
+            this.shardId = shardId;
+            this.shardRequestIndex = shardRequestIndex;
+            this.aliasFilter = aliasFilter;
+            this.indexBoost = indexBoost;
+            this.readerId = readerId;
+            this.keepAlive = keepAlive;
+            assert keepAlive == null || readerId != null : "readerId: " + readerId + " keepAlive: " + keepAlive;
+        }
+
+        public Shard(StreamInput in) throws IOException {
+            shardId = new ShardId(in);
+            shardRequestIndex = in.readVInt();
+            aliasFilter = new AliasFilter(in);
+            indexBoost = in.readFloat();
+            readerId = in.readOptionalWriteable(ShardSearchContextId::new);
+            keepAlive = in.readOptionalTimeValue();
+            assert keepAlive == null || readerId != null : "readerId: " + readerId + " keepAlive: " + keepAlive;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            shardId.writeTo(out);
+            out.writeVInt(shardRequestIndex);
+            aliasFilter.writeTo(out);
+            out.writeFloat(indexBoost);
+            out.writeOptionalWriteable(readerId);
+            out.writeOptionalTimeValue(keepAlive);
+        }
+
+        public int getShardRequestIndex() {
+            return shardRequestIndex;
+        }
+
+        public ShardId shardId() {
+            return shardId;
+        }
+    }
+
+    public CanMatchRequest(
+        OriginalIndices originalIndices,
+        SearchRequest searchRequest,
+        List<Shard> shards,
+        int numberOfShards,
+        long nowInMillis,
+        @Nullable String clusterAlias
+        ) {
+        this.originalIndices = originalIndices;
+        this.source = searchRequest.source();
+        this.shards = new ArrayList<>(shards);
+        this.searchType = searchRequest.searchType();
+        this.requestCache = searchRequest.requestCache();
+        // If allowPartialSearchResults is unset (ie null), the cluster-level default should have been substituted
+        // at this stage. Any NPEs in the above are therefore an error in request preparation logic.
+        assert searchRequest.allowPartialSearchResults() != null;
+        this.allowPartialSearchResults = searchRequest.allowPartialSearchResults();
+        this.scroll = searchRequest.scroll();
+        this.numberOfShards = numberOfShards;
+        this.nowInMillis = nowInMillis;
+        this.clusterAlias = clusterAlias;
+    }
+
+    public CanMatchRequest(StreamInput in) throws IOException {
+        super(in);
+        source = in.readOptionalWriteable(SearchSourceBuilder::new);
+        originalIndices = OriginalIndices.readOriginalIndices(in);
+        searchType = SearchType.fromId(in.readByte());
+        scroll = in.readOptionalWriteable(Scroll::new);
+        requestCache = in.readOptionalBoolean();
+        allowPartialSearchResults = in.readBoolean();
+        numberOfShards = in.readVInt();
+        nowInMillis = in.readVLong();
+        clusterAlias = in.readOptionalString();
+        shards = in.readList(Shard::new);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeOptionalWriteable(source);
+        OriginalIndices.writeOriginalIndices(originalIndices, out);
+        out.writeByte(searchType.id());
+        out.writeOptionalWriteable(scroll);
+        out.writeOptionalBoolean(requestCache);
+        out.writeBoolean(allowPartialSearchResults);
+        out.writeVInt(numberOfShards);
+        out.writeVLong(nowInMillis);
+        out.writeOptionalString(clusterAlias);
+        out.writeList(shards);
+    }
+
+    @Override
+    public String[] indices() {
+        if (originalIndices == null) {
+            return null;
+        }
+        return originalIndices.indices();
+    }
+
+    @Override
+    public IndicesOptions indicesOptions() {
+        if (originalIndices == null) {
+            return null;
+        }
+        return originalIndices.indicesOptions();
+    }
+
+    public List<Shard> getShardLevelRequests() {
+        return shards;
+    }
+
+    public List<ShardSearchRequest> createShardSearchRequests() {
+        return shards.stream().map(this::createShardSearchRequest).collect(Collectors.toList());
+    }
+
+    public ShardSearchRequest createShardSearchRequest(Shard r) {
+        ShardSearchRequest shardSearchRequest = new ShardSearchRequest(
+            originalIndices, r.shardId, r.shardRequestIndex, numberOfShards, searchType,
+            source, requestCache, r.aliasFilter, r.indexBoost, allowPartialSearchResults, scroll,
+            nowInMillis, clusterAlias, r.readerId, r.keepAlive
+        );
+        shardSearchRequest.setParentTask(getParentTask());
+        return shardSearchRequest;
+    }
+
+    @Override
+    public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+        return new SearchShardTask(id, type, action, getDescription(), parentTaskId, headers);
+    }
+
+    @Override
+    public String getDescription() {
+        // Shard id is enough here, the request itself can be found by looking at the parent task description
+        return "shardIds[" + shards.stream().map(slr -> slr.shardId).collect(Collectors.toList()) + "]";
+    }
+
+}

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

@@ -0,0 +1,89 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.action.search;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.search.CanMatchShardResponse;
+import org.elasticsearch.transport.TransportResponse;
+
+import java.io.IOException;
+import java.util.List;
+
+public class CanMatchResponse extends TransportResponse {
+
+    private final List<ResponseOrFailure> responses;
+
+    public CanMatchResponse(StreamInput in) throws IOException {
+        super(in);
+        responses = in.readList(ResponseOrFailure::new);
+    }
+
+    public CanMatchResponse(List<ResponseOrFailure> responses) {
+        this.responses = responses;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeList(responses);
+    }
+
+    public List<ResponseOrFailure> getResponses() {
+        return responses;
+    }
+
+    public static class ResponseOrFailure implements Writeable {
+
+        public ResponseOrFailure(CanMatchShardResponse response) {
+            this.response = response;
+            this.exception = null;
+        }
+
+        public ResponseOrFailure(Exception exception) {
+            this.exception = exception;
+            this.response = null;
+        }
+
+        @Nullable
+        public CanMatchShardResponse getResponse() {
+            return response;
+        }
+
+        @Nullable
+        public Exception getException() {
+            return exception;
+        }
+
+        private final CanMatchShardResponse response;
+        private final Exception exception;
+
+        public ResponseOrFailure(StreamInput in) throws IOException {
+            if (in.readBoolean()) {
+                response = new CanMatchShardResponse(in);
+                exception = null;
+            } else {
+                exception = in.readException();
+                response = null;
+            }
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            final boolean hasResponse = response != null;
+            out.writeBoolean(hasResponse);
+            if (hasResponse) {
+                response.writeTo(out);
+            } else {
+                out.writeException(exception);
+            }
+        }
+    }
+}

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

@@ -10,6 +10,7 @@
 import org.elasticsearch.core.CheckedRunnable;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.Objects;
 
 /**
@@ -28,4 +29,12 @@ protected SearchPhase(String name) {
     public String getName() {
         return name;
     }
+
+    public void start() {
+        try {
+            run();
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
 }