Introduce single prefix byte TSID layout (elastic#143955)

The current TSID layout:

```
byte 0 = hash(dimension_names)
bytes 1–4 = hash(value_0) … hash(value_3)
bytes 5–20 = 128-bit hash (uniqueness)
```

The first 5 bytes are derived from dimension names and a subset of 
values. Since these are often static for a given metric, partitioning by
prefix typically yields only a single partition.

This change introduces a 16-byte TSID layout with a single prefix byte 
to better partition time-series of a single metric:

```
byte 0 = hash(metric_name) for OTel,
         hash(labels.__name__) for Prometheus,
         hash(all dimension names and values) for generic TSDB
bytes 1–15 = 15-byte hash (uniqueness + within-metric ordering)
---
Total: 16 bytes (2 longs)
```

The new layout groups time-series by metric, enabling partitioned rate 
aggregations across at least 256 partitions.

Author: dnhatn

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

@@ -581,7 +581,7 @@ public BytesRef buildTsid(XContentType sourceType, BytesReference source) {
                 } catch (IOException | ParsingException e) {
                     throw new IllegalArgumentException("Error extracting tsid: " + e.getMessage(), e);
                 }
-                return b.buildTsid();
+                return b.buildTsid(creationVersion);
             }
         }
     }

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

@@ -13,6 +13,9 @@
 import org.elasticsearch.common.hash.BufferedMurmur3Hasher;
 import org.elasticsearch.common.hash.MurmurHash3;
 import org.elasticsearch.common.util.ByteUtils;
+import org.elasticsearch.common.util.FeatureFlag;
+import org.elasticsearch.index.IndexVersion;
+import org.elasticsearch.index.IndexVersions;
 import org.elasticsearch.index.mapper.RoutingPathFields;
 import org.elasticsearch.xcontent.XContentString;
 
@@ -31,12 +34,17 @@
  */
 public class TsidBuilder {
 
+    public static final boolean SINGLE_PREFIX_BYTE_ENABLED = new FeatureFlag("tsid_layout_single_prefix_byte").isEnabled();
+
     /**
      * The maximum number of fields to use for the value similarity part of the TSID.
      * This is a trade-off between clustering similar time series together and the size of the TSID.
      * More fields improve clustering but also increase the size of the TSID.
      */
     private static final int MAX_TSID_VALUE_SIMILARITY_FIELDS = 4;
+    static final String OTEL_METRIC_FIELD = "_metric_names_hash";
+    static final String PROMETHEUS_LABEL_FIELD = "labels.__name__";
+
     private final BufferedMurmur3Hasher murmur3Hasher = new BufferedMurmur3Hasher(0L);
 
     private final List<Dimension> dimensions;
@@ -215,6 +223,18 @@ public MurmurHash3.Hash128 hash() {
         return murmur3Hasher.digestHash();
     }
 
+    public final BytesRef buildTsid(IndexVersion indexVersion) {
+        if (useSingleBytePrefixLayout(indexVersion)) {
+            return buildSingleBytePrefixTsid();
+        } else {
+            return buildMultiBytePrefixTsid();
+        }
+    }
+
+    public static boolean useSingleBytePrefixLayout(IndexVersion indexVersion) {
+        return SINGLE_PREFIX_BYTE_ENABLED && indexVersion.onOrAfter(IndexVersions.TSID_SINGLE_PREFIX_BYTE_FEATURE_FLAG);
+    }
+
     /**
      * Builds a time series identifier (TSID) based on the dimensions added to this builder.
      * This is a slight adaptation of {@link RoutingPathFields#buildHash()} but creates shorter tsids.
@@ -233,18 +253,16 @@ public MurmurHash3.Hash128 hash() {
      *         This is to avoid hash collisions.
      *     </li>
      * </ul>
-     *
-     * @return a BytesRef containing the TSID
-     * @throws IllegalArgumentException if no dimensions have been added
+     * Note that this layout has been used with indices created before {@link IndexVersions#TSID_SINGLE_PREFIX_BYTE_FEATURE_FLAG}
      */
-    public BytesRef buildTsid() {
+    private BytesRef buildMultiBytePrefixTsid() {
         throwIfEmpty();
+        Collections.sort(dimensions);
+
         int numberOfValues = Math.min(MAX_TSID_VALUE_SIMILARITY_FIELDS, dimensions.size());
         byte[] hash = new byte[1 + numberOfValues + 16];
         int index = 0;
 
-        Collections.sort(dimensions);
-
         MurmurHash3.Hash128 hashBuffer = new MurmurHash3.Hash128();
         murmur3Hasher.reset();
         // similarity hash for dimension names
@@ -280,6 +298,55 @@ public BytesRef buildTsid() {
         return new BytesRef(hash, 0, index);
     }
 
+    private BytesRef buildSingleBytePrefixTsid() {
+        throwIfEmpty();
+        Collections.sort(dimensions);
+
+        final byte[] tsid = new byte[16];
+        murmur3Hasher.reset();
+        MurmurHash3.Hash128 hashBuffer = new MurmurHash3.Hash128();
+        // hash of all dimension names and values for uniqueness
+        for (Dimension dim : dimensions) {
+            murmur3Hasher.addLongs(dim.pathHash.h1, dim.pathHash.h2, dim.valueHash.h1, dim.valueHash.h2);
+        }
+        murmur3Hasher.digestHash(hashBuffer);
+        ByteUtils.writeLongLE(hashBuffer.h2, tsid, 0);
+        ByteUtils.writeLongLE(hashBuffer.h1, tsid, 8);
+        tsid[0] = computeSingleBytePrefix(hashBuffer);
+        return new BytesRef(tsid);
+    }
+
+    private byte computeSingleBytePrefix(MurmurHash3.Hash128 scratch) {
+        murmur3Hasher.reset();
+        Dimension otelMetric = findDimension(dimensions, OTEL_METRIC_FIELD);
+        if (otelMetric != null) {
+            murmur3Hasher.addLong(otelMetric.valueHash().h1 ^ otelMetric.valueHash().h2);
+            return (byte) murmur3Hasher.digestHash(scratch).h1;
+        }
+        Dimension prometheusLabel = findDimension(dimensions, PROMETHEUS_LABEL_FIELD);
+        if (prometheusLabel != null) {
+            murmur3Hasher.addLong(prometheusLabel.valueHash().h1 ^ prometheusLabel.valueHash().h2);
+            return (byte) murmur3Hasher.digestHash(scratch).h1;
+        }
+        // similarity hash for dimension names
+        for (Dimension dim : dimensions) {
+            murmur3Hasher.addLong(dim.pathHash.h1 ^ dim.pathHash.h2);
+        }
+        return (byte) murmur3Hasher.digestHash(scratch).h1;
+    }
+
+    private static Dimension findDimension(List<Dimension> sortedDimensions, String name) {
+        for (Dimension dim : sortedDimensions) {
+            int cmp = dim.path.compareTo(name);
+            if (cmp > 0) {
+                return null;
+            } else if (cmp == 0) {
+                return dim;
+            }
+        }
+        return null;
+    }
+
     private void throwIfEmpty() {
         if (dimensions.isEmpty()) {
             throw new IllegalArgumentException("Dimensions are empty");

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

@@ -238,7 +238,7 @@ private static Version parseUnchecked(String version) {
     public static final IndexVersion FLATTENED_FIELD_NO_ROOT_DOC_VALUES = def(9_077_0_00, Version.LUCENE_10_4_0);
     public static final IndexVersion IGNORED_SOURCE_AS_DOC_VALUES = def(9_078_0_00, Version.LUCENE_10_4_0);
     public static final IndexVersion TIME_SERIES_USE_SYNTHETIC_ID_DEFAULT = def(9_079_0_00, Version.LUCENE_10_4_0);
-
+    public static final IndexVersion TSID_SINGLE_PREFIX_BYTE_FEATURE_FLAG = def(9_080_00_0, Version.LUCENE_10_4_0);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/test/java/org/elasticsearch/cluster/routing/IndexRoutingTests.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.index.mapper.TsidExtractingIdFieldMapper;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.index.IndexVersionUtils;
 import org.elasticsearch.xcontent.DeprecationHandler;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.elasticsearch.xcontent.XContentType;
@@ -686,11 +687,15 @@ public void testRoutingPathBwc() throws IOException {
     public void testRoutingPathBwcAfterTsidBasedRouting() throws IOException {
         boolean useSyntheticId = IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG && randomBoolean();
         TimeSeriesRoutingFixture fixture = indexRoutingForTimeSeriesDimensions(
-            IndexVersion.current(),
+            IndexVersionUtils.randomVersionBetween(
+                IndexVersions.TIME_SERIES_USE_SYNTHETIC_ID_94,
+                IndexVersionUtils.getPreviousVersion(IndexVersions.TSID_SINGLE_PREFIX_BYTE_FEATURE_FLAG)
+            ),
             8,
             "dim.*,other.*,top",
             useSyntheticId
         );
+        assertFalse(TsidBuilder.useSingleBytePrefixLayout(fixture.routing.creationVersion));
         /*
          * These are the expected shards after tsid based routing. If these values change
          * time series will be routed to unexpected shards. You may modify
@@ -712,6 +717,28 @@ public void testRoutingPathBwcAfterTsidBasedRouting() throws IOException {
         assertIndexShard(fixture, Map.of("dim.a", "true"), 6);
     }
 
+    public void testRoutingPathWithSingleBytePrefixTsid() throws IOException {
+        boolean useSyntheticId = IndexSettings.TSDB_SYNTHETIC_ID_FEATURE_FLAG && randomBoolean();
+        TimeSeriesRoutingFixture fixture = indexRoutingForTimeSeriesDimensions(
+            IndexVersionUtils.randomVersionOnOrAfter(IndexVersions.TSID_SINGLE_PREFIX_BYTE_FEATURE_FLAG),
+            8,
+            "dim.*,other.*,top",
+            useSyntheticId
+        );
+        assumeTrue("require single-byte-prefix tsid", TsidBuilder.useSingleBytePrefixLayout(fixture.routing.creationVersion));
+        assertIndexShard(fixture, Map.of("dim", Map.of("a", "a")), 5);
+        assertIndexShard(fixture, Map.of("dim", Map.of("a", "b")), 3);
+        assertIndexShard(fixture, Map.of("dim", Map.of("c", "d")), 7);
+        assertIndexShard(fixture, Map.of("other", Map.of("a", "a")), 1);
+        assertIndexShard(fixture, Map.of("top", "a"), 6);
+        assertIndexShard(fixture, Map.of("dim", Map.of("c", "d"), "top", "b"), 0);
+        assertIndexShard(fixture, Map.of("dim.a", "a"), 5);
+        assertIndexShard(fixture, Map.of("dim.a", 1), 2);
+        assertIndexShard(fixture, Map.of("dim.a", "1"), 0);
+        assertIndexShard(fixture, Map.of("dim.a", true), 3);
+        assertIndexShard(fixture, Map.of("dim.a", "true"), 1);
+    }
+
     public void testRoutingPathReadWithInvalidString() {
         int shards = between(2, 1000);
         IndexRouting indexRouting = indexRoutingForPath(shards, "foo").routing();
@@ -1236,7 +1263,7 @@ private int expectedShard(IndexRouting routing, List<Object> keysAndValues, int
      */
     private int hash(IndexRouting routing, List<Object> keysAndValues) {
         if (routing instanceof IndexRouting.ExtractFromSource.ForIndexDimensions) {
-            return tsidBasedRoutingHash(keysAndValues);
+            return tsidBasedRoutingHash(keysAndValues, routing.creationVersion);
         }
         return legacyRoutingHash(keysAndValues);
     }
@@ -1252,7 +1279,7 @@ private int legacyRoutingHash(List<Object> keysAndValues) {
         return hash;
     }
 
-    private static int tsidBasedRoutingHash(List<Object> keysAndValues) {
+    private static int tsidBasedRoutingHash(List<Object> keysAndValues, IndexVersion indexVersion) {
         TsidBuilder tsidBuilder = new TsidBuilder();
         for (int i = 0; i < keysAndValues.size(); i += 2) {
             String key = keysAndValues.get(i).toString();
@@ -1271,6 +1298,6 @@ private static int tsidBasedRoutingHash(List<Object> keysAndValues) {
                 throw new IllegalArgumentException("Unsupported value type for TSID routing: " + value.getClass());
             }
         }
-        return StringHelper.murmurhash3_x86_32(tsidBuilder.buildTsid(), 0);
+        return StringHelper.murmurhash3_x86_32(tsidBuilder.buildTsid(indexVersion), 0);
     }
 }