Skip to content

Commit 93788a8

Browse files
authored
Fix counting skipped shards with filters (#131737) (#131770)
* Fix counting skipped shards with filters (cherry picked from commit a345f56) # Conflicts: # muted-tests.yml # 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
1 parent 575065e commit 93788a8

File tree

4 files changed

+47
-14
lines changed

4 files changed

+47
-14
lines changed

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

Lines changed: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@ record Doc(int id, String color, long data) {
7272
List<Doc> localDocs = List.of();
7373
final String remoteIndex = "test-remote-index";
7474
List<Doc> remoteDocs = List.of();
75+
private Boolean shouldCheckShardCounts = null;
7576

7677
@Before
7778
public void setUpIndices() throws Exception {
@@ -164,6 +165,17 @@ private Map<String, Object> runEsql(RestEsqlTestCase.RequestObjectBuilder reques
164165
}
165166
}
166167

168+
private boolean checkShardCounts() {
169+
if (shouldCheckShardCounts == null) {
170+
try {
171+
shouldCheckShardCounts = capabilitiesSupportedNewAndOld(List.of("correct_skipped_shard_count"));
172+
} catch (IOException e) {
173+
shouldCheckShardCounts = false;
174+
}
175+
}
176+
return shouldCheckShardCounts;
177+
}
178+
167179
private <C, V> void assertResultMapForLike(
168180
boolean includeCCSMetadata,
169181
Map<String, Object> result,
@@ -295,11 +307,16 @@ private void assertClusterDetailsMap(Map<String, Object> result, boolean remoteO
295307
assertThat(
296308
remoteClusterShards,
297309
matchesMap().entry("total", greaterThanOrEqualTo(0))
298-
.entry("successful", remoteClusterShards.get("total"))
310+
.entry("successful", greaterThanOrEqualTo(0))
299311
.entry("skipped", greaterThanOrEqualTo(0))
300312
.entry("failed", 0)
301313
);
302-
314+
if (checkShardCounts()) {
315+
assertThat(
316+
(int) remoteClusterShards.get("successful") + (int) remoteClusterShards.get("skipped"),
317+
equalTo(remoteClusterShards.get("total"))
318+
);
319+
}
303320
if (remoteOnly == false) {
304321
@SuppressWarnings("unchecked")
305322
Map<String, Object> localCluster = (Map<String, Object>) details.get("(local)");
@@ -313,10 +330,16 @@ private void assertClusterDetailsMap(Map<String, Object> result, boolean remoteO
313330
assertThat(
314331
localClusterShards,
315332
matchesMap().entry("total", greaterThanOrEqualTo(0))
316-
.entry("successful", localClusterShards.get("total"))
333+
.entry("successful", greaterThanOrEqualTo(0))
317334
.entry("skipped", greaterThanOrEqualTo(0))
318335
.entry("failed", 0)
319336
);
337+
if (checkShardCounts()) {
338+
assertThat(
339+
(int) localClusterShards.get("successful") + (int) localClusterShards.get("skipped"),
340+
equalTo(localClusterShards.get("total"))
341+
);
342+
}
320343
}
321344
}
322345

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -62,8 +62,10 @@ protected void assertClusterMetadata(EsqlExecutionInfo.Cluster clusterMetatata,
6262
protected void assertClusterMetadataSuccess(EsqlExecutionInfo.Cluster clusterMetatata, int shards, long took, String indexExpression) {
6363
assertClusterMetadata(clusterMetatata, took, indexExpression, Status.SUCCESSFUL);
6464
assertThat(clusterMetatata.getTotalShards(), equalTo(shards));
65-
assertThat(clusterMetatata.getSuccessfulShards(), equalTo(shards));
66-
assertThat(clusterMetatata.getSkippedShards(), equalTo(0));
65+
// We should have at least one successful shard for data
66+
assertThat(clusterMetatata.getSuccessfulShards(), greaterThanOrEqualTo(1));
67+
// Some shards may be skipped, but total sum of the shards should match up
68+
assertThat(clusterMetatata.getSkippedShards() + clusterMetatata.getSuccessfulShards(), equalTo(shards));
6769
}
6870

6971
protected void assertClusterMetadataNoShards(EsqlExecutionInfo.Cluster clusterMetatata, long took, String indexExpression) {
@@ -81,7 +83,7 @@ protected void assertClusterMetadataSkippedShards(
8183
) {
8284
assertClusterMetadata(clusterMetatata, took, indexExpression, Status.SUCCESSFUL);
8385
assertThat(clusterMetatata.getTotalShards(), equalTo(shards));
84-
assertThat(clusterMetatata.getSuccessfulShards(), equalTo(shards));
86+
assertThat(clusterMetatata.getSuccessfulShards(), equalTo(0));
8587
assertThat(clusterMetatata.getSkippedShards(), equalTo(shards));
8688
}
8789

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1035,7 +1035,12 @@ public enum Cap {
10351035
* Forbid usage of brackets in unquoted index and enrich policy names
10361036
* https://github.com/elastic/elasticsearch/issues/130378
10371037
*/
1038-
NO_BRACKETS_IN_UNQUOTED_INDEX_NAMES;
1038+
NO_BRACKETS_IN_UNQUOTED_INDEX_NAMES,
1039+
1040+
/**
1041+
* Support correct counting of skipped shards.
1042+
*/
1043+
CORRECT_SKIPPED_SHARDS_COUNT;
10391044

10401045
private final boolean enabled;
10411046

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

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -129,17 +129,20 @@ final void startComputeOnDataNodes(Set<String> concreteIndices, Runnable runOnTa
129129
var computeListener = new ComputeListener(
130130
transportService.getThreadPool(),
131131
runOnTaskFailure,
132-
listener.map(
133-
completionInfo -> new ComputeResponse(
132+
listener.map(completionInfo -> {
133+
final int totalSkipShards = targetShards.skippedShards() + skippedShards.get();
134+
final int failedShards = shardFailures.size();
135+
final int successfulShards = targetShards.totalShards() - totalSkipShards - failedShards;
136+
return new ComputeResponse(
134137
completionInfo,
135138
timeValueNanos(System.nanoTime() - startTimeInNanos),
136139
targetShards.totalShards(),
137-
targetShards.totalShards() - shardFailures.size() - skippedShards.get(),
138-
targetShards.skippedShards() + skippedShards.get(),
139-
shardFailures.size(),
140+
successfulShards,
141+
totalSkipShards,
142+
failedShards,
140143
selectFailures()
141-
)
142-
)
144+
);
145+
})
143146
)
144147
) {
145148
pendingShardIds.addAll(order(targetShards));

0 commit comments

Comments
 (0)