Use a new synthetic _id format for time-series datastreams

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

@@ -26,9 +26,11 @@
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.IndexSettings;
 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 +323,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 +336,9 @@ 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 = trackTimeSeriesRoutingHash
+                && metadata.getCreationVersion().onOrAfter(IndexVersions.TIME_SERIES_USE_SYNTHETIC_ID)
+                && IndexSettings.USE_SYNTHETIC_ID.get(metadata.getSettings());
             addIdWithRoutingHash = indexMode == IndexMode.LOGSDB;
             this.parserConfig = XContentParserConfiguration.EMPTY.withFiltering(null, Set.copyOf(includePaths), null, true);
         }
@@ -417,10 +423,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 +525,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/common/lucene/uid/VersionsAndSeqNoResolver.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/index/IndexMode.java

@@ -42,15 +42,13 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 import java.util.function.BooleanSupplier;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import static java.util.stream.Collectors.toSet;
 
 /**
  * "Mode" that controls which behaviors and settings an index supports.
@@ -141,7 +139,16 @@ void validateWithOtherSettings(Map<Setting<?>, Object> settings) {
                 throw new IllegalArgumentException(error(IndexMetadata.INDEX_ROUTING_PARTITION_SIZE_SETTING));
             }
 
-            var settingsWithIndexMode = Settings.builder().put(IndexSettings.MODE.getKey(), getName()).build();
+            Settings settingsWithIndexMode;
+            if (IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG) {
+                settingsWithIndexMode = Settings.builder()
+                    .put(IndexSettings.MODE.getKey(), getName())
+                    // Default values of some index sort settings depend of the feature flag and USE_SYNTHETIC_ID setting
+                    .put(IndexSettings.USE_SYNTHETIC_ID.getKey(), (Boolean) settings.get(IndexSettings.USE_SYNTHETIC_ID))
+                    .build();
+            } else {
+                settingsWithIndexMode = Settings.builder().put(IndexSettings.MODE.getKey(), getName()).build();
+            }
 
             for (Setting<?> unsupported : TIME_SERIES_UNSUPPORTED) {
                 if (false == Objects.equals(unsupported.getDefault(settingsWithIndexMode), settings.get(unsupported))) {
@@ -460,20 +467,22 @@ private static CompressedXContent createDefaultMapping(boolean includeHostName)
         IndexSortConfig.INDEX_SORT_MISSING_SETTING
     );
 
-    static final List<Setting<?>> VALIDATE_WITH_SETTINGS = List.copyOf(
-        Stream.concat(
-            Stream.of(
-                IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING,
-                IndexMetadata.INDEX_ROUTING_PARTITION_SIZE_SETTING,
-                IndexMetadata.INDEX_ROUTING_PATH,
-                IndexMetadata.INDEX_DIMENSIONS,
-                IndexSettings.LOGSDB_ROUTE_ON_SORT_FIELDS,
-                IndexSettings.TIME_SERIES_START_TIME,
-                IndexSettings.TIME_SERIES_END_TIME
-            ),
-            TIME_SERIES_UNSUPPORTED.stream()
-        ).collect(toSet())
-    );
+    static final List<Setting<?>> VALIDATE_WITH_SETTINGS;
+    static {
+        var settings = new HashSet<Setting<?>>();
+        settings.add(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING);
+        settings.add(IndexMetadata.INDEX_ROUTING_PARTITION_SIZE_SETTING);
+        settings.add(IndexMetadata.INDEX_ROUTING_PATH);
+        settings.add(IndexMetadata.INDEX_DIMENSIONS);
+        settings.add(IndexSettings.LOGSDB_ROUTE_ON_SORT_FIELDS);
+        settings.add(IndexSettings.TIME_SERIES_START_TIME);
+        settings.add(IndexSettings.TIME_SERIES_END_TIME);
+        if (IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG) {
+            settings.add(IndexSettings.USE_SYNTHETIC_ID);
+        }
+        settings.addAll(TIME_SERIES_UNSUPPORTED);
+        VALIDATE_WITH_SETTINGS = List.copyOf(settings);
+    }
 
     private final String name;
 

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

@@ -107,7 +107,7 @@ public final class IndexSortConfig {
     );
 
     public static class IndexSortConfigDefaults {
-        public static final FieldSortSpec[] TIME_SERIES_SORT, HOSTNAME_TIMESTAMP_BWC_SORT;
+        public static final FieldSortSpec[] TIME_SERIES_SORT, TIME_SERIES_WITH_SYNTHETIC_ID_SORT, HOSTNAME_TIMESTAMP_BWC_SORT;
 
         private static final FieldSortSpec HOSTNAME_SPEC, MESSAGE_PATTERN_SPEC, TIMESTAMP_SPEC;
 
@@ -116,6 +116,10 @@ public static class IndexSortConfigDefaults {
             TIMESTAMP_SPEC.order = SortOrder.DESC;
             TIME_SERIES_SORT = new FieldSortSpec[] { new FieldSortSpec(TimeSeriesIdFieldMapper.NAME), TIMESTAMP_SPEC };
 
+            TIME_SERIES_WITH_SYNTHETIC_ID_SORT = new FieldSortSpec[] {
+                new FieldSortSpec(TimeSeriesIdFieldMapper.NAME),
+                new FieldSortSpec(DataStreamTimestampFieldMapper.DEFAULT_PATH) };
+
             HOSTNAME_SPEC = new FieldSortSpec(IndexMode.HOST_NAME);
             HOSTNAME_SPEC.order = SortOrder.ASC;
             HOSTNAME_SPEC.missingValue = "_last";

server/src/main/java/org/elasticsearch/index/IndexVersions.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/codec/tsdb/TSDBSyntheticIdCodec.java

@@ -27,6 +27,7 @@
 import static org.elasticsearch.index.codec.tsdb.TSDBSyntheticIdPostingsFormat.SYNTHETIC_ID;
 import static org.elasticsearch.index.codec.tsdb.TSDBSyntheticIdPostingsFormat.TIMESTAMP;
 import static org.elasticsearch.index.codec.tsdb.TSDBSyntheticIdPostingsFormat.TS_ID;
+import static org.elasticsearch.index.codec.tsdb.TSDBSyntheticIdPostingsFormat.TS_ROUTING_HASH;
 
 /**
  * Special codec for time-series datastreams that use synthetic ids.
@@ -83,6 +84,13 @@ private void ensureSyntheticIdFields(FieldInfos fieldInfos) {
                 assert false : message;
                 throw new IllegalArgumentException(message);
             }
+            // Ensure _ts_routing_hash exists
+            fi = fieldInfos.fieldInfo(TS_ROUTING_HASH);
+            if (fi == null) {
+                var message = "Field [" + TS_ROUTING_HASH + "] does not exist";
+                assert false : message;
+                throw new IllegalArgumentException(message);
+            }
             // Ensure _id exists and not indexed
             fi = fieldInfos.fieldInfo(SYNTHETIC_ID);
             if (fi == null) {
@@ -102,6 +110,49 @@ private void ensureSyntheticIdFields(FieldInfos fieldInfos) {
         @Override
         public void write(Directory directory, SegmentInfo segmentInfo, String segmentSuffix, FieldInfos fieldInfos, IOContext context)
             throws IOException {
+
+            // Change the _id field index options from IndexOptions.DOCS to IndexOptions.NONE
+            final var infos = new FieldInfo[fieldInfos.size()];
+            int i = 0;
+            for (FieldInfo fi : fieldInfos) {
+                if (SYNTHETIC_ID.equals(fi.getName())) {
+                    final var attributes = new HashMap<>(fi.attributes());
+
+                    // Assert that PerFieldPostingsFormat are not present or have the expected format and suffix
+                    assert attributes.get(PerFieldPostingsFormat.PER_FIELD_FORMAT_KEY) == null
+                        || TSDBSyntheticIdPostingsFormat.FORMAT_NAME.equals(attributes.get(PerFieldPostingsFormat.PER_FIELD_FORMAT_KEY));
+                    assert attributes.get(PerFieldPostingsFormat.PER_FIELD_SUFFIX_KEY) == null
+                        || TSDBSyntheticIdPostingsFormat.SUFFIX.equals(attributes.get(PerFieldPostingsFormat.PER_FIELD_SUFFIX_KEY));
+
+                    // Remove attributes if present
+                    attributes.remove(PerFieldPostingsFormat.PER_FIELD_FORMAT_KEY);
+                    attributes.remove(PerFieldPostingsFormat.PER_FIELD_SUFFIX_KEY);
+
+                    fi = new FieldInfo(
+                        fi.getName(),
+                        fi.getFieldNumber(),
+                        fi.hasTermVectors(),
+                        true,
+                        fi.hasPayloads(),
+                        IndexOptions.NONE,
+                        fi.getDocValuesType(),
+                        fi.docValuesSkipIndexType(),
+                        fi.getDocValuesGen(),
+                        attributes,
+                        fi.getPointDimensionCount(),
+                        fi.getPointIndexDimensionCount(),
+                        fi.getPointNumBytes(),
+                        fi.getVectorDimension(),
+                        fi.getVectorEncoding(),
+                        fi.getVectorSimilarityFunction(),
+                        fi.isSoftDeletesField(),
+                        fi.isParentField()
+                    );
+                }
+                infos[i++] = fi;
+            }
+
+            fieldInfos = new FieldInfos(infos);
             ensureSyntheticIdFields(fieldInfos);
             delegate.write(directory, segmentInfo, segmentSuffix, fieldInfos, context);
         }