[ES|QL] Inference command: support for CCS. (elastic#139244)

Author: afoucret

docs/changelog/139244.yaml

@@ -0,0 +1,6 @@
+pr: 139244
+summary: "Inference command: support for CCS"
+area: ES|QL
+type: enhancement
+issues:
+ - 136860

docs/reference/query-languages/esql/_snippets/functions/examples/text_embedding.md

@@ -75,7 +75,12 @@ protected boolean enableRoundingDoubleValuesOnAsserting() {
     }
 
     @Override
-    protected boolean supportsInferenceTestService() {
+    protected boolean supportsSemanticTextInference() {
+        return false;
+    }
+
+    @Override
+    protected boolean supportsInferenceTestServiceOnLocalCluster() {
         return false;
     }
 

docs/reference/query-languages/esql/kibana/definition/functions/text_embedding.json

@@ -21,6 +21,8 @@ dependencies {
   javaRestTestImplementation project(xpackModule('esql:qa:testFixtures'))
   javaRestTestImplementation project(xpackModule('esql:qa:server'))
   javaRestTestImplementation project(xpackModule('esql'))
+
+  clusterPlugins project(':x-pack:plugin:inference:qa:test-service-plugin')
 }
 
 def supportedVersion = bwcVersion -> {

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

@@ -54,6 +54,9 @@ public static ElasticsearchCluster localCluster(ElasticsearchCluster remoteClust
         if (supportRetryOnShardFailures(version) == false) {
             cluster.setting("cluster.routing.rebalance.enable", "none");
         }
+        if (localClusterSupportsInferenceTestService()) {
+            cluster.plugin("inference-service-test");
+        }
         return cluster.build();
     }
 
@@ -73,6 +76,22 @@ public static org.elasticsearch.Version bwcVersion() {
         return local.before(remote) ? local : remote;
     }
 
+    public static boolean localClusterSupportsInferenceTestService() {
+        return isNewToOld();
+    }
+
+    /**
+     * Returns true if the current task is a "newToOld" BWC test.
+     * Checks the tests.task system property to determine the task type.
+     */
+    private static boolean isNewToOld() {
+        String taskName = System.getProperty("tests.task");
+        if (taskName == null) {
+            return false;
+        }
+        return taskName.endsWith("#newToOld");
+    }
+
     private static Version distributionVersion(String key) {
         final String val = System.getProperty(key);
         return val != null ? Version.fromString(val) : Version.CURRENT;

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

@@ -48,6 +48,7 @@
 import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.CSV_DATASET_MAP;
 import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.ENRICH_SOURCE_INDICES;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.classpathResources;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.COMPLETION;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.ENABLE_FORK_FOR_REMOTE_INDICES_V2;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.ENABLE_LOOKUP_JOIN_ON_REMOTE;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.FORK_V9;
@@ -56,7 +57,9 @@
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.JOIN_LOOKUP_V12;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.JOIN_PLANNING_V1;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.METADATA_FIELDS_REMOTE_TEST;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.RERANK;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.SUBQUERY_IN_FROM_COMMAND;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.TEXT_EMBEDDING_FUNCTION;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.UNMAPPED_FIELDS;
 import static org.elasticsearch.xpack.esql.qa.rest.RestEsqlTestCase.hasCapabilities;
 import static org.mockito.ArgumentMatchers.any;
@@ -82,6 +85,12 @@ public class MultiClusterSpecIT extends EsqlSpecTestCase {
     private static RestClient remoteClusterClient;
     private static DataLocation dataLocation = null;
 
+    private static final Set<String> LOCAL_ONLY_INFERENCE_CAPABILITIES = Set.of(
+        RERANK.capabilityName(),
+        COMPLETION.capabilityName(),
+        TEXT_EMBEDDING_FUNCTION.capabilityName()
+    );
+
     @ParametersFactory(argumentFormatting = "csv-spec:%2$s.%3$s")
     public static List<Object[]> readScriptSpec() throws Exception {
         List<URL> urls = classpathResources("/*.csv-spec");
@@ -134,8 +143,15 @@ protected void shouldSkipTest(String testName) throws IOException {
                 .filter(c -> c.equals("metadata_fields_remote_test") == false)
                 .toList();
         }
+        // Check all capabilities on the local cluster first.
         super.shouldSkipTest(testName);
-        checkCapabilities(remoteClusterClient(), remoteFeaturesService(), testName, testCase);
+
+        // Filter out capabilities that are required only on the local cluster and then check the remaining on the remote cluster.
+        List<String> remoteCapabilities = testCase.requiredCapabilities.stream()
+            .filter(c -> LOCAL_ONLY_INFERENCE_CAPABILITIES.contains(c) == false)
+            .toList();
+        checkCapabilities(remoteClusterClient(), remoteFeaturesService(), testName, remoteCapabilities);
+
         // Do not run tests including "METADATA _index" unless marked with metadata_fields_remote_test,
         // because they may produce inconsistent results with multiple clusters.
         assumeFalse("can't test with _index metadata", (remoteMetadata == false) && hasIndexMetadata(testCase.query));
@@ -230,10 +246,12 @@ protected RestClient buildClient(Settings settings, HttpHost[] localHosts) throw
         .collect(Collectors.toSet());
 
     /**
-     * Creates a new mock client that dispatches every request to both the local and remote clusters, excluding _bulk and _query requests.
+     * Creates a new mock client that dispatches every request to both the local and remote clusters, excluding _bulk, _query,
+     *  and _inference requests :
      * - '_bulk' requests are randomly sent to either the local or remote cluster to populate data. Some spec tests, such as AVG,
      *   prevent the splitting of bulk requests.
      * - '_query' requests are dispatched to the local cluster only, as we are testing cross-cluster queries.
+     * - '_inference' requests are dispatched to the local cluster only, as inference endpoints are not available on remote clusters.
      */
     static RestClient twoClients(RestClient localClient, RestClient remoteClient) throws IOException {
         RestClient twoClients = mock(RestClient.class);
@@ -245,6 +263,8 @@ static RestClient twoClients(RestClient localClient, RestClient remoteClient) th
             String endpoint = request.getEndpoint();
             if (endpoint.startsWith("/_query")) {
                 return localClient.performRequest(request);
+            } else if (endpoint.startsWith("/_inference")) {
+                return localClient.performRequest(request);
             } else if (endpoint.endsWith("/_bulk") && METADATA_INDICES.stream().anyMatch(i -> endpoint.equals("/" + i + "/_bulk"))) {
                 return remoteClient.performRequest(request);
             } else if (endpoint.endsWith("/_bulk")
@@ -340,10 +360,15 @@ protected boolean enableRoundingDoubleValuesOnAsserting() {
     }
 
     @Override
-    protected boolean supportsInferenceTestService() {
+    protected boolean supportsSemanticTextInference() {
         return false;
     }
 
+    @Override
+    protected boolean supportsInferenceTestServiceOnLocalCluster() {
+        return Clusters.localClusterSupportsInferenceTestService();
+    }
+
     @Override
     protected boolean supportsIndexModeLookup() throws IOException {
         return hasCapabilities(adminClient(), List.of(JOIN_LOOKUP_V12.capabilityName()));

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

@@ -823,7 +823,7 @@ protected boolean preserveClusterUponCompletion() {
 
     private void setUpTextEmbeddingInferenceEndpoint() throws IOException {
         setupEmbeddings = true;
-        Request request = new Request("PUT", "_inference/text_embedding/test");
+        Request request = new Request("PUT", "/_inference/text_embedding/test");
         request.setJsonEntity("""
                   {
                    "service": "text_embedding_test_service",

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

@@ -548,7 +548,7 @@ public void setUpTextEmbeddingInferenceEndpoint() throws IOException {
             return;
         }
         setupEmbeddings = true;
-        Request request = new Request("PUT", "_inference/text_embedding/test");
+        Request request = new Request("PUT", "/_inference/text_embedding/test");
         request.setJsonEntity("""
                   {
                    "service": "text_embedding_test_service",

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

@@ -175,14 +175,15 @@ public void setup() {
         assumeTrue("test clusters were broken", testClustersOk);
         INGEST.protectedBlock(() -> {
             // Inference endpoints must be created before ingesting any datasets that rely on them (mapping of inference_id)
-            if (supportsInferenceTestService()) {
+            // If multiple clusters are used, only create endpoints on the local cluster if it supports the inference test service.
+            if (supportsInferenceTestServiceOnLocalCluster()) {
                 createInferenceEndpoints(adminClient());
             }
             loadDataSetIntoEs(
                 client(),
                 supportsIndexModeLookup(),
                 supportsSourceFieldMapping(),
-                supportsInferenceTestService(),
+                supportsSemanticTextInference(),
                 false,
                 supportsExponentialHistograms(),
                 supportsTDigestField()
@@ -236,8 +237,11 @@ protected void ensureTestClustersAreOk(Exception failure) {
 
     protected void shouldSkipTest(String testName) throws IOException {
         assumeTrue("test clusters were broken", testClustersOk);
-        if (requiresInferenceEndpoint()) {
-            assumeTrue("Inference test service needs to be supported", supportsInferenceTestService());
+        if (requiresSemanticTextInference()) {
+            assumeTrue("Inference test service needs to be supported", supportsSemanticTextInference());
+        }
+        if (requiresInferenceEndpointOnLocalCluster()) {
+            assumeTrue("Inference test service needs to be supported", supportsInferenceTestServiceOnLocalCluster());
         }
         checkCapabilities(adminClient(), testFeatureService, testName, testCase);
         assumeTrue("Test " + testName + " is not enabled", isEnabled(testName, instructions, Version.CURRENT));
@@ -252,26 +256,42 @@ protected static void checkCapabilities(
         String testName,
         CsvTestCase testCase
     ) {
-        if (hasCapabilities(client, testCase.requiredCapabilities)) {
+        checkCapabilities(client, testFeatureService, testName, testCase.requiredCapabilities);
+    }
+
+    protected static void checkCapabilities(
+        RestClient client,
+        TestFeatureService testFeatureService,
+        String testName,
+        List<String> requiredCapabilities
+    ) {
+        if (hasCapabilities(client, requiredCapabilities)) {
             return;
         }
 
         var features = new EsqlFeatures().getFeatures().stream().map(NodeFeature::id).collect(Collectors.toSet());
 
-        for (String feature : testCase.requiredCapabilities) {
+        for (String feature : requiredCapabilities) {
             var esqlFeature = "esql." + feature;
             assumeTrue("Requested capability " + feature + " is an ESQL cluster feature", features.contains(esqlFeature));
             assumeTrue("Test " + testName + " requires " + feature, testFeatureService.clusterHasFeature(esqlFeature));
         }
     }
 
-    protected boolean supportsInferenceTestService() {
+    protected boolean supportsSemanticTextInference() {
+        return true;
+    }
+
+    protected boolean supportsInferenceTestServiceOnLocalCluster() {
         return true;
     }
 
-    protected boolean requiresInferenceEndpoint() {
+    protected boolean requiresSemanticTextInference() {
+        return testCase.requiredCapabilities.contains(SEMANTIC_TEXT_FIELD_CAPS.capabilityName());
+    }
+
+    protected boolean requiresInferenceEndpointOnLocalCluster() {
         return Stream.of(
-            SEMANTIC_TEXT_FIELD_CAPS.capabilityName(),
             RERANK.capabilityName(),
             COMPLETION.capabilityName(),
             KNN_FUNCTION_V5.capabilityName(),