CCS metadata is opt-in in ESQL JSON responses

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

@@ -67,6 +67,11 @@ precedence.
 `false`. The API only supports this parameter for CBOR, JSON, SMILE, and YAML
 responses. See <<esql-rest-columnar>>.
 
+`include_ccs_metadata`::
+(Optional, boolean) If `true`, cross-cluster searches will include metadata about the query
+on each cluster. Defaults to `false`. The API only supports this parameter for CBOR, JSON, SMILE,
+and YAML responses. See <<ccq-cluster-details>>.
+
 `locale`::
 (Optional, string) Returns results (especially dates) formatted per the conventions of the locale.
 For syntax, refer to <<esql-locale-param>>.
@@ -85,6 +90,7 @@ https://en.wikipedia.org/wiki/Query_plan[EXPLAIN PLAN].
 `query`::
 (Required, string) {esql} query to run. For syntax, refer to <<esql-syntax>>.
 
+
 ifeval::["{release-state}"=="unreleased"]
 `table`::
 (Optional, object) Named "table" parameters that can be referenced by the <<esql-lookup>> command.
@@ -108,6 +114,13 @@ returned if `drop_null_columns` is sent with the request.
 (array of arrays)
 Values for the search results.
 
+`_clusters`::
+(object)
+Metadata about clusters involved in the execution of a cross-cluster query. Only returned (1) for
+cross-cluster searches and (2) when `include_ccs_metadata` is sent in the body and set to `true`
+and (3) when `format` of the response is set to JSON (the default), CBOR, SMILE, or YAML.
+See <<ccq-cluster-details>> for more information.
+
 `profile`::
 (object)
 Profile describing the execution of the query. Only returned if `profile` was sent in the body.

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

@@ -238,7 +238,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion FAST_REFRESH_RCO = def(8_762_00_0);
     public static final TransportVersion TEXT_SIMILARITY_RERANKER_QUERY_REWRITE = def(8_763_00_0);
     public static final TransportVersion SIMULATE_INDEX_TEMPLATES_SUBSTITUTIONS = def(8_764_00_0);
-    public static final TransportVersion OPT_IN_ESQL_CCS_EXECUTION_INFO = def(8_765_00_0);
+    public static final TransportVersion RETRIEVERS_TELEMETRY_ADDED = def(8_765_00_0);
+    public static final TransportVersion OPT_IN_ESQL_CCS_EXECUTION_INFO = def(8_766_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -135,6 +135,14 @@ private Map<String, Object> run(String query, boolean includeCCSMetadata) throws
         return resp;
     }
 
+    private Map<String, Object> runWithColumnarAndIncludeCCSMetadata(String query) throws IOException {
+        Map<String, Object> resp = runEsql(
+            new RestEsqlTestCase.RequestObjectBuilder().query(query).includeCCSMetadata(true).columnar(true).build()
+        );
+        logger.info("--> query {} response {}", query, resp);
+        return resp;
+    }
+
     protected boolean supportsAsync() {
         return false; // TODO: Version.CURRENT.onOrAfter(Version.V_8_13_0); ?? // the Async API was introduced in 8.13.0
     }
@@ -205,7 +213,6 @@ public void testUngroupedAggs() throws Exception {
             long sum = remoteDocs.stream().mapToLong(d -> d.data).sum();
             var values = List.of(List.of(Math.toIntExact(sum)));
 
-            // check all sections of map except _cluster/details
             MapMatcher mapMatcher = matchesMap();
             if (includeCCSMetadata) {
                 mapMatcher = mapMatcher.entry("_clusters", any(Map.class));
@@ -215,6 +222,22 @@ public void testUngroupedAggs() throws Exception {
                 assertClusterDetailsMap(result, true);
             }
         }
+        {
+            Map<String, Object> result = runWithColumnarAndIncludeCCSMetadata("FROM *:test-remote-index | STATS total = SUM(data)");
+            var columns = List.of(Map.of("name", "total", "type", "long"));
+            long sum = remoteDocs.stream().mapToLong(d -> d.data).sum();
+            var values = List.of(List.of(Math.toIntExact(sum)));
+
+            MapMatcher mapMatcher = matchesMap();
+            assertMap(
+                result,
+                mapMatcher.entry("columns", columns)
+                    .entry("values", values)
+                    .entry("took", greaterThanOrEqualTo(0))
+                    .entry("_clusters", any(Map.class))
+            );
+            assertClusterDetailsMap(result, true);
+        }
     }
 
     private void assertClusterDetailsMap(Map<String, Object> result, boolean remoteOnly) {

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClustersEnrichIT.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.ingest.common.IngestCommonPlugin;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.license.LicenseService;
@@ -220,7 +221,7 @@ static String enrichVendors(Enrich.Mode mode) {
     public void testWithHostsPolicy() {
         for (var mode : Enrich.Mode.values()) {
             String query = "FROM events | eval ip= TO_STR(host) | " + enrichHosts(mode) + " | stats c = COUNT(*) by os | SORT os";
-            try (EsqlQueryResponse resp = runQuery(query)) {
+            try (EsqlQueryResponse resp = runQuery(query, null)) {
                 List<List<Object>> rows = getValuesList(resp);
                 assertThat(
                     rows,
@@ -237,9 +238,14 @@ public void testWithHostsPolicy() {
                 assertFalse(resp.getExecutionInfo().isCrossClusterSearch());
             }
         }
+
+        Tuple<Boolean, Boolean> includeCCSMetadata = randomIncludeCCSMetadata();
+        Boolean requestIncludeMeta = includeCCSMetadata.v1();
+        boolean responseExpectMeta = includeCCSMetadata.v2();
+
         for (var mode : Enrich.Mode.values()) {
             String query = "FROM *:events | eval ip= TO_STR(host) | " + enrichHosts(mode) + " | stats c = COUNT(*) by os | SORT os";
-            try (EsqlQueryResponse resp = runQuery(query)) {
+            try (EsqlQueryResponse resp = runQuery(query, requestIncludeMeta)) {
                 List<List<Object>> rows = getValuesList(resp);
                 assertThat(
                     rows,
@@ -255,14 +261,15 @@ public void testWithHostsPolicy() {
                     )
                 );
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+                assertThat(executionInfo.includeCCSMetadata(), equalTo(responseExpectMeta));
                 assertThat(executionInfo.clusterAliases(), equalTo(Set.of("c1", "c2")));
                 assertCCSExecutionInfoDetails(executionInfo);
             }
         }
 
         for (var mode : Enrich.Mode.values()) {
             String query = "FROM *:events,events | eval ip= TO_STR(host) | " + enrichHosts(mode) + " | stats c = COUNT(*) by os | SORT os";
-            try (EsqlQueryResponse resp = runQuery(query)) {
+            try (EsqlQueryResponse resp = runQuery(query, requestIncludeMeta)) {
                 List<List<Object>> rows = getValuesList(resp);
                 assertThat(
                     rows,
@@ -278,13 +285,18 @@ public void testWithHostsPolicy() {
                     )
                 );
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+                assertThat(executionInfo.includeCCSMetadata(), equalTo(responseExpectMeta));
                 assertThat(executionInfo.clusterAliases(), equalTo(Set.of("", "c1", "c2")));
                 assertCCSExecutionInfoDetails(executionInfo);
             }
         }
     }
 
     public void testEnrichHostsAggThenEnrichVendorCoordinator() {
+        Tuple<Boolean, Boolean> includeCCSMetadata = randomIncludeCCSMetadata();
+        Boolean requestIncludeMeta = includeCCSMetadata.v1();
+        boolean responseExpectMeta = includeCCSMetadata.v2();
+
         for (var hostMode : Enrich.Mode.values()) {
             String query = String.format(Locale.ROOT, """
                 FROM *:events,events
@@ -295,7 +307,7 @@ public void testEnrichHostsAggThenEnrichVendorCoordinator() {
                 | stats c = SUM(c) by vendor
                 | sort vendor
                 """, enrichHosts(hostMode), enrichVendors(Enrich.Mode.COORDINATOR));
-            try (EsqlQueryResponse resp = runQuery(query)) {
+            try (EsqlQueryResponse resp = runQuery(query, requestIncludeMeta)) {
                 assertThat(
                     getValuesList(resp),
                     equalTo(
@@ -309,13 +321,18 @@ public void testEnrichHostsAggThenEnrichVendorCoordinator() {
                     )
                 );
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+                assertThat(executionInfo.includeCCSMetadata(), equalTo(responseExpectMeta));
                 assertThat(executionInfo.clusterAliases(), equalTo(Set.of("", "c1", "c2")));
                 assertCCSExecutionInfoDetails(executionInfo);
             }
         }
     }
 
     public void testEnrichTwiceThenAggs() {
+        Tuple<Boolean, Boolean> includeCCSMetadata = randomIncludeCCSMetadata();
+        Boolean requestIncludeMeta = includeCCSMetadata.v1();
+        boolean responseExpectMeta = includeCCSMetadata.v2();
+
         for (var hostMode : Enrich.Mode.values()) {
             String query = String.format(Locale.ROOT, """
                 FROM *:events,events
@@ -325,7 +342,7 @@ public void testEnrichTwiceThenAggs() {
                 | stats c = COUNT(*) by vendor
                 | sort vendor
                 """, enrichHosts(hostMode), enrichVendors(Enrich.Mode.COORDINATOR));
-            try (EsqlQueryResponse resp = runQuery(query)) {
+            try (EsqlQueryResponse resp = runQuery(query, requestIncludeMeta)) {
                 assertThat(
                     getValuesList(resp),
                     equalTo(
@@ -339,13 +356,18 @@ public void testEnrichTwiceThenAggs() {
                     )
                 );
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+                assertThat(executionInfo.includeCCSMetadata(), equalTo(responseExpectMeta));
                 assertThat(executionInfo.clusterAliases(), equalTo(Set.of("", "c1", "c2")));
                 assertCCSExecutionInfoDetails(executionInfo);
             }
         }
     }
 
     public void testEnrichCoordinatorThenAny() {
+        Tuple<Boolean, Boolean> includeCCSMetadata = randomIncludeCCSMetadata();
+        Boolean requestIncludeMeta = includeCCSMetadata.v1();
+        boolean responseExpectMeta = includeCCSMetadata.v2();
+
         String query = String.format(Locale.ROOT, """
             FROM *:events,events
             | eval ip= TO_STR(host)
@@ -354,7 +376,7 @@ public void testEnrichCoordinatorThenAny() {
             | stats c = COUNT(*) by vendor
             | sort vendor
             """, enrichHosts(Enrich.Mode.COORDINATOR), enrichVendors(Enrich.Mode.ANY));
-        try (EsqlQueryResponse resp = runQuery(query)) {
+        try (EsqlQueryResponse resp = runQuery(query, requestIncludeMeta)) {
             assertThat(
                 getValuesList(resp),
                 equalTo(
@@ -368,12 +390,17 @@ public void testEnrichCoordinatorThenAny() {
                 )
             );
             EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+            assertThat(executionInfo.includeCCSMetadata(), equalTo(responseExpectMeta));
             assertThat(executionInfo.clusterAliases(), equalTo(Set.of("", "c1", "c2")));
             assertCCSExecutionInfoDetails(executionInfo);
         }
     }
 
     public void testEnrichCoordinatorWithVendor() {
+        Tuple<Boolean, Boolean> includeCCSMetadata = randomIncludeCCSMetadata();
+        Boolean requestIncludeMeta = includeCCSMetadata.v1();
+        boolean responseExpectMeta = includeCCSMetadata.v2();
+
         for (Enrich.Mode hostMode : Enrich.Mode.values()) {
             String query = String.format(Locale.ROOT, """
                 FROM *:events,events
@@ -383,7 +410,7 @@ public void testEnrichCoordinatorWithVendor() {
                 | stats c = COUNT(*) by vendor
                 | sort vendor
                 """, enrichHosts(hostMode), enrichVendors(Enrich.Mode.COORDINATOR));
-            try (EsqlQueryResponse resp = runQuery(query)) {
+            try (EsqlQueryResponse resp = runQuery(query, requestIncludeMeta)) {
                 assertThat(
                     getValuesList(resp),
                     equalTo(
@@ -397,6 +424,7 @@ public void testEnrichCoordinatorWithVendor() {
                     )
                 );
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+                assertThat(executionInfo.includeCCSMetadata(), equalTo(responseExpectMeta));
                 assertThat(executionInfo.clusterAliases(), equalTo(Set.of("", "c1", "c2")));
                 assertCCSExecutionInfoDetails(executionInfo);
             }
@@ -405,6 +433,10 @@ public void testEnrichCoordinatorWithVendor() {
     }
 
     public void testEnrichRemoteWithVendor() {
+        Tuple<Boolean, Boolean> includeCCSMetadata = randomIncludeCCSMetadata();
+        Boolean requestIncludeMeta = includeCCSMetadata.v1();
+        boolean responseExpectMeta = includeCCSMetadata.v2();
+
         for (Enrich.Mode hostMode : List.of(Enrich.Mode.ANY, Enrich.Mode.REMOTE)) {
             var query = String.format(Locale.ROOT, """
                 FROM *:events,events
@@ -414,7 +446,7 @@ public void testEnrichRemoteWithVendor() {
                 | stats c = COUNT(*) by vendor
                 | sort vendor
                 """, enrichHosts(hostMode), enrichVendors(Enrich.Mode.REMOTE));
-            try (EsqlQueryResponse resp = runQuery(query)) {
+            try (EsqlQueryResponse resp = runQuery(query, requestIncludeMeta)) {
                 assertThat(
                     getValuesList(resp),
                     equalTo(
@@ -430,6 +462,7 @@ public void testEnrichRemoteWithVendor() {
                     )
                 );
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
+                assertThat(executionInfo.includeCCSMetadata(), equalTo(responseExpectMeta));
                 assertThat(executionInfo.clusterAliases(), equalTo(Set.of("", "c1", "c2")));
                 assertCCSExecutionInfoDetails(executionInfo);
             }
@@ -444,7 +477,7 @@ public void testTopNThenEnrichRemote() {
             | LIMIT 5
             | %s
             """, enrichHosts(Enrich.Mode.REMOTE));
-        var error = expectThrows(VerificationException.class, () -> runQuery(query).close());
+        var error = expectThrows(VerificationException.class, () -> runQuery(query, randomBoolean()).close());
         assertThat(error.getMessage(), containsString("ENRICH with remote policy can't be executed after LIMIT"));
     }
 
@@ -455,7 +488,7 @@ public void testLimitThenEnrichRemote() {
             | eval ip= TO_STR(host)
             | %s
             """, enrichHosts(Enrich.Mode.REMOTE));
-        var error = expectThrows(VerificationException.class, () -> runQuery(query).close());
+        var error = expectThrows(VerificationException.class, () -> runQuery(query, randomBoolean()).close());
         assertThat(error.getMessage(), containsString("ENRICH with remote policy can't be executed after LIMIT"));
     }
 
@@ -468,7 +501,7 @@ public void testAggThenEnrichRemote() {
             | %s
             | sort vendor
             """, enrichHosts(Enrich.Mode.ANY), enrichVendors(Enrich.Mode.REMOTE));
-        var error = expectThrows(VerificationException.class, () -> runQuery(query).close());
+        var error = expectThrows(VerificationException.class, () -> runQuery(query, randomBoolean()).close());
         assertThat(error.getMessage(), containsString("ENRICH with remote policy can't be executed after STATS"));
     }
 
@@ -480,20 +513,23 @@ public void testEnrichCoordinatorThenEnrichRemote() {
             | %s
             | sort vendor
             """, enrichHosts(Enrich.Mode.COORDINATOR), enrichVendors(Enrich.Mode.REMOTE));
-        var error = expectThrows(VerificationException.class, () -> runQuery(query).close());
+        var error = expectThrows(VerificationException.class, () -> runQuery(query, randomBoolean()).close());
         assertThat(
             error.getMessage(),
             containsString("ENRICH with remote policy can't be executed after another ENRICH with coordinator policy")
         );
     }
 
-    protected EsqlQueryResponse runQuery(String query) {
+    protected EsqlQueryResponse runQuery(String query, Boolean ccsMetadataInResponse) {
         EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
         request.query(query);
         request.pragmas(AbstractEsqlIntegTestCase.randomPragmas());
         if (randomBoolean()) {
             request.profile(true);
         }
+        if (ccsMetadataInResponse != null) {
+            request.includeCCSMetadata(ccsMetadataInResponse);
+        }
         return client(LOCAL_CLUSTER).execute(EsqlQueryAction.INSTANCE, request).actionGet(30, TimeUnit.SECONDS);
     }
 
@@ -516,6 +552,15 @@ private static void assertCCSExecutionInfoDetails(EsqlExecutionInfo executionInf
         }
     }
 
+    public static Tuple<Boolean, Boolean> randomIncludeCCSMetadata() {
+        return switch (randomIntBetween(1, 3)) {
+            case 1 -> new Tuple<>(Boolean.TRUE, Boolean.TRUE);
+            case 2 -> new Tuple<>(Boolean.FALSE, Boolean.FALSE);
+            case 3 -> new Tuple<>(null, Boolean.FALSE);
+            default -> throw new AssertionError("should not get here");
+        };
+    }
+
     public static class LocalStateEnrich extends LocalStateCompositeXPackPlugin {
 
         public LocalStateEnrich(final Settings settings, final Path configPath) throws Exception {