Fix counting skipped shards with filters (#131737) (#131769)

* Fix counting skipped shards with filters

(cherry picked from commit a345f56)

# Conflicts:
#	x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClustersIT.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

Author: smalyshev

muted-tests.yml

@@ -330,9 +330,6 @@ tests:
 - class: org.elasticsearch.xpack.search.CrossClusterAsyncSearchIT
   method: testCCSClusterDetailsWhereAllShardsSkippedInCanMatch
   issue: https://github.com/elastic/elasticsearch/issues/128418
-- class: org.elasticsearch.xpack.esql.action.CrossClusterQueryWithFiltersIT
-  method: testTimestampFilterFromQuery
-  issue: https://github.com/elastic/elasticsearch/issues/127332
 - class: org.elasticsearch.xpack.esql.plugin.DataNodeRequestSenderIT
   method: testSearchWhileRelocating
   issue: https://github.com/elastic/elasticsearch/issues/128500

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClustersIT.java

@@ -72,6 +72,7 @@ record Doc(int id, String color, long data) {
     List<Doc> localDocs = List.of();
     final String remoteIndex = "test-remote-index";
     List<Doc> remoteDocs = List.of();
+    private Boolean shouldCheckShardCounts = null;
 
     @Before
     public void setUpIndices() throws Exception {
@@ -164,6 +165,17 @@ private Map<String, Object> runEsql(RestEsqlTestCase.RequestObjectBuilder reques
         }
     }
 
+    private boolean checkShardCounts() {
+        if (shouldCheckShardCounts == null) {
+            try {
+                shouldCheckShardCounts = capabilitiesSupportedNewAndOld(List.of("correct_skipped_shard_count"));
+            } catch (IOException e) {
+                shouldCheckShardCounts = false;
+            }
+        }
+        return shouldCheckShardCounts;
+    }
+
     private <C, V> void assertResultMapForLike(
         boolean includeCCSMetadata,
         Map<String, Object> result,
@@ -295,11 +307,16 @@ private void assertClusterDetailsMap(Map<String, Object> result, boolean remoteO
         assertThat(
             remoteClusterShards,
             matchesMap().entry("total", greaterThanOrEqualTo(0))
-                .entry("successful", remoteClusterShards.get("total"))
+                .entry("successful", greaterThanOrEqualTo(0))
                 .entry("skipped", greaterThanOrEqualTo(0))
                 .entry("failed", 0)
         );
-
+        if (checkShardCounts()) {
+            assertThat(
+                (int) remoteClusterShards.get("successful") + (int) remoteClusterShards.get("skipped"),
+                equalTo(remoteClusterShards.get("total"))
+            );
+        }
         if (remoteOnly == false) {
             @SuppressWarnings("unchecked")
             Map<String, Object> localCluster = (Map<String, Object>) details.get("(local)");
@@ -313,10 +330,16 @@ private void assertClusterDetailsMap(Map<String, Object> result, boolean remoteO
             assertThat(
                 localClusterShards,
                 matchesMap().entry("total", greaterThanOrEqualTo(0))
-                    .entry("successful", localClusterShards.get("total"))
+                    .entry("successful", greaterThanOrEqualTo(0))
                     .entry("skipped", greaterThanOrEqualTo(0))
                     .entry("failed", 0)
             );
+            if (checkShardCounts()) {
+                assertThat(
+                    (int) localClusterShards.get("successful") + (int) localClusterShards.get("skipped"),
+                    equalTo(localClusterShards.get("total"))
+                );
+            }
         }
     }
 

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

@@ -62,8 +62,10 @@ protected void assertClusterMetadata(EsqlExecutionInfo.Cluster clusterMetatata,
     protected void assertClusterMetadataSuccess(EsqlExecutionInfo.Cluster clusterMetatata, int shards, long took, String indexExpression) {
         assertClusterMetadata(clusterMetatata, took, indexExpression, Status.SUCCESSFUL);
         assertThat(clusterMetatata.getTotalShards(), equalTo(shards));
-        assertThat(clusterMetatata.getSuccessfulShards(), equalTo(shards));
-        assertThat(clusterMetatata.getSkippedShards(), equalTo(0));
+        // We should have at least one successful shard for data
+        assertThat(clusterMetatata.getSuccessfulShards(), greaterThanOrEqualTo(1));
+        // Some shards may be skipped, but total sum of the shards should match up
+        assertThat(clusterMetatata.getSkippedShards() + clusterMetatata.getSuccessfulShards(), equalTo(shards));
     }
 
     protected void assertClusterMetadataNoShards(EsqlExecutionInfo.Cluster clusterMetatata, long took, String indexExpression) {
@@ -81,7 +83,7 @@ protected void assertClusterMetadataSkippedShards(
     ) {
         assertClusterMetadata(clusterMetatata, took, indexExpression, Status.SUCCESSFUL);
         assertThat(clusterMetatata.getTotalShards(), equalTo(shards));
-        assertThat(clusterMetatata.getSuccessfulShards(), equalTo(shards));
+        assertThat(clusterMetatata.getSuccessfulShards(), equalTo(0));
         assertThat(clusterMetatata.getSkippedShards(), equalTo(shards));
     }
 

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

@@ -1250,7 +1250,12 @@ public enum Cap {
          * Forbid usage of brackets in unquoted index and enrich policy names
          * https://github.com/elastic/elasticsearch/issues/130378
          */
-        NO_BRACKETS_IN_UNQUOTED_INDEX_NAMES;
+        NO_BRACKETS_IN_UNQUOTED_INDEX_NAMES,
+
+        /**
+         * Support correct counting of skipped shards.
+         */
+        CORRECT_SKIPPED_SHARDS_COUNT;
 
         private final boolean enabled;
 

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

@@ -134,17 +134,20 @@ final void startComputeOnDataNodes(Set<String> concreteIndices, Runnable runOnTa
                 var computeListener = new ComputeListener(
                     transportService.getThreadPool(),
                     runOnTaskFailure,
-                    listener.map(
-                        completionInfo -> new ComputeResponse(
+                    listener.map(completionInfo -> {
+                        final int totalSkipShards = targetShards.skippedShards() + skippedShards.get();
+                        final int failedShards = shardFailures.size();
+                        final int successfulShards = targetShards.totalShards() - totalSkipShards - failedShards;
+                        return new ComputeResponse(
                             completionInfo,
                             timeValueNanos(System.nanoTime() - startTimeInNanos),
                             targetShards.totalShards(),
-                            targetShards.totalShards() - shardFailures.size() - skippedShards.get(),
-                            targetShards.skippedShards() + skippedShards.get(),
-                            shardFailures.size(),
+                            successfulShards,
+                            totalSkipShards,
+                            failedShards,
                             selectFailures()
-                        )
-                    )
+                        );
+                    })
                 )
             ) {
                 pendingShardIds.addAll(order(targetShards));