Sharding and Routing — Intensive

A document arrives at a coordinating node with no shard stamped on it. By the time that write returns 201 Created, the document lives in exactly one primary shard's Lucene index, replicated to its replicas, and every future read of that document will deterministically find the same shard — without a lookup table, without a directory service, without asking any other node. That determinism is the whole trick of sharding, and it is produced by a few lines of arithmetic on a hash.

This masterclass nails down that arithmetic. It is the deeper companion to two chapters you should have read first: the Sharding, Scaling, and Reader/Writer Separation engineering chapter (shard sizing and the read/write-scaling story) and the Shard Allocation deep-dive (which node a shard lives on). Those answer "how big should a shard be" and "where does it live." This one answers the question underneath both: given a document, which shard? — and then the three resize operations (_split, _shrink, _clone) that the answer makes possible.

After this masterclass you can:

• Derive the routing formula from first principles and compute a document's destination shard by hand, matching what the cluster does to the integer.
• Explain why a document's partition is fixed for life, and why that single invariant is what makes split possible without moving the whole index.
• Distinguish split / shrink / clone at the level of "what happens to the segment files on disk."
• Use custom _routing and index.routing_partition_size to control fan-out, and reason about the hotspot trade-off.
• Read the search-side routing path (OperationRouting#searchShards, GroupShardsIterator, adaptive replica selection, preference).
• Follow the active in-place shard splitting design effort and have an informed opinion about it.

The three labs that follow make you do each of these on a running cluster: Lab SR1 (compute a shard by hand and verify), Lab SR2 (resize and inspect the hard-linked segments), Lab SR3 (custom routing and partitioning).

Prerequisites. You can drive a cluster with curl, read _cat output, and read Java. You understand that a shard is a Lucene index and that search fans out to one copy of every shard. The allocation machinery (which node) is assumed from the Shard Allocation deep-dive and is not re-derived here — this masterclass is about which shard, not which node.

First principles: why hash, and why this hash

You need a function shard = f(document) with four properties:

A naive hash(id) % number_of_shards satisfies the first three and fails the fourth catastrophically: change number_of_shards from 8 to 16 and almost every document maps to a different shard, because the modulus changed. That is the classic rehashing problem, and it is exactly why you "can't change shard count" in the naive design — you'd have to reindex the world.

OpenSearch's answer is a layer of indirection borrowed from consistent hashing: hash into a large, fixed space of routing shards (a.k.a. partitions), and then scale that partition down to an actual shard. The partition space never changes for the life of the index. The actual shard count can change, as long as it stays a divisor or multiple of the partition count — and because each shard owns a contiguous block of partitions, a resize only has to move the partitions that change owner, never recompute every document's hash.

flowchart LR
    DOC["document: _id (or _routing)"] --> H["Murmur3 32-bit hash"]
    H --> P["partition = hash mod number_of_routing_shards<br/>(the fixed hash space)"]
    P --> S["shard = partition / routingFactor<br/>(scale down to number_of_shards)"]
    S --> LUCENE["primary shard's Lucene index"]

The two numbers that govern everything:

• index.number_of_routing_shards — the size of the partition space. Fixed at index creation. Call it R.
• index.number_of_shards — the actual shard count. Call it N. Also fixed at creation, but _split/_shrink create a new index with a different N (same R, as long as it divides cleanly).

The routing factor is routingFactor = R / N. With 8 shards and R = 1024, routingFactor = 128: each shard owns 128 contiguous partitions.

The routing formula, exactly

The hash function is MurmurHash3 x86 32-bit over the UTF-8 bytes of the routing string. Grep the real class to confirm it hasn't moved or changed seed in your checkout:

# The hash itself — a static murmur3 over the routing bytes.
grep -rn "class Murmur3HashFunction\|StringHelper.murmurhash3\|public static int hash" \
  server/src/main/java/org/opensearch/cluster/routing/Murmur3HashFunction.java

# The scaling from a partition to an actual shard id.
grep -rn "calculateScaledShardId\|routingFactor\|getRoutingFactor\|number_of_routing_shards" \
  server/src/main/java/org/opensearch/cluster/routing/OperationRouting.java \
  server/src/main/java/org/opensearch/cluster/routing/IndexRouting.java

# Where the routing-shard number is stored / defaulted.
grep -rn "getRoutingNumShards\|getRoutingFactor\|INDEX_NUMBER_OF_ROUTING_SHARDS" \
  server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

Note: In current lines the per-write routing math lives in IndexRouting (the write path) which delegates the scaling arithmetic that historically lived in OperationRouting.calculateScaledShardId. The names drift between releases; trust the grep. The math below is the invariant — it is what IndexRouting and OperationRouting both compute.

The computation, as the cluster does it, for a document with effective routing value routing (which is _routing if you supplied one, else the document _id):

hash       = Murmur3HashFunction.hash(routing)        // signed 32-bit int
partition  = floorMod(hash, R)                        // 0 .. R-1   (R = number_of_routing_shards)
shardId    = partition / routingFactor                // 0 .. N-1   (routingFactor = R / N)

Three subtleties that trip people up, and that you must get exactly right to match the cluster (you will in Lab SR1):

1. floorMod, not %. Murmur3's 32-bit result is a signed Java int and is frequently negative. Java's % keeps the sign of the dividend, so hash % R can be negative — an invalid shard id. OpenSearch uses a floored modulo (Math.floorMod, equivalently masking when R is a power of two) so the result is always in [0, R). Grep for it:

   grep -rn "floorMod\|& (.*- 1)\|MathUtil.mod" \
     server/src/main/java/org/opensearch/cluster/routing/
   

2. The scale is integer division, truncating. partition / routingFactor floors. Because partitions [k·routingFactor, (k+1)·routingFactor) all divide to the same k, each shard owns exactly routingFactor contiguous partitions. That contiguity is the property split exploits.

3. R is fixed for the life of the index. A _split/_shrink produces a new index that inherits the same R and only changes N. A document's hash and therefore its partition are identical before and after a resize. Only the partition → shardId division changes, because routingFactor changed.

That last point is the keystone of the whole chapter, so say it precisely: a document's partition is invariant for the life of the index; only the mapping from partition to shard changes on resize.

A full worked numeric example

Take the spec's canonical layout: number_of_shards = 8, number_of_routing_shards = 1024. Then:

R             = 1024
N             = 8
routingFactor = R / N = 128         // 128 partitions per shard

Pick a document with no custom routing, _id = "user-42". We need Murmur3HashFunction.hash("user-42"). You will compute this exactly in Lab SR1 with a tiny standalone Java program that calls the real StringHelper.murmurhash3_x86_32; here we walk the shape of the result so the arithmetic is concrete. Suppose the 32-bit hash comes out as the signed value:

hash      = -1,899,304,452            // example signed 32-bit murmur3 result
partition = floorMod(-1899304452, 1024)
          = -1899304452 - 1024 * floor(-1899304452 / 1024)
          = -1899304452 - 1024 * (-1854790)
          = -1899304452 + 1899304960
          = 508
shardId   = 508 / 128 = 3            // integer division floors: 3.96 -> 3

So user-42 lives in shard 3, because its partition 508 falls in shard 3's contiguous block [3·128, 4·128) = [384, 512). The full partition→shard map:

The number 508 (the partition) is fixed for user-42 forever. If you later split this index 8 → 16, routingFactor becomes 1024/16 = 64, and 508 / 64 = 7 — so user-42 moves from shard 3 to shard 7. But its partition is still 508. No hash was recomputed; the document moved because partition 508 now belongs to a different, narrower shard. And critically, every partition in old shard 3's range [384, 512) splits cleanly: [384, 448) → new shard 6, [448, 512) → new shard 7. Old shard 3 becomes exactly new shards 6 and 7, with no partition crossing a boundary. That is why split never has to move a document between unrelated shards — it only ever subdivides one shard into a contiguous set of new ones.

Note: Whether user-42 actually lands in shard 3 depends on the real murmur3 output, which we compute for real in the lab. The point of this section is the arithmetic shape — hash → floorMod → divide — and the contiguity, both of which are exact.

Why split works, drawn

flowchart TD
    subgraph "Before: N = 4, R = 1024, factor = 256"
      S0["shard 0<br/>partitions 0–255"]
      S1["shard 1<br/>partitions 256–511"]
      S2["shard 2<br/>partitions 512–767"]
      S3["shard 3<br/>partitions 768–1023"]
    end
    subgraph "After split x2: N = 8, R = 1024, factor = 128"
      N0["shard 0<br/>0–127"]
      N1["shard 1<br/>128–255"]
      N2["shard 2<br/>256–383"]
      N3["shard 3<br/>384–511"]
      N4["shard 4<br/>512–639"]
      N5["shard 5<br/>640–767"]
      N6["shard 6<br/>768–895"]
      N7["shard 7<br/>896–1023"]
    end
    S0 --> N0
    S0 --> N1
    S1 --> N2
    S1 --> N3
    S2 --> N4
    S2 --> N5
    S3 --> N6
    S3 --> N7

Old shard 0's partitions [0,256) divide into exactly new shards 0 and 1 ([0,128) and [128,256)). No old shard's partitions are scattered across non-adjacent new shards, and no new shard pulls partitions from two old shards. This is the consistent-hashing payoff: the resize is a local subdivision, so OpenSearch implements split by copying the parent shard's Lucene segments into the two children and then deleting the documents that don't belong (a delete-by-query on the wrong-partition documents, which later merges reclaim). No document is hashed again; no document crosses to an unrelated shard.

The factor constraint follows directly: split must be by a multiplicative factor that divides R evenly and is a multiple of the old N. You can go 4 → 8 → 16 → … up to R. You cannot go 4 → 6, because partitions wouldn't divide into contiguous equal blocks. R defaults to a value that permits repeated doubling up to ~1024 shards (grep INDEX_NUMBER_OF_ROUTING_SHARDS / getRoutingNumShards to see the default heuristic for a given N).

Shrink and clone, by contrast

Shrink is the inverse: reduce N by a factor (e.g. 8 → 4 → 2 → 1). The new shard count must divide the old one. Here the merge is trivial and beautiful: several old shards' partition ranges concatenate into one new shard's range, so a new shard is just the union of its source shards' documents. OpenSearch implements shrink by hard-linking every source shard's segment files into the target shard's directory (a hard link, not a copy — same inode, same bytes on disk), then committing. No data is rewritten; the new shard's Lucene index is literally the old shards' segment files sharing the same blocks on disk. You verify the shared inodes with find/ls -i in Lab SR2.

Clone is the degenerate 1:1 case: same N, hard-link all segments, get an independent copy of the index that can then evolve separately (different settings, different lifecycle). It's the cheapest of the three — no per-document work at all.

flowchart LR
    subgraph SPLIT["split (N grows by factor)"]
      direction TB
      sp["copy parent segments<br/>to each child, then<br/>delete wrong-partition docs"]
    end
    subgraph SHRINK["shrink (N divides by factor)"]
      direction TB
      sh["hard-link all source<br/>segments into one target<br/>(shared inodes), commit"]
    end
    subgraph CLONE["clone (N unchanged, 1:1)"]
      direction TB
      cl["hard-link all segments<br/>1:1 into the new index"]
    end

All three run through the cluster manager (formerly master) as a create-index variant. Grep the service that implements the resize:

grep -rn "ResizeType\|resizeIndex\|SHRINK\|SPLIT\|CLONE\|recoverFromLocalShards\|prepareResizeIndexSettings" \
  server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java
grep -rn "class ResizeRequest\|class ResizeAction\|TransportResizeAction" \
  server/src/main/java/org/opensearch/action/admin/indices/shrink/

Resize prerequisites (why the API yells at you)

Both split and shrink require the source index to be made read-only first (index.blocks.write: true), because the operation copies/links a consistent snapshot of segments and must not race with new writes. Shrink additionally requires all primaries of the source on a single node (so the hard links can be made locally — you can't hard-link across machines), enforced by setting index.routing.allocation.require._name to that node and waiting for relocation. You drive both of these by hand in Lab SR2. The target index must also have a shard count consistent with the factor rules above, or the request fails validation in MetadataCreateIndexService.

Custom _routing and routing_partition_size

By default the routing value is the _id, so documents scatter uniformly. Supply a custom _routing and you control which value gets hashed — so you can co-locate related documents on one shard. Index every document for one tenant with ?routing=tenant-7, and they all hash to the same partition and therefore the same shard. A search filtered to that tenant can then be routed to that one shard (?routing=tenant-7), turning an N-shard fan-out into a 1-shard query. That is the single biggest fan-out win available, and you measure it in Lab SR3.

The cost: the _routing value must be supplied on every get/update/delete of that document, because without it the cluster would hash the _id and look on the wrong shard. And co-location risks a hotspot — if one tenant is 40% of your data, its shard is 40% of your index, and the balancer can't fix that (it places whole shards, not partitions).

index.routing_partition_size softens the hotspot. Instead of a routing value mapping to one shard, it maps to a contiguous window of routing_partition_size shards, and the _id picks which one inside the window. Concretely the shard becomes:

shardId = (floorMod(hash(_routing), R) + floorMod(hash(_id), routing_partition_size)) ... / routingFactor

— i.e. the _routing selects a base partition and the _id jitters it within a window of routing_partition_size shards. So one tenant's documents spread across several shards (less hotspotting) but a routed search still only touches that window of shards, not all N (still less fan-out than unrouted). It must satisfy 1 ≤ routing_partition_size < number_of_shards, and when it's set, _routing is required on every document (the _id alone is no longer enough to find a document). Grep the validation and the math:

grep -rn "routing_partition_size\|partitionSize\|INDEX_ROUTING_PARTITION_SIZE\|partitionOffset" \
  server/src/main/java/org/opensearch/cluster/routing/ \
  server/src/main/java/org/opensearch/cluster/metadata/IndexMetadata.java

Search-side routing

Writes target the primary of one shard. Reads are different: a search must hit one copy (primary or replica) of every shard the query could match, then merge. That selection is OperationRouting#searchShards, which returns a GroupShardsIterator<ShardIterator> — one iterator per shard group, each listing the candidate copies in the order they should be tried.

grep -rn "searchShards\|GroupShardsIterator\|ShardIterator\|computeTargetedShards\|class OperationRouting" \
  server/src/main/java/org/opensearch/cluster/routing/OperationRouting.java | head

Three levers shape that selection:

• ?routing= on a search. If you pass a routing value, searchShards computes only the shard(s) that value maps to (one shard, or a routing_partition_size window) and skips the rest. This is the fan-out reduction you measure with _search_shards in Lab SR3. The _search_shards API is searchShards made observable — it returns the exact shard list a search would hit, without running it.

• preference. By default OpenSearch picks among a shard's copies using Adaptive Replica Selection (ARS) — it ranks copies by recently observed response time, queue length, and service time, so slow/overloaded nodes get less traffic. preference overrides this: _primary, _replica, _local, a custom string (sticky hashing so the same client repeatedly hits the same copy — good for cache locality and consistent pagination), or _only_nodes:….

  grep -rn "class ResponseCollectorService\|adaptive\|computeRanks\|ARS\|preference\|_primary\|_local" \
    server/src/main/java/org/opensearch/cluster/routing/OperationRouting.java \
    server/src/main/java/org/opensearch/node/ResponseCollectorService.java | head
  

• The merge. Once searchShards yields the per-shard iterators, the search action issues query-phase requests to one copy per shard and merges (see search execution). Routing's only job is copy selection; everything after is the search path you already know.

flowchart TD
    REQ["search request (maybe ?routing=, ?preference=)"] --> OR["OperationRouting.searchShards"]
    OR -->|routing set| SUB["only the mapped shard(s) / window"]
    OR -->|no routing| ALL["all shards"]
    SUB --> GSI["GroupShardsIterator: one ShardIterator per shard"]
    ALL --> GSI
    GSI --> ARS["per shard, pick a copy: ARS rank or preference"]
    ARS --> QP["query phase to chosen copies, then merge"]

The active in-place shard-splitting design

Today's _split builds a new index and recovers the children from the parent's segments — it is not an in-place mutation of the running index, and it requires the source to be read-only for the duration. There is an active, public design effort to support splitting a shard in place — growing an index's shard count without a full create-and-recover, ideally online. Read these in order; they are a textbook public design trail (cite by full URL, per the book's citation discipline):

The routing RFC (#13925) is the one to read closely after this masterclass: it is exactly the partition → shard mapping above, generalized so that a child shard can own a subset of a parent's partition range while the parent still exists, without a new index and without rehashing. The partition-reassignment picture:

flowchart TD
    P["parent shard S<br/>owns partitions [384, 512)"] -->|in-place split| C1["child S.0<br/>keeps [384, 448)"]
    P -->|in-place split| C2["child S.1<br/>takes [448, 512)"]
    note["routing must now map a partition to S.0 or S.1<br/>based on which sub-range it falls in —<br/>no rehash, contiguous subdivision (issue #13925)"]

If you finish this masterclass and want a real contribution surface, the routing algorithm in #13925 is directly downstream of everything you just learned — and the vector-aware allocation capstone is the allocation-side sibling.

Common bugs and symptoms

Validation: prove you understand this

1. Write the three-line routing formula (hash → partition → shardId) and explain why it uses floorMod rather than %, citing the sign of the murmur3 result.
2. For number_of_shards = 8, number_of_routing_shards = 1024, state the routingFactor and the exact partition range each shard owns. Given a document whose partition is 700, which shard does it land in?
3. Explain, in terms of partitions, why a document's shard can change on _split but the document is never rehashed — and why each old shard becomes a contiguous set of new shards.
4. Contrast split, shrink, and clone by (a) the change to N, (b) what happens to the segment files on disk, and (c) the per-document work. Which two share inodes?
5. State the two prerequisites _shrink enforces on the source index and why each is necessary (consistency; local hard links).
6. A tenant's documents are all indexed with ?routing=tenant-7. Write the search that hits only that shard, and explain what routing_partition_size=3 would change about both the fan-out and the hotspot.
7. Name the method that selects shards for a search and the structure it returns, and describe what Adaptive Replica Selection decides and how preference overrides it.
8. Summarize what the routing RFC #13925 must solve that today's new-index _split avoids.

Next: Lab SR1 — Trace Document Routing to a Shard. Then Lab SR2 — Shrink and Split and Lab SR3 — Custom Routing and Partitioning. For the node-placement half of the story, see the Shard Allocation deep-dive and the vector-aware allocation capstone.