Use a new synthetic _id format for time-series datastreams (#137274)

This pull request follows #136810 and introduces a new format for 
documents _id fields in time-series datastreams. In order to test this 
new format, the TSDB synthetic terms and postings format 
implementations are also changed.

The current _id format is composed of a routing hash, the hashed 
value of the _tsid and a timestamp. While it is possible to extract the 
routing hash and the timestamp from a document _id value, it is 
not possible to extract the original _tsid value.

This is an issue for synthetic _id as the document _id value is not
indexed anymore: instead the synthetic _id is computed at runtime 
from the value of the routing hash, _tsid and timestamp. Therefore 
we need to be able to extract the routing hash/_tsid/timestamp 
values from the _id value and vice-versa.

The format for the synthetic _id in this pull request has been changed to be:
    _tsid (variable length)
    Long.MAX_VALUE - timestamp (unsigned long on 8 bytes encoded using big endian)
    routing hash _ts_routing_hash (4 bytes)

Extracting the values from the _id is then used for routing GETs and DELETEs 
requests to the appropriate shard or setting the
 _tsid/timestamp/_ts_routing_hash fields in tombstone documents.

Note that in searches, the document _id is built using the existing 
TsIdLoader (that has been adjusted).

It is also important that the generated _id can be sorted lexicographically, 
as Lucene stops applying doc values updates when it seeks to a term that 
is greater than the one used in the soft-update. The ordering of the arrays 
of bytes representing the _id must match the ordering of documents in 
the segment, and to do that the timestamp value is stored in the array 
as Long.MAX_VALUE - timestamp. Docs with higher timestamp are then 
sorted first.

The SyntheticIdTermsEnum and SyntheticIdPostingsEnum introduced in 
#136810 had to be adjusted for the new _id format. We expect their 
implementation to be somewhat slow as it requires several lookups to 
work. In a different change we'll add a bloom filter on top of these 
enumerations to avoid costly lookups.

Relates #136304

Author: tlrx

docs/changelog/137274.yaml

@@ -0,0 +1,5 @@
+pr: 137274
+summary: Use a new synthetic `_id` format for time-series datastreams
+area: TSDB
+type: enhancement
+issues: []

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

@@ -705,6 +705,8 @@ public Iterator<Setting<?>> settings() {
     @Nullable
     private final IndexReshardingMetadata reshardingMetadata;
 
+    private final boolean useTimeSeriesSyntheticId;
+
     private IndexMetadata(
         final Index index,
         final long version,
@@ -754,7 +756,8 @@ private IndexMetadata(
         @Nullable final IndexMetadataStats stats,
         @Nullable final Double writeLoadForecast,
         @Nullable Long shardSizeInBytesForecast,
-        @Nullable IndexReshardingMetadata reshardingMetadata
+        @Nullable IndexReshardingMetadata reshardingMetadata,
+        final boolean useTimeSeriesSyntheticId
     ) {
         this.index = index;
         this.version = version;
@@ -815,6 +818,7 @@ private IndexMetadata(
         this.shardSizeInBytesForecast = shardSizeInBytesForecast;
         assert numberOfShards * routingFactor == routingNumShards : routingNumShards + " must be a multiple of " + numberOfShards;
         this.reshardingMetadata = reshardingMetadata;
+        this.useTimeSeriesSyntheticId = useTimeSeriesSyntheticId;
     }
 
     IndexMetadata withMappingMetadata(MappingMetadata mapping) {
@@ -870,7 +874,8 @@ IndexMetadata withMappingMetadata(MappingMetadata mapping) {
             this.stats,
             this.writeLoadForecast,
             this.shardSizeInBytesForecast,
-            this.reshardingMetadata
+            this.reshardingMetadata,
+            this.useTimeSeriesSyntheticId
         );
     }
 
@@ -933,7 +938,8 @@ public IndexMetadata withInSyncAllocationIds(int shardId, Set<String> inSyncSet)
             this.stats,
             this.writeLoadForecast,
             this.shardSizeInBytesForecast,
-            this.reshardingMetadata
+            this.reshardingMetadata,
+            this.useTimeSeriesSyntheticId
         );
     }
 
@@ -1004,7 +1010,8 @@ public IndexMetadata withSetPrimaryTerm(int shardId, long primaryTerm) {
             this.stats,
             this.writeLoadForecast,
             this.shardSizeInBytesForecast,
-            this.reshardingMetadata
+            this.reshardingMetadata,
+            this.useTimeSeriesSyntheticId
         );
     }
 
@@ -1066,7 +1073,8 @@ public IndexMetadata withTimestampRanges(IndexLongFieldRange timestampRange, Ind
             this.stats,
             this.writeLoadForecast,
             this.shardSizeInBytesForecast,
-            this.reshardingMetadata
+            this.reshardingMetadata,
+            this.useTimeSeriesSyntheticId
         );
     }
 
@@ -1123,7 +1131,8 @@ public IndexMetadata withIncrementedVersion() {
             this.stats,
             this.writeLoadForecast,
             this.shardSizeInBytesForecast,
-            this.reshardingMetadata
+            this.reshardingMetadata,
+            this.useTimeSeriesSyntheticId
         );
     }
 
@@ -1314,6 +1323,13 @@ public Instant getTimeSeriesEnd() {
         return timeSeriesEnd;
     }
 
+    /**
+     * @return whether the index is a time-series index that uses synthetic ids or not.
+     */
+    public boolean useTimeSeriesSyntheticId() {
+        return useTimeSeriesSyntheticId;
+    }
+
     /**
      * Return the concrete mapping for this index or {@code null} if this index has no mappings at all.
      */
@@ -2497,6 +2513,16 @@ IndexMetadata build(boolean repair) {
             String indexModeString = settings.get(IndexSettings.MODE.getKey());
             final IndexMode indexMode = indexModeString != null ? IndexMode.fromString(indexModeString.toLowerCase(Locale.ROOT)) : null;
             final boolean isTsdb = indexMode == IndexMode.TIME_SERIES;
+            boolean useTimeSeriesSyntheticId = false;
+            if (isTsdb
+                && IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG
+                && indexCreatedVersion.onOrAfter(IndexVersions.TIME_SERIES_USE_SYNTHETIC_ID)) {
+                var setting = settings.get(IndexSettings.USE_SYNTHETIC_ID.getKey());
+                if (setting != null && setting.equalsIgnoreCase(Boolean.TRUE.toString())) {
+                    assert IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG;
+                    useTimeSeriesSyntheticId = true;
+                }
+            }
             return new IndexMetadata(
                 new Index(index, uuid),
                 version,
@@ -2546,7 +2572,8 @@ IndexMetadata build(boolean repair) {
                 stats,
                 indexWriteLoadForecast,
                 shardSizeInBytesForecast,
-                reshardingMetadata
+                reshardingMetadata,
+                useTimeSeriesSyntheticId
             );
         }
 

server/src/main/java/org/elasticsearch/cluster/metadata/IndexMetadata.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.index.mapper.TimeSeriesRoutingHashFieldMapper;
+import org.elasticsearch.index.mapper.TsidExtractingIdFieldMapper;
 import org.elasticsearch.transport.Transports;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
@@ -321,6 +322,7 @@ public abstract static class ExtractFromSource extends IndexRouting {
         protected final XContentParserConfiguration parserConfig;
         private final IndexMode indexMode;
         private final boolean trackTimeSeriesRoutingHash;
+        private final boolean useTimeSeriesSyntheticId;
         private final boolean addIdWithRoutingHash;
         private int hash = Integer.MAX_VALUE;
 
@@ -333,6 +335,7 @@ public abstract static class ExtractFromSource extends IndexRouting {
             assert indexMode != null : "Index mode must be set for ExtractFromSource routing";
             this.trackTimeSeriesRoutingHash = indexMode == IndexMode.TIME_SERIES
                 && metadata.getCreationVersion().onOrAfter(IndexVersions.TIME_SERIES_ROUTING_HASH_IN_ID);
+            this.useTimeSeriesSyntheticId = metadata.useTimeSeriesSyntheticId();
             addIdWithRoutingHash = indexMode == IndexMode.LOGSDB;
             this.parserConfig = XContentParserConfiguration.EMPTY.withFiltering(null, Set.copyOf(includePaths), null, true);
         }
@@ -417,10 +420,19 @@ private int idToHash(String id) {
             if (idBytes.length < 4) {
                 throw new ResourceNotFoundException("invalid id [{}] for index [{}] in " + indexMode.getName() + " mode", id, indexName);
             }
-            // For TSDB, the hash is stored as the id prefix.
-            // For LogsDB with routing on sort fields, the routing hash is stored in the range[id.length - 9, id.length - 5] of the id,
-            // see IndexRequest#autoGenerateTimeBasedId.
-            return hashToShardId(ByteUtils.readIntLE(idBytes, addIdWithRoutingHash ? idBytes.length - 9 : 0));
+            int hash;
+            if (addIdWithRoutingHash) {
+                // For LogsDB with routing on sort fields, the routing hash is stored in the range[id.length - 9, id.length - 5] of the id,
+                // see IndexRequest#autoGenerateTimeBasedId.
+                hash = ByteUtils.readIntLE(idBytes, idBytes.length - 9);
+            } else if (useTimeSeriesSyntheticId) {
+                // For TSDB with synthetic ids, the hash is stored as the id suffix.
+                hash = TsidExtractingIdFieldMapper.extractRoutingHashFromSyntheticId(new BytesRef(idBytes));
+            } else {
+                // For TSDB, the hash is stored as the id prefix.
+                hash = ByteUtils.readIntLE(idBytes, 0);
+            }
+            return hashToShardId(hash);
         }
 
         @Override
@@ -510,7 +522,6 @@ public static class ForIndexDimensions extends ExtractFromSource {
 
             @Override
             protected int hashSource(IndexRequest indexRequest) {
-                // System.out.println("hashSource for tsid");
                 BytesRef tsid = indexRequest.tsid();
                 if (tsid == null) {
                     tsid = buildTsid(indexRequest.getContentType(), indexRequest.indexSource().bytes());

server/src/main/java/org/elasticsearch/cluster/routing/IndexRouting.java

@@ -14,9 +14,9 @@
 import org.apache.lucene.index.LeafReaderContext;
 import org.apache.lucene.util.BytesRef;
 import org.apache.lucene.util.CloseableThreadLocal;
-import org.elasticsearch.common.util.ByteUtils;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.core.Assertions;
+import org.elasticsearch.index.mapper.TsidExtractingIdFieldMapper;
 
 import java.io.IOException;
 import java.util.Base64;
@@ -153,22 +153,30 @@ public static DocIdAndVersion timeSeriesLoadDocIdAndVersion(IndexReader reader,
      * This allows this method to know whether there is no document with the specified id without loading the docid for
      * the specified id.
      *
-     * @param reader    The reader load docid, version and seqno from.
-     * @param uid       The term that describes the uid of the document to load docid, version and seqno for.
-     * @param id        The id that contains the encoded timestamp. The timestamp is used to skip checking the id for entire segments.
-     * @param loadSeqNo Whether to load sequence number from _seq_no doc values field.
+     * @param reader         The reader load docid, version and seqno from.
+     * @param uid            The term that describes the uid of the document to load docid, version and seqno for.
+     * @param id             The id that contains the encoded timestamp. The timestamp is used to skip checking the id for entire segments.
+     * @param loadSeqNo      Whether to load sequence number from _seq_no doc values field.
+     * @param useSyntheticId Whether the id is a synthetic (true) or standard (false ) document id.
      * @return the internal doc ID and version for the specified term from the specified reader or
      *         returning <code>null</code> if no document was found for the specified id
      * @throws IOException In case of an i/o related failure
      */
-    public static DocIdAndVersion timeSeriesLoadDocIdAndVersion(IndexReader reader, BytesRef uid, String id, boolean loadSeqNo)
-        throws IOException {
-        byte[] idAsBytes = Base64.getUrlDecoder().decode(id);
-        assert idAsBytes.length == 20;
-        // id format: [4 bytes (basic hash routing fields), 8 bytes prefix of 128 murmurhash dimension fields, 8 bytes
-        // @timestamp)
-        long timestamp = ByteUtils.readLongBE(idAsBytes, 12);
-
+    public static DocIdAndVersion timeSeriesLoadDocIdAndVersion(
+        IndexReader reader,
+        BytesRef uid,
+        String id,
+        boolean loadSeqNo,
+        boolean useSyntheticId
+    ) throws IOException {
+        final long timestamp;
+        if (useSyntheticId) {
+            assert uid.equals(new BytesRef(Base64.getUrlDecoder().decode(id)));
+            timestamp = TsidExtractingIdFieldMapper.extractTimestampFromSyntheticId(uid);
+        } else {
+            byte[] idAsBytes = Base64.getUrlDecoder().decode(id);
+            timestamp = TsidExtractingIdFieldMapper.extractTimestampFromId(idAsBytes);
+        }
         PerThreadIDVersionAndSeqNoLookup[] lookups = getLookupState(reader, true);
         List<LeafReaderContext> leaves = reader.leaves();
         // iterate in default order, the segments should be sorted by DataStream#TIMESERIES_LEAF_READERS_SORTER

server/src/main/java/org/elasticsearch/common/lucene/uid/VersionsAndSeqNoResolver.java

@@ -690,7 +690,19 @@ public boolean isES87TSDBCodecEnabled() {
         false,
         new Setting.Validator<>() {
             @Override
-            public void validate(Boolean value) {}
+            public void validate(Boolean enabled) {
+                if (enabled) {
+                    if (TSDB_SYNTHETIC_ID_FEATURE_FLAG == false) {
+                        throw new IllegalArgumentException(
+                            String.format(
+                                Locale.ROOT,
+                                "The setting [%s] is only permitted when the feature flag is enabled.",
+                                USE_SYNTHETIC_ID.getKey()
+                            )
+                        );
+                    }
+                }
+            }
 
             @Override
             public void validate(Boolean enabled, Map<Setting<?>, Object> settings) {
@@ -983,7 +995,7 @@ private void setRetentionLeaseMillis(final TimeValue retentionLease) {
     private final boolean recoverySourceEnabled;
     private final boolean recoverySourceSyntheticEnabled;
     private final boolean useDocValuesSkipper;
-    private final boolean tsdbSyntheticId;
+    private final boolean useTimeSeriesSyntheticId;
 
     /**
      * The maximum number of refresh listeners allows on this shard.
@@ -1170,8 +1182,28 @@ 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;
+        final var useSyntheticId = IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG && scopedSettings.get(USE_SYNTHETIC_ID);
+        if (indexMetadata.useTimeSeriesSyntheticId() != useSyntheticId) {
+            assert false;
+            throw new IllegalArgumentException(
+                String.format(
+                    Locale.ROOT,
+                    "The setting [%s] is set to [%s] but index metadata has a different value [%s].",
+                    USE_SYNTHETIC_ID.getKey(),
+                    useSyntheticId,
+                    indexMetadata.useTimeSeriesSyntheticId()
+                )
+            );
+        }
+        if (useSyntheticId) {
+            assert TSDB_SYNTHETIC_ID_FEATURE_FLAG;
+            assert indexMetadata.useTimeSeriesSyntheticId();
+            assert indexMetadata.getIndexMode() == IndexMode.TIME_SERIES : indexMetadata.getIndexMode();
+            assert indexMetadata.getCreationVersion().onOrAfter(IndexVersions.TIME_SERIES_USE_SYNTHETIC_ID);
+            useTimeSeriesSyntheticId = true;
+        } else {
+            useTimeSeriesSyntheticId = false;
+        }
         if (recoverySourceSyntheticEnabled) {
             if (DiscoveryNode.isStateless(settings)) {
                 throw new IllegalArgumentException("synthetic recovery source is only allowed in stateful");
@@ -1907,8 +1939,8 @@ public boolean useDocValuesSkipper() {
     /**
      * @return Whether the index is a time-series index that use synthetic ids.
      */
-    public boolean useTsdbSyntheticId() {
-        return tsdbSyntheticId;
+    public boolean useTimeSeriesSyntheticId() {
+        return useTimeSeriesSyntheticId;
     }
 
     /**

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

@@ -192,6 +192,7 @@ private static Version parseUnchecked(String version) {
 
     public static final IndexVersion REENABLED_TIMESTAMP_DOC_VALUES_SPARSE_INDEX = def(9_042_0_00, Version.LUCENE_10_3_1);
     public static final IndexVersion SKIPPERS_ENABLED_BY_DEFAULT = def(9_043_0_00, Version.LUCENE_10_3_1);
+    public static final IndexVersion TIME_SERIES_USE_SYNTHETIC_ID = def(9_044_0_00, Version.LUCENE_10_3_1);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -67,7 +67,7 @@ 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();
+        final boolean useTsdbSyntheticId = mapperService != null && mapperService.getIndexSettings().useTimeSeriesSyntheticId();
         assert useTsdbSyntheticId == false || mapperService.getIndexSettings().getMode() == IndexMode.TIME_SERIES;
 
         this.codecs = codecs.entrySet().stream().collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> {