Introduce TimeSeriesRoutingIdFieldMapper and use it to create TSDB ids

modules/lang-painless/src/main/java/org/elasticsearch/painless/action/PainlessExecuteAction.java

@@ -56,11 +56,13 @@
 import org.elasticsearch.index.Index;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexService;
+import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.index.mapper.DocumentMapper;
 import org.elasticsearch.index.mapper.OnScriptError;
 import org.elasticsearch.index.mapper.ParsedDocument;
 import org.elasticsearch.index.mapper.SourceToParse;
+import org.elasticsearch.index.mapper.TimeSeriesRoutingHashFieldMapper;
 import org.elasticsearch.index.query.AbstractQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.SearchExecutionContext;
@@ -808,13 +810,18 @@ private static Response prepareRamIndex(
                 try (IndexWriter indexWriter = new IndexWriter(directory, new IndexWriterConfig(defaultAnalyzer))) {
                     BytesReference document = request.contextSetup.document;
                     XContentType xContentType = request.contextSetup.xContentType;
-                    String id;
-                    if (indexService.getIndexSettings().getMode() == IndexMode.TIME_SERIES) {
-                        id = null; // The id gets auto generated for time series indices.
-                    } else {
-                        id = "_id";
-                    }
-                    SourceToParse sourceToParse = new SourceToParse(id, document, xContentType);
+
+                    SourceToParse sourceToParse = (indexService.getIndexSettings().getMode() == IndexMode.TIME_SERIES)
+                        ? new SourceToParse(
+                            null,
+                            document,
+                            xContentType,
+                            indexService.getIndexSettings().getIndexVersionCreated().onOrAfter(IndexVersions.TIME_SERIES_ROUTING_HASH_IN_ID)
+                                ? TimeSeriesRoutingHashFieldMapper.DUMMY_ENCODED_VALUE
+                                : null
+                        )
+                        : new SourceToParse("_id", document, xContentType);
+
                     DocumentMapper documentMapper = indexService.mapperService().documentMapper();
                     if (documentMapper == null) {
                         documentMapper = DocumentMapper.createEmpty(indexService.mapperService());

modules/parent-join/src/main/java/org/elasticsearch/join/mapper/ParentJoinFieldMapper.java

@@ -293,7 +293,7 @@ public void parse(DocumentParserContext context) throws IOException {
             if (parent == null) {
                 throw new IllegalArgumentException("[parent] is missing for join field [" + name() + "]");
             }
-            if (context.sourceToParse().routing() == null) {
+            if (context.routing() == null) {
                 throw new IllegalArgumentException("[routing] is missing for join field [" + name() + "]");
             }
             String fieldName = fieldType().joiner.parentJoinField(name);

server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -689,9 +689,7 @@ private static Engine.Result performOpOnReplica(
                     indexRequest.id(),
                     indexRequest.source(),
                     indexRequest.getContentType(),
-                    indexRequest.routing(),
-                    Map.of(),
-                    DocumentSizeObserver.EMPTY_INSTANCE
+                    indexRequest.routing()
                 );
                 result = replica.applyIndexOperationOnReplica(
                     primaryResponse.getSeqNo(),

server/src/main/java/org/elasticsearch/action/index/IndexRequest.java

@@ -881,7 +881,7 @@ public Index getConcreteWriteIndex(IndexAbstraction ia, Metadata metadata) {
 
     @Override
     public int route(IndexRouting indexRouting) {
-        return indexRouting.indexShard(id, routing, contentType, source);
+        return indexRouting.indexShard(id, routing, contentType, source, this::routing);
     }
 
     public IndexRequest setRequireAlias(boolean requireAlias) {

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

@@ -21,6 +21,8 @@
 import org.elasticsearch.common.util.ByteUtils;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.index.IndexVersions;
+import org.elasticsearch.index.mapper.TimeSeriesRoutingHashFieldMapper;
 import org.elasticsearch.transport.Transports;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParser.Token;
@@ -35,6 +37,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Consumer;
 import java.util.function.IntConsumer;
 import java.util.function.IntSupplier;
 import java.util.function.Predicate;
@@ -74,7 +77,13 @@ private IndexRouting(IndexMetadata metadata) {
      * Called when indexing a document to generate the shard id that should contain
      * a document with the provided parameters.
      */
-    public abstract int indexShard(String id, @Nullable String routing, XContentType sourceType, BytesReference source);
+    public abstract int indexShard(
+        String id,
+        @Nullable String routing,
+        XContentType sourceType,
+        BytesReference source,
+        Consumer<String> routingHashSetter
+    );
 
     /**
      * Called when updating a document to generate the shard id that should contain
@@ -153,7 +162,13 @@ public void process(IndexRequest indexRequest) {
         }
 
         @Override
-        public int indexShard(String id, @Nullable String routing, XContentType sourceType, BytesReference source) {
+        public int indexShard(
+            String id,
+            @Nullable String routing,
+            XContentType sourceType,
+            BytesReference source,
+            Consumer<String> routingHashSetter
+        ) {
             if (id == null) {
                 throw new IllegalStateException("id is required and should have been set by process");
             }
@@ -237,12 +252,14 @@ public void collectSearchShards(String routing, IntConsumer consumer) {
     public static class ExtractFromSource extends IndexRouting {
         private final Predicate<String> isRoutingPath;
         private final XContentParserConfiguration parserConfig;
+        private final boolean trackTimeSeriesRoutingHash;
 
         ExtractFromSource(IndexMetadata metadata) {
             super(metadata);
             if (metadata.isRoutingPartitionedIndex()) {
                 throw new IllegalArgumentException("routing_partition_size is incompatible with routing_path");
             }
+            trackTimeSeriesRoutingHash = metadata.getCreationVersion().onOrAfter(IndexVersions.TIME_SERIES_ROUTING_HASH_IN_ID);
             List<String> routingPaths = metadata.getRoutingPaths();
             isRoutingPath = Regex.simpleMatcher(routingPaths.toArray(String[]::new));
             this.parserConfig = XContentParserConfiguration.EMPTY.withFiltering(Set.copyOf(routingPaths), null, true);
@@ -256,10 +273,20 @@ public boolean matchesField(String fieldName) {
         public void process(IndexRequest indexRequest) {}
 
         @Override
-        public int indexShard(String id, @Nullable String routing, XContentType sourceType, BytesReference source) {
+        public int indexShard(
+            String id,
+            @Nullable String routing,
+            XContentType sourceType,
+            BytesReference source,
+            Consumer<String> routingHashSetter
+        ) {
             assert Transports.assertNotTransportThread("parsing the _source can get slow");
             checkNoRouting(routing);
-            return hashToShardId(hashSource(sourceType, source).buildHash(IndexRouting.ExtractFromSource::defaultOnEmpty));
+            int hash = hashSource(sourceType, source).buildHash(IndexRouting.ExtractFromSource::defaultOnEmpty);
+            if (trackTimeSeriesRoutingHash) {
+                routingHashSetter.accept(TimeSeriesRoutingHashFieldMapper.encode(hash));
+            }
+            return hashToShardId(hash);
         }
 
         public String createId(XContentType sourceType, BytesReference source, byte[] suffix) {
@@ -334,16 +361,13 @@ private void extractItem(String path, XContentParser source) throws IOException
                         source.nextToken();
                         break;
                     case VALUE_STRING:
+                    case VALUE_NUMBER:
                         hashes.add(new NameAndHash(new BytesRef(path), hash(new BytesRef(source.text()))));
                         source.nextToken();
                         break;
                     case VALUE_NULL:
                         source.nextToken();
                         break;
-                    case VALUE_NUMBER: // allow parsing numbers assuming routing fields are always keyword fields
-                        hashes.add(new NameAndHash(new BytesRef(path), hash(new BytesRef(source.text()))));
-                        source.nextToken();
-                        break;
                     default:
                         throw new ParsingException(
                             source.getTokenLocation(),

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

@@ -28,6 +28,7 @@
 import org.elasticsearch.index.mapper.RoutingFieldMapper;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.index.mapper.TimeSeriesIdFieldMapper;
+import org.elasticsearch.index.mapper.TimeSeriesRoutingHashFieldMapper;
 import org.elasticsearch.index.mapper.TsidExtractingIdFieldMapper;
 
 import java.io.IOException;
@@ -92,6 +93,12 @@ public MetadataFieldMapper timeSeriesIdFieldMapper() {
             return null;
         }
 
+        @Override
+        public MetadataFieldMapper timeSeriesRoutingHashFieldMapper() {
+            // non time-series indices must not have a TimeSeriesRoutingIdFieldMapper
+            return null;
+        }
+
         @Override
         public IdFieldMapper idFieldMapperWithoutFieldData() {
             return ProvidedIdFieldMapper.NO_FIELD_DATA;
@@ -185,6 +192,11 @@ public MetadataFieldMapper timeSeriesIdFieldMapper() {
             return TimeSeriesIdFieldMapper.INSTANCE;
         }
 
+        @Override
+        public MetadataFieldMapper timeSeriesRoutingHashFieldMapper() {
+            return TimeSeriesRoutingHashFieldMapper.INSTANCE;
+        }
+
         public IdFieldMapper idFieldMapperWithoutFieldData() {
             return TsidExtractingIdFieldMapper.INSTANCE;
         }
@@ -322,6 +334,13 @@ public String getName() {
      */
     public abstract MetadataFieldMapper timeSeriesIdFieldMapper();
 
+    /**
+     * Return an instance of the {@link TimeSeriesRoutingHashFieldMapper} that generates
+     * the _ts_routing_hash field. The field mapper will be added to the list of the metadata
+     * field mappers for the index.
+     */
+    public abstract MetadataFieldMapper timeSeriesRoutingHashFieldMapper();
+
     /**
      * How {@code time_series_dimension} fields are handled by indices in this mode.
      */

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

@@ -102,6 +102,7 @@ private static IndexVersion def(int id, Version luceneVersion) {
     public static final IndexVersion UPGRADE_LUCENE_9_9_2 = def(8_502_00_0, Version.LUCENE_9_9_2);
     public static final IndexVersion TIME_SERIES_ID_HASHING = def(8_502_00_1, Version.LUCENE_9_9_2);
     public static final IndexVersion UPGRADE_TO_LUCENE_9_10 = def(8_503_00_0, Version.LUCENE_9_10_0);
+    public static final IndexVersion TIME_SERIES_ROUTING_HASH_IN_ID = def(8_504_00_0, Version.LUCENE_9_10_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -115,8 +115,8 @@ boolean useTSDBDocValuesFormat(final String field) {
 
     private boolean excludeFields(String fieldName) {
         // Avoid using tsdb codec for fields like _seq_no, _primary_term.
-        // But _tsid should always use the tesbd codec.
-        return fieldName.startsWith("_") && fieldName.equals("_tsid") == false;
+        // But _tsid and _ts_routing_hash should always use the tsdb codec.
+        return fieldName.startsWith("_") && fieldName.equals("_tsid") == false && fieldName.equals("_ts_routing_hash") == false;
     }
 
     private boolean isTimeSeriesModeIndex() {

server/src/main/java/org/elasticsearch/index/engine/TranslogDirectoryReader.java

@@ -59,11 +59,9 @@
 import org.elasticsearch.index.mapper.VersionFieldMapper;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.translog.Translog;
-import org.elasticsearch.plugins.internal.DocumentSizeObserver;
 
 import java.io.IOException;
 import java.util.Collections;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -254,14 +252,7 @@ private LeafReader getDelegate() {
         private LeafReader createInMemoryLeafReader() {
             assert Thread.holdsLock(this);
             final ParsedDocument parsedDocs = documentParser.parseDocument(
-                new SourceToParse(
-                    operation.id(),
-                    operation.source(),
-                    XContentHelper.xContentType(operation.source()),
-                    operation.routing(),
-                    Map.of(),
-                    DocumentSizeObserver.EMPTY_INSTANCE
-                ),
+                new SourceToParse(operation.id(), operation.source(), XContentHelper.xContentType(operation.source()), operation.routing()),
                 mappingLookup
             );
 

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

@@ -99,7 +99,7 @@ public ParsedDocument parseDocument(SourceToParse source, MappingLookup mappingL
             context.version(),
             context.seqID(),
             context.id(),
-            source.routing(),
+            context.routing(),
             context.reorderParentAndGetDocs(),
             context.sourceToParse().source(),
             context.sourceToParse().getXContentType(),

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

@@ -12,6 +12,7 @@
 import org.apache.lucene.document.StringField;
 import org.apache.lucene.index.IndexableField;
 import org.elasticsearch.common.time.DateFormatter;
+import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.IndexSettings;
 import org.elasticsearch.index.analysis.IndexAnalyzers;
 import org.elasticsearch.xcontent.FilterXContentParserWrapper;
@@ -231,6 +232,10 @@ public final SourceToParse sourceToParse() {
         return this.sourceToParse;
     }
 
+    public final String routing() {
+        return mappingParserContext.getIndexSettings().getMode() == IndexMode.TIME_SERIES ? null : sourceToParse.routing();
+    }
+
     /**
      * Add the given {@code field} to the set of ignored fields.
      */

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

@@ -66,21 +66,24 @@ final class TsIdLoader implements IdLoader {
         }
 
         public IdLoader.Leaf leaf(LeafStoredFieldLoader loader, LeafReader reader, int[] docIdsInLeaf) throws IOException {
-            IndexRouting.ExtractFromSource.Builder[] builders = new IndexRouting.ExtractFromSource.Builder[docIdsInLeaf.length];
-            for (int i = 0; i < builders.length; i++) {
-                builders[i] = indexRouting.builder();
-            }
+            IndexRouting.ExtractFromSource.Builder[] builders = null;
+            if (indexRouting != null) {
+                builders = new IndexRouting.ExtractFromSource.Builder[docIdsInLeaf.length];
+                for (int i = 0; i < builders.length; i++) {
+                    builders[i] = indexRouting.builder();
+                }
 
-            for (String routingField : routingPaths) {
-                // Routing field must always be keyword fields, so it is ok to use SortedSetDocValues directly here.
-                SortedSetDocValues dv = DocValues.getSortedSet(reader, routingField);
-                for (int i = 0; i < docIdsInLeaf.length; i++) {
-                    int docId = docIdsInLeaf[i];
-                    var builder = builders[i];
-                    if (dv.advanceExact(docId)) {
-                        for (int j = 0; j < dv.docValueCount(); j++) {
-                            BytesRef routingValue = dv.lookupOrd(dv.nextOrd());
-                            builder.addMatching(routingField, routingValue);
+                for (String routingField : routingPaths) {
+                    // Routing field must always be keyword fields, so it is ok to use SortedSetDocValues directly here.
+                    SortedSetDocValues dv = DocValues.getSortedSet(reader, routingField);
+                    for (int i = 0; i < docIdsInLeaf.length; i++) {
+                        int docId = docIdsInLeaf[i];
+                        var builder = builders[i];
+                        if (dv.advanceExact(docId)) {
+                            for (int j = 0; j < dv.docValueCount(); j++) {
+                                BytesRef routingValue = dv.lookupOrd(dv.nextOrd());
+                                builder.addMatching(routingField, routingValue);
+                            }
                         }
                     }
                 }
@@ -100,9 +103,17 @@ public IdLoader.Leaf leaf(LeafStoredFieldLoader loader, LeafReader reader, int[]
                 assert found;
                 assert timestampDocValues.docValueCount() == 1;
                 long timestamp = timestampDocValues.nextValue();
-
-                var routingBuilder = builders[i];
-                ids[i] = TsidExtractingIdFieldMapper.createId(false, routingBuilder, tsid, timestamp, new byte[16]);
+                if (builders != null) {
+                    var routingBuilder = builders[i];
+                    ids[i] = TsidExtractingIdFieldMapper.createId(false, routingBuilder, tsid, timestamp, new byte[16]);
+                } else {
+                    SortedDocValues routingHashDocValues = DocValues.getSorted(reader, TimeSeriesRoutingHashFieldMapper.NAME);
+                    found = routingHashDocValues.advanceExact(docId);
+                    assert found;
+                    BytesRef routingHashBytes = routingHashDocValues.lookupOrd(routingHashDocValues.ordValue());
+                    int routingHash = TimeSeriesRoutingHashFieldMapper.decode(Uid.decodeId(routingHashBytes.bytes));
+                    ids[i] = TsidExtractingIdFieldMapper.createId(routingHash, tsid, timestamp);
+                }
             }
             return new TsIdLeaf(docIdsInLeaf, ids);
         }

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

@@ -108,7 +108,7 @@ public boolean required() {
 
     @Override
     public void preParse(DocumentParserContext context) {
-        String routing = context.sourceToParse().routing();
+        String routing = context.routing();
         if (routing != null) {
             context.doc().add(new StringField(fieldType().name(), routing, Field.Store.YES));
             context.addToFieldNames(fieldType().name());

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

@@ -52,6 +52,10 @@ public SourceToParse(String id, BytesReference source, XContentType xContentType
         this(id, source, xContentType, null, Map.of(), DocumentSizeObserver.EMPTY_INSTANCE);
     }
 
+    public SourceToParse(String id, BytesReference source, XContentType xContentType, String routing) {
+        this(id, source, xContentType, routing, Map.of(), DocumentSizeObserver.EMPTY_INSTANCE);
+    }
+
     public BytesReference source() {
         return this.source;
     }

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

@@ -140,7 +140,13 @@ public void postParse(DocumentParserContext context) throws IOException {
             ? timeSeriesIdBuilder.buildLegacyTsid().toBytesRef()
             : timeSeriesIdBuilder.buildTsidHash().toBytesRef();
         context.doc().add(new SortedDocValuesField(fieldType().name(), timeSeriesId));
-        TsidExtractingIdFieldMapper.createField(context, timeSeriesIdBuilder.routingBuilder, timeSeriesId);
+        TsidExtractingIdFieldMapper.createField(
+            context,
+            getIndexVersionCreated(context).before(IndexVersions.TIME_SERIES_ROUTING_HASH_IN_ID)
+                ? timeSeriesIdBuilder.routingBuilder
+                : null,
+            timeSeriesId
+        );
     }
 
     private IndexVersion getIndexVersionCreated(final DocumentParserContext context) {