ES|QL: Make _tsid available in metadata

docs/changelog/135204.yaml

@@ -0,0 +1,6 @@
+pr: 135204
+summary: Make `_tsid` available in metadata
+area: ES|QL
+type: enhancement
+issues:
+ - 133205

server/src/main/java/org/elasticsearch/index/mapper/TimeSeriesIdFieldMapper.java

@@ -134,6 +134,15 @@ public Query termQuery(Object value, SearchExecutionContext context) {
  throw new IllegalArgumentException("[" + NAME + "] is not searchable");
  }
 
+ @Override
+ public Object valueForDisplay(Object value) {
+ if (value == null) {
+ return null;
+ }
+ BytesRef binaryValue = (BytesRef) value;
+ return TimeSeriesIdFieldMapper.encodeTsid(binaryValue);
+ }
+
  @Override
  public BlockLoader blockLoader(BlockLoaderContext blContext) {
  return new BlockDocValuesReader.BytesRefsFromOrdsBlockLoader(name());

server/src/test/java/org/elasticsearch/index/mapper/TimeSeriesIdFieldMapperTests.java

@@ -768,4 +768,23 @@ public void testParseWithDynamicMappingInvalidRoutingHash() {
  });
  assertThat(failure.getMessage(), equalTo("[5:1] failed to parse: Illegal base64 character 20"));
  }
+
+ public void testValueForDisplay() throws Exception {
+ DocumentMapper docMapper = createDocumentMapper("a", mapping(b -> {
+ b.startObject("a").field("type", "keyword").field("time_series_dimension", true).endObject();
+ b.startObject("b").field("type", "long").field("time_series_dimension", true).endObject();
+ }));
+
+ ParsedDocument doc = parseDocument(docMapper, b -> b.field("a", "value").field("b", 100));
+ BytesRef tsidBytes = doc.rootDoc().getBinaryValue("_tsid");
+ assertThat(tsidBytes, not(nullValue()));
+
+ TimeSeriesIdFieldMapper.TimeSeriesIdFieldType fieldType = TimeSeriesIdFieldMapper.FIELD_TYPE;
+ Object displayValue = fieldType.valueForDisplay(tsidBytes);
+ Object encodedValue = TimeSeriesIdFieldMapper.encodeTsid(tsidBytes);
+
+ assertThat(displayValue, equalTo(encodedValue));
+ assertThat(displayValue.getClass(), is(String.class));
+ assertThat(fieldType.valueForDisplay(null), nullValue());
+ }
 }

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/expression/MetadataAttribute.java

@@ -45,7 +45,8 @@ public class MetadataAttribute extends TypedAttribute {
  Map.entry(IgnoredFieldMapper.NAME, new MetadataAttributeConfiguration(DataType.KEYWORD, true)),
  Map.entry(SourceFieldMapper.NAME, new MetadataAttributeConfiguration(DataType.SOURCE, false)),
  Map.entry(IndexModeFieldMapper.NAME, new MetadataAttributeConfiguration(DataType.KEYWORD, true)),
- Map.entry(SCORE, new MetadataAttributeConfiguration(DataType.DOUBLE, false))
+ Map.entry(SCORE, new MetadataAttributeConfiguration(DataType.DOUBLE, false)),
+ Map.entry(TSID_FIELD, new MetadataAttributeConfiguration(DataType.TSID_DATA_TYPE, false))
  );
 
  private record MetadataAttributeConfiguration(DataType dataType, boolean searchable) {}

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/DataType.java

@@ -414,7 +414,7 @@ public enum DataType implements Writeable {
  }
 
  private static final Collection<DataType> TYPES = Arrays.stream(values())
- .filter(d -> d != DOC_DATA_TYPE && d != TSID_DATA_TYPE)
+ .filter(d -> d != DOC_DATA_TYPE)
  .sorted(Comparator.comparing(DataType::typeName))
  .toList();
 

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java

@@ -45,6 +45,7 @@
 import org.elasticsearch.geometry.utils.Geohash;
 import org.elasticsearch.h3.H3;
 import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.mapper.RoutingPathFields;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.aggregations.bucket.geogrid.GeoTileUtils;
@@ -901,8 +902,9 @@ public static Literal randomLiteral(DataType type) {
  throw new UncheckedIOException(e);
  }
  }
+ case TSID_DATA_TYPE -> randomTsId().toBytesRef();
  case DENSE_VECTOR -> Arrays.asList(randomArray(10, 10, i -> new Float[10], ESTestCase::randomFloat));
- case UNSUPPORTED, OBJECT, DOC_DATA_TYPE, TSID_DATA_TYPE, PARTIAL_AGG -> throw new IllegalArgumentException(
+ case UNSUPPORTED, OBJECT, DOC_DATA_TYPE, PARTIAL_AGG -> throw new IllegalArgumentException(
  "can't make random values for [" + type.typeName() + "]"
  );
  }, type);
@@ -918,6 +920,22 @@ static Version randomVersion() {
  };
  }
 
+ static BytesReference randomTsId() {
+ RoutingPathFields routingPathFields = new RoutingPathFields(null);
+
+ int numDimensions = randomIntBetween(1, 4);
+ for (int i = 0; i < numDimensions; i++) {
+ String fieldName = "dim" + i;
+ if (randomBoolean()) {
+ routingPathFields.addString(fieldName, randomAlphaOfLength(randomIntBetween(3, 10)));
+ } else {
+ routingPathFields.addLong(fieldName, randomLongBetween(1, 1000));
+ }
+ }
+
+ return routingPathFields.buildHash();
+ }
+
  public static WildcardLike wildcardLike(Expression left, String exp) {
  return new WildcardLike(EMPTY, left, new WildcardPattern(exp), false);
  }

x-pack/plugin/esql/qa/testFixtures/src/main/resources/k8s-timeseries.csv-spec

@@ -525,3 +525,37 @@ true | two | 2024-05-10T00:18:00.000Z
 false | two | 2024-05-10T00:20:00.000Z
 false | two | 2024-05-10T00:22:00.000Z
 ;
+
+tsidMetadataAttributeCount
+required_capability: ts_command_v0
+required_capability: metadata_tsid_field
+
+TS k8s METADATA _tsid
+| STATS cnt = count_distinct(_tsid)
+;
+
+cnt:long
+9
+;
+
+tsidMetadataAttributeAggregation
+required_capability: ts_command_v0
+required_capability: metadata_tsid_field
+
+TS k8s METADATA _tsid
+| STATS cnt = count_distinct(_tsid) BY cluster, pod
+| SORT cluster
+;
+ignoreOrder:true
+
+cnt:long | cluster:keyword | pod:keyword
+1 | staging | one
+1 | staging | two
+1 | staging | three
+1 | qa | one
+1 | qa | two
+1 | qa | three
+1 | prod | one
+1 | prod | two
+1 | prod | three
+;

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

@@ -0,0 +1,168 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.action;
+
+import org.elasticsearch.common.Randomness;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.elasticsearch.index.mapper.DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.getValuesList;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasSize;
+
+// @TestLogging(value = "org.elasticsearch.xpack.esql.session:DEBUG", reason = "to better understand planning")
+public class CrossClusterTimeSeriesIT extends AbstractCrossClusterTestCase {
+
+ private static final String INDEX_NAME = "hosts";
+
+ record Doc(String host, String cluster, long timestamp, int requestCount, double cpu, ByteSizeValue memory) {}
+
+ public void testTsIdMetadataInResponse() {
+ populateTimeSeriesIndex(LOCAL_CLUSTER, INDEX_NAME);
+ populateTimeSeriesIndex(REMOTE_CLUSTER_1, INDEX_NAME);
+
+ try (EsqlQueryResponse resp = runQuery("TS hosts, cluster-a:hosts METADATA _tsid", Boolean.TRUE)) {
+ assertNotNull(
+ resp.columns().stream().map(ColumnInfoImpl::name).filter(name -> name.equalsIgnoreCase("_tsid")).findFirst().orElse(null)
+ );
+
+ assertCCSExecutionInfoDetails(resp.getExecutionInfo(), 2);
+ }
+ }
+
+ public void testTsIdMetadataInResponseWithFailure() {
+ populateTimeSeriesIndex(LOCAL_CLUSTER, INDEX_NAME);
+ populateTimeSeriesIndex(REMOTE_CLUSTER_1, INDEX_NAME);
+
+ try (
+ EsqlQueryResponse resp = runQuery(
+ "TS hosts, cluster-a:hosts METADATA _tsid | WHERE host IS NOT NULL | STATS cnt = count_distinct(_tsid)",
+ Boolean.TRUE
+ )
+ ) {
+ List<List<Object>> values = getValuesList(resp);
+ assertThat(values, hasSize(1));
+ assertNotNull(values.getFirst().getFirst());
+ assertCCSExecutionInfoDetails(resp.getExecutionInfo(), 2);
+ }
+ }
+
+ private void populateTimeSeriesIndex(String clusterAlias, String indexName) {
+ int numShards = randomIntBetween(1, 5);
+ String clusterTag = Strings.isEmpty(clusterAlias) ? "local" : clusterAlias;
+ Settings settings = Settings.builder()
+ .put("mode", "time_series")
+ .putList("routing_path", List.of("host", "cluster"))
+ .put("index.number_of_shards", numShards)
+ .build();
+
+ client(clusterAlias).admin()
+ .indices()
+ .prepareCreate(indexName)
+ .setSettings(settings)
+ .setMapping(
+ "@timestamp",
+ "type=date",
+ "host",
+ "type=keyword,time_series_dimension=true",
+ "cluster",
+ "type=keyword,time_series_dimension=true",
+ "cpu",
+ "type=double,time_series_metric=gauge",
+ "memory",
+ "type=long,time_series_metric=gauge",
+ "request_count",
+ "type=integer,time_series_metric=counter",
+ "cluster_tag",
+ "type=keyword"
+ )
+ .get();
+
+ final List<Doc> docs = getRandomDocs();
+
+ for (Doc doc : docs) {
+ client().prepareIndex(indexName)
+ .setSource(
+ "@timestamp",
+ doc.timestamp,
+ "host",
+ doc.host,
+ "cluster",
+ doc.cluster,
+ "cpu",
+ doc.cpu,
+ "memory",
+ doc.memory.getBytes(),
+ "request_count",
+ doc.requestCount,
+ "cluster_tag",
+ clusterTag
+ )
+ .get();
+ }
+ client().admin().indices().prepareRefresh(indexName).get();
+ }
+
+ private List<Doc> getRandomDocs() {
+ final List<Doc> docs = new ArrayList<>();
+
+ Map<String, String> hostToClusters = new HashMap<>();
+ for (int i = 0; i < 5; i++) {
+ hostToClusters.put("p" + i, randomFrom("qa", "prod"));
+ }
+ long timestamp = DEFAULT_DATE_TIME_FORMATTER.parseMillis("2024-04-15T00:00:00Z");
+
+ Map<String, Integer> requestCounts = new HashMap<>();
+ int numDocs = between(20, 100);
+ for (int i = 0; i < numDocs; i++) {
+ List<String> hosts = randomSubsetOf(between(1, hostToClusters.size()), hostToClusters.keySet());
+ timestamp += between(1, 10) * 1000L;
+ for (String host : hosts) {
+ var requestCount = requestCounts.compute(host, (k, curr) -> {
+ if (curr == null || randomInt(100) <= 20) {
+ return randomIntBetween(0, 10);
+ } else {
+ return curr + randomIntBetween(1, 10);
+ }
+ });
+ int cpu = randomIntBetween(0, 100);
+ ByteSizeValue memory = ByteSizeValue.ofBytes(randomIntBetween(1024, 1024 * 1024));
+ docs.add(new Doc(host, hostToClusters.get(host), timestamp, requestCount, cpu, memory));
+ }
+ }
+
+ Randomness.shuffle(docs);
+
+ return docs;
+ }
+
+ private static void assertCCSExecutionInfoDetails(EsqlExecutionInfo executionInfo, int expectedNumClusters) {
+ assertNotNull(executionInfo);
+ assertThat(executionInfo.overallTook().millis(), greaterThanOrEqualTo(0L));
+ assertTrue(executionInfo.isCrossClusterSearch());
+ assertThat(executionInfo.getClusters().size(), equalTo(expectedNumClusters));
+
+ List<EsqlExecutionInfo.Cluster> clusters = executionInfo.clusterAliases().stream().map(executionInfo::getCluster).toList();
+
+ for (EsqlExecutionInfo.Cluster cluster : clusters) {
+ assertThat(cluster.getTook().millis(), greaterThanOrEqualTo(0L));
+ assertThat(cluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL));
+ assertThat(cluster.getSkippedShards(), equalTo(0));
+ assertThat(cluster.getFailedShards(), equalTo(0));
+ }
+ }
+
+}

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

@@ -686,4 +686,21 @@ public void testNullMetricsAreSkipped() {
  assertEquals("Did not filter nulls on counter type", 50, resp.documentsFound());
  }
  }
+
+ public void testTSIDMetadataAttribute() {
+ List<ColumnInfoImpl> columns = List.of(
+ new ColumnInfoImpl("_tsid", DataType.TSID_DATA_TYPE, null),
+ new ColumnInfoImpl("cluster", DataType.KEYWORD, null)
+ );
+
+ try (EsqlQueryResponse resp = run(" TS hosts METADATA _tsid | KEEP _tsid, cluster | LIMIT 1")) {
+ assertThat(resp.columns(), equalTo(columns));
+
+ List<List<Object>> values = EsqlTestUtils.getValuesList(resp);
+ assertThat(values, hasSize(1));
+ assertThat(values.getFirst().get(0), Matchers.notNullValue());
+ assertThat(values.getFirst().get(1), Matchers.notNullValue());
+ }
+ }
+
 }

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

@@ -1558,9 +1558,12 @@ public enum Cap {
 
  INLINE_STATS_FIX_OPTIMIZED_AS_LOCAL_RELATION(INLINESTATS_V11.enabled),
 
- DENSE_VECTOR_AGG_METRIC_DOUBLE_IF_FNS
+ DENSE_VECTOR_AGG_METRIC_DOUBLE_IF_FNS,
 
- ;
+ /**
+ * Support for requesting the "_tsid" metadata field.
+ */
+ METADATA_TSID_FIELD;
 
  private final boolean enabled;
 

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

@@ -18,6 +18,7 @@
 import org.elasticsearch.compute.data.FloatBlock;
 import org.elasticsearch.compute.data.IntBlock;
 import org.elasticsearch.compute.data.LongBlock;
+import org.elasticsearch.index.mapper.TimeSeriesIdFieldMapper;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xcontent.XContentParser;
@@ -199,8 +200,16 @@ protected XContentBuilder valueToXContent(XContentBuilder builder, ToXContent.Pa
  return builder.value(((FloatBlock) block).getFloat(valueIndex));
  }
  };
- case DATE_PERIOD, TIME_DURATION, DOC_DATA_TYPE, TSID_DATA_TYPE, SHORT, BYTE, OBJECT, FLOAT, HALF_FLOAT, SCALED_FLOAT,
- PARTIAL_AGG -> throw new IllegalArgumentException("can't convert values of type [" + columnInfo.type() + "]");
+ case TSID_DATA_TYPE -> new PositionToXContent(block) {
+ @Override
+ protected XContentBuilder valueToXContent(XContentBuilder builder, ToXContent.Params params, int valueIndex)
+ throws IOException {
+ BytesRef bytesRef = ((BytesRefBlock) block).getBytesRef(valueIndex, scratch);
+ return builder.value(TimeSeriesIdFieldMapper.encodeTsid(bytesRef));
+ }
+ };
+ case DATE_PERIOD, TIME_DURATION, DOC_DATA_TYPE, SHORT, BYTE, OBJECT, FLOAT, HALF_FLOAT, SCALED_FLOAT, PARTIAL_AGG ->
+ throw new IllegalArgumentException("can't convert values of type [" + columnInfo.type() + "]");
  };
  }
 }