Introduce TimeSeriesRoutingIdFieldMapper and use it to create TSDB ids (elastic#106080)

Supporting non-keyword fields requires updating non-keyword fields in
the routing path to be included in routing calculations. Routing is
performed in coordinating nodes that lack mappings (or mappings haven't
been created yet, for dynamically-defined dimensions), so the routing
hash they calculate are passed to data nodes and stored in a new fields,
namely _ts_routind_hash. This is included in the _id field, in turn, so
that it can consistently reach the right shard for get-by-id and
delete-by-id operations.

A few interesting points:

- The hash is passed from the coordinating to data nodes using the `routing` field in `IndexRequest`; adding another field to the latter requires updating dozens of classes.
- We explicitly skip (double-) storing the hash to the routing field, as the latter is not optimized for storage using the TSDB codec.
- The routing hash may not be available in Translog operations, it can then be retrieved from the `id` prefix.

Related to elastic#103567

Author: kkrik-es

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(),