No special handling rules for skip_unavailable=true clusters at plan time other than disconnected exceptions (elastic#120236)

For ES|QL, we are moving to limit the scope of the skip_unavailable setting for remote clusters.
Going forward, skip_unavailable will be considered for two scenarios:

1) inability to connect to a remote cluster ("unavailable")
2) whether to fail on execution time errors or not (inline with the upcoming 
allow_partial_search_results work for ES|QL).

This PR reverses the special plan-time handling for skip_unavailable=true clusters
that was added in elastic#116348. Remote clusters, regardless of their skip_unavailable setting, 
will now use the same logic as the local cluster for index expression analysis at plan time, namely:

1) If any concrete index specified is missing from the cluster, a VerificationException will be thrown
2) If no matching index/alias/datastream was found on any cluster (even if all were specified with a wildcard), 
a VerificationException will be thrown

Thus, we no longer require at least one matching index expression for skip_unavailable=false 
clusters either, as was done in the previous PR referenced above.

Author: quux00

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClustersQueryIT.java

@@ -91,35 +91,32 @@ public void testFailed() throws Exception {
         assertThat(telemetry.getSuccessCount(), equalTo(0L));
         assertThat(telemetry.getByRemoteCluster().size(), equalTo(0));
 
-        // One remote is skipped, one is not
+        // Errors from both remotes
         telemetry = getTelemetryFromFailedQuery("from logs-*,c*:no_such_index | stats sum (v)");
 
         assertThat(telemetry.getTotalCount(), equalTo(1L));
         assertThat(telemetry.getSuccessCount(), equalTo(0L));
-        assertThat(telemetry.getByRemoteCluster().size(), equalTo(1));
+        assertThat(telemetry.getByRemoteCluster().size(), equalTo(0));
         assertThat(telemetry.getRemotesPerSearchAvg(), equalTo(2.0));
         assertThat(telemetry.getRemotesPerSearchMax(), equalTo(2L));
-        assertThat(telemetry.getSearchCountWithSkippedRemotes(), equalTo(1L));
+        assertThat(telemetry.getSearchCountWithSkippedRemotes(), equalTo(0L));
         Map<String, Long> expectedFailure = Map.of(CCSUsageTelemetry.Result.NOT_FOUND.getName(), 1L);
         assertThat(telemetry.getFailureReasons(), equalTo(expectedFailure));
-        // cluster-b should be skipped
-        assertThat(telemetry.getByRemoteCluster().get(REMOTE2).getCount(), equalTo(0L));
-        assertThat(telemetry.getByRemoteCluster().get(REMOTE2).getSkippedCount(), equalTo(1L));
 
         // this is only for cluster-a so no skipped remotes
         telemetry = getTelemetryFromFailedQuery("from logs-*,cluster-a:no_such_index | stats sum (v)");
         assertThat(telemetry.getTotalCount(), equalTo(2L));
         assertThat(telemetry.getSuccessCount(), equalTo(0L));
-        assertThat(telemetry.getByRemoteCluster().size(), equalTo(1));
+        assertThat(telemetry.getByRemoteCluster().size(), equalTo(0));
         assertThat(telemetry.getRemotesPerSearchAvg(), equalTo(2.0));
         assertThat(telemetry.getRemotesPerSearchMax(), equalTo(2L));
-        assertThat(telemetry.getSearchCountWithSkippedRemotes(), equalTo(1L));
+        assertThat(telemetry.getSearchCountWithSkippedRemotes(), equalTo(0L));
         expectedFailure = Map.of(CCSUsageTelemetry.Result.NOT_FOUND.getName(), 2L);
         assertThat(telemetry.getFailureReasons(), equalTo(expectedFailure));
-        assertThat(telemetry.getByRemoteCluster().size(), equalTo(1));
+        assertThat(telemetry.getByRemoteCluster().size(), equalTo(0));
     }
 
-    // TODO: enable when skip-up patch is merged
+    // TODO: enable when skip-un patch is merged
     // public void testSkipAllRemotes() throws Exception {
     // var telemetry = getTelemetryFromQuery("from logs-*,c*:no_such_index | stats sum (v)", "unknown");
     //

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClustersUsageTelemetryIT.java

@@ -199,15 +199,13 @@ static void updateExecutionInfoWithClustersWithNoMatchingIndices(EsqlExecutionIn
 
         /**
          * Rules enforced at planning time around non-matching indices
-         * P1. fail query if no matching indices on any cluster (VerificationException) - that is handled elsewhere (TODO: document where)
-         * P2. fail query if a skip_unavailable:false cluster has no matching indices (the local cluster already has this rule
-         *     enforced at planning time)
-         * P3. fail query if the local cluster has no matching indices and a concrete index was specified
+         * 1. fail query if no matching indices on any cluster (VerificationException) - that is handled elsewhere
+         * 2. fail query if a cluster has no matching indices *and* a concrete index was specified - handled here
          */
         String fatalErrorMessage = null;
         /*
          * These are clusters in the original request that are not present in the field-caps response. They were
-         * specified with an index expression matched no indices, so the search on that cluster is done.
+         * specified with an index expression that matched no indices, so the search on that cluster is done.
          * Mark it as SKIPPED with 0 shards searched and took=0.
          */
         for (String c : clustersWithNoMatchingIndices) {
@@ -216,7 +214,7 @@ static void updateExecutionInfoWithClustersWithNoMatchingIndices(EsqlExecutionIn
                 continue;
             }
             final String indexExpression = executionInfo.getCluster(c).getIndexExpression();
-            if (missingIndicesIsFatal(c, executionInfo)) {
+            if (concreteIndexRequested(executionInfo.getCluster(c).getIndexExpression())) {
                 String error = Strings.format(
                     "Unknown index [%s]",
                     (c.equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY) ? indexExpression : c + ":" + indexExpression)
@@ -227,10 +225,11 @@ static void updateExecutionInfoWithClustersWithNoMatchingIndices(EsqlExecutionIn
                     fatalErrorMessage += "; " + error;
                 }
             } else {
-                // handles local cluster (when no concrete indices requested) and skip_unavailable=true clusters
+                // no matching indices and no concrete index requested - just skip it, no error
                 EsqlExecutionInfo.Cluster.Status status;
                 ShardSearchFailure failure;
                 if (c.equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
+                    // never mark local cluster as SKIPPED
                     status = EsqlExecutionInfo.Cluster.Status.SUCCESSFUL;
                     failure = null;
                 } else {
@@ -257,15 +256,7 @@ static void updateExecutionInfoWithClustersWithNoMatchingIndices(EsqlExecutionIn
     }
 
     // visible for testing
-    static boolean missingIndicesIsFatal(String clusterAlias, EsqlExecutionInfo executionInfo) {
-        // missing indices on local cluster is fatal only if a concrete index requested
-        if (clusterAlias.equals(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY)) {
-            return concreteIndexRequested(executionInfo.getCluster(clusterAlias).getIndexExpression());
-        }
-        return executionInfo.getCluster(clusterAlias).isSkipUnavailable() == false;
-    }
-
-    private static boolean concreteIndexRequested(String indexExpression) {
+    static boolean concreteIndexRequested(String indexExpression) {
         if (Strings.isNullOrBlank(indexExpression)) {
             return false;
         }

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

@@ -94,7 +94,6 @@ public void resolveAsMergedMapping(
     public IndexResolution mergedMappings(String indexPattern, FieldCapabilitiesResponse fieldCapsResponse) {
         assert ThreadPool.assertCurrentThreadPool(ThreadPool.Names.SEARCH_COORDINATION); // too expensive to run this on a transport worker
         if (fieldCapsResponse.getIndexResponses().isEmpty()) {
-            // TODO in follow-on PR, handle the case where remotes were specified with non-existent indices, according to skip_unavailable
             return IndexResolution.notFound(indexPattern);
         }