Ensure the set of remote clusters is consistent over the life of ES|QL query

Author: smalyshev

server/src/main/java/org/elasticsearch/indices/IndicesExpressionGrouper.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.common.Strings;
 
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Interface for grouping index expressions, along with IndicesOptions by cluster alias.
@@ -29,23 +30,36 @@
 public interface IndicesExpressionGrouper {
 
     /**
+     * @param remoteClusterNames Set of configured remote cluster names.
      * @param indicesOptions IndicesOptions to clarify how the index expression should be parsed/applied
      * @param indexExpressionCsv Multiple index expressions as CSV string (with no spaces), e.g., "logs1,logs2,cluster-a:logs1".
      *                           A single index expression is also supported.
      * @return Map where the key is the cluster alias (for "local" cluster, it is RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)
      *         and the value for that cluster from the index expression is an OriginalIndices object.
      */
-    default Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String indexExpressionCsv) {
-        return groupIndices(indicesOptions, Strings.splitStringByCommaToArray(indexExpressionCsv));
+    default Map<String, OriginalIndices> groupIndices(
+        Set<String> remoteClusterNames,
+        IndicesOptions indicesOptions,
+        String indexExpressionCsv
+    ) {
+        return groupIndices(remoteClusterNames, indicesOptions, Strings.splitStringByCommaToArray(indexExpressionCsv));
     }
 
     /**
      * Same behavior as the other groupIndices, except the incoming multiple index expressions must already be
      * parsed into a String array.
+     * @param remoteClusterNames Set of configured remote cluster names.
      * @param indicesOptions IndicesOptions to clarify how the index expressions should be parsed/applied
      * @param indexExpressions Multiple index expressions as string[].
      * @return Map where the key is the cluster alias (for "local" cluster, it is RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)
      *         and the value for that cluster from the index expression is an OriginalIndices object.
      */
-    Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indexExpressions);
+    Map<String, OriginalIndices> groupIndices(Set<String> remoteClusterNames, IndicesOptions indicesOptions, String[] indexExpressions);
+
+    /**
+     * Returns a set of currently configured remote clusters.
+     */
+    default Set<String> getConfiguredClusters() {
+        return Set.of();
+    }
 }

server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java

@@ -185,6 +185,7 @@ boolean isRemoteNodeConnected(final String remoteCluster, final DiscoveryNode no
 
     /**
      * Group indices by cluster alias mapped to OriginalIndices for that cluster.
+     * @param remoteClusterNames Set of configured remote cluster names.
      * @param indicesOptions IndicesOptions to clarify how the index expressions should be parsed/applied
      * @param indices Multiple index expressions as string[].
      * @param returnLocalAll whether to support the _all functionality needed by _search
@@ -193,9 +194,14 @@ boolean isRemoteNodeConnected(final String remoteCluster, final DiscoveryNode no
      *        If false, an empty map is returned when no indices are specified.
      * @return Map keyed by cluster alias having OriginalIndices as the map value parsed from the String[] indices argument
      */
-    public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indices, boolean returnLocalAll) {
+    public Map<String, OriginalIndices> groupIndices(
+        Set<String> remoteClusterNames,
+        IndicesOptions indicesOptions,
+        String[] indices,
+        boolean returnLocalAll
+    ) {
         final Map<String, OriginalIndices> originalIndicesMap = new HashMap<>();
-        final Map<String, List<String>> groupedIndices = groupClusterIndices(getRemoteClusterNames(), indices);
+        final Map<String, List<String>> groupedIndices = groupClusterIndices(remoteClusterNames, indices);
         if (groupedIndices.isEmpty()) {
             if (returnLocalAll) {
                 // search on _all in the local cluster if neither local indices nor remote indices were specified
@@ -214,12 +220,26 @@ public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions,
     /**
      * If no indices are specified, then a Map with one entry for the local cluster with an empty index array is returned.
      * For details see {@code groupIndices(IndicesOptions indicesOptions, String[] indices, boolean returnLocalAll)}
+     * @param remoteClusterNames Set of configured remote cluster names.
      * @param indicesOptions IndicesOptions to clarify how the index expressions should be parsed/applied
      * @param indices Multiple index expressions as string[].
      * @return Map keyed by cluster alias having OriginalIndices as the map value parsed from the String[] indices argument
      */
+    public Map<String, OriginalIndices> groupIndices(Set<String> remoteClusterNames, IndicesOptions indicesOptions, String[] indices) {
+        return groupIndices(remoteClusterNames, indicesOptions, indices, true);
+    }
+
+    public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indices, boolean returnLocalAll) {
+        return groupIndices(getRemoteClusterNames(), indicesOptions, indices, returnLocalAll);
+    }
+
     public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indices) {
-        return groupIndices(indicesOptions, indices, true);
+        return groupIndices(getRemoteClusterNames(), indicesOptions, indices, true);
+    }
+
+    @Override
+    public Set<String> getConfiguredClusters() {
+        return getRemoteClusterNames();
     }
 
     /**

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlExecutionInfo.java

@@ -77,6 +77,7 @@ public class EsqlExecutionInfo implements ChunkedToXContentObject, Writeable {
     private transient TimeValue planningTookTime;  // time elapsed since start of query to calling ComputeService.execute
     private volatile boolean isPartial; // Does this request have partial results?
     private transient volatile boolean isStopped; // Have we received stop command?
+    private transient Set<String> configuredClusters; // Set of configured remote clusters
 
     public EsqlExecutionInfo(boolean includeCCSMetadata) {
         this(Predicates.always(), includeCCSMetadata);  // default all clusters to skip_unavailable=true
@@ -354,6 +355,14 @@ public boolean isStopped() {
         return isStopped;
     }
 
+    public Set<String> getConfiguredClusters() {
+        return configuredClusters;
+    }
+
+    public void setConfiguredClusters(Set<String> configuredClusters) {
+        this.configuredClusters = configuredClusters;
+    }
+
     /**
      * Represents the search metadata about a particular cluster involved in a cross-cluster search.
      * The Cluster object can represent either the local cluster or a remote cluster.

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java

@@ -446,8 +446,8 @@ protected void getRemoteConnection(String cluster, ActionListener<Transport.Conn
         );
     }
 
-    public Map<String, List<String>> groupIndicesPerCluster(String[] indices) {
-        return remoteClusterService.groupIndices(SearchRequest.DEFAULT_INDICES_OPTIONS, indices)
+    public Map<String, List<String>> groupIndicesPerCluster(Set<String> remoteClusterNames, String[] indices) {
+        return remoteClusterService.groupIndices(remoteClusterNames, SearchRequest.DEFAULT_INDICES_OPTIONS, indices)
             .entrySet()
             .stream()
             .collect(Collectors.toMap(Map.Entry::getKey, e -> Arrays.asList(e.getValue().indices())));

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlCCSUtils.java

@@ -283,7 +283,11 @@ public static void checkForCcsLicense(
         for (TableInfo tableInfo : indices) {
             Map<String, OriginalIndices> groupedIndices;
             try {
-                groupedIndices = indicesGrouper.groupIndices(IndicesOptions.DEFAULT, tableInfo.id().indexPattern());
+                groupedIndices = indicesGrouper.groupIndices(
+                    executionInfo.getConfiguredClusters(),
+                    IndicesOptions.DEFAULT,
+                    tableInfo.id().indexPattern()
+                );
             } catch (NoSuchRemoteClusterException e) {
                 if (EsqlLicenseChecker.isCcsAllowed(licenseState)) {
                     throw e;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java

@@ -343,6 +343,8 @@ public void analyzedPlan(
             plan.setAnalyzed();
             return plan;
         };
+        // Capture configured remotes list to ensure consistency throughout the session
+        executionInfo.setConfiguredClusters(indicesExpressionGrouper.getConfiguredClusters());
 
         PreAnalyzer.PreAnalysis preAnalysis = preAnalyzer.preAnalyze(parsed);
         var unresolvedPolicies = preAnalysis.enriches.stream()
@@ -358,6 +360,7 @@ public void analyzedPlan(
         EsqlCCSUtils.checkForCcsLicense(executionInfo, indices, indicesExpressionGrouper, verifier.licenseState());
 
         final Set<String> targetClusters = enrichPolicyResolver.groupIndicesPerCluster(
+            executionInfo.getConfiguredClusters(),
             indices.stream()
                 .flatMap(t -> Arrays.stream(Strings.commaDelimitedListToStringArray(t.id().indexPattern())))
                 .toArray(String[]::new)
@@ -378,7 +381,7 @@ public void analyzedPlan(
             // invalid index resolution to updateExecutionInfo
             if (result.indices.isValid()) {
                 // CCS indices and skip_unavailable cluster values can stop the analysis right here
-                if (analyzeCCSIndices(executionInfo, targetClusters, unresolvedPolicies, result, logicalPlanListener, l)) return;
+                if (allCCSClustersSkipped(executionInfo, result, logicalPlanListener)) return;
             }
             // whatever tuple we have here (from CCS-special handling or from the original pre-analysis), pass it on to the next step
             l.onResponse(result);
@@ -442,6 +445,7 @@ private void preAnalyzeIndices(
             IndexPattern table = tableInfo.id();
 
             Map<String, OriginalIndices> clusterIndices = indicesExpressionGrouper.groupIndices(
+                executionInfo.getConfiguredClusters(),
                 IndicesOptions.DEFAULT,
                 table.indexPattern()
             );
@@ -501,13 +505,14 @@ private void preAnalyzeIndices(
         }
     }
 
-    private boolean analyzeCCSIndices(
+    /**
+     * Check if there are any clusters to search.
+     * @return true if there are no clusters to search, false otherwise
+     */
+    private boolean allCCSClustersSkipped(
         EsqlExecutionInfo executionInfo,
-        Set<String> targetClusters,
-        Set<EnrichPolicyResolver.UnresolvedPolicy> unresolvedPolicies,
         PreAnalysisResult result,
-        ActionListener<LogicalPlan> logicalPlanListener,
-        ActionListener<PreAnalysisResult> l
+        ActionListener<LogicalPlan> logicalPlanListener
     ) {
         IndexResolution indexResolution = result.indices;
         EsqlCCSUtils.updateExecutionInfoWithClustersWithNoMatchingIndices(executionInfo, indexResolution);
@@ -520,22 +525,6 @@ private boolean analyzeCCSIndices(
             return true;
         }
 
-        Set<String> newClusters = enrichPolicyResolver.groupIndicesPerCluster(
-            indexResolution.get().concreteIndices().toArray(String[]::new)
-        ).keySet();
-        // If new clusters appear when resolving the main indices, we need to resolve the enrich policies again
-        // or exclude main concrete indices. Since this is rare, it's simpler to resolve the enrich policies again.
-        // TODO: add a test for this
-        if (targetClusters.containsAll(newClusters) == false
-            // do not bother with a re-resolution if only remotes were requested and all were offline
-            && executionInfo.getClusterStates(EsqlExecutionInfo.Cluster.Status.RUNNING).findAny().isPresent()) {
-            enrichPolicyResolver.resolvePolicies(
-                newClusters,
-                unresolvedPolicies,
-                l.map(enrichResolution -> result.withEnrichResolution(enrichResolution))
-            );
-            return true;
-        }
         return false;
     }
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/session/EsqlCCSUtilsTests.java

@@ -825,7 +825,11 @@ private void assertLicenseCheckFails(
 
     static class TestIndicesExpressionGrouper implements IndicesExpressionGrouper {
         @Override
-        public Map<String, OriginalIndices> groupIndices(IndicesOptions indicesOptions, String[] indexExpressions) {
+        public Map<String, OriginalIndices> groupIndices(
+            Set<String> remoteClusterNames,
+            IndicesOptions indicesOptions,
+            String[] indexExpressions
+        ) {
             final Map<String, OriginalIndices> originalIndicesMap = new HashMap<>();
             final String localKey = RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY;
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/telemetry/PlanExecutorMetricsTests.java

@@ -157,7 +157,7 @@ public void testFailedMetric() {
         // test a failed query: xyz field doesn't exist
         request.query("from test | stats m = max(xyz)");
         EsqlSession.PlanRunner runPhase = (p, r) -> fail("this shouldn't happen");
-        IndicesExpressionGrouper groupIndicesByCluster = (indicesOptions, indexExpressions) -> Map.of(
+        IndicesExpressionGrouper groupIndicesByCluster = (remoteClusterNames, indicesOptions, indexExpressions) -> Map.of(
             "",
             new OriginalIndices(new String[] { "test" }, IndicesOptions.DEFAULT)
         );