ES|QL async queries: Partial result on demand (#118122)

Add capability to stop async query on demand
The theory:

- User initiates async search request
- User sends the stop request (POST _query/async/<ID>/stop)
- If the async is finished by that time, it's like regular async get
- If it's not finished, the sinks are closed and the request is forcefully finished

Author: smalyshev

docs/changelog/118122.yaml

@@ -0,0 +1,5 @@
+pr: 118122
+summary: "ES|QL: Partial result on demand for async queries"
+area: ES|QL
+type: enhancement
+issues: []

docs/reference/esql/esql-across-clusters.asciidoc

@@ -210,6 +210,7 @@ Which returns:
 {
   "is_running": false,
   "took": 42,  <1>
+  "is_partial": false, <7>
   "columns" : [
     {
       "name" : "COUNT(http.response.status_code)",
@@ -275,8 +276,9 @@ Which returns:
 <2> This section of counters shows all possible cluster search states and how many cluster
 searches are currently in that state. The clusters can have one of the following statuses: *running*,
 *successful* (searches on all shards were successful), *skipped* (the search
-failed on a cluster marked with `skip_unavailable`=`true`) or *failed* (the search
-failed on a cluster marked with `skip_unavailable`=`false`).
+failed on a cluster marked with `skip_unavailable`=`true`), *failed* (the search
+failed on a cluster marked with `skip_unavailable`=`false`) or **partial** (the search was
+<<esql-async-query-stop-api, interrupted>> before finishing).
 <3> The `_clusters/details` section shows metadata about the search on each cluster.
 <4> If you included indices from the local cluster you sent the request to in your {ccs},
 it is identified as "(local)".
@@ -285,6 +287,8 @@ which clusters have slower response times than others.
 <6> The shard details for the search on that cluster, including a count of shards that were
 skipped due to the can-match phase results. Shards are skipped when they cannot have any matching data
 and therefore are not included in the full ES|QL query.
+<7> The `is_partial` field is set to `true` if the search has partial results for any reason,
+for example if it was interrupted before finishing using the <<esql-async-query-stop-api,async query stop API>>.
 
 
 The cross-cluster metadata can be used to determine whether any data came back from a cluster.
@@ -314,6 +318,7 @@ Which returns:
 {
   "is_running": false,
   "took": 55,
+  "is_partial": false,
   "columns": [
      ... // not shown
   ],

docs/reference/esql/esql-apis.asciidoc

@@ -17,6 +17,7 @@ overview of {esql} and related tutorials, see <<esql>>.
 * <<esql-async-query-api>>
 * <<esql-async-query-get-api>>
 * <<esql-async-query-delete-api>>
+* <<esql-async-query-stop-api>>
 
 
 include::esql-query-api.asciidoc[]
@@ -26,3 +27,5 @@ include::esql-async-query-api.asciidoc[]
 include::esql-async-query-get-api.asciidoc[]
 
 include::esql-async-query-delete-api.asciidoc[]
+
+include::esql-async-query-stop-api.asciidoc[]

docs/reference/esql/esql-async-query-api.asciidoc

@@ -170,3 +170,10 @@ API>> to get the current status and available results for the query.
 (Boolean)
 If `true`, the query request is still executing.
 --
+
+`is_partial`::
++
+--
+(Boolean)
+If `true`, the query has partial results - for example, as a result of using the <<esql-async-query-stop-api, async query stop API>>.
+--

docs/reference/esql/esql-async-query-stop-api.asciidoc

@@ -0,0 +1,49 @@
+[[esql-async-query-stop-api]]
+=== {esql} async query stop API
+++++
+<titleabbrev>{esql} async query stop API</titleabbrev>
+++++
+
+.New API reference
+[sidebar]
+--
+For the most up-to-date API details, refer to {api-es}/group/endpoint-esql[ES|QL APIs].
+--
+
+The <<esql,{esql}>> async query stop API is used to manually stop an async query. Once the stop command is issued,
+the query stops processing new data and returns the results that have been already processed. Note that due to the pipelined
+nature of {esql} queries, the stop operation is not immediate and may take time to return results.
+
+The results are returned in <<esql-query-api-response-body,the same format>> as the
+<<esql-async-query-get-api,{esql} async query get API>>.
+If the query has been finished by the time the stop command is issued, the results are returned immediately.
+
+If the query processing has not finished by the time the stop command is issued, the response will have the `is_partial`
+field set to `true`.
+
+[source,console]
+----
+POST /query/async/FkpMRkJGS1gzVDRlM3g4ZzMyRGlLbkEaTXlJZHdNT09TU2VTZVBoNDM3cFZMUToxMDM=/stop
+----
+// TEST[skip: no access to query ID]
+
+[[esql-async-query-stop-api-request]]
+==== {api-request-title}
+
+`POST /_query/async/<query_id>/stop`
+
+[[esql-async-query-stop-api-prereqs]]
+==== {api-prereq-title}
+
+* If the {es} {security-features} are enabled, only the authenticated user that submitted the original query request
+can stop the query.
+
+[[esql-async-query-stop-api-path-params]]
+==== {api-path-parms-title}
+
+`<query_id>`::
+(Required, string)
+Identifier for the query to stop.
++
+A query ID is provided in the <<esql-async-query-api,{esql} async query API>>'s
+response for a query that does not complete in the awaited time.

docs/reference/esql/esql-rest.asciidoc

@@ -193,6 +193,7 @@ Which returns:
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     {"name": "author", "type": "text"},
     {"name": "name", "type": "text"},

docs/reference/esql/multivalued-fields.asciidoc

@@ -27,6 +27,7 @@ Multivalued fields come back as a JSON array:
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     { "name": "a", "type": "long"},
     { "name": "b", "type": "long"}
@@ -78,6 +79,7 @@ And {esql} sees that removal:
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     { "name": "a", "type": "long"},
     { "name": "b", "type": "keyword"}
@@ -122,6 +124,7 @@ And {esql} also sees that:
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     { "name": "a", "type": "long"},
     { "name": "b", "type": "long"}
@@ -165,6 +168,7 @@ POST /_query
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     { "name": "a", "type": "long"},
     { "name": "b", "type": "keyword"}
@@ -198,6 +202,7 @@ POST /_query
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     { "name": "a", "type": "long"},
   ],
@@ -241,6 +246,7 @@ POST /_query
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     { "name": "a",   "type": "long"},
     { "name": "b",   "type": "long"},
@@ -278,6 +284,7 @@ POST /_query
 ----
 {
   "took": 28,
+  "is_partial": false,
   "columns": [
     { "name": "a",   "type": "long"},
     { "name": "b",   "type": "long"},

rest-api-spec/src/main/resources/rest-api-spec/api/esql.async_query_stop.json

@@ -0,0 +1,31 @@
+{
+  "esql.async_query_stop": {
+    "documentation": {
+      "url": "https://www.elastic.co/guide/en/elasticsearch/reference/master/esql-async-query-stop-api.html",
+      "description": "Stops a previously submitted async query request given its ID and collects the results."
+    },
+    "stability": "stable",
+    "visibility": "public",
+    "headers": {
+      "accept": [
+        "application/json"
+      ]
+    },
+    "url": {
+      "paths": [
+        {
+          "path": "/_query/async/{id}/stop",
+          "methods": [
+            "POST"
+          ],
+          "parts": {
+            "id": {
+              "type": "string",
+              "description": "The async query ID"
+            }
+          }
+        }
+      ]
+    }
+  }
+}

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -163,6 +163,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion RESOLVE_CLUSTER_NO_INDEX_EXPRESSION = def(8_829_00_0);
     public static final TransportVersion ML_ROLLOVER_LEGACY_INDICES = def(8_830_00_0);
     public static final TransportVersion ADD_INCLUDE_FAILURE_INDICES_OPTION = def(8_831_00_0);
+    public static final TransportVersion ESQL_RESPONSE_PARTIAL = def(8_832_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -86,7 +86,7 @@ public void skipOnAborted() {
     public void testSortByManyLongsSuccess() throws IOException {
         initManyLongs();
         Response response = sortByManyLongs(500);
-        Map<?, ?> map = responseAsMap(response);
+        Map<String, Object> map = responseAsMap(response);
         ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long"))
             .item(matchesMap().entry("name", "b").entry("type", "long"));
         ListMatcher values = matchesList();
@@ -95,8 +95,7 @@ public void testSortByManyLongsSuccess() throws IOException {
                 values = values.item(List.of(0, b));
             }
         }
-        MapMatcher mapMatcher = matchesMap();
-        assertMap(map, mapMatcher.entry("columns", columns).entry("values", values).entry("took", greaterThanOrEqualTo(0)));
+        assertResultMap(map, columns, values);
     }
 
     /**
@@ -236,11 +235,10 @@ private StringBuilder makeSortByManyLongs(int count) {
     public void testGroupOnSomeLongs() throws IOException {
         initManyLongs();
         Response resp = groupOnManyLongs(200);
-        Map<?, ?> map = responseAsMap(resp);
+        Map<String, Object> map = responseAsMap(resp);
         ListMatcher columns = matchesList().item(matchesMap().entry("name", "MAX(a)").entry("type", "long"));
         ListMatcher values = matchesList().item(List.of(9));
-        MapMatcher mapMatcher = matchesMap();
-        assertMap(map, mapMatcher.entry("columns", columns).entry("values", values).entry("took", greaterThanOrEqualTo(0)));
+        assertResultMap(map, columns, values);
     }
 
     /**
@@ -249,11 +247,10 @@ public void testGroupOnSomeLongs() throws IOException {
     public void testGroupOnManyLongs() throws IOException {
         initManyLongs();
         Response resp = groupOnManyLongs(5000);
-        Map<?, ?> map = responseAsMap(resp);
+        Map<String, Object> map = responseAsMap(resp);
         ListMatcher columns = matchesList().item(matchesMap().entry("name", "MAX(a)").entry("type", "long"));
         ListMatcher values = matchesList().item(List.of(9));
-        MapMatcher mapMatcher = matchesMap();
-        assertMap(map, mapMatcher.entry("columns", columns).entry("values", values).entry("took", greaterThanOrEqualTo(0)));
+        assertResultMap(map, columns, values);
     }
 
     private Response groupOnManyLongs(int count) throws IOException {
@@ -279,12 +276,11 @@ private StringBuilder makeManyLongs(int count) {
     public void testSmallConcat() throws IOException {
         initSingleDocIndex();
         Response resp = concat(2);
-        Map<?, ?> map = responseAsMap(resp);
+        Map<String, Object> map = responseAsMap(resp);
         ListMatcher columns = matchesList().item(matchesMap().entry("name", "a").entry("type", "long"))
             .item(matchesMap().entry("name", "str").entry("type", "keyword"));
         ListMatcher values = matchesList().item(List.of(1, "1".repeat(100)));
-        MapMatcher mapMatcher = matchesMap();
-        assertMap(map, mapMatcher.entry("columns", columns).entry("values", values).entry("took", greaterThanOrEqualTo(0)));
+        assertResultMap(map, columns, values);
     }
 
     public void testHugeConcat() throws IOException {
@@ -465,7 +461,7 @@ private void assertManyStrings(Response resp, int strings) throws IOException {
     public void testManyEval() throws IOException {
         initManyLongs();
         Response resp = manyEval(1);
-        Map<?, ?> map = responseAsMap(resp);
+        Map<String, Object> map = responseAsMap(resp);
         ListMatcher columns = matchesList();
         columns = columns.item(matchesMap().entry("name", "a").entry("type", "long"));
         columns = columns.item(matchesMap().entry("name", "b").entry("type", "long"));
@@ -475,8 +471,7 @@ public void testManyEval() throws IOException {
         for (int i = 0; i < 20; i++) {
             columns = columns.item(matchesMap().entry("name", "i0" + i).entry("type", "long"));
         }
-        MapMatcher mapMatcher = matchesMap();
-        assertMap(map, mapMatcher.entry("columns", columns).entry("values", hasSize(10_000)).entry("took", greaterThanOrEqualTo(0)));
+        assertResultMap(map, columns, hasSize(10_000));
     }
 
     public void testTooManyEval() throws IOException {