Introduce basic support for synthetic ids (#136810)

This change introduces basic support for the synthetic id feature, in which 
document identifiers (_id) are not indexed but instead computed at runtime 
from other document fields (_tsid and @timestamp).

This pull request adds a new tsdb_synthetic_id flag along with a new 
index.mapping.use_synthetic_id index setting that can only be applied to 
time series indices. When both are enabled, the index uses a specialized 
Lucene codec that ensures the _id field is never indexed, and also overrides 
the field informations when a segment is opened to expose the _id field as 
if it has postings. This way it is still possible to soft-update Lucene documents
 using a term value while not paying the price of indexing such a field. Note 
that the synthetic _id field is still declared as stored since we plan to use the 
stored value in the future in a bloom filter.

This change also introduces a new posting format to support synthetic _id. 
This implementation is not important for now: it only exists to demonstrate
 that doc values updates can work with a synthetic id posting format. The 
implementation uses the current _id format of TSDB indices which is hashed 
from a routing hash value, the tsid and the timestamp. This format cannot be 
unhashed to extract the real _tsid value and is also not naturally sorted, 
something that Lucene expects when applying doc values updates. To work 
around this the integration test only updates the first documents of the 
segments. In a follow up we'd like to change the _id format for TSDS to be
 [_tsid, @timestamp] and maybe also include the routing hash.

Relates ES-13191
Relates #136304

Author: tlrx

modules/data-streams/src/internalClusterTest/java/org/elasticsearch/datastreams/TSDBSyntheticIdsIT.java

@@ -0,0 +1,279 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.datastreams;
+
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.diskusage.AnalyzeIndexDiskUsageRequest;
+import org.elasticsearch.action.admin.indices.diskusage.AnalyzeIndexDiskUsageTestUtils;
+import org.elasticsearch.action.admin.indices.diskusage.IndexDiskUsageStats;
+import org.elasticsearch.action.admin.indices.diskusage.TransportAnalyzeIndexDiskUsageAction;
+import org.elasticsearch.action.admin.indices.template.put.TransportPutComposableIndexTemplateAction;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
+import org.elasticsearch.cluster.metadata.Template;
+import org.elasticsearch.common.compress.CompressedXContent;
+import org.elasticsearch.common.time.DateFormatter;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.IndexSettings;
+import org.elasticsearch.index.mapper.IdFieldMapper;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.test.ESIntegTestCase;
+import org.elasticsearch.test.InternalSettingsPlugin;
+import org.elasticsearch.test.junit.annotations.TestLogging;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentFactory;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Random;
+
+import static org.elasticsearch.common.time.FormatNames.STRICT_DATE_OPTIONAL_TIME;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+
+/**
+ * Test suite for time series indices that use synthetic ids for documents.
+ * <p>
+ * Synthetic _id fields are not indexed in Lucene, instead they are generated on demand by concatenating the values of two other fields of
+ * the document (typically the {@code @timestamp} and {@code _tsid} fields).
+ * </p>
+ */
+@LuceneTestCase.SuppressCodecs("*") // requires codecs used in production only
+public class TSDBSyntheticIdsIT extends ESIntegTestCase {
+
+    private static final DateFormatter DATE_FORMATTER = DateFormatter.forPattern(STRICT_DATE_OPTIONAL_TIME.getName());
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        var plugins = new ArrayList<>(super.nodePlugins());
+        plugins.add(InternalSettingsPlugin.class);
+        plugins.add(DataStreamsPlugin.class);
+        return plugins;
+    }
+
+    public void testInvalidIndexMode() {
+        assumeTrue("Test should only run with feature flag", IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG);
+        final var indexName = randomIdentifier();
+        var randomNonTsdbIndexMode = randomValueOtherThan(IndexMode.TIME_SERIES, () -> randomFrom(IndexMode.values()));
+
+        var exception = expectThrows(
+            IllegalArgumentException.class,
+            () -> createIndex(
+                indexName,
+                indexSettings(1, 0).put(IndexSettings.MODE.getKey(), randomNonTsdbIndexMode)
+                    .put(IndexSettings.USE_SYNTHETIC_ID.getKey(), true)
+                    .build()
+            )
+        );
+        assertThat(
+            exception.getMessage(),
+            containsString(
+                "The setting ["
+                    + IndexSettings.USE_SYNTHETIC_ID.getKey()
+                    + "] is only permitted when [index.mode] is set to [TIME_SERIES]. Current mode: ["
+                    + randomNonTsdbIndexMode.getName().toUpperCase(Locale.ROOT)
+                    + "]."
+            )
+        );
+    }
+
+    @TestLogging(reason = "debug", value = "org.elasticsearch.index.engine.Engine:TRACE")
+    public void testSyntheticId() throws Exception {
+        assumeTrue("Test should only run with feature flag", IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG);
+        final var indexName = randomIdentifier();
+        putDataStreamTemplate(random(), indexName);
+
+        final var timestamp = Instant.now();
+
+        // Index 5 docs in datastream
+        var results = createDocuments(
+            indexName,
+            document(timestamp, "vm-dev01", "cpu-load", 0),                                // will be updated
+            document(timestamp.plusSeconds(2), "vm-dev01", "cpu-load", 1),    // will be deleted
+            document(timestamp, "vm-dev02", "cpu-load", 2),
+            document(timestamp.plusSeconds(2), "vm-dev03", "cpu-load", 3),
+            document(timestamp.plusSeconds(3), "vm-dev03", "cpu-load", 4)
+        );
+
+        // Verify documents
+        assertThat(results[0].getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
+        assertThat(results[0].getVersion(), equalTo(1L));
+
+        assertThat(results[1].getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
+        assertThat(results[1].getVersion(), equalTo(1L));
+
+        assertThat(results[2].getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
+        assertThat(results[2].getVersion(), equalTo(1L));
+
+        assertThat(results[3].getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
+        assertThat(results[3].getVersion(), equalTo(1L));
+
+        assertThat(results[4].getResponse().getResult(), equalTo(DocWriteResponse.Result.CREATED));
+        assertThat(results[4].getVersion(), equalTo(1L));
+
+        final var docIndex = results[1].getIndex();
+        final var docId = results[1].getId();
+
+        enum Operation {
+            FLUSH,
+            REFRESH,
+            NONE
+        }
+        switch (randomFrom(Operation.values())) {
+            case FLUSH:
+                flush(indexName);
+                break;
+            case REFRESH:
+                refresh(indexName);
+                break;
+            case NONE:
+            default:
+                break;
+        }
+
+        // Get by synthetic _id
+        // Note: before synthetic _id this would have required postings on disks
+        var getResponse = client().prepareGet(docIndex, docId).setFetchSource(true).execute().actionGet();
+        assertThat(getResponse.isExists(), equalTo(true));
+        assertThat(getResponse.getVersion(), equalTo(1L));
+        var source = asInstanceOf(Map.class, getResponse.getSourceAsMap().get("metric"));
+        assertThat(asInstanceOf(Integer.class, source.get("value")), equalTo(1));
+
+        // Update by synthetic _id
+        // Note: it doesn't work, is that expected? Is is blocked by IndexRouting.ExtractFromSource.updateShard
+        var exception = expectThrows(IllegalArgumentException.class, () -> {
+            var doc = document(timestamp, "vm-dev01", "cpu-load", 10); // update
+            client().prepareUpdate(docIndex, docId).setDoc(doc).get();
+        });
+        assertThat(
+            exception.getMessage(),
+            containsString("update is not supported because the destination index [" + docIndex + "] is in time_series mode")
+        );
+
+        // Delete by synthetic _id
+        var deleteResponse = client().prepareDelete(docIndex, docId).get();
+        assertThat(deleteResponse.getId(), equalTo(docId));
+        assertThat(deleteResponse.getResult(), equalTo(DocWriteResponse.Result.DELETED));
+        assertThat(deleteResponse.getVersion(), equalTo(2L));
+
+        // Index more docs
+        // TODO Randomize this to have segments only composed of deleted docs
+        createDocuments(
+            indexName,
+            document(timestamp.plusSeconds(4), "vm-dev03", "cpu-load", 5),
+            document(timestamp.plusSeconds(5), "vm-dev03", "cpu-load", 6)
+        );
+
+        flushAndRefresh(indexName);
+
+        // Check that synthetic _id field has no postings on disk
+        var diskUsage = diskUsage(docIndex);
+        var diskUsageIdField = AnalyzeIndexDiskUsageTestUtils.getPerFieldDiskUsage(diskUsage, IdFieldMapper.NAME);
+        assertThat("_id field should not have postings on disk", diskUsageIdField.getInvertedIndexBytes(), equalTo(0L));
+
+        // TODO Search datastream and count hits
+    }
+
+    private static XContentBuilder document(Instant timestamp, String hostName, String metricField, Integer metricValue)
+        throws IOException {
+        var source = XContentFactory.jsonBuilder();
+        source.startObject();
+        {
+            source.field("@timestamp", DATE_FORMATTER.format(timestamp));
+            source.field("hostname", hostName);
+            source.startObject("metric");
+            {
+                source.field("field", metricField);
+                source.field("value", metricValue);
+
+            }
+            source.endObject();
+        }
+        source.endObject();
+        return source;
+    }
+
+    private static BulkItemResponse[] createDocuments(String indexName, XContentBuilder... docs) throws IOException {
+        assertThat(docs, notNullValue());
+        final var client = client();
+        var bulkRequest = client.prepareBulk();
+        for (var doc : docs) {
+            bulkRequest.add(client.prepareIndex(indexName).setOpType(DocWriteRequest.OpType.CREATE).setSource(doc));
+        }
+        var bulkResponse = bulkRequest.get();
+        assertNoFailures(bulkResponse);
+        return bulkResponse.getItems();
+    }
+
+    private static void putDataStreamTemplate(Random random, String indexPattern) throws IOException {
+        final var settings = indexSettings(1, 0).put(IndexSettings.MODE.getKey(), IndexMode.TIME_SERIES.getName())
+            .put(IndexSettings.BLOOM_FILTER_ID_FIELD_ENABLED_SETTING.getKey(), false)
+            .put(IndexSettings.INDEX_REFRESH_INTERVAL_SETTING.getKey(), -1)
+            .put(IndexSettings.USE_SYNTHETIC_ID.getKey(), true);
+
+        final var mappings = """
+            {
+                "_doc": {
+                    "properties": {
+                        "@timestamp": {
+                            "type": "date"
+                        },
+                        "hostname": {
+                            "type": "keyword",
+                            "time_series_dimension": true
+                        },
+                        "metric": {
+                            "properties": {
+                                "field": {
+                                    "type": "keyword",
+                                    "time_series_dimension": true
+                                },
+                                "value": {
+                                    "type": "integer",
+                                    "time_series_metric": "counter"
+                                }
+                            }
+                        }
+                    }
+                }
+            }""";
+
+        var putTemplateRequest = new TransportPutComposableIndexTemplateAction.Request(getTestClass().getName().toLowerCase(Locale.ROOT))
+            .indexTemplate(
+                ComposableIndexTemplate.builder()
+                    .indexPatterns(List.of(indexPattern))
+                    .template(new Template(settings.build(), new CompressedXContent(mappings), null))
+                    .dataStreamTemplate(new ComposableIndexTemplate.DataStreamTemplate(false, false))
+                    .build()
+            );
+        assertAcked(client().execute(TransportPutComposableIndexTemplateAction.TYPE, putTemplateRequest).actionGet());
+    }
+
+    private static IndexDiskUsageStats diskUsage(String indexName) {
+        var diskUsageResponse = client().execute(
+            TransportAnalyzeIndexDiskUsageAction.TYPE,
+            new AnalyzeIndexDiskUsageRequest(new String[] { indexName }, AnalyzeIndexDiskUsageRequest.DEFAULT_INDICES_OPTIONS, false)
+        ).actionGet();
+
+        var indexDiskUsageStats = AnalyzeIndexDiskUsageTestUtils.getIndexStats(diskUsageResponse, indexName);
+        assertNotNull(indexDiskUsageStats);
+        return indexDiskUsageStats;
+    }
+}

server/src/main/java/org/elasticsearch/common/settings/IndexScopedSettings.java

@@ -246,6 +246,9 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
         if (IndexSettings.DOC_VALUES_SKIPPER) {
             settings.add(IndexSettings.USE_DOC_VALUES_SKIPPER);
         }
+        if (IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG) {
+            settings.add(IndexSettings.USE_SYNTHETIC_ID);
+        }
         settings.add(IndexSettings.INDEX_MAPPING_EXCLUDE_SOURCE_VECTORS_SETTING);
         BUILT_IN_INDEX_SETTINGS = Collections.unmodifiableSet(settings);
     };

server/src/main/java/org/elasticsearch/index/IndexSettings.java

@@ -676,6 +676,44 @@ public boolean isES87TSDBCodecEnabled() {
         Property.Final
     );
 
+    public static final boolean TSDB_SYNTHETIC_ID_FEATURE_FLAG = new FeatureFlag("tsdb_synthetic_id").isEnabled();
+    public static final Setting<Boolean> USE_SYNTHETIC_ID = Setting.boolSetting(
+        "index.mapping.use_synthetic_id",
+        false,
+        new Setting.Validator<>() {
+            @Override
+            public void validate(Boolean value) {}
+
+            @Override
+            public void validate(Boolean enabled, Map<Setting<?>, Object> settings) {
+                if (enabled) {
+                    // Verify if index mode is TIME_SERIES
+                    var indexMode = (IndexMode) settings.get(MODE);
+                    if (indexMode != IndexMode.TIME_SERIES) {
+                        throw new IllegalArgumentException(
+                            String.format(
+                                Locale.ROOT,
+                                "The setting [%s] is only permitted when [%s] is set to [%s]. Current mode: [%s].",
+                                USE_SYNTHETIC_ID.getKey(),
+                                MODE.getKey(),
+                                IndexMode.TIME_SERIES.name(),
+                                indexMode.name()
+                            )
+                        );
+                    }
+                }
+            }
+
+            @Override
+            public Iterator<Setting<?>> settings() {
+                List<Setting<?>> list = List.of(MODE);
+                return list.iterator();
+            }
+        },
+        Property.IndexScope,
+        Property.Final
+    );
+
     /**
      * The {@link IndexMode "mode"} of the index.
      */
@@ -937,6 +975,7 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
     private final boolean recoverySourceEnabled;
     private final boolean recoverySourceSyntheticEnabled;
     private final boolean useDocValuesSkipper;
+    private final boolean tsdbSyntheticId;
 
     /**
      * The maximum number of refresh listeners allows on this shard.
@@ -1123,6 +1162,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             && scopedSettings.get(RECOVERY_USE_SYNTHETIC_SOURCE_SETTING);
         useDocValuesSkipper = DOC_VALUES_SKIPPER && scopedSettings.get(USE_DOC_VALUES_SKIPPER);
         seqNoIndexOptions = scopedSettings.get(SEQ_NO_INDEX_OPTIONS_SETTING);
+        tsdbSyntheticId = TSDB_SYNTHETIC_ID_FEATURE_FLAG && scopedSettings.get(USE_SYNTHETIC_ID);
+        assert tsdbSyntheticId == false || mode == IndexMode.TIME_SERIES : mode;
         if (recoverySourceSyntheticEnabled) {
             if (DiscoveryNode.isStateless(settings)) {
                 throw new IllegalArgumentException("synthetic recovery source is only allowed in stateful");
@@ -1855,6 +1896,13 @@ public boolean useDocValuesSkipper() {
         return useDocValuesSkipper;
     }
 
+    /**
+     * @return Whether the index is a time-series index that use synthetic ids.
+     */
+    public boolean useTsdbSyntheticId() {
+        return tsdbSyntheticId;
+    }
+
     /**
      * The bounds for {@code @timestamp} on this index or
      * {@code null} if there are no bounds.

server/src/main/java/org/elasticsearch/index/codec/CodecService.java

@@ -16,6 +16,8 @@
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.codec.tsdb.TSDBSyntheticIdCodec;
 import org.elasticsearch.index.codec.zstd.Zstd814StoredFieldsFormat;
 import org.elasticsearch.index.mapper.MapperService;
 
@@ -65,12 +67,20 @@ public CodecService(@Nullable MapperService mapperService, BigArrays bigArrays)
         for (String codec : Codec.availableCodecs()) {
             codecs.put(codec, Codec.forName(codec));
         }
+        final boolean useTsdbSyntheticId = mapperService != null && mapperService.getIndexSettings().useTsdbSyntheticId();
+        assert useTsdbSyntheticId == false || mapperService.getIndexSettings().getMode() == IndexMode.TIME_SERIES;
+
         this.codecs = codecs.entrySet().stream().collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> {
-            var codec = e.getValue();
-            if (codec instanceof DeduplicateFieldInfosCodec) {
-                return codec;
+            Codec codec;
+            if (e.getValue() instanceof DeduplicateFieldInfosCodec dedupCodec) {
+                codec = dedupCodec;
+            } else {
+                codec = new DeduplicateFieldInfosCodec(e.getValue().getName(), e.getValue());
+            }
+            if (useTsdbSyntheticId && codec instanceof TSDBSyntheticIdCodec == false) {
+                codec = new TSDBSyntheticIdCodec(codec.getName(), codec);
             }
-            return new DeduplicateFieldInfosCodec(codec.getName(), codec);
+            return codec;
         }));
     }