Merge branch 'main' into esql-inference-command-serialization

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

@@ -36,6 +36,7 @@ public final class SystemPrivilege extends Privilege {
         "indices:admin/template/put", // needed for the TemplateUpgradeService
         "indices:admin/template/delete", // needed for the TemplateUpgradeService
         "indices:admin/seq_no/global_checkpoint_sync*", // needed for global checkpoint syncs
+        "indices:admin/reshard/split*", // needed for reshard operations
         RetentionLeaseSyncAction.ACTION_NAME + "*", // needed for retention lease syncs
         RetentionLeaseBackgroundSyncAction.ACTION_NAME + "*", // needed for background retention lease syncs
         RetentionLeaseActions.ADD.name() + "*", // needed for CCR to add retention leases

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

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

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/SystemPrivilege.java

@@ -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",