Integrate CCS with new search_shards API (#95894)

This PR integrates CCS with the new search_shards API. With this change, we will be able to skip shards on the coordinator on remote clusters using the timestamps stored in the cluster state. Relates #94534 Closes #93730

Author: dnhatn

docs/changelog/95894.yaml

@@ -0,0 +1,6 @@
+pr: 95894
+summary: Integrate CCS with new `search_shards` API
+area: Search
+type: enhancement
+issues:
+ - 93730

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

@@ -15,10 +15,6 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.Version;
-import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsAction;
-import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsGroup;
-import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsRequest;
-import org.elasticsearch.action.admin.cluster.shards.ClusterSearchShardsResponse;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateAction;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateRequest;
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
@@ -27,6 +23,9 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.SearchScrollRequest;
+import org.elasticsearch.action.search.SearchShardsAction;
+import org.elasticsearch.action.search.SearchShardsRequest;
+import org.elasticsearch.action.search.SearchShardsResponse;
 import org.elasticsearch.action.search.ShardSearchFailure;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
@@ -103,17 +102,11 @@ private static MockTransportService startTransport(
  MockTransportService newService = MockTransportService.createNewService(s, version, transportVersion, threadPool, null);
  try {
  newService.registerRequestHandler(
- ClusterSearchShardsAction.NAME,
+ SearchShardsAction.NAME,
  ThreadPool.Names.SAME,
- ClusterSearchShardsRequest::new,
+ SearchShardsRequest::new,
  (request, channel, task) -> {
- channel.sendResponse(
- new ClusterSearchShardsResponse(
- new ClusterSearchShardsGroup[0],
- knownNodes.toArray(new DiscoveryNode[0]),
- Collections.emptyMap()
- )
- );
+ channel.sendResponse(new SearchShardsResponse(List.of(), List.of(), Collections.emptyMap()));
  }
  );
  newService.registerRequestHandler(SearchAction.NAME, ThreadPool.Names.SAME, SearchRequest::new, (request, channel, task) -> {

server/src/internalClusterTest/java/org/elasticsearch/search/ccs/CCSCanMatchIT.java

@@ -0,0 +1,193 @@
+/*
+ * 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.search.ccs;
+
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.PointValues;
+import org.elasticsearch.action.search.CanMatchNodeRequest;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchTransportService;
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.engine.EngineConfig;
+import org.elasticsearch.index.engine.EngineFactory;
+import org.elasticsearch.index.engine.InternalEngine;
+import org.elasticsearch.index.engine.InternalEngineFactory;
+import org.elasticsearch.index.query.RangeQueryBuilder;
+import org.elasticsearch.index.shard.IndexLongFieldRange;
+import org.elasticsearch.index.shard.ShardLongFieldRange;
+import org.elasticsearch.plugins.EnginePlugin;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.elasticsearch.test.AbstractMultiClustersTestCase;
+import org.elasticsearch.test.hamcrest.ElasticsearchAssertions;
+import org.elasticsearch.test.transport.MockTransportService;
+import org.elasticsearch.transport.TransportService;
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.in;
+
+public class CCSCanMatchIT extends AbstractMultiClustersTestCase {
+ static final String REMOTE_CLUSTER = "cluster_a";
+
+ @Override
+ protected Collection<String> remoteClusterAlias() {
+ return List.of("cluster_a");
+ }
+
+ private static class EngineWithExposingTimestamp extends InternalEngine {
+ EngineWithExposingTimestamp(EngineConfig engineConfig) {
+ super(engineConfig);
+ assert IndexMetadata.INDEX_BLOCKS_WRITE_SETTING.get(config().getIndexSettings().getSettings()) : "require read-only index";
+ }
+
+ @Override
+ public ShardLongFieldRange getRawFieldRange(String field) {
+ try (Searcher searcher = acquireSearcher("test")) {
+ final DirectoryReader directoryReader = searcher.getDirectoryReader();
+
+ final byte[] minPackedValue = PointValues.getMinPackedValue(directoryReader, field);
+ final byte[] maxPackedValue = PointValues.getMaxPackedValue(directoryReader, field);
+ if (minPackedValue == null || maxPackedValue == null) {
+ assert minPackedValue == null && maxPackedValue == null
+ : Arrays.toString(minPackedValue) + "-" + Arrays.toString(maxPackedValue);
+ return ShardLongFieldRange.EMPTY;
+ }
+
+ return ShardLongFieldRange.of(LongPoint.decodeDimension(minPackedValue, 0), LongPoint.decodeDimension(maxPackedValue, 0));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ }
+
+ public static class ExposingTimestampEnginePlugin extends Plugin implements EnginePlugin {
+
+ @Override
+ public Optional<EngineFactory> getEngineFactory(IndexSettings indexSettings) {
+ if (IndexMetadata.INDEX_BLOCKS_WRITE_SETTING.get(indexSettings.getSettings())) {
+ return Optional.of(EngineWithExposingTimestamp::new);
+ } else {
+ return Optional.of(new InternalEngineFactory());
+ }
+ }
+ }
+
+ @Override
+ protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
+ return CollectionUtils.appendToCopy(super.nodePlugins(clusterAlias), ExposingTimestampEnginePlugin.class);
+ }
+
+ int createIndexAndIndexDocs(String cluster, String index, int numberOfShards, long timestamp, boolean exposeTimestamp)
+ throws Exception {
+ Client client = client(cluster);
+ ElasticsearchAssertions.assertAcked(
+ client.admin()
+ .indices()
+ .prepareCreate(index)
+ .setSettings(
+ Settings.builder()
+ .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, numberOfShards)
+ .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+ )
+ .setMapping("@timestamp", "type=date", "position", "type=long")
+ );
+ int numDocs = between(100, 500);
+ for (int i = 0; i < numDocs; i++) {
+ client.prepareIndex(index).setSource("position", i, "@timestamp", timestamp + i).get();
+ }
+ if (exposeTimestamp) {
+ client.admin().indices().prepareClose(index).get();
+ client.admin()
+ .indices()
+ .prepareUpdateSettings(index)
+ .setSettings(Settings.builder().put(IndexMetadata.INDEX_BLOCKS_WRITE_SETTING.getKey(), true).build())
+ .get();
+ client.admin().indices().prepareOpen(index).get();
+ assertBusy(() -> {
+ IndexLongFieldRange timestampRange = cluster(cluster).clusterService().state().metadata().index(index).getTimestampRange();
+ assertTrue(Strings.toString(timestampRange), timestampRange.containsAllShardRanges());
+ });
+ } else {
+ client.admin().indices().prepareRefresh(index).get();
+ }
+ return numDocs;
+ }
+
+ public void testCanMatchOnTimeRange() throws Exception {
+ long timestamp = randomLongBetween(10_000_000, 50_000_000);
+ int oldLocalNumShards = randomIntBetween(1, 5);
+ createIndexAndIndexDocs(LOCAL_CLUSTER, "local_old_index", oldLocalNumShards, timestamp - 10_000, true);
+ int oldRemoteNumShards = randomIntBetween(1, 5);
+ createIndexAndIndexDocs(REMOTE_CLUSTER, "remote_old_index", oldRemoteNumShards, timestamp - 10_000, true);
+
+ int newLocalNumShards = randomIntBetween(1, 5);
+ int localDocs = createIndexAndIndexDocs(LOCAL_CLUSTER, "local_new_index", newLocalNumShards, timestamp, randomBoolean());
+ int newRemoteNumShards = randomIntBetween(1, 5);
+ int remoteDocs = createIndexAndIndexDocs(REMOTE_CLUSTER, "remote_new_index", newRemoteNumShards, timestamp, randomBoolean());
+
+ for (String cluster : List.of(LOCAL_CLUSTER, REMOTE_CLUSTER)) {
+ for (TransportService ts : cluster(cluster).getInstances(TransportService.class)) {
+ MockTransportService mockTransportService = (MockTransportService) ts;
+ mockTransportService.addSendBehavior((connection, requestId, action, request, options) -> {
+ if (action.equals(SearchTransportService.QUERY_CAN_MATCH_NODE_NAME)) {
+ CanMatchNodeRequest canMatchNodeRequest = (CanMatchNodeRequest) request;
+ List<String> indices = canMatchNodeRequest.getShardLevelRequests()
+ .stream()
+ .map(r -> r.shardId().getIndexName())
+ .toList();
+ assertThat("old indices should be prefiltered on coordinator node", "local_old_index", Matchers.not(in(indices)));
+ assertThat("old indices should be prefiltered on coordinator node", "remote_old_index", Matchers.not(in(indices)));
+ if (cluster.equals(LOCAL_CLUSTER)) {
+ DiscoveryNode targetNode = connection.getNode();
+ DiscoveryNodes remoteNodes = cluster(REMOTE_CLUSTER).clusterService().state().nodes();
+ assertNull("No can_match requests sent across clusters", remoteNodes.get(targetNode.getId()));
+ }
+ }
+ connection.sendRequest(requestId, action, request, options);
+ });
+ }
+ }
+ try {
+ for (boolean minimizeRoundTrips : List.of(true, false)) {
+ SearchSourceBuilder source = new SearchSourceBuilder().query(new RangeQueryBuilder("@timestamp").from(timestamp));
+ SearchRequest request = new SearchRequest("local_*", "*:remote_*");
+ request.source(source).setCcsMinimizeRoundtrips(minimizeRoundTrips);
+ SearchResponse searchResp = client().search(request).actionGet();
+ ElasticsearchAssertions.assertHitCount(searchResp, localDocs + remoteDocs);
+ int totalShards = oldLocalNumShards + newLocalNumShards + oldRemoteNumShards + newRemoteNumShards;
+ assertThat(searchResp.getTotalShards(), equalTo(totalShards));
+ assertThat(searchResp.getSkippedShards(), equalTo(oldLocalNumShards + oldRemoteNumShards));
+ }
+ } finally {
+ for (String cluster : List.of(LOCAL_CLUSTER, REMOTE_CLUSTER)) {
+ for (TransportService ts : cluster(cluster).getInstances(TransportService.class)) {
+ MockTransportService mockTransportService = (MockTransportService) ts;
+ mockTransportService.clearAllRules();
+ }
+ }
+ }
+ }
+}

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

@@ -158,6 +158,12 @@ private void runCoordinatorRewritePhase() {
  searchShardIterator.getClusterAlias()
  );
  final ShardSearchRequest request = canMatchNodeRequest.createShardSearchRequest(buildShardLevelRequest(searchShardIterator));
+ if (searchShardIterator.prefiltered()) {
+ CanMatchShardResponse result = new CanMatchShardResponse(searchShardIterator.skip() == false, null);
+ result.setShardIndex(request.shardRequestIndex());
+ results.consumeResult(result, () -> {});
+ continue;
+ }
  boolean canMatch = true;
  CoordinatorRewriteContext coordinatorRewriteContext = coordinatorRewriteContextProvider.getCoordinatorRewriteContext(
  request.shardId().getIndex()
@@ -510,8 +516,10 @@ private GroupShardsIterator<SearchShardIterator> getIterator(
  // shards available in order to produce a valid search result.
  int shardIndexToQuery = 0;
  for (int i = 0; i < shardsIts.size(); i++) {
- if (shardsIts.get(i).size() > 0) {
+ SearchShardIterator it = shardsIts.get(i);
+ if (it.size() > 0) {
  shardIndexToQuery = i;
+ it.skip(false); // un-skip which is needed when all the remote shards were skipped by the remote can_match
  break;
  }
  }
@@ -520,10 +528,12 @@ private GroupShardsIterator<SearchShardIterator> getIterator(
  SearchSourceBuilder source = request.source();
  int i = 0;
  for (SearchShardIterator iter : shardsIts) {
- if (possibleMatches.get(i++)) {
- iter.reset();
+ iter.reset();
+ boolean match = possibleMatches.get(i++);
+ if (match) {
+ assert iter.skip() == false;
  } else {
- iter.resetAndSkip();
+ iter.skip(true);
  }
  }
  if (shouldSortShards(results.minAndMaxes) == false) {

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

@@ -34,7 +34,8 @@ public final class SearchShardIterator implements Comparable<SearchShardIterator
  private final OriginalIndices originalIndices;
  private final String clusterAlias;
  private final ShardId shardId;
- private boolean skip = false;
+ private boolean skip;
+ private final boolean prefiltered;
 
  private final ShardSearchContextId searchContextId;
  private final TimeValue searchContextKeepAlive;
@@ -50,16 +51,30 @@ public final class SearchShardIterator implements Comparable<SearchShardIterator
  * @param originalIndices the indices that the search request originally related to (before any rewriting happened)
  */
  public SearchShardIterator(@Nullable String clusterAlias, ShardId shardId, List<ShardRouting> shards, OriginalIndices originalIndices) {
- this(clusterAlias, shardId, shards.stream().map(ShardRouting::currentNodeId).toList(), originalIndices, null, null);
+ this(clusterAlias, shardId, shards.stream().map(ShardRouting::currentNodeId).toList(), originalIndices, null, null, false, false);
  }
 
+ /**
+ * Creates a {@link PlainShardIterator} instance that iterates over a subset of the given shards
+ *
+ * @param clusterAlias the alias of the cluster where the shard is located
+ * @param shardId shard id of the group
+ * @param targetNodeIds the list of nodes hosting shard copies
+ * @param originalIndices the indices that the search request originally related to (before any rewriting happened)
+ * @param searchContextId the point-in-time specified for this group if exists
+ * @param searchContextKeepAlive the time interval that data nodes should extend the keep alive of the point-in-time
+ * @param prefiltered if true, then this group already executed the can_match phase
+ * @param skip if true, then this group won't have matches, and it can be safely skipped from the search
+ */
  public SearchShardIterator(
  @Nullable String clusterAlias,
  ShardId shardId,
  List<String> targetNodeIds,
  OriginalIndices originalIndices,
  ShardSearchContextId searchContextId,
- TimeValue searchContextKeepAlive
+ TimeValue searchContextKeepAlive,
+ boolean prefiltered,
+ boolean skip
  ) {
  this.shardId = shardId;
  this.targetNodesIterator = new PlainIterator<>(targetNodeIds);
@@ -68,6 +83,9 @@ public SearchShardIterator(
  this.searchContextId = searchContextId;
  this.searchContextKeepAlive = searchContextKeepAlive;
  assert searchContextKeepAlive == null || searchContextId != null;
+ this.prefiltered = prefiltered;
+ this.skip = skip;
+ assert skip == false || prefiltered : "only prefiltered shards are skip-able";
  }
 
  /**
@@ -112,15 +130,6 @@ List<String> getTargetNodeIds() {
  return targetNodesIterator.asList();
  }
 
- /**
- * Reset the iterator and mark it as skippable
- * @see #skip()
- */
- void resetAndSkip() {
- reset();
- skip = true;
- }
-
  void reset() {
  targetNodesIterator.reset();
  }
@@ -132,6 +141,20 @@ boolean skip() {
  return skip;
  }
 
+ /**
+ * Specifies if the search execution should skip this shard copies
+ */
+ void skip(boolean skip) {
+ this.skip = skip;
+ }
+
+ /**
+ * Returns {@code true} if this iterator was applied pre-filtered
+ */
+ boolean prefiltered() {
+ return prefiltered;
+ }
+
  @Override
  public int size() {
  return targetNodesIterator.size();