ES|QL: add support for include_execution_metadata parameter

docs/changelog/134446.yaml

@@ -0,0 +1,5 @@
+pr: 134446
+summary: Add support for `include_execution_metadata` parameter
+area: ES|QL
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java

@@ -82,7 +82,7 @@ public boolean isRemoteClusterServerEnabled() {
     private final Map<ProjectId, Map<String, RemoteClusterConnection>> remoteClusters;
     private final RemoteClusterCredentialsManager remoteClusterCredentialsManager;
     private final ProjectResolver projectResolver;
-    private final boolean canUseSkipUnavailable;
+    private final boolean crossProjectEnabled;
 
     RemoteClusterService(Settings settings, TransportService transportService, ProjectResolver projectResolver) {
         super(settings);
@@ -103,7 +103,7 @@ public boolean isRemoteClusterServerEnabled() {
          * TODO: This is not the right way to check if we're in CPS context and is more of a temporary measure since
          *  the functionality to do it the right way is not yet ready -- replace this code when it's ready.
          */
-        this.canUseSkipUnavailable = settings.getAsBoolean("serverless.cross_project.enabled", false) == false;
+        this.crossProjectEnabled = settings.getAsBoolean("serverless.cross_project.enabled", false);
     }
 
     /**
@@ -220,13 +220,17 @@ void ensureConnected(String clusterAlias, ActionListener<Void> listener) {
      * it returns an empty value where we default/fall back to true.
      */
     public Optional<Boolean> isSkipUnavailable(String clusterAlias) {
-        if (canUseSkipUnavailable == false) {
+        if (crossProjectEnabled) {
             return Optional.empty();
         } else {
             return Optional.of(getRemoteClusterConnection(clusterAlias).isSkipUnavailable());
         }
     }
 
+    public boolean crossProjectEnabled() {
+        return crossProjectEnabled;
+    }
+
     /**
      * Signifies if an error can be skipped for the specified cluster based on skip_unavailable, or,
      * allow_partial_search_results if in CPS-like environment.

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

@@ -350,7 +350,11 @@ protected EsqlQueryResponse runQuery(String query, Boolean ccsMetadataInResponse
         request.profile(randomInt(5) == 2);
         request.columnar(randomBoolean());
         if (ccsMetadataInResponse != null) {
-            request.includeCCSMetadata(ccsMetadataInResponse);
+            if (randomBoolean()) {
+                request.includeExecutionMetadata(ccsMetadataInResponse);
+            } else {
+                request.includeCCSMetadata(ccsMetadataInResponse);
+            }
         }
         return runQuery(request);
     }

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

@@ -55,8 +55,8 @@
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
 public class CrossClusterQueryIT extends AbstractCrossClusterTestCase {
-    private static final String IDX_ALIAS = "alias1";
-    private static final String FILTERED_IDX_ALIAS = "alias-filtered-1";
+    protected static final String IDX_ALIAS = "alias1";
+    protected static final String FILTERED_IDX_ALIAS = "alias-filtered-1";
 
     @Override
     protected Map<String, Boolean> skipUnavailableForRemoteClusters() {
@@ -234,6 +234,10 @@ public void testSearchesAgainstIndicesWithNoMappingsSkipUnavailableTrue() throws
     }
 
     public void testSearchesAgainstNonMatchingIndices() throws Exception {
+        testSearchesAgainstNonMatchingIndices(true);
+    }
+
+    protected void testSearchesAgainstNonMatchingIndices(boolean exceptionWithSkipUnavailableFalse) throws Exception {
         int numClusters = 3;
         Map<String, Object> testClusterInfo = setupClusters(numClusters);
         int localNumShards = (Integer) testClusterInfo.get("local.num_shards");
@@ -260,9 +264,11 @@ public void testSearchesAgainstNonMatchingIndices() throws Exception {
         {
             String q = "FROM logs*,cluster-a:nomatch";
             String expectedError = "Unknown index [cluster-a:nomatch]";
-            setSkipUnavailable(REMOTE_CLUSTER_1, false);
-            expectVerificationExceptionForQuery(q, expectedError, requestIncludeMeta);
-            setSkipUnavailable(REMOTE_CLUSTER_1, true);
+            if (exceptionWithSkipUnavailableFalse) {
+                setSkipUnavailable(REMOTE_CLUSTER_1, false);
+                expectVerificationExceptionForQuery(q, expectedError, requestIncludeMeta);
+                setSkipUnavailable(REMOTE_CLUSTER_1, true);
+            }
             try (EsqlQueryResponse resp = runQuery(q, requestIncludeMeta)) {
                 assertThat(getValuesList(resp).size(), greaterThanOrEqualTo(1));
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
@@ -406,9 +412,11 @@ public void testSearchesAgainstNonMatchingIndices() throws Exception {
             String remote2IndexName = randomFrom(remote2Index, IDX_ALIAS, FILTERED_IDX_ALIAS);
             String q = Strings.format("FROM %s*,cluster-a:nomatch,%s:%s*", localIndexName, REMOTE_CLUSTER_2, remote2IndexName);
             String expectedError = "Unknown index [cluster-a:nomatch]";
-            setSkipUnavailable(REMOTE_CLUSTER_1, false);
-            expectVerificationExceptionForQuery(q, expectedError, requestIncludeMeta);
-            setSkipUnavailable(REMOTE_CLUSTER_1, true);
+            if (exceptionWithSkipUnavailableFalse) {
+                setSkipUnavailable(REMOTE_CLUSTER_1, false);
+                expectVerificationExceptionForQuery(q, expectedError, requestIncludeMeta);
+                setSkipUnavailable(REMOTE_CLUSTER_1, true);
+            }
             try (EsqlQueryResponse resp = runQuery(q, requestIncludeMeta)) {
                 assertThat(getValuesList(resp).size(), greaterThanOrEqualTo(1));
                 EsqlExecutionInfo executionInfo = resp.getExecutionInfo();
@@ -443,7 +451,7 @@ record ExpectedCluster(String clusterAlias, String indexExpression, EsqlExecutio
      * Runs the provided query, expecting a VerificationError. It then runs the same query with a "| LIMIT 0"
      * extra processing step to ensure that ESQL coordinator-only operations throw the same VerificationError.
      */
-    private void expectVerificationExceptionForQuery(String query, String error, Boolean requestIncludeMeta) {
+    protected void expectVerificationExceptionForQuery(String query, String error, Boolean requestIncludeMeta) {
         VerificationException e = expectThrows(VerificationException.class, () -> runQuery(query, requestIncludeMeta));
         assertThat(e.getDetailedMessage(), containsString(error));
 
@@ -1017,4 +1025,24 @@ public void testMultiTypes() throws Exception {
             }
         }
     }
+
+    public void testNoBothIncludeCcsMetadataAndIncludeExecutionMetadata() throws Exception {
+        setupTwoClusters();
+        var query = "from logs-*,c*:logs-* | stats sum (v)";
+        EsqlQueryRequest request = EsqlQueryRequest.syncEsqlQueryRequest();
+        request.query(query);
+        request.pragmas(AbstractEsqlIntegTestCase.randomPragmas());
+        request.profile(randomInt(5) == 2);
+        request.columnar(randomBoolean());
+        request.includeCCSMetadata(randomBoolean());
+        request.includeExecutionMetadata(randomBoolean());
+
+        assertThat(
+            expectThrows(VerificationException.class, () -> runQuery(request)).getMessage(),
+            containsString(
+                "Both [include_execution_metadata] and [include_ccs_metadata] query parameters are set. "
+                    + "Use only [include_execution_metadata]"
+            )
+        );
+    }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlQueryRequest.java

@@ -45,7 +45,8 @@ public class EsqlQueryRequest extends org.elasticsearch.xpack.core.esql.action.E
     private String query;
     private boolean columnar;
     private boolean profile;
-    private boolean includeCCSMetadata;
+    private Boolean includeCCSMetadata;
+    private Boolean includeExecutionMetadata;
     private Locale locale;
     private QueryBuilder filter;
     private QueryPragmas pragmas = new QueryPragmas(Settings.EMPTY);
@@ -134,14 +135,22 @@ public void profile(boolean profile) {
         this.profile = profile;
     }
 
-    public void includeCCSMetadata(boolean include) {
+    public void includeCCSMetadata(Boolean include) {
         this.includeCCSMetadata = include;
     }
 
-    public boolean includeCCSMetadata() {
+    public Boolean includeCCSMetadata() {
         return includeCCSMetadata;
     }
 
+    public void includeExecutionMetadata(Boolean include) {
+        this.includeExecutionMetadata = include;
+    }
+
+    public Boolean includeExecutionMetadata() {
+        return includeExecutionMetadata;
+    }
+
     /**
      * Is profiling enabled?
      */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/RequestXContent.java

@@ -80,6 +80,7 @@ String fields() {
     private static final ParseField PROFILE_FIELD = new ParseField("profile");
     private static final ParseField ACCEPT_PRAGMA_RISKS = new ParseField("accept_pragma_risks");
     private static final ParseField INCLUDE_CCS_METADATA_FIELD = new ParseField("include_ccs_metadata");
+    private static final ParseField INCLUDE_EXECUTION_METADATA_FIELD = new ParseField("include_execution_metadata");
     static final ParseField TABLES_FIELD = new ParseField("tables");
 
     static final ParseField WAIT_FOR_COMPLETION_TIMEOUT = new ParseField("wait_for_completion_timeout");
@@ -105,6 +106,7 @@ private static void objectParserCommon(ObjectParser<EsqlQueryRequest, ?> parser)
         parser.declareObject(EsqlQueryRequest::filter, (p, c) -> AbstractQueryBuilder.parseTopLevelQuery(p), FILTER_FIELD);
         parser.declareBoolean(EsqlQueryRequest::acceptedPragmaRisks, ACCEPT_PRAGMA_RISKS);
         parser.declareBoolean(EsqlQueryRequest::includeCCSMetadata, INCLUDE_CCS_METADATA_FIELD);
+        parser.declareBoolean(EsqlQueryRequest::includeExecutionMetadata, INCLUDE_EXECUTION_METADATA_FIELD);
         parser.declareObject(
             EsqlQueryRequest::pragmas,
             (p, c) -> new QueryPragmas(Settings.builder().loadFromMap(p.map()).build()),

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/EsqlMediaTypeParser.java

@@ -45,6 +45,7 @@ public static MediaType getResponseMediaType(RestRequest request, EsqlQueryReque
         var mediaType = getResponseMediaType(request, (MediaType) null);
         validateColumnarRequest(esqlRequest.columnar(), mediaType);
         validateIncludeCCSMetadata(esqlRequest.includeCCSMetadata(), mediaType);
+        validateIncludeExecutionMetadata(esqlRequest.includeExecutionMetadata(), mediaType);
         validateProfile(esqlRequest.profile(), mediaType);
         return checkNonNullMediaType(mediaType, request);
     }
@@ -72,12 +73,18 @@ private static void validateColumnarRequest(boolean requestIsColumnar, MediaType
         }
     }
 
-    private static void validateIncludeCCSMetadata(boolean includeCCSMetadata, MediaType fromMediaType) {
-        if (includeCCSMetadata && fromMediaType instanceof TextFormat) {
+    private static void validateIncludeCCSMetadata(Boolean includeCCSMetadata, MediaType fromMediaType) {
+        if (Boolean.TRUE.equals(includeCCSMetadata) && fromMediaType instanceof TextFormat) {
             throw invalid("include_ccs_metadata");
         }
     }
 
+    private static void validateIncludeExecutionMetadata(Boolean includeExecutionMetadata, MediaType fromMediaType) {
+        if (Boolean.TRUE.equals(includeExecutionMetadata) && fromMediaType instanceof TextFormat) {
+            throw invalid("include_execution_metadata");
+        }
+    }
+
     private static void validateProfile(boolean profile, MediaType fromMediaType) {
         if (profile && fromMediaType instanceof TextFormat) {
             throw invalid("profile");

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/TransportEsqlQueryAction.java

@@ -356,10 +356,19 @@ private EsqlExecutionInfo getOrCreateExecutionInfo(Task task, EsqlQueryRequest r
     }
 
     private EsqlExecutionInfo createEsqlExecutionInfo(EsqlQueryRequest request) {
-        return new EsqlExecutionInfo(
-            clusterAlias -> remoteClusterService.isSkipUnavailable(clusterAlias).orElse(true),
-            request.includeCCSMetadata()
-        );
+        if (request.includeCCSMetadata() != null && request.includeExecutionMetadata() != null) {
+            throw new VerificationException(
+                "Both [include_execution_metadata] and [include_ccs_metadata] query parameters are set. "
+                    + "Use only [include_execution_metadata]"
+            );
+        }
+
+        Boolean includeCcsMetadata = request.includeExecutionMetadata();
+        if (includeCcsMetadata == null) {
+            // include_ccs_metadata is considered only if include_execution_metadata is not set
+            includeCcsMetadata = Boolean.TRUE.equals(request.includeCCSMetadata());
+        }
+        return new EsqlExecutionInfo(clusterAlias -> remoteClusterService.isSkipUnavailable(clusterAlias).orElse(true), includeCcsMetadata);
     }
 
     private EsqlQueryResponse toResponse(Task task, EsqlQueryRequest request, Configuration configuration, Result result) {

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

@@ -93,6 +93,18 @@ public void testIncludeCCSMetadataWithAcceptText() {
         );
     }
 
+    public void testIncludeExecutionMetadataWithAcceptText() {
+        var accept = randomFrom("text/plain", "text/csv", "text/tab-separated-values");
+        IllegalArgumentException e = expectThrows(
+            IllegalArgumentException.class,
+            () -> getResponseMediaType(reqWithAccept(accept), createCpsTestInstance(false, true, false))
+        );
+        assertEquals(
+            "Invalid use of [include_execution_metadata] argument: cannot be used in combination with [txt, csv, tsv] formats",
+            e.getMessage()
+        );
+    }
+
     public void testColumnarWithParamText() {
         IllegalArgumentException e = expectThrows(
             IllegalArgumentException.class,
@@ -121,6 +133,26 @@ public void testIncludeCCSMetadataWithNonJSONMediaTypesInParams() {
         }
     }
 
+    public void testIncludeExecutionMetadataWithNonJSONMediaTypesInParams() {
+        {
+            RestRequest restRequest = reqWithParams(Map.of("format", randomFrom("txt", "csv", "tsv")));
+            IllegalArgumentException e = expectThrows(
+                IllegalArgumentException.class,
+                () -> getResponseMediaType(restRequest, createCpsTestInstance(false, true, false))
+            );
+            assertEquals(
+                "Invalid use of [include_execution_metadata] argument: cannot be used in combination with [txt, csv, tsv] formats",
+                e.getMessage()
+            );
+        }
+        {
+            // check that no exception is thrown for the XContent types
+            RestRequest restRequest = reqWithParams(Map.of("format", randomFrom("SMILE", "YAML", "CBOR", "JSON")));
+            MediaType responseMediaType = getResponseMediaType(restRequest, createCpsTestInstance(true, true, false));
+            assertNotNull(responseMediaType);
+        }
+    }
+
     public void testProfileWithNonJSONMediaTypesInParams() {
         {
             RestRequest restRequest = reqWithParams(Map.of("format", randomFrom("txt", "csv", "tsv")));
@@ -180,4 +212,11 @@ protected EsqlQueryRequest createTestInstance(boolean columnar, boolean includeC
         request.profile(profile);
         return request;
     }
+
+    protected EsqlQueryRequest createCpsTestInstance(boolean columnar, boolean includeExecutionMetadata, boolean profile) {
+        var request = createTestInstance(columnar);
+        request.includeExecutionMetadata(includeExecutionMetadata);
+        request.profile(profile);
+        return request;
+    }
 }