Skip to content

Commit 7869509

Browse files
smalyshevsmalyshev
committed
Report failures on partial results (#124823)
* Report failures on partial results (cherry picked from commit 0e6d6f4)
1 parent 2f71f6f commit 7869509

File tree

5 files changed

+117
-3
lines changed

5 files changed

+117
-3
lines changed

‎docs/changelog/124823.yaml

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 124823
2+
summary: Report failures on partial results
3+
area: "ES|QL"
4+
type: enhancement
5+
issues: []

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

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
import java.util.concurrent.atomic.AtomicLong;
4141

4242
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
43+
import static org.hamcrest.Matchers.aMapWithSize;
4344
import static org.hamcrest.Matchers.equalTo;
4445
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
4546

@@ -143,7 +144,27 @@ protected static void assertClusterMetadataInResponse(EsqlQueryResponse resp, bo
143144
assertThat((int) inner.get("total"), equalTo(numClusters));
144145
assertTrue(inner.containsKey("details"));
145146
} else {
146-
assertNull(clusters);
147+
final Object partial = esqlResponseAsMap.get("is_partial");
148+
if (partial != null && (Boolean) partial) {
149+
// If we have partial response, we could have cluster metadata, it should contain details.
150+
// Details should not be empty, and it should contain clusters with failures.
151+
if (clusters != null) {
152+
@SuppressWarnings("unchecked")
153+
Map<String, Object> inner = (Map<String, Object>) clusters;
154+
assertThat(inner, aMapWithSize(1));
155+
assertTrue(inner.containsKey("details"));
156+
@SuppressWarnings("unchecked")
157+
Map<String, Object> details = (Map<String, Object>) inner.get("details");
158+
assertThat(details.size(), greaterThanOrEqualTo(1));
159+
details.forEach((k, v) -> {
160+
@SuppressWarnings("unchecked")
161+
Map<String, Object> cluster = (Map<String, Object>) v;
162+
assertTrue(cluster.containsKey("failures"));
163+
});
164+
}
165+
} else {
166+
assertNull(clusters);
167+
}
147168
}
148169
} catch (IOException e) {
149170
fail("Could not convert ESQLQueryResponse to Map: " + e);

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

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -215,6 +215,15 @@ public boolean isCrossClusterSearch() {
215215
|| clusterInfo.size() == 1 && clusterInfo.containsKey(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY) == false;
216216
}
217217

218+
/**
219+
* Is there any metadata to report in the response?
220+
* This is true on cross-cluster search with includeCCSMetadata=true or when there are partial failures.
221+
*/
222+
public boolean hasMetadataToReport() {
223+
return isCrossClusterSearch() && includeCCSMetadata
224+
|| (isPartial && clusterInfo.values().stream().anyMatch(c -> c.getFailures().isEmpty() == false));
225+
}
226+
218227
public Cluster getCluster(String clusterAlias) {
219228
return clusterInfo.get(clusterAlias);
220229
}
@@ -254,9 +263,13 @@ public Cluster swapCluster(String clusterAlias, BiFunction<String, Cluster, Clus
254263

255264
@Override
256265
public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params) {
257-
if (isCrossClusterSearch() == false || clusterInfo.isEmpty()) {
266+
if (clusterInfo.isEmpty()) {
258267
return Collections.emptyIterator();
259268
}
269+
if (includeCCSMetadata == false) {
270+
// If includeCCSMetadata is false, the only reason we're here is partial failures, so just report them.
271+
return onlyFailuresToXContent(params);
272+
}
260273
return ChunkedToXContent.builder(params).object(b -> {
261274
b.field(TOTAL_FIELD.getPreferredName(), clusterInfo.size());
262275
b.field(SUCCESSFUL_FIELD.getPreferredName(), getClusterStateCount(Cluster.Status.SUCCESSFUL));
@@ -269,6 +282,15 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params
269282
});
270283
}
271284

285+
private Iterator<? extends ToXContent> onlyFailuresToXContent(ToXContent.Params params) {
286+
Iterator<Cluster> failuresIterator = clusterInfo.values().stream().filter(c -> (c.getFailures().isEmpty() == false)).iterator();
287+
if (failuresIterator.hasNext()) {
288+
return ChunkedToXContent.builder(params).object(b -> { b.xContentObject("details", failuresIterator); });
289+
} else {
290+
return Collections.emptyIterator();
291+
}
292+
}
293+
272294
/**
273295
* @param status the status you want a count of
274296
* @return how many clusters are currently in a specific state

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params
214214
b.append(ResponseXContentUtils.allColumns(columns, "columns"));
215215
}
216216
b.array("values", ResponseXContentUtils.columnValues(this.columns, this.pages, columnar, nullColumns));
217-
if (executionInfo != null && executionInfo.isCrossClusterSearch() && executionInfo.includeCCSMetadata()) {
217+
if (executionInfo != null && executionInfo.hasMetadataToReport()) {
218218
b.field("_clusters", executionInfo);
219219
}
220220
if (profile != null) {

‎x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/action/EsqlExecutionInfoTests.java

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License
4+
* 2.0; you may not use this file except in compliance with the Elastic License
5+
* 2.0.
6+
*/
7+
8+
package org.elasticsearch.xpack.esql.action;
9+
10+
import org.elasticsearch.action.search.ShardSearchFailure;
11+
import org.elasticsearch.test.ESTestCase;
12+
import org.elasticsearch.transport.RemoteClusterService;
13+
14+
import java.util.List;
15+
16+
public class EsqlExecutionInfoTests extends ESTestCase {
17+
18+
static final EsqlExecutionInfo.Cluster localCluster = new EsqlExecutionInfo.Cluster(
19+
RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY,
20+
"test"
21+
);
22+
static final EsqlExecutionInfo.Cluster remoteCluster = new EsqlExecutionInfo.Cluster("remote", "test");
23+
24+
public void testHasMetadataInclude() {
25+
// includeCCSMetadata + non-local clusters will produce true
26+
EsqlExecutionInfo info = new EsqlExecutionInfo(true);
27+
assertFalse(info.hasMetadataToReport());
28+
info.swapCluster(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY, (k, v) -> localCluster);
29+
assertFalse(info.hasMetadataToReport());
30+
info.swapCluster("remote", (k, v) -> remoteCluster);
31+
assertTrue(info.hasMetadataToReport());
32+
// Only remote is enough
33+
info = new EsqlExecutionInfo(true);
34+
info.swapCluster("remote", (k, v) -> remoteCluster);
35+
assertTrue(info.hasMetadataToReport());
36+
}
37+
38+
public void testHasMetadataIncludeFalse() {
39+
// If includeCCSMetadata is false, then it should always return false
40+
EsqlExecutionInfo info = new EsqlExecutionInfo(false);
41+
assertFalse(info.hasMetadataToReport());
42+
assertFalse(info.hasMetadataToReport());
43+
info.swapCluster(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY, (k, v) -> localCluster);
44+
assertFalse(info.hasMetadataToReport());
45+
info.swapCluster("remote", (k, v) -> remoteCluster);
46+
assertFalse(info.hasMetadataToReport());
47+
}
48+
49+
public void testHasMetadataPartial() {
50+
EsqlExecutionInfo info = new EsqlExecutionInfo(false);
51+
String key = randomFrom(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY, "remote");
52+
info.swapCluster(key, (k, v) -> new EsqlExecutionInfo.Cluster(k, "test", false, EsqlExecutionInfo.Cluster.Status.SUCCESSFUL));
53+
assertFalse(info.isPartial());
54+
assertFalse(info.hasMetadataToReport());
55+
info.swapCluster(key, (k, v) -> new EsqlExecutionInfo.Cluster(k, "test", false, EsqlExecutionInfo.Cluster.Status.PARTIAL));
56+
assertTrue(info.isPartial());
57+
assertFalse(info.hasMetadataToReport());
58+
info.swapCluster(key, (k, v) -> {
59+
EsqlExecutionInfo.Cluster.Builder builder = new EsqlExecutionInfo.Cluster.Builder(v);
60+
builder.setFailures(List.of(new ShardSearchFailure(new IllegalStateException("shard failure"))));
61+
return builder.build();
62+
});
63+
assertTrue(info.hasMetadataToReport());
64+
}
65+
66+
}

0 commit comments

Comments
 (0)