ES|QL: add support for include_execution_metadata parameter (#134446)

Author: luigidellaquila

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;
+ }
 }