Skip to content

Commit da2cc30

Browse files
committed
Allow partial results by default in ES|QL - Take 2 (elastic#127351)
* Revert "ESQL: Revert "Allow partial results by default in ES|QL (elastic#125060)" (elastic#126286)" This reverts commit 8f38b13. Restore changes from elastic#125060 now that the breakage should be fixed. (cherry picked from commit eb479e5) # Conflicts: # docs/release-notes/breaking-changes.md # x-pack/plugin/build.gradle # x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java
1 parent 30d5a9d commit da2cc30

File tree

14 files changed

+80
-62
lines changed

14 files changed

+80
-62
lines changed

docs/changelog/127351.yaml

+16
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
pr: 127351
2+
summary: Allow partial results by default in ES|QL
3+
area: ES|QL
4+
type: breaking
5+
issues: [122802]
6+
7+
breaking:
8+
title: Allow partial results by default in ES|QL
9+
area: ES|QL
10+
details: >-
11+
In earlier versions of {es}, ES|QL would fail the entire query if it encountered any error. ES|QL now returns partial results instead of failing when encountering errors.
12+
13+
impact: >-
14+
Callers should check the `is_partial` flag returned in the response to determine if the result is partial or complete. If returning partial results is not desired, this option can be overridden per request via an `allow_partial_results` parameter in the query URL or globally via the cluster setting `esql.query.allow_partial_results`.
15+
16+
notable: true

test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/Clusters.java

+1
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ static ElasticsearchCluster buildCluster() {
2121
.module("test-esql-heap-attack")
2222
.setting("xpack.security.enabled", "false")
2323
.setting("xpack.license.self_generated.type", "trial")
24+
.setting("esql.query.allow_partial_results", "false")
2425
.jvmArg("-Xmx512m");
2526
String javaVersion = JvmInfo.jvmInfo().version();
2627
if (javaVersion.equals("20") || javaVersion.equals("21")) {

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,6 @@ private RestClient remoteClusterClient() throws IOException {
8383

8484
@Before
8585
public void skipTestOnOldVersions() {
86-
assumeTrue("skip on old versions", Clusters.localClusterVersion().equals(Version.V_8_16_0));
86+
assumeTrue("skip on old versions", Clusters.localClusterVersion().equals(Version.V_8_19_0));
8787
}
8888
}

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

+7
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.junit.rules.TestRule;
3131

3232
import java.io.IOException;
33+
import java.util.List;
3334
import java.util.Map;
3435

3536
import static org.elasticsearch.test.MapMatcher.assertMap;
@@ -94,6 +95,12 @@ protected String from(String... indexName) {
9495

9596
@Override
9697
public Map<String, Object> runEsql(RestEsqlTestCase.RequestObjectBuilder requestObject) throws IOException {
98+
if (requestObject.allowPartialResults() != null) {
99+
assumeTrue(
100+
"require allow_partial_results on local cluster",
101+
clusterHasCapability("POST", "/_query", List.of(), List.of("support_partial_results")).orElse(false)
102+
);
103+
}
97104
requestObject.includeCCSMetadata(true);
98105
return super.runEsql(requestObject);
99106
}

x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/RestEsqlIT.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,7 @@ public void testInvalidPragma() throws IOException {
110110
request.setJsonEntity("{\"f\":" + i + "}");
111111
assertOK(client().performRequest(request));
112112
}
113-
RequestObjectBuilder builder = requestObjectBuilder().query("from test-index | limit 1 | keep f");
113+
RequestObjectBuilder builder = requestObjectBuilder().query("from test-index | limit 1 | keep f").allowPartialResults(false);
114114
builder.pragmas(Settings.builder().put("data_partitioning", "invalid-option").build());
115115
ResponseException re = expectThrows(ResponseException.class, () -> runEsqlSync(builder));
116116
assertThat(EntityUtils.toString(re.getResponse().getEntity()), containsString("No enum constant"));

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/RestEsqlTestCase.java

+9-5
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,7 @@ public static class RequestObjectBuilder {
132132
private Boolean includeCCSMetadata = null;
133133

134134
private CheckedConsumer<XContentBuilder, IOException> filter;
135-
private Boolean allPartialResults = null;
135+
private Boolean allowPartialResults = null;
136136

137137
public RequestObjectBuilder() throws IOException {
138138
this(randomFrom(XContentType.values()));
@@ -210,11 +210,15 @@ public RequestObjectBuilder filter(CheckedConsumer<XContentBuilder, IOException>
210210
return this;
211211
}
212212

213-
public RequestObjectBuilder allPartialResults(boolean allPartialResults) {
214-
this.allPartialResults = allPartialResults;
213+
public RequestObjectBuilder allowPartialResults(boolean allowPartialResults) {
214+
this.allowPartialResults = allowPartialResults;
215215
return this;
216216
}
217217

218+
public Boolean allowPartialResults() {
219+
return allowPartialResults;
220+
}
221+
218222
public RequestObjectBuilder build() throws IOException {
219223
if (isBuilt == false) {
220224
if (tables != null) {
@@ -1369,8 +1373,8 @@ protected static Request prepareRequestWithOptions(RequestObjectBuilder requestO
13691373
requestObject.build();
13701374
Request request = prepareRequest(mode);
13711375
String mediaType = attachBody(requestObject, request);
1372-
if (requestObject.allPartialResults != null) {
1373-
request.addParameter("allow_partial_results", String.valueOf(requestObject.allPartialResults));
1376+
if (requestObject.allowPartialResults != null) {
1377+
request.addParameter("allow_partial_results", String.valueOf(requestObject.allowPartialResults));
13741378
}
13751379

13761380
RequestOptions.Builder options = request.getOptions().toBuilder();

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

+6
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.elasticsearch.transport.RemoteClusterAware;
2626
import org.elasticsearch.xcontent.XContentBuilder;
2727
import org.elasticsearch.xcontent.json.JsonXContent;
28+
import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
2829
import org.junit.After;
2930
import org.junit.Before;
3031

@@ -76,6 +77,11 @@ protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
7677
return plugins;
7778
}
7879

80+
@Override
81+
protected Settings nodeSettings() {
82+
return Settings.builder().put(super.nodeSettings()).put(EsqlPlugin.QUERY_ALLOW_PARTIAL_RESULTS.getKey(), false).build();
83+
}
84+
7985
public static class InternalExchangePlugin extends Plugin {
8086
@Override
8187
public List<Setting<?>> getSettings() {

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

+8
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,14 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
139139
return CollectionUtils.appendToCopy(super.nodePlugins(), EsqlPlugin.class);
140140
}
141141

142+
@Override
143+
protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
144+
return Settings.builder()
145+
.put(super.nodeSettings(nodeOrdinal, otherSettings))
146+
.put(EsqlPlugin.QUERY_ALLOW_PARTIAL_RESULTS.getKey(), false)
147+
.build();
148+
}
149+
142150
protected void setRequestCircuitBreakerLimit(ByteSizeValue limit) {
143151
if (limit != null) {
144152
assertAcked(

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

+7-36
Original file line numberDiff line numberDiff line change
@@ -14,25 +14,20 @@
1414
import org.elasticsearch.action.index.IndexRequest;
1515
import org.elasticsearch.action.support.PlainActionFuture;
1616
import org.elasticsearch.action.support.WriteRequest;
17-
import org.elasticsearch.common.settings.Setting;
17+
import org.elasticsearch.common.settings.Settings;
1818
import org.elasticsearch.common.transport.TransportAddress;
1919
import org.elasticsearch.compute.operator.DriverTaskRunner;
2020
import org.elasticsearch.compute.operator.exchange.ExchangeService;
2121
import org.elasticsearch.core.TimeValue;
22-
import org.elasticsearch.plugins.Plugin;
2322
import org.elasticsearch.tasks.TaskCancelledException;
2423
import org.elasticsearch.tasks.TaskInfo;
25-
import org.elasticsearch.test.AbstractMultiClustersTestCase;
2624
import org.elasticsearch.transport.TransportService;
2725
import org.elasticsearch.xcontent.XContentBuilder;
2826
import org.elasticsearch.xcontent.json.JsonXContent;
2927
import org.elasticsearch.xpack.esql.EsqlTestUtils;
3028
import org.elasticsearch.xpack.esql.plugin.ComputeService;
31-
import org.junit.After;
32-
import org.junit.Before;
3329

3430
import java.util.ArrayList;
35-
import java.util.Collection;
3631
import java.util.List;
3732
import java.util.concurrent.TimeUnit;
3833

@@ -44,7 +39,7 @@
4439
import static org.hamcrest.Matchers.hasSize;
4540
import static org.hamcrest.Matchers.instanceOf;
4641

47-
public class CrossClusterCancellationIT extends AbstractMultiClustersTestCase {
42+
public class CrossClusterCancellationIT extends AbstractCrossClusterTestCase {
4843
private static final String REMOTE_CLUSTER = "cluster-a";
4944

5045
@Override
@@ -53,35 +48,11 @@ protected Collection<String> remoteClusterAlias() {
5348
}
5449

5550
@Override
56-
protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
57-
List<Class<? extends Plugin>> plugins = new ArrayList<>(super.nodePlugins(clusterAlias));
58-
plugins.add(EsqlPluginWithEnterpriseOrTrialLicense.class);
59-
plugins.add(InternalExchangePlugin.class);
60-
plugins.add(SimplePauseFieldPlugin.class);
61-
return plugins;
62-
}
63-
64-
public static class InternalExchangePlugin extends Plugin {
65-
@Override
66-
public List<Setting<?>> getSettings() {
67-
return List.of(
68-
Setting.timeSetting(
69-
ExchangeService.INACTIVE_SINKS_INTERVAL_SETTING,
70-
TimeValue.timeValueMillis(between(3000, 4000)),
71-
Setting.Property.NodeScope
72-
)
73-
);
74-
}
75-
}
76-
77-
@Before
78-
public void resetPlugin() {
79-
SimplePauseFieldPlugin.resetPlugin();
80-
}
81-
82-
@After
83-
public void releasePlugin() {
84-
SimplePauseFieldPlugin.release();
51+
protected Settings nodeSettings() {
52+
return Settings.builder()
53+
.put(super.nodeSettings())
54+
.put(ExchangeService.INACTIVE_SINKS_INTERVAL_SETTING, TimeValue.timeValueMillis(between(3000, 4000)))
55+
.build();
8556
}
8657

8758
@Override

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

+1
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
6969
@Override
7070
protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
7171
return Settings.builder()
72+
.put(super.nodeSettings(nodeOrdinal, otherSettings))
7273
.put(ExchangeService.INACTIVE_SINKS_INTERVAL_SETTING, TimeValue.timeValueMillis(between(3000, 5000)))
7374
.build();
7475
}

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

+6-14
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77

88
package org.elasticsearch.xpack.esql.plugin;
99

10-
import org.elasticsearch.ExceptionsHelper;
1110
import org.elasticsearch.action.ActionListener;
1211
import org.elasticsearch.action.ActionListenerResponseHandler;
1312
import org.elasticsearch.action.OriginalIndices;
@@ -17,7 +16,6 @@
1716
import org.elasticsearch.compute.operator.exchange.ExchangeSourceHandler;
1817
import org.elasticsearch.core.Releasable;
1918
import org.elasticsearch.core.TimeValue;
20-
import org.elasticsearch.index.IndexNotFoundException;
2119
import org.elasticsearch.tasks.CancellableTask;
2220
import org.elasticsearch.tasks.Task;
2321
import org.elasticsearch.tasks.TaskCancelledException;
@@ -90,18 +88,12 @@ void startComputeOnRemoteCluster(
9088
if (receivedResults == false && EsqlCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
9189
EsqlCCSUtils.markClusterWithFinalStateAndNoShards(executionInfo, clusterAlias, EsqlExecutionInfo.Cluster.Status.SKIPPED, e);
9290
l.onResponse(List.of());
93-
} else if (configuration.allowPartialResults()
94-
&& (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) == false) {
95-
EsqlCCSUtils.markClusterWithFinalStateAndNoShards(
96-
executionInfo,
97-
clusterAlias,
98-
EsqlExecutionInfo.Cluster.Status.PARTIAL,
99-
e
100-
);
101-
l.onResponse(List.of());
102-
} else {
103-
l.onFailure(e);
104-
}
91+
} else if (configuration.allowPartialResults() && EsqlCCSUtils.canAllowPartial(e)) {
92+
EsqlCCSUtils.markClusterWithFinalStateAndNoShards(executionInfo, clusterAlias, EsqlExecutionInfo.Cluster.Status.PARTIAL, e);
93+
l.onResponse(List.of());
94+
} else {
95+
l.onFailure(e);
96+
}
10597
});
10698
ExchangeService.openExchange(
10799
transportService,

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

+2-4
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77

88
package org.elasticsearch.xpack.esql.plugin;
99

10-
import org.elasticsearch.ExceptionsHelper;
1110
import org.elasticsearch.action.ActionListener;
1211
import org.elasticsearch.action.OriginalIndices;
1312
import org.elasticsearch.action.search.SearchRequest;
@@ -28,7 +27,6 @@
2827
import org.elasticsearch.core.Releasables;
2928
import org.elasticsearch.core.TimeValue;
3029
import org.elasticsearch.core.Tuple;
31-
import org.elasticsearch.index.IndexNotFoundException;
3230
import org.elasticsearch.index.query.SearchExecutionContext;
3331
import org.elasticsearch.logging.LogManager;
3432
import org.elasticsearch.logging.Logger;
@@ -59,6 +57,7 @@
5957
import org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner;
6058
import org.elasticsearch.xpack.esql.planner.PlannerUtils;
6159
import org.elasticsearch.xpack.esql.session.Configuration;
60+
import org.elasticsearch.xpack.esql.session.EsqlCCSUtils;
6261
import org.elasticsearch.xpack.esql.session.Result;
6362

6463
import java.util.ArrayList;
@@ -276,8 +275,7 @@ public void execute(
276275
);
277276
dataNodesListener.onResponse(r.getProfiles());
278277
}, e -> {
279-
if (configuration.allowPartialResults()
280-
&& (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) == false) {
278+
if (configuration.allowPartialResults() && EsqlCCSUtils.canAllowPartial(e)) {
281279
execInfo.swapCluster(
282280
LOCAL_CLUSTER,
283281
(k, v) -> new EsqlExecutionInfo.Cluster.Builder(v).setStatus(

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

+1-1
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ public class EsqlPlugin extends Plugin implements ActionPlugin {
105105

106106
public static final Setting<Boolean> QUERY_ALLOW_PARTIAL_RESULTS = Setting.boolSetting(
107107
"esql.query.allow_partial_results",
108-
false,
108+
true,
109109
Setting.Property.NodeScope,
110110
Setting.Property.Dynamic
111111
);

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

+14
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77

88
package org.elasticsearch.xpack.esql.session;
99

10+
import org.elasticsearch.ElasticsearchSecurityException;
1011
import org.elasticsearch.ExceptionsHelper;
1112
import org.elasticsearch.action.ActionListener;
1213
import org.elasticsearch.action.OriginalIndices;
@@ -16,6 +17,7 @@
1617
import org.elasticsearch.common.Strings;
1718
import org.elasticsearch.common.util.set.Sets;
1819
import org.elasticsearch.core.Nullable;
20+
import org.elasticsearch.index.IndexNotFoundException;
1921
import org.elasticsearch.index.query.QueryBuilder;
2022
import org.elasticsearch.indices.IndicesExpressionGrouper;
2123
import org.elasticsearch.license.XPackLicenseState;
@@ -367,4 +369,16 @@ public static boolean shouldIgnoreRuntimeError(EsqlExecutionInfo executionInfo,
367369

368370
return ExceptionsHelper.isRemoteUnavailableException(e);
369371
}
372+
373+
/**
374+
* Check whether this exception can be tolerated when partial results are on, or should be treated as fatal.
375+
* @return true if the exception can be tolerated, false if it should be treated as fatal.
376+
*/
377+
public static boolean canAllowPartial(Exception e) {
378+
Throwable unwrapped = ExceptionsHelper.unwrapCause(e);
379+
if (unwrapped instanceof IndexNotFoundException || unwrapped instanceof ElasticsearchSecurityException) {
380+
return false;
381+
}
382+
return true;
383+
}
370384
}

0 commit comments

Comments
 (0)