Allow partial results by default in ES|QL - Take 2 (#127351)

smalyshev · web-flow · commit eb479e5457c0 · 2025-04-28T10:01:22.000-06:00
* Revert "ESQL: Revert "Allow partial results by default in ES|QL (#125060)" (#126286)" This reverts commit 8f38b13. Restore changes from #125060 now that the breakage should be fixed.
diff --git a/docs/changelog/126286.yaml b/docs/changelog/126286.yaml
diff --git a/docs/changelog/127351.yaml b/docs/changelog/127351.yaml
@@ -0,0 +1,16 @@
+pr: 127351
+summary: Allow partial results by default in ES|QL
+area: ES|QL
+type: breaking
+issues: [122802]
+
+breaking:
+  title: Allow partial results by default in ES|QL
+  area: ES|QL
+  details: >-
+    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.
+
+  impact: >-
+    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`.
+
+  notable: true
diff --git a/docs/release-notes/breaking-changes.md b/docs/release-notes/breaking-changes.md
@@ -12,6 +12,11 @@ If you are migrating from a version prior to version 9.0, you must first upgrade
 
 % ## Next version [elasticsearch-nextversion-breaking-changes]
 
+## 9.1.0 [elasticsearch-910-breaking-changes]
+
+ES|QL
+:   * Allow partial results by default in ES|QL [#125060](https://github.com/elastic/elasticsearch/pull/125060)
+
 ## 9.0.0 [elasticsearch-900-breaking-changes]
 
 Aggregations:
diff --git a/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/Clusters.java b/test/external-modules/esql-heap-attack/src/javaRestTest/java/org/elasticsearch/xpack/esql/heap_attack/Clusters.java
@@ -21,6 +21,7 @@ static ElasticsearchCluster buildCluster() {
             .module("test-esql-heap-attack")
             .setting("xpack.security.enabled", "false")
             .setting("xpack.license.self_generated.type", "trial")
+            .setting("esql.query.allow_partial_results", "false")
             .jvmArg("-Xmx512m");
         String javaVersion = JvmInfo.jvmInfo().version();
         if (javaVersion.equals("20") || javaVersion.equals("21")) {
diff --git a/x-pack/plugin/build.gradle b/x-pack/plugin/build.gradle
@@ -125,6 +125,7 @@ tasks.named("yamlRestCompatTestTransform").configure({ task ->
   task.replaceValueInMatch("Size", 49, "Test flamegraph from profiling-events")
   task.replaceValueInMatch("Size", 49, "Test flamegraph from test-events")
   task.skipTest("esql/90_non_indexed/fetch", "Temporary until backported")
+  task.skipTest("esql/63_enrich_int_range/Invalid age as double", "TODO: require disable allow_partial_results")
 })
 
 tasks.named('yamlRestCompatTest').configure {
diff --git a/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/EsqlRestValidationIT.java b/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/EsqlRestValidationIT.java
@@ -83,6 +83,6 @@ private RestClient remoteClusterClient() throws IOException {
 
     @Before
     public void skipTestOnOldVersions() {
-        assumeTrue("skip on old versions", Clusters.localClusterVersion().equals(Version.V_8_16_0));
+        assumeTrue("skip on old versions", Clusters.localClusterVersion().equals(Version.V_8_19_0));
     }
 }
diff --git a/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/RequestIndexFilteringIT.java b/x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/RequestIndexFilteringIT.java
@@ -28,6 +28,7 @@
 import org.junit.rules.TestRule;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Map;
 
 import static org.elasticsearch.test.MapMatcher.assertMap;
@@ -87,6 +88,12 @@ protected String from(String... indexName) {
 
     @Override
     public Map<String, Object> runEsql(RestEsqlTestCase.RequestObjectBuilder requestObject) throws IOException {
+        if (requestObject.allowPartialResults() != null) {
+            assumeTrue(
+                "require allow_partial_results on local cluster",
+                clusterHasCapability("POST", "/_query", List.of(), List.of("support_partial_results")).orElse(false)
+            );
+        }
         requestObject.includeCCSMetadata(true);
         return super.runEsql(requestObject);
     }
diff --git a/x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/RestEsqlIT.java b/x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/RestEsqlIT.java
@@ -111,7 +111,7 @@ public void testInvalidPragma() throws IOException {
             request.setJsonEntity("{\"f\":" + i + "}");
             assertOK(client().performRequest(request));
         }
-        RequestObjectBuilder builder = requestObjectBuilder().query("from test-index | limit 1 | keep f");
+        RequestObjectBuilder builder = requestObjectBuilder().query("from test-index | limit 1 | keep f").allowPartialResults(false);
         builder.pragmas(Settings.builder().put("data_partitioning", "invalid-option").build());
         ResponseException re = expectThrows(ResponseException.class, () -> runEsqlSync(builder));
         assertThat(EntityUtils.toString(re.getResponse().getEntity()), containsString("No enum constant"));
diff --git a/x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/RestEsqlTestCase.java b/x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/RestEsqlTestCase.java
@@ -131,7 +131,7 @@ public static class RequestObjectBuilder {
         private Boolean includeCCSMetadata = null;
 
         private CheckedConsumer<XContentBuilder, IOException> filter;
-        private Boolean allPartialResults = null;
+        private Boolean allowPartialResults = null;
 
         public RequestObjectBuilder() throws IOException {
             this(randomFrom(XContentType.values()));
@@ -209,11 +209,15 @@ public RequestObjectBuilder filter(CheckedConsumer<XContentBuilder, IOException>
             return this;
         }
 
-        public RequestObjectBuilder allPartialResults(boolean allPartialResults) {
-            this.allPartialResults = allPartialResults;
+        public RequestObjectBuilder allowPartialResults(boolean allowPartialResults) {
+            this.allowPartialResults = allowPartialResults;
             return this;
         }
 
+        public Boolean allowPartialResults() {
+            return allowPartialResults;
+        }
+
         public RequestObjectBuilder build() throws IOException {
             if (isBuilt == false) {
                 if (tables != null) {
@@ -1376,8 +1380,8 @@ protected static Request prepareRequestWithOptions(RequestObjectBuilder requestO
         requestObject.build();
         Request request = prepareRequest(mode);
         String mediaType = attachBody(requestObject, request);
-        if (requestObject.allPartialResults != null) {
-            request.addParameter("allow_partial_results", String.valueOf(requestObject.allPartialResults));
+        if (requestObject.allowPartialResults != null) {
+            request.addParameter("allow_partial_results", String.valueOf(requestObject.allowPartialResults));
         }
 
         RequestOptions.Builder options = request.getOptions().toBuilder();
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractCrossClusterTestCase.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractCrossClusterTestCase.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.json.JsonXContent;
+import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.junit.After;
 import org.junit.Before;
 
@@ -76,6 +77,11 @@ protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
         return plugins;
     }
 
+    @Override
+    protected Settings nodeSettings() {
+        return Settings.builder().put(super.nodeSettings()).put(EsqlPlugin.QUERY_ALLOW_PARTIAL_RESULTS.getKey(), false).build();
+    }
+
     public static class InternalExchangePlugin extends Plugin {
         @Override
         public List<Setting<?>> getSettings() {
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractEsqlIntegTestCase.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/AbstractEsqlIntegTestCase.java
@@ -139,6 +139,14 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
         return CollectionUtils.appendToCopy(super.nodePlugins(), EsqlPlugin.class);
     }
 
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal, otherSettings))
+            .put(EsqlPlugin.QUERY_ALLOW_PARTIAL_RESULTS.getKey(), false)
+            .build();
+    }
+
     protected void setRequestCircuitBreakerLimit(ByteSizeValue limit) {
         if (limit != null) {
             assertAcked(
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterCancellationIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterCancellationIT.java
@@ -14,25 +14,20 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.WriteRequest;
-import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.compute.operator.DriverTaskRunner;
 import org.elasticsearch.compute.operator.exchange.ExchangeService;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.tasks.TaskInfo;
-import org.elasticsearch.test.AbstractMultiClustersTestCase;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.plugin.ComputeService;
-import org.junit.After;
-import org.junit.Before;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
@@ -44,7 +39,7 @@
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
 
-public class CrossClusterCancellationIT extends AbstractMultiClustersTestCase {
+public class CrossClusterCancellationIT extends AbstractCrossClusterTestCase {
     private static final String REMOTE_CLUSTER = "cluster-a";
 
     @Override
@@ -53,35 +48,11 @@ protected List<String> remoteClusterAlias() {
     }
 
     @Override
-    protected Collection<Class<? extends Plugin>> nodePlugins(String clusterAlias) {
-        List<Class<? extends Plugin>> plugins = new ArrayList<>(super.nodePlugins(clusterAlias));
-        plugins.add(EsqlPluginWithEnterpriseOrTrialLicense.class);
-        plugins.add(InternalExchangePlugin.class);
-        plugins.add(SimplePauseFieldPlugin.class);
-        return plugins;
-    }
-
-    public static class InternalExchangePlugin extends Plugin {
-        @Override
-        public List<Setting<?>> getSettings() {
-            return List.of(
-                Setting.timeSetting(
-                    ExchangeService.INACTIVE_SINKS_INTERVAL_SETTING,
-                    TimeValue.timeValueMillis(between(3000, 4000)),
-                    Setting.Property.NodeScope
-                )
-            );
-        }
-    }
-
-    @Before
-    public void resetPlugin() {
-        SimplePauseFieldPlugin.resetPlugin();
-    }
-
-    @After
-    public void releasePlugin() {
-        SimplePauseFieldPlugin.release();
+    protected Settings nodeSettings() {
+        return Settings.builder()
+            .put(super.nodeSettings())
+            .put(ExchangeService.INACTIVE_SINKS_INTERVAL_SETTING, TimeValue.timeValueMillis(between(3000, 4000)))
+            .build();
     }
 
     @Override
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/ManyShardsIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/ManyShardsIT.java
@@ -71,6 +71,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
     @Override
     protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
         return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal, otherSettings))
             .put(ExchangeService.INACTIVE_SINKS_INTERVAL_SETTING, TimeValue.timeValueMillis(between(3000, 5000)))
             .build();
     }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java
@@ -7,7 +7,6 @@
 
 package org.elasticsearch.xpack.esql.plugin;
 
-import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.OriginalIndices;
@@ -17,7 +16,6 @@
 import org.elasticsearch.compute.operator.exchange.ExchangeSourceHandler;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskCancelledException;
@@ -90,18 +88,12 @@ void startComputeOnRemoteCluster(
             if (receivedResults == false && EsqlCCSUtils.shouldIgnoreRuntimeError(executionInfo, clusterAlias, e)) {
                 EsqlCCSUtils.markClusterWithFinalStateAndNoShards(executionInfo, clusterAlias, EsqlExecutionInfo.Cluster.Status.SKIPPED, e);
                 l.onResponse(DriverCompletionInfo.EMPTY);
-            } else if (configuration.allowPartialResults()
-                && (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) == false) {
-                    EsqlCCSUtils.markClusterWithFinalStateAndNoShards(
-                        executionInfo,
-                        clusterAlias,
-                        EsqlExecutionInfo.Cluster.Status.PARTIAL,
-                        e
-                    );
-                    l.onResponse(DriverCompletionInfo.EMPTY);
-                } else {
-                    l.onFailure(e);
-                }
+            } else if (configuration.allowPartialResults() && EsqlCCSUtils.canAllowPartial(e)) {
+                EsqlCCSUtils.markClusterWithFinalStateAndNoShards(executionInfo, clusterAlias, EsqlExecutionInfo.Cluster.Status.PARTIAL, e);
+                l.onResponse(DriverCompletionInfo.EMPTY);
+            } else {
+                l.onFailure(e);
+            }
         });
         ExchangeService.openExchange(
             transportService,
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java
@@ -7,7 +7,6 @@
 
 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;
@@ -31,7 +30,6 @@
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.Tuple;
-import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.logging.LogManager;
 import org.elasticsearch.logging.Logger;
@@ -63,6 +61,7 @@
 import org.elasticsearch.xpack.esql.planner.LocalExecutionPlanner;
 import org.elasticsearch.xpack.esql.planner.PlannerUtils;
 import org.elasticsearch.xpack.esql.session.Configuration;
+import org.elasticsearch.xpack.esql.session.EsqlCCSUtils;
 import org.elasticsearch.xpack.esql.session.Result;
 
 import java.util.ArrayList;
@@ -433,8 +432,7 @@ public void executePlan(
                                 );
                                 dataNodesListener.onResponse(r.getCompletionInfo());
                             }, e -> {
-                                if (configuration.allowPartialResults()
-                                    && (ExceptionsHelper.unwrapCause(e) instanceof IndexNotFoundException) == false) {
+                                if (configuration.allowPartialResults() && EsqlCCSUtils.canAllowPartial(e)) {
                                     execInfo.swapCluster(
                                         LOCAL_CLUSTER,
                                         (k, v) -> new EsqlExecutionInfo.Cluster.Builder(v).setStatus(
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlPlugin.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlPlugin.java
@@ -107,7 +107,7 @@ public class EsqlPlugin extends Plugin implements ActionPlugin {
 
     public static final Setting<Boolean> QUERY_ALLOW_PARTIAL_RESULTS = Setting.boolSetting(
         "esql.query.allow_partial_results",
-        false,
+        true,
         Setting.Property.NodeScope,
         Setting.Property.Dynamic
     );
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlCCSUtils.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlCCSUtils.java
@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.session;
 
+import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.OriginalIndices;
@@ -17,6 +18,7 @@
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.compute.operator.DriverCompletionInfo;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.indices.IndicesExpressionGrouper;
 import org.elasticsearch.license.XPackLicenseState;
@@ -368,4 +370,16 @@ public static boolean shouldIgnoreRuntimeError(EsqlExecutionInfo executionInfo,
 
         return ExceptionsHelper.isRemoteUnavailableException(e);
     }
+
+    /**
+     * Check whether this exception can be tolerated when partial results are on, or should be treated as fatal.
+     * @return true if the exception can be tolerated, false if it should be treated as fatal.
+     */
+    public static boolean canAllowPartial(Exception e) {
+        Throwable unwrapped = ExceptionsHelper.unwrapCause(e);
+        if (unwrapped instanceof IndexNotFoundException || unwrapped instanceof ElasticsearchSecurityException) {
+            return false;
+        }
+        return true;
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -83,6 +83,6 @@ private RestClient remoteClusterClient() throws IOException {`
`83`	`83`
`84`	`84`	`@Before`
`85`	`85`	`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));`
`87`	`87`	`}`
`88`	`88`	`}`
Original file line number	Diff line number	Diff line change
`@@ -111,7 +111,7 @@ public void testInvalidPragma() throws IOException {`
`111`	`111`	`request.setJsonEntity("{\"f\":" + i + "}");`
`112`	`112`	`assertOK(client().performRequest(request));`
`113`	`113`	`}`
`114`		`- RequestObjectBuilder builder = requestObjectBuilder().query("from test-index \| limit 1 \| keep f");`
	`114`	`+ RequestObjectBuilder builder = requestObjectBuilder().query("from test-index \| limit 1 \| keep f").allowPartialResults(false);`
`115`	`115`	`builder.pragmas(Settings.builder().put("data_partitioning", "invalid-option").build());`
`116`	`116`	`ResponseException re = expectThrows(ResponseException.class, () -> runEsqlSync(builder));`
`117`	`117`	`assertThat(EntityUtils.toString(re.getResponse().getEntity()), containsString("No enum constant"));`