ESQL: Make esql version required in REST requests

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

@@ -56,12 +56,13 @@
  * crash Elasticsearch.
  */
 public class HeapAttackIT extends ESRestTestCase {
-
     @ClassRule
     public static ElasticsearchCluster cluster = Clusters.buildCluster();
 
     static volatile boolean SUITE_ABORTED = false;
 
+    private static String ESQL_VERSION = "2024.04.01";
+
     @Override
     protected String getTestRestCluster() {
         return cluster.getHttpAddresses();
@@ -155,8 +156,8 @@ private Response groupOnManyLongs(int count) throws IOException {
     }
 
     private StringBuilder makeManyLongs(int count) {
-        StringBuilder query = new StringBuilder();
-        query.append("{\"query\":\"FROM manylongs\\n| EVAL i0 = a + b, i1 = b + i0");
+        StringBuilder query = startQueryWithVersion(ESQL_VERSION);
+        query.append("FROM manylongs\\n| EVAL i0 = a + b, i1 = b + i0");
         for (int i = 2; i < count; i++) {
             query.append(", i").append(i).append(" = i").append(i - 2).append(" + ").append(i - 1);
         }
@@ -186,8 +187,8 @@ public void testHugeConcat() throws IOException {
     }
 
     private Response concat(int evals) throws IOException {
-        StringBuilder query = new StringBuilder();
-        query.append("{\"query\":\"FROM single | EVAL str = TO_STRING(a)");
+        StringBuilder query = startQueryWithVersion(ESQL_VERSION);
+        query.append("FROM single | EVAL str = TO_STRING(a)");
         for (int e = 0; e < evals; e++) {
             query.append("\n| EVAL str=CONCAT(")
                 .append(IntStream.range(0, 10).mapToObj(i -> "str").collect(Collectors.joining(", ")))
@@ -223,8 +224,8 @@ public void testHugeManyConcat() throws IOException {
      * Tests that generate many moderately long strings.
      */
     private Response manyConcat(int strings) throws IOException {
-        StringBuilder query = new StringBuilder();
-        query.append("{\"query\":\"FROM manylongs | EVAL str = CONCAT(");
+        StringBuilder query = startQueryWithVersion(ESQL_VERSION);
+        query.append("FROM manylongs | EVAL str = CONCAT(");
         query.append(
             Arrays.stream(new String[] { "a", "b", "c", "d", "e" })
                 .map(f -> "TO_STRING(" + f + ")")
@@ -274,8 +275,8 @@ public void testTooManyEval() throws IOException {
     }
 
     private Response manyEval(int evalLines) throws IOException {
-        StringBuilder query = new StringBuilder();
-        query.append("{\"query\":\"FROM manylongs");
+        StringBuilder query = startQueryWithVersion(ESQL_VERSION);
+        query.append("FROM manylongs");
         for (int e = 0; e < evalLines; e++) {
             query.append("\n| EVAL ");
             for (int i = 0; i < 10; i++) {
@@ -356,7 +357,9 @@ public void testFetchTooManyBigFields() throws IOException {
      * Fetches documents containing 1000 fields which are {@code 1kb} each.
      */
     private void fetchManyBigFields(int docs) throws IOException {
-        Response response = query("{\"query\": \"FROM manybigfields | SORT f000 | LIMIT " + docs + "\"}", "columns");
+        StringBuilder query = startQueryWithVersion(ESQL_VERSION);
+        query.append("FROM manybigfields | SORT f000 | LIMIT " + docs + "\"}");
+        Response response = query(query.toString(), "columns");
         Map<?, ?> map = responseAsMap(response);
         ListMatcher columns = matchesList();
         for (int f = 0; f < 1000; f++) {
@@ -383,11 +386,12 @@ public void testAggTooManyMvLongs() throws IOException {
     }
 
     private Response aggMvLongs(int fields) throws IOException {
-        StringBuilder builder = new StringBuilder("{\"query\": \"FROM mv_longs | STATS MAX(f00) BY f00");
+        StringBuilder query = startQueryWithVersion(ESQL_VERSION);
+        query.append("FROM mv_longs | STATS MAX(f00) BY f00");
         for (int f = 1; f < fields; f++) {
-            builder.append(", f").append(String.format(Locale.ROOT, "%02d", f));
+            query.append(", f").append(String.format(Locale.ROOT, "%02d", f));
         }
-        return query(builder.append("\"}").toString(), "columns");
+        return query(query.append("\"}").toString(), "columns");
     }
 
     public void testFetchMvLongs() throws IOException {
@@ -408,7 +412,9 @@ public void testFetchTooManyMvLongs() throws IOException {
     }
 
     private Response fetchMvLongs() throws IOException {
-        return query("{\"query\": \"FROM mv_longs\"}", "columns");
+        StringBuilder query = startQueryWithVersion(ESQL_VERSION);
+        query.append("FROM mv_longs\"}");
+        return query(query.toString(), "columns");
     }
 
     private void initManyLongs() throws IOException {
@@ -576,4 +582,12 @@ public void assertRequestBreakerEmpty() throws Exception {
             }
         });
     }
+
+    private static StringBuilder startQueryWithVersion(String version) {
+        StringBuilder query = new StringBuilder();
+        query.append("{\"version\":\"" + version + "\",");
+        query.append("\"query\":\"");
+
+        return query;
+    }
 }

test/yaml-rest-runner/src/main/java/org/elasticsearch/test/rest/yaml/ImpersonateOfficialClientTestClient.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.test.rest.yaml;
+
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.NodeSelector;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.common.CheckedSupplier;
+import org.elasticsearch.test.rest.yaml.restspec.ClientYamlSuiteRestApi;
+import org.elasticsearch.test.rest.yaml.restspec.ClientYamlSuiteRestSpec;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiPredicate;
+
+/**
+ * Impersonates an official test client by setting the @{code x-elastic-client-meta} header.
+ */
+public class ImpersonateOfficialClientTestClient extends ClientYamlTestClient {
+    private final String meta;
+
+    public ImpersonateOfficialClientTestClient(
+        ClientYamlSuiteRestSpec restSpec,
+        RestClient restClient,
+        List<HttpHost> hosts,
+        CheckedSupplier<RestClientBuilder, IOException> clientBuilderWithSniffedNodes,
+        String meta
+    ) {
+        super(restSpec, restClient, hosts, clientBuilderWithSniffedNodes);
+        this.meta = meta;
+    }
+
+    @Override
+    public ClientYamlTestResponse callApi(
+        String apiName,
+        Map<String, String> params,
+        HttpEntity entity,
+        Map<String, String> headers,
+        NodeSelector nodeSelector,
+        BiPredicate<ClientYamlSuiteRestApi, ClientYamlSuiteRestApi.Path> pathPredicate
+    ) throws IOException {
+        headers.put("x-elastic-client-meta", meta);
+        return super.callApi(apiName, params, entity, headers, nodeSelector, pathPredicate);
+    }
+}

x-pack/plugin/esql/qa/security/src/javaRestTest/java/org/elasticsearch/xpack/esql/EsqlAsyncSecurityIT.java

@@ -90,6 +90,7 @@ private Response runAsync(String user, String command) throws IOException {
         }
         XContentBuilder json = JsonXContent.contentBuilder();
         json.startObject();
+        json.field("version", ESQL_VERSION);
         json.field("query", command);
         addRandomPragmas(json);
         json.field("wait_for_completion_timeout", timeValueNanos(randomIntBetween(1, 1000)));

x-pack/plugin/esql/qa/security/src/javaRestTest/java/org/elasticsearch/xpack/esql/EsqlSecurityIT.java

@@ -38,6 +38,7 @@
 import static org.hamcrest.Matchers.equalTo;
 
 public class EsqlSecurityIT extends ESRestTestCase {
+    static String ESQL_VERSION = "2024.04.01.🚀";
 
     @ClassRule
     public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
@@ -354,6 +355,7 @@ protected Response runESQLCommand(String user, String command) throws IOExceptio
         }
         XContentBuilder json = JsonXContent.contentBuilder();
         json.startObject();
+        json.field("version", ESQL_VERSION);
         json.field("query", command);
         addRandomPragmas(json);
         json.endObject();

x-pack/plugin/esql/qa/server/mixed-cluster/build.gradle

@@ -25,20 +25,27 @@ dependencies {
 
 GradleUtils.extendSourceSet(project, "javaRestTest", "yamlRestTest")
 
+// ESQL is available in 8.11 or later
 def supportedVersion = bwcVersion -> {
-  // ESQL is available in 8.11 or later
   return bwcVersion.onOrAfter(Version.fromString("8.11.0"));
 }
 
+// Versions on and after 8.13.3 will get a `version` parameter
+def versionUnsupported = bwcVersion -> {
+  return bwcVersion.before(Version.fromString("8.13.3"));
+}
+
 BuildParams.bwcVersions.withWireCompatible(supportedVersion) { bwcVersion, baseName ->
   def javaRestTest = tasks.register("v${bwcVersion}#javaRestTest", StandaloneRestIntegTestTask) {
     usesBwcDistribution(bwcVersion)
     systemProperty("tests.old_cluster_version", bwcVersion)
+    systemProperty("tests.version_parameter_unsupported", versionUnsupported(bwcVersion))
   }
 
   def yamlRestTest = tasks.register("v${bwcVersion}#yamlRestTest", StandaloneRestIntegTestTask) {
     usesBwcDistribution(bwcVersion)
     systemProperty("tests.old_cluster_version", bwcVersion)
+    systemProperty("tests.version_parameter_unsupported", versionUnsupported(bwcVersion))
     testClassesDirs = sourceSets.yamlRestTest.output.classesDirs
     classpath = sourceSets.yamlRestTest.runtimeClasspath
   }

x-pack/plugin/esql/qa/server/mixed-cluster/src/yamlRestTest/java/org/elasticsearch/xpack/esql/qa/mixed/EsqlClientYamlIT.java

@@ -9,14 +9,28 @@
 
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.RestClient;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
+import org.elasticsearch.test.rest.yaml.ClientYamlTestClient;
 import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+import org.elasticsearch.test.rest.yaml.ImpersonateOfficialClientTestClient;
+import org.elasticsearch.test.rest.yaml.restspec.ClientYamlSuiteRestSpec;
+import org.elasticsearch.test.rest.yaml.section.ApiCallSection;
+import org.elasticsearch.test.rest.yaml.section.ClientYamlTestSection;
+import org.elasticsearch.test.rest.yaml.section.DoSection;
+import org.elasticsearch.test.rest.yaml.section.ExecutableSection;
 import org.elasticsearch.xpack.esql.qa.rest.EsqlSpecTestCase;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.ClassRule;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+
 public class EsqlClientYamlIT extends ESClientYamlSuiteTestCase {
     @ClassRule
     public static ElasticsearchCluster cluster = Clusters.mixedVersionCluster();
@@ -32,6 +46,9 @@ public EsqlClientYamlIT(final ClientYamlTestCandidate testCandidate) {
 
     @ParametersFactory
     public static Iterable<Object[]> parameters() throws Exception {
+        if (EsqlSpecTestCase.availableVersions().isEmpty()) {
+            return updateEsqlQueryDoSections(createParameters(), EsqlClientYamlIT::stripVersion);
+        }
         return createParameters();
     }
 
@@ -40,4 +57,63 @@ public static Iterable<Object[]> parameters() throws Exception {
     public void assertRequestBreakerEmpty() throws Exception {
         EsqlSpecTestCase.assertRequestBreakerEmpty();
     }
+
+    @Override
+    protected ClientYamlTestClient initClientYamlTestClient(
+        final ClientYamlSuiteRestSpec restSpec,
+        final RestClient restClient,
+        final List<HttpHost> hosts
+    ) {
+        if (EsqlSpecTestCase.availableVersions().isEmpty()) {
+            return new ImpersonateOfficialClientTestClient(restSpec, restClient, hosts, this::getClientBuilderWithSniffedHosts, "es=8.13");
+        }
+        return super.initClientYamlTestClient(restSpec, restClient, hosts);
+    }
+
+    static DoSection stripVersion(DoSection doSection) {
+        ApiCallSection copy = doSection.getApiCallSection().copyWithNewApi(doSection.getApiCallSection().getApi());
+        for (Map<String, Object> body : copy.getBodies()) {
+            body.remove("version");
+        }
+        doSection.setApiCallSection(copy);
+        return doSection;
+    }
+
+    // TODO: refactor, copied from single-node's AbstractEsqlClientYamlIt
+    public static Iterable<Object[]> updateEsqlQueryDoSections(Iterable<Object[]> parameters, Function<DoSection, ExecutableSection> modify)
+        throws Exception {
+        List<Object[]> result = new ArrayList<>();
+        for (Object[] orig : parameters) {
+            assert orig.length == 1;
+            ClientYamlTestCandidate candidate = (ClientYamlTestCandidate) orig[0];
+            try {
+                ClientYamlTestSection modified = new ClientYamlTestSection(
+                    candidate.getTestSection().getLocation(),
+                    candidate.getTestSection().getName(),
+                    candidate.getTestSection().getPrerequisiteSection(),
+                    candidate.getTestSection().getExecutableSections().stream().map(e -> modifyExecutableSection(e, modify)).toList()
+                );
+                result.add(new Object[] { new ClientYamlTestCandidate(candidate.getRestTestSuite(), modified) });
+            } catch (IllegalArgumentException e) {
+                throw new IllegalArgumentException("error modifying " + candidate + ": " + e.getMessage(), e);
+            }
+        }
+        return result;
+    }
+
+    // TODO: refactor, copied from single-node's AbstractEsqlClientYamlIt
+    private static ExecutableSection modifyExecutableSection(ExecutableSection e, Function<DoSection, ExecutableSection> modify) {
+        if (false == (e instanceof DoSection)) {
+            return e;
+        }
+        DoSection doSection = (DoSection) e;
+        String api = doSection.getApiCallSection().getApi();
+        return switch (api) {
+            case "esql.query" -> modify.apply(doSection);
+            // case "esql.async_query", "esql.async_query_get" -> throw new IllegalArgumentException(
+            // "The esql yaml tests can't contain async_query or async_query_get because we modify them on the fly and *add* those."
+            // );
+            default -> e;
+        };
+    }
 }

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

@@ -18,6 +18,7 @@
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
 import org.elasticsearch.test.rest.ESRestTestCase;
 import org.elasticsearch.test.rest.TestFeatureService;
+import org.elasticsearch.xpack.esql.EsqlTestUtils;
 import org.elasticsearch.xpack.esql.qa.rest.RestEsqlTestCase;
 import org.junit.After;
 import org.junit.Before;
@@ -122,7 +123,9 @@ void indexDocs(RestClient client, String index, List<Doc> docs) throws IOExcepti
     }
 
     private Map<String, Object> run(String query) throws IOException {
-        Map<String, Object> resp = runEsql(new RestEsqlTestCase.RequestObjectBuilder().query(query).build());
+        Map<String, Object> resp = runEsql(
+            new RestEsqlTestCase.RequestObjectBuilder().query(query).version(EsqlTestUtils.latestEsqlVersionOrSnapshot()).build()
+        );
         logger.info("--> query {} response {}", query, resp);
         return resp;
     }

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

@@ -69,7 +69,7 @@ public void testBasicEsql() throws IOException {
         Response response = client().performRequest(bulk);
         Assert.assertEquals("{\"errors\":false}", EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8));
 
-        RequestObjectBuilder builder = new RequestObjectBuilder().query(fromIndex() + " | stats avg(value)");
+        RequestObjectBuilder builder = requestObjectBuilder().query(fromIndex() + " | stats avg(value)");
         if (Build.current().isSnapshot()) {
             builder.pragmas(Settings.builder().put("data_partitioning", "shard").build());
         }
@@ -89,7 +89,7 @@ public void testInvalidPragma() throws IOException {
             request.setJsonEntity("{\"f\":" + i + "}");
             assertOK(client().performRequest(request));
         }
-        RequestObjectBuilder builder = new RequestObjectBuilder().query("from test-index | limit 1 | keep f");
+        RequestObjectBuilder builder = requestObjectBuilder().query("from test-index | limit 1 | keep f");
         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"));
@@ -99,7 +99,7 @@ public void testInvalidPragma() throws IOException {
 
     public void testPragmaNotAllowed() throws IOException {
         assumeFalse("pragma only disabled on release builds", Build.current().isSnapshot());
-        RequestObjectBuilder builder = new RequestObjectBuilder().query("row a = 1, b = 2");
+        RequestObjectBuilder builder = requestObjectBuilder().query("row a = 1, b = 2");
         builder.pragmas(Settings.builder().put("data_partitioning", "shard").build());
         ResponseException re = expectThrows(ResponseException.class, () -> runEsqlSync(builder));
         assertThat(EntityUtils.toString(re.getResponse().getEntity()), containsString("[pragma] only allowed in snapshot builds"));
@@ -197,7 +197,7 @@ public void testIncompatibleMappingsErrors() throws IOException {
     }
 
     private void assertException(String query, String... errorMessages) throws IOException {
-        ResponseException re = expectThrows(ResponseException.class, () -> runEsqlSync(new RequestObjectBuilder().query(query)));
+        ResponseException re = expectThrows(ResponseException.class, () -> runEsqlSync(requestObjectBuilder().query(query)));
         assertThat(re.getResponse().getStatusLine().getStatusCode(), equalTo(400));
         for (var error : errorMessages) {
             assertThat(re.getMessage(), containsString(error));

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

@@ -61,9 +61,8 @@ public void testTimeSeriesQuerying() throws IOException {
         Response response = client().performRequest(bulk);
         assertEquals("{\"errors\":false}", EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8));
 
-        RestEsqlTestCase.RequestObjectBuilder builder = new RestEsqlTestCase.RequestObjectBuilder().query(
-            "FROM k8s | KEEP k8s.pod.name, @timestamp"
-        );
+        RestEsqlTestCase.RequestObjectBuilder builder = RestEsqlTestCase.requestObjectBuilder()
+            .query("FROM k8s | KEEP k8s.pod.name, @timestamp");
         builder.pragmas(Settings.builder().put("time_series", true).build());
         Map<String, Object> result = runEsqlSync(builder);
         @SuppressWarnings("unchecked")