Add mixed-cluster tests for ESQL (#102198)

This PR adds mixed-cluster tests for ESQL and introduces a version 
skipping instruction for CSV tests. This version skipping is similar to
the YAML skipping version. For example, `aggTest#[skip:-8.11.99]` should
guide the test runner to skip the aggTest until version 8.12.0.

Currently, it only supports CSV spec tests, but I plan to support YAML 
and other rest tests and address the workaround in a Gradle build soon.

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/RestrictedBuildApiService.java

@@ -143,6 +143,7 @@ public abstract class RestrictedBuildApiService implements BuildService<Restrict
         map.put(LegacyRestTestBasePlugin.class, ":x-pack:plugin:esql:qa:security");
         map.put(LegacyRestTestBasePlugin.class, ":x-pack:plugin:esql:qa:server:multi-node");
         map.put(LegacyRestTestBasePlugin.class, ":x-pack:plugin:esql:qa:server:single-node");
+        map.put(LegacyRestTestBasePlugin.class, ":x-pack:plugin:esql:qa:server:mixed-cluster");
         map.put(LegacyRestTestBasePlugin.class, ":x-pack:plugin:fleet:qa:rest");
         map.put(LegacyRestTestBasePlugin.class, ":x-pack:plugin:graph:qa:with-security");
         map.put(LegacyRestTestBasePlugin.class, ":x-pack:plugin:identity-provider:qa:idp-rest-tests");

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

@@ -19,7 +19,7 @@ subprojects {
   }
 
 
-  if (project.name != 'security') {
+  if (project.name != 'security' && project.name != 'mixed-cluster' ) {
     // The security project just configures its subprojects
     apply plugin: 'elasticsearch.legacy-java-rest-test'
 

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

@@ -0,0 +1,50 @@
+
+import org.elasticsearch.gradle.Version
+import org.elasticsearch.gradle.VersionProperties
+import org.elasticsearch.gradle.internal.info.BuildParams
+import org.elasticsearch.gradle.testclusters.StandaloneRestIntegTestTask
+
+apply plugin: 'elasticsearch.internal-testclusters'
+apply plugin: 'elasticsearch.standalone-rest-test'
+apply plugin: 'elasticsearch.bwc-test'
+apply plugin: 'elasticsearch.rest-resources'
+
+dependencies {
+  testImplementation project(xpackModule('esql:qa:testFixtures'))
+  testImplementation project(xpackModule('esql:qa:server'))
+}
+
+restResources {
+  restApi {
+    include '_common', 'bulk', 'indices', 'esql', 'xpack', 'enrich'
+  }
+}
+
+BuildParams.bwcVersions.withWireCompatible { bwcVersion, baseName ->
+
+  if (bwcVersion != VersionProperties.getElasticsearchVersion() && bwcVersion.onOrAfter(Version.fromString("8.11.0"))) {
+    /* This project runs the ESQL spec tests against a 4 node cluster where two of the nodes has a different minor.  */
+    def baseCluster = testClusters.register(baseName) {
+      versions = [bwcVersion.toString(), bwcVersion.toString(), project.version, project.version]
+      numberOfNodes = 4
+      testDistribution = 'DEFAULT'
+      setting 'xpack.license.self_generated.type', 'trial'
+      setting 'xpack.security.enabled', 'false'
+    }
+
+    tasks.register("${baseName}#mixedClusterTest", StandaloneRestIntegTestTask) {
+      useCluster baseCluster
+      mustRunAfter("precommit")
+      nonInputProperties.systemProperty('tests.rest.cluster', baseCluster.map(c -> c.allHttpSocketURI.join(",")))
+      nonInputProperties.systemProperty('tests.clustername', baseName)
+      systemProperty 'tests.bwc_nodes_version', bwcVersion.toString().replace('-SNAPSHOT', '')
+      systemProperty 'tests.new_nodes_version', project.version.toString().replace('-SNAPSHOT', '')
+      onlyIf("BWC tests disabled") { project.bwc_tests_enabled }
+    }
+
+    tasks.register(bwcTaskName(bwcVersion)) {
+      dependsOn "${baseName}#mixedClusterTest"
+    }
+  }
+}
+

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

@@ -0,0 +1,30 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.qa.mixed;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.xpack.esql.qa.rest.EsqlSpecTestCase;
+import org.elasticsearch.xpack.ql.CsvSpecReader.CsvTestCase;
+
+import static org.elasticsearch.xpack.esql.CsvTestUtils.isEnabled;
+
+public class MixedClusterEsqlSpecIT extends EsqlSpecTestCase {
+
+    static final Version bwcVersion = Version.fromString(System.getProperty("tests.bwc_nodes_version"));
+    static final Version newVersion = Version.fromString(System.getProperty("tests.new_nodes_version"));
+
+    public MixedClusterEsqlSpecIT(String fileName, String groupName, String testName, Integer lineNumber, CsvTestCase testCase) {
+        super(fileName, groupName, testName, lineNumber, testCase);
+    }
+
+    @Override
+    protected void shouldSkipTest(String testName) {
+        assumeTrue("Test " + testName + " is skipped on " + bwcVersion, isEnabled(testName, bwcVersion));
+        assumeTrue("Test " + testName + " is skipped on " + newVersion, isEnabled(testName, newVersion));
+    }
+}

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

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.esql.qa.rest;
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
 import org.apache.http.HttpEntity;
+import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.common.xcontent.XContentHelper;
@@ -90,13 +91,17 @@ public abstract class EsqlSpecTestCase extends ESRestTestCase {
 
     public final void test() throws Throwable {
         try {
-            assumeTrue("Test " + testName + " is not enabled", isEnabled(testName));
+            shouldSkipTest(testName);
             doTest();
         } catch (Exception e) {
             throw reworkException(e);
         }
     }
 
+    protected void shouldSkipTest(String testName) {
+        assumeTrue("Test " + testName + " is not enabled", isEnabled(testName, Version.CURRENT));
+    }
+
     protected final void doTest() throws Throwable {
         RequestObjectBuilder builder = new RequestObjectBuilder(randomFrom(XContentType.values()));
         Map<String, Object> answer = runEsql(builder.query(testCase.query).build(), testCase.expectedWarnings);

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/CsvTestUtils.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.esql;
 
 import org.apache.lucene.sandbox.document.HalfFloatPoint;
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.time.DateFormatters;
 import org.elasticsearch.compute.data.Block;
@@ -22,9 +23,9 @@ import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.logging.Logger;
+import org.elasticsearch.test.VersionUtils;
 import org.elasticsearch.xpack.esql.action.EsqlQueryResponse;
 import org.elasticsearch.xpack.ql.util.StringUtils;
-import org.elasticsearch.xpack.versionfield.Version;
 import org.supercsv.io.CsvListReader;
 import org.supercsv.prefs.CsvPreference;
 
@@ -42,6 +43,8 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import static org.elasticsearch.common.Strings.delimitedListToStringArray;
 import static org.elasticsearch.common.logging.LoggerMessageFormat.format;
@@ -57,8 +60,51 @@ public final class CsvTestUtils {
 
     private CsvTestUtils() {}
 
-    public static boolean isEnabled(String testName) {
-        return testName.endsWith("-Ignore") == false;
+    public static boolean isEnabled(String testName, Version version) {
+        if (testName.endsWith("-Ignore")) {
+            return false;
+        }
+        Tuple<Version, Version> skipRange = skipVersionRange(testName);
+        if (skipRange != null && version.onOrAfter(skipRange.v1()) && version.onOrBefore(skipRange.v2())) {
+            return false;
+        }
+        return true;
+    }
+
+    private static final Pattern INSTRUCTION_PATTERN = Pattern.compile("#\\[(.*?)]");
+
+    public static Map<String, String> extractInstructions(String testName) {
+        Matcher matcher = INSTRUCTION_PATTERN.matcher(testName);
+        Map<String, String> pairs = new HashMap<>();
+        if (matcher.find()) {
+            String[] groups = matcher.group(1).split(",");
+            for (String group : groups) {
+                String[] kv = group.split(":");
+                if (kv.length != 2) {
+                    throw new IllegalArgumentException("expected instruction in [k1:v1,k2:v2] format; got " + matcher.group(1));
+                }
+                pairs.put(kv[0].trim(), kv[1].trim());
+            }
+        }
+        return pairs;
+    }
+
+    public static Tuple<Version, Version> skipVersionRange(String testName) {
+        Map<String, String> pairs = extractInstructions(testName);
+        String versionRange = pairs.get("skip");
+        if (versionRange != null) {
+            String[] skipVersions = versionRange.split("-");
+            if (skipVersions.length != 2) {
+                throw new IllegalArgumentException("malformed version range : " + versionRange);
+            }
+            String lower = skipVersions[0].trim();
+            String upper = skipVersions[1].trim();
+            return Tuple.tuple(
+                lower.isEmpty() ? VersionUtils.getFirstVersion() : Version.fromString(lower),
+                upper.isEmpty() ? Version.CURRENT : Version.fromString(upper)
+            );
+        }
+        return null;
     }
 
     public static Tuple<Page, List<String>> loadPageFromCsv(URL source) throws Exception {
@@ -333,7 +379,7 @@ public final class CsvTestUtils {
                 : ((BytesRef) l).compareTo((BytesRef) r),
             BytesRef.class
         ),
-        VERSION(v -> new Version(v).toBytesRef(), BytesRef.class),
+        VERSION(v -> new org.elasticsearch.xpack.versionfield.Version(v).toBytesRef(), BytesRef.class),
         NULL(s -> null, Void.class),
         DATETIME(
             x -> x == null ? null : DateFormatters.from(UTC_DATE_TIME_FORMATTER.parse(x)).toInstant().toEpochMilli(),

x-pack/plugin/esql/qa/testFixtures/src/main/resources/date.csv-spec

@@ -182,7 +182,7 @@ string:keyword                                          |datetime:date
 // end::to_datetime-str-result[]
 ;
 
-convertFromUnsignedLong
+convertFromUnsignedLong#[skip:-8.11.99, reason:ql exceptions were updated in 8.12]
 row ul = [9223372036854775808, 520128000000] | eval dt = to_datetime(ul);
 warning:Line 1:58: evaluation of [to_datetime(ul)] failed, treating result as null. Only first 20 failures recorded.
 warning:Line 1:58: org.elasticsearch.xpack.ql.InvalidArgumentException: [9223372036854775808] out of [long] range

x-pack/plugin/esql/qa/testFixtures/src/main/resources/ints.csv-spec

@@ -63,7 +63,7 @@ long:long                    |ul:ul
 [501379200000, 520128000000] |[501379200000, 520128000000]
 ;
 
-convertDoubleToUL
+convertDoubleToUL#[skip:-8.11.99, reason:ql exceptions updated in 8.12]
 row d = 123.4 | eval ul = to_ul(d), overflow = to_ul(1e20);
 warning:Line 1:48: evaluation of [to_ul(1e20)] failed, treating result as null. Only first 20 failures recorded.
 warning:Line 1:48: org.elasticsearch.xpack.ql.InvalidArgumentException: [1.0E20] out of [unsigned_long] range
@@ -120,7 +120,7 @@ int:integer       |long:long
 [5013792, 520128] |[5013792, 520128]
 ;
 
-convertULToLong
+convertULToLong#[skip:-8.11.99, reason:ql exceptions were updated in 8.12]
 row ul = [9223372036854775807, 9223372036854775808] | eval long = to_long(ul);
 warning:Line 1:67: evaluation of [to_long(ul)] failed, treating result as null. Only first 20 failures recorded.
 warning:Line 1:67: org.elasticsearch.xpack.ql.InvalidArgumentException: [9223372036854775808] out of [long] range
@@ -161,7 +161,7 @@ str1:keyword |str2:keyword |str3:keyword |long1:long  |long2:long |long3:long
 // end::to_long-str-result[]
 ;
 
-convertDoubleToLong
+convertDoubleToLong#[skip:-8.11.99, reason:ql exceptions were updated in 8.12]
 row d = 123.4 | eval d2l = to_long(d), overflow = to_long(1e19);
 warning:Line 1:51: evaluation of [to_long(1e19)] failed, treating result as null. Only first 20 failures recorded.
 warning:Line 1:51: org.elasticsearch.xpack.ql.InvalidArgumentException: [1.0E19] out of [long] range
@@ -179,7 +179,7 @@ int:integer       |ii:integer
 [5013792, 520128] |[5013792, 520128]
 ;
 
-convertLongToInt
+convertLongToInt#[skip:-8.11.99, reason:ql exceptions were updated in 8.12]
 // tag::to_int-long[]
 ROW long = [5013792, 2147483647, 501379200000]
 | EVAL int = TO_INTEGER(long)
@@ -194,7 +194,7 @@ long:long                           |int:integer
 // end::to_int-long-result[]
 ;
 
-convertULToInt
+convertULToInt#[skip:-8.11.99, reason:ql exceptions were updated in 8.12]
 row ul = [2147483647, 9223372036854775808] | eval int = to_int(ul);
 warning:Line 1:57: evaluation of [to_int(ul)] failed, treating result as null. Only first 20 failures recorded.
 // UL conversion to int dips into long; not the most efficient, but it's how SQL does it too.
@@ -229,7 +229,7 @@ int_str:keyword  |int_dbl_str:keyword |is2i:integer|ids2i:integer   |overflow:in
 2147483647       |2147483647.2        |2147483647  |2147483647      |null             |null
 ;
 
-convertDoubleToInt
+convertDoubleToInt#[skip:-8.11.99, reason:ql exceptions were updated in 8.12]
 row d = 123.4 | eval d2i = to_integer(d), overflow = to_integer(1e19);
 warning:Line 1:54: evaluation of [to_integer(1e19)] failed, treating result as null. Only first 20 failures recorded.
 warning:Line 1:54: org.elasticsearch.xpack.ql.InvalidArgumentException: [1.0E19] out of [long] range
@@ -473,7 +473,7 @@ ROW deg = [90, 180, 270]
 [90, 180, 270] | [1.5707963267948966, 3.141592653589793, 4.71238898038469]
 ;
 
-warningWithFromSource
+warningWithFromSource#[skip:-8.11.99, reason:ql exceptions were updated in 8.12]
 from employees | sort emp_no | limit 1 | eval x = to_long(emp_no) * 10000000 | eval y = to_int(x) > 1 | keep y;
 warning:Line 1:89: evaluation of [to_int(x)] failed, treating result as null. Only first 20 failures recorded.
 warning:Line 1:89: org.elasticsearch.xpack.ql.InvalidArgumentException: [100010000000] out of [integer] range

x-pack/plugin/esql/qa/testFixtures/src/main/resources/show.csv-spec

@@ -5,7 +5,7 @@ v:long
 1
 ;
 
-showFunctions
+showFunctions#[skip:-8.11.99]
 show functions;
 
        name:keyword      |                        synopsis:keyword          |       argNames:keyword  | argTypes:keyword |             argDescriptions:keyword                |returnType:keyword   |    description:keyword  | optionalArgs:boolean | variadic:boolean                
@@ -94,7 +94,7 @@ trim                     |? trim(arg1:?)
 ;
 
 
-showFunctionsSynopsis
+showFunctionsSynopsis#[skip:-8.11.99]
 show functions | keep synopsis;
 
 synopsis:keyword          

x-pack/plugin/esql/qa/testFixtures/src/main/resources/stats.csv-spec

@@ -682,14 +682,14 @@ c:l |  job_positions:s
 4   |Tech Lead    
 ;
 
-duplicateAggregationsWithoutGrouping
+duplicateAggregationsWithoutGrouping#[skip:-8.11.99]
 from employees | eval x = salary | stats c = count(), m = min(x), m1 = min(salary), c1 = count(1);
 
 c:l | m:i | m1:i | c1:l
 100 | 25324 | 25324  | 100
 ;
 
-duplicateAggregationsWithGrouping
+duplicateAggregationsWithGrouping#[skip:-8.11.99]
 from employees | eval x = salary | stats c = count(), m = min(x), m1 = min(salary), c1 = count(1) by gender | sort gender;
 
 c:l| m:i   | m1:i  | c1:l| gender:s

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/CsvTests.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.esql;
 
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.common.Randomness;
@@ -215,7 +216,7 @@ public class CsvTests extends ESTestCase {
 
     public final void test() throws Throwable {
         try {
-            assumeTrue("Test " + testName + " is not enabled", isEnabled(testName));
+            assumeTrue("Test " + testName + " is not enabled", isEnabled(testName, Version.CURRENT));
             doTest();
         } catch (Throwable th) {
             throw reworkException(th);