Skip to content

Fail request when all target shards fail in runtime #131177

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jul 15, 2025
Merged

Fail request when all target shards fail in runtime #131177

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6c34c05
Fail request when all target shards fail in runtime
dnhatn Jul 15, 2025
a1f002b
reword
dnhatn Jul 15, 2025
b961e9b
assertion
dnhatn Jul 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 40 additions & 15 deletions ...s/error-query/src/javaRestTest/java/org/elasticsearch/test/esql/EsqlPartialResultsIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@

import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.lessThanOrEqualTo;

public class EsqlPartialResultsIT extends ESRestTestCase {
@ClassRule
Expand Down Expand Up @@ -106,7 +106,11 @@ public void testPartialResult() throws Exception {
Set<String> okIds = populateIndices();
String query = """
{
"query": "FROM ok-index,failing-index | LIMIT 100 | KEEP fail_me,v"
"query": "FROM ok-index,failing-index | LIMIT 100 | KEEP fail_me,v",
"pragma": {
"max_concurrent_shards_per_node": 1
},
"accept_pragma_risks": true
}
""";
// allow_partial_results = true
Expand All @@ -123,19 +127,18 @@ public void testPartialResult() throws Exception {
List<?> columns = (List<?>) results.get("columns");
assertThat(columns, equalTo(List.of(Map.of("name", "fail_me", "type", "long"), Map.of("name", "v", "type", "long"))));
List<?> values = (List<?>) results.get("values");
assertThat(values.size(), lessThanOrEqualTo(okIds.size()));
assertThat(values.size(), equalTo(okIds.size()));
Map<String, Object> localInfo = (Map<String, Object>) XContentMapValues.extractValue(
results,
"_clusters",
"details",
"(local)"
);
assertNotNull(localInfo);
assertThat(XContentMapValues.extractValue(localInfo, "_shards", "successful"), equalTo(0));
assertThat(
XContentMapValues.extractValue(localInfo, "_shards", "failed"),
equalTo(XContentMapValues.extractValue(localInfo, "_shards", "total"))
);
Integer successfulShards = (Integer) XContentMapValues.extractValue(localInfo, "_shards", "successful");
Integer failedShards = (Integer) XContentMapValues.extractValue(localInfo, "_shards", "failed");
assertThat(successfulShards, greaterThan(0));
assertThat(failedShards, greaterThan(0));
List<Map<String, Object>> failures = (List<Map<String, Object>>) XContentMapValues.extractValue(localInfo, "failures");
assertThat(failures, hasSize(1));
assertThat(
Expand Down Expand Up @@ -167,7 +170,11 @@ public void testFailureFromRemote() throws Exception {
Set<String> okIds = populateIndices();
String query = """
{
"query": "FROM *:ok-index,*:failing-index | LIMIT 100 | KEEP fail_me,v"
"query": "FROM *:ok-index,*:failing-index | LIMIT 100 | KEEP fail_me,v",
"pragma": {
"max_concurrent_shards_per_node": 1
},
"accept_pragma_risks": true
}
""";
// allow_partial_results = true
Expand All @@ -183,19 +190,18 @@ public void testFailureFromRemote() throws Exception {
List<?> columns = (List<?>) results.get("columns");
assertThat(columns, equalTo(List.of(Map.of("name", "fail_me", "type", "long"), Map.of("name", "v", "type", "long"))));
List<?> values = (List<?>) results.get("values");
assertThat(values.size(), lessThanOrEqualTo(okIds.size()));
assertThat(values.size(), equalTo(okIds.size()));
Map<String, Object> remoteCluster = (Map<String, Object>) XContentMapValues.extractValue(
results,
"_clusters",
"details",
"cluster_one"
);
assertNotNull(remoteCluster);
assertThat(XContentMapValues.extractValue(remoteCluster, "_shards", "successful"), equalTo(0));
assertThat(
XContentMapValues.extractValue(remoteCluster, "_shards", "failed"),
equalTo(XContentMapValues.extractValue(remoteCluster, "_shards", "total"))
);
Integer successfulShards = (Integer) XContentMapValues.extractValue(remoteCluster, "_shards", "successful");
Integer failedShards = (Integer) XContentMapValues.extractValue(remoteCluster, "_shards", "failed");
assertThat(successfulShards, greaterThan(0));
assertThat(failedShards, greaterThan(0));
List<Map<String, Object>> failures = (List<Map<String, Object>>) XContentMapValues.extractValue(remoteCluster, "failures");
assertThat(failures, hasSize(1));
assertThat(
Expand All @@ -207,6 +213,25 @@ public void testFailureFromRemote() throws Exception {
}
}

public void testAllShardsFailed() throws Exception {
setupRemoteClusters();
populateIndices();
try {
for (boolean allowPartialResults : List.of(Boolean.TRUE, Boolean.FALSE)) {
for (String index : List.of("failing*", "*:failing*", "*:failing*,failing*")) {
Request request = new Request("POST", "/_query");
request.setJsonEntity("{\"query\": \"FROM " + index + " | LIMIT 100\"}");
request.addParameter("allow_partial_results", Boolean.toString(allowPartialResults));
var error = expectThrows(ResponseException.class, () -> client().performRequest(request));
Response resp = error.getResponse();
assertThat(EntityUtils.toString(resp.getEntity()), containsString("Accessing failing field"));
}
}
} finally {
removeRemoteCluster();
}
}

private void setupRemoteClusters() throws IOException {
String settings = String.format(Locale.ROOT, """
{
Expand Down
66 changes: 35 additions & 31 deletions ...l/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlNodeFailureIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import org.elasticsearch.xcontent.json.JsonXContent;
import org.elasticsearch.xpack.esql.EsqlTestUtils;
import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
import org.elasticsearch.xpack.esql.plugin.QueryPragmas;

import java.util.ArrayList;
import java.util.Collection;
Expand All @@ -30,6 +31,7 @@
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.in;
import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.hamcrest.Matchers.not;
Expand Down Expand Up @@ -98,38 +100,30 @@ public void testFailureLoadingFields() throws Exception {

public void testPartialResults() throws Exception {
Set<String> okIds = populateIndices();
{
EsqlQueryRequest request = new EsqlQueryRequest();
request.query("FROM fail,ok | LIMIT 100");
request.allowPartialResults(true);
request.pragmas(randomPragmas());
try (EsqlQueryResponse resp = run(request)) {
assertTrue(resp.isPartial());
List<List<Object>> rows = EsqlTestUtils.getValuesList(resp);
assertThat(rows.size(), lessThanOrEqualTo(okIds.size()));
}
}
{
EsqlQueryRequest request = new EsqlQueryRequest();
request.query("FROM fail,ok METADATA _id | KEEP _id, fail_me | LIMIT 100");
request.allowPartialResults(true);
request.pragmas(randomPragmas());
try (EsqlQueryResponse resp = run(request)) {
assertTrue(resp.isPartial());
List<List<Object>> rows = EsqlTestUtils.getValuesList(resp);
assertThat(rows.size(), lessThanOrEqualTo(okIds.size()));
Set<String> actualIds = new HashSet<>();
for (List<Object> row : rows) {
assertThat(row.size(), equalTo(2));
String id = (String) row.getFirst();
assertThat(id, in(okIds));
assertTrue(actualIds.add(id));
}
EsqlExecutionInfo.Cluster localInfo = resp.getExecutionInfo().getCluster(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY);
assertThat(localInfo.getFailures(), not(empty()));
assertThat(localInfo.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.PARTIAL));
assertThat(localInfo.getFailures().get(0).reason(), containsString("Accessing failing field"));
EsqlQueryRequest request = new EsqlQueryRequest();
request.query("FROM fail,ok METADATA _id | KEEP _id, fail_me | LIMIT 100");
request.allowPartialResults(true);
// have to run one shard at a time to avoid failing all shards
QueryPragmas pragma = new QueryPragmas(
Settings.builder().put(randomPragmas().getSettings()).put(QueryPragmas.MAX_CONCURRENT_SHARDS_PER_NODE.getKey(), 1).build()
);
request.pragmas(pragma);
request.acceptedPragmaRisks(true);
try (EsqlQueryResponse resp = run(request)) {
assertTrue(resp.isPartial());
List<List<Object>> rows = EsqlTestUtils.getValuesList(resp);
assertThat(rows.size(), equalTo(okIds.size()));
Set<String> actualIds = new HashSet<>();
for (List<Object> row : rows) {
assertThat(row.size(), equalTo(2));
String id = (String) row.getFirst();
assertThat(id, in(okIds));
assertTrue(actualIds.add(id));
}
EsqlExecutionInfo.Cluster localInfo = resp.getExecutionInfo().getCluster(RemoteClusterService.LOCAL_CLUSTER_GROUP_KEY);
assertThat(localInfo.getFailures(), not(empty()));
assertThat(localInfo.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.PARTIAL));
assertThat(localInfo.getFailures().get(0).reason(), containsString("Accessing failing field"));
}
}

Expand All @@ -147,13 +141,23 @@ public void testDefaultPartialResults() throws Exception {
EsqlQueryRequest request = new EsqlQueryRequest();
request.query("FROM fail,ok | LIMIT 100");
request.pragmas(randomPragmas());
// have to run one shard at a time to avoid failing all shards
QueryPragmas pragma = new QueryPragmas(
Settings.builder()
.put(randomPragmas().getSettings())
.put(QueryPragmas.MAX_CONCURRENT_SHARDS_PER_NODE.getKey(), 1)
.build()
);
request.pragmas(pragma);
request.acceptedPragmaRisks(true);
if (randomBoolean()) {
request.allowPartialResults(true);
}
try (EsqlQueryResponse resp = run(request)) {
assertTrue(resp.isPartial());
List<List<Object>> rows = EsqlTestUtils.getValuesList(resp);
assertThat(rows.size(), lessThanOrEqualTo(okIds.size()));
assertThat(rows.size(), greaterThan(0));
}
}
// allow_partial_results = false
Expand Down
46 changes: 44 additions & 2 deletions x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

package org.elasticsearch.xpack.esql.plugin;

import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.OriginalIndices;
import org.elasticsearch.action.search.SearchRequest;
Expand Down Expand Up @@ -375,9 +376,10 @@ public void executePlan(
var computeListener = new ComputeListener(
transportService.getThreadPool(),
cancelQueryOnFailure,
listener.map(completionInfo -> {
listener.delegateFailureAndWrap((l, completionInfo) -> {
failIfAllShardsFailed(execInfo, collectedPages);
execInfo.markEndQuery(); // TODO: revisit this time recording model as part of INLINESTATS improvements
return new Result(outputAttributes, collectedPages, completionInfo, execInfo);
l.onResponse(new Result(outputAttributes, collectedPages, completionInfo, execInfo));
})
)
) {
Expand Down Expand Up @@ -540,6 +542,46 @@ private static void updateExecutionInfoAfterCoordinatorOnlyQuery(EsqlExecutionIn
}
}

/**
* If all of target shards excluding the skipped shards failed, then we should fail the entire query regardless of the partial_results
* configuration or skip_unavailable setting. This behavior doesn't fully align with the search API as the skip_unavailable
* would ignore all the failures from the remote clusters; hence, only fail the request when all shards in the local cluster failed.
*/
static void failIfAllShardsFailed(EsqlExecutionInfo execInfo, List<Page> finalResults) {
Copy link
Contributor

@idegtiarenko idegtiarenko Jul 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this method concerning only remote shards/failures or local too?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method treats both local and remote failures uniformly. This is where the behavior in ES|QL differs slightly from the search API. Here, we iterate over all execution info in each cluster (both local and remotes) to accumulate successful shards (excluding skipped shards) and failed shards. The request only fails if there are no successful shards, some failed shards, and no rows produced in the final results.

// do not fail if any final result has results
if (finalResults.stream().anyMatch(p -> p.getPositionCount() > 0)) {
return;
}
Copy link
Contributor

@idegtiarenko idegtiarenko Jul 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure if this will behave correctly.
I am imagining a case with a single index and no rows in it.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The request will not fail as long as the single target shard does not fail.

int totalFailedShards = 0;
for (EsqlExecutionInfo.Cluster cluster : execInfo.clusterInfo.values()) {
final Integer successfulShards = cluster.getSuccessfulShards();
if (successfulShards != null && successfulShards > 0) {
return;
}
if (cluster.getFailedShards() != null) {
totalFailedShards += cluster.getFailedShards();
}
}
if (totalFailedShards == 0) {
return;
}
final var failureCollector = new FailureCollector();
for (EsqlExecutionInfo.Cluster cluster : execInfo.clusterInfo.values()) {
var failedShards = cluster.getFailedShards();
if (failedShards != null && failedShards > 0) {
assert cluster.getFailures().isEmpty() == false : "expected failures for cluster [" + cluster.getClusterAlias() + "]";
for (ShardSearchFailure failure : cluster.getFailures()) {
if (failure.getCause() instanceof Exception e) {
failureCollector.unwrapAndCollect(e);
} else {
failureCollector.unwrapAndCollect(failure);
}
}
}
}
ExceptionsHelper.reThrowIfNotNull(failureCollector.getFailure());
}

void runCompute(CancellableTask task, ComputeContext context, PhysicalPlan plan, ActionListener<DriverCompletionInfo> listener) {
listener = ActionListener.runBefore(listener, () -> Releasables.close(context.searchContexts()));
List<EsPhysicalOperationProviders.ShardContext> contexts = new ArrayList<>(context.searchContexts().size());
Expand Down