Billion-scale sharding, routing and cold start

What it is

The set of decisions that appear when an index stops fitting on one machine: how documents are partitioned across shards, how a query reaches the right shards, and what happens when a shard is cold.

Two partitioning strategies, and they are not equivalent:

DOCUMENT PARTITIONING (local index)
  Each shard holds a complete index over a SUBSET of documents.
  Query: broadcast to all shards, each returns its top k,
         merge.
  + Writes are local to one shard. Adding a document touches
    one shard.
  + A shard failure loses a fraction of results, not a term.
  - Every query touches every shard: the tail-at-scale problem.
  - IDF is per shard unless corrected.

TERM PARTITIONING (global index)
  Each shard holds the complete postings list for a SUBSET of
  terms.
  Query: route to only the shards holding the query's terms.
  + A query touches only as many shards as it has terms.
  - Writes touch every shard containing any of the document's
    terms. A document with 200 distinct terms touches many
    shards.
  - Catastrophic skew: the shard holding a common term serves
    a huge share of traffic.

Essentially every production search system uses document partitioning, and the reason is the write path: term partitioning makes indexing a scatter operation and creates irreducible hot shards on common terms. Knowing that both exist and why one won is a better answer than knowing only one.

Commonly confused with replication. Sharding splits the data for capacity; replication copies it for throughput and availability. A billion-document index is typically both: sharded to fit, then each shard replicated for QPS.

The problem it solves

1 billion documents, ~2 KB each of indexed text.
  Lucene index at ~0.5x:                    ~1 TB
  Vectors, 768-dim int8 + HNSW graph:       ~1 TB
  Total serving footprint:                  ~2 TB

That does not fit in memory on one machine, and serving from
disk costs 20 µs per random read against 100 ns from memory,
so a query touching a few thousand postings entries goes from
microseconds to tens of milliseconds.

The sharding decision is forced by memory, and everything downstream follows from it: scatter-gather latency, per-shard IDF, cold-start behaviour on a restart, and how a rebuild is staged.

Mechanics

Sizing shards, and the two competing pressures

BIGGER SHARDS                    SMALLER SHARDS
+ Fewer shards to fan out to     + Faster per-shard query
+ Better IDF statistics          + Faster recovery and rebalance
+ Less merge coordination        + Finer capacity granularity
- Slower per-shard query         - Worse tail: p99 is the max
- Slow recovery (copying 200 GB)   over more shards
- Coarse capacity steps          - More coordination overhead

The rule that resolves it: target 20 to 50 GB per shard, and let the shard count fall out. Not a fixed shard count, which is the common error.

2 TB / 40 GB = 50 shards.
At 3 replicas each: 150 serving units.
On 64 GB nodes with one shard each: 150 nodes.
On 256 GB nodes with 4 shards each: ~38 nodes.

And the arithmetic that governs the choice of shard count is the tail:

If each shard is slow (above p99) with probability p, a query
touching N shards is slow with probability 1 - (1-p)^N.

N = 10,  p = 1%:  9.6%  of queries hit a slow shard
N = 50,  p = 1%:  39.5%
N = 100, p = 1%:  63.4%

To keep the QUERY p99 at the target with N = 50, each shard
needs roughly its p99.98.

This is the tail at scale, and it is the strongest argument against over-sharding. Every additional shard makes the aggregate p99 worse, and the fix (hedged requests, backup requests after the p95) costs extra load.

Routing: reducing the fan-out

The most valuable optimisation available, because it attacks the tail directly.

BROADCAST (the default)
  Every query touches every shard. Simple, and the tail scales
  with shard count.

PARTITION-AWARE ROUTING
  Shard by a field that appears in most queries, so a query
  touches one shard.
    by tenant     -> a tenant's query touches one shard
    by locale     -> a French query touches the French shards
    by category   -> if queries are category-scoped
  *** This turns a 50-shard scatter into a 1-shard lookup and
      removes the tail problem entirely for those queries. ***

TIERED ROUTING
  Split the corpus by quality or popularity:
    tier 1: the best 5% of documents, on fast nodes
    tier 2: the rest
  Query tier 1 first; if it yields enough good results, stop.
  Only fall through to tier 2 when tier 1 is insufficient.

Tiered routing is the technique large web search engines use and it is under-known. The observation behind it: for most queries, the answer is in the popular subset, so the majority of queries can be served from a small, hot, fast tier and never touch the bulk of the index.

Measured shape (illustrative):
  70% of queries satisfied entirely by tier 1 (5% of corpus)
  30% fall through to tier 2

Latency: tier-1-only queries touch 3 shards instead of 50.
Cost: tier 2 can be on cheaper, denser, colder hardware.

The failure to design for: knowing when tier 1 is insufficient. A cheap heuristic (fewer than $k$ results above a score threshold) works, and it must be tuned, because falling through too often loses the benefit and too rarely loses recall.

The IDF problem in a sharded index

BM25's IDF uses N (total documents) and n_t (documents
containing t). In a sharded index, each shard knows only its
own counts.

Shard A: 20M docs, "kubernetes" in 800k     -> IDF = 3.24
Shard B: 20M docs, "kubernetes" in 40       -> IDF = 13.1

The same term scores 4x higher on shard B, so shard B's
results dominate the merged top-k for no relevance reason.

With random document assignment this is usually a small effect, because term frequencies are similar across shards by the law of large numbers. With any non-random assignment it is severe, and non-random assignment is exactly what partition-aware routing introduces.

Sharded by locale: "the" is common in English shards and rare
in Japanese ones. Sharded by category: "engine" is common in
automotive and rare in software. Both produce large IDF
skew.

The fixes:
  dfs_query_then_fetch   gather global term statistics in an
                         extra round trip, then score. Correct,
                         and it costs a round trip on every query.
  Periodic global stats  broadcast approximate global document
                         frequencies to every shard, refreshed
                         hourly. Cheap, approximately correct,
                         and what most large systems do.
  Rank-based fusion      merge by rank rather than score, which
                         is immune to the incomparability.

Periodic global statistics is the right default: term frequencies change slowly, so an hourly broadcast is accurate enough and costs nothing per query.

Cold start: the failure that takes down a fleet

The most operationally important part of this topic.

A shard restarts. Its index is on disk; its page cache is empty.

  Warm:  postings in page cache, query 15 ms
  Cold:  every postings read is a disk seek, query 400 ms+

A cold shard in a 50-shard scatter makes EVERY query slow,
because the query waits for the slowest shard.

And the cascade, which is the actual outage:

1. Shard 17 restarts, cold.
2. Every query touching shard 17 (all of them) takes 400 ms.
3. Client timeouts fire; clients retry.
4. Retries multiply load on an already-slow fleet.
5. Other shards saturate and their latency rises.
6. More timeouts, more retries. The fleet is now down.

The mitigations, in the order they matter:

1. DO NOT SERVE TRAFFIC UNTIL WARM.
   The health check must fail while cold, so the load balancer
   does not route to it. This is the single most important
   line and it is routinely missing: a process that has
   started is not a process that can serve.

2. WARM DELIBERATELY.
   Replay a sample of recent production queries against the
   shard before marking it healthy. A few thousand queries
   populates the page cache for the hot postings.

3. RESTART SLOWLY.
   One shard at a time, with a delay. A rolling restart of 50
   shards with no delay is 50 cold shards in sequence, and the
   fleet never recovers between them.

4. HEDGE.
   Send a backup request to a replica after the p95. A cold
   replica loses the race to a warm one, so the cold shard
   costs nothing while it warms.

5. mmap AND PRELOAD.
   Lucene mmaps its index files; a preload hint (or reading
   the files) faults them in before serving.

Point 1 is the one to say first, because it converts a fleet outage into a slow rollout.

Rebuilding a sharded index

Full rebuild of 50 shards, HNSW at ~6 hours per shard:
  serially:  300 hours. Not viable.
  in parallel across a build fleet: 6 hours plus scheduling.

The staging, which is the alias-swap pattern per shard:
  1. Build all 50 new shards offline, in parallel.
  2. Verify: recall on a held-out set, per shard, against the
     current index. A shard that regressed does not ship.
  3. Swap the alias, per shard, gradually, watching quality
     metrics between shards.
  4. Keep the old shards for the rollback window.

Swapping per shard rather than all at once is the important detail, because it makes the rollout gradual and gives you a quality signal before full exposure. And it introduces a constraint: the old and new shards must be query-compatible during the transition, which for vector indexes means the embedding model cannot change in the same operation, since mixing vectors from two model versions is meaningless.

A worked example: 1 billion documents

CORPUS       1B documents, ~2 KB indexed text each
TRAFFIC      8,000 QPS peak
LATENCY      p99 under 300 ms

STORAGE
  Lexical index at 0.5x:                       1.0 TB
  Vectors 768-dim int8 + HNSW graph (M=32):    1.0 TB
  Total:                                       2.0 TB

SHARDING
  At 40 GB per shard: 50 shards.
  Tail check: 50 shards at 1% slow each -> 39.5% of queries
  hit a slow shard. That will not meet a p99 target without
  help.
  -> Reduce effective fan-out with routing, and hedge.

ROUTING DECISION
  Query logs show 82% of queries carry a locale filter.
  Shard primarily by locale, secondarily by hash within locale:
    en: 24 shards, de: 7, fr: 5, es: 4, ja: 3, other: 7
  A typical query now touches 24 shards (English) or fewer,
  not 50.
  -> And the IDF problem is now SEVERE, because locale
     partitioning is maximally non-random. Global term
     statistics broadcast hourly.

TIERING
  Within English, split by document quality:
    tier 1: top 5% by quality score, 2 shards, on memory-rich
            nodes
    tier 2: the remaining 22 shards
  Route to tier 1 first; fall through when fewer than 30
  results clear the score threshold.
  Measured: ~65% of queries never touch tier 2.
  -> Effective fan-out for most queries: 2 shards.
  Tail at 2 shards, p=1%: 2% of queries hit a slow shard,
  against 39.5% at 50. *** This is the whole win. ***

REPLICATION AND FLEET
  Each shard x3 for QPS and AZ tolerance: 150 units.
  8,000 QPS / (queries mostly touching 2-3 shards) is
  comfortably served; size from Little's Law on the per-shard
  service time.

COLD START
  Health check fails until 2,000 warm-up queries have run.
  Rolling restarts: one shard per minute, 50 minutes total.
  Hedge at the p95 to a replica.

The lesson to state: the routing decision did more for latency than any amount of per-shard optimisation. Going from a 50-shard broadcast to a 2-shard tier-1 lookup took the probability of hitting a slow shard from 39.5 percent to 2 percent, and no query tuning achieves that.

Production evidence

Barroso, Dean and Hölzle, "Web Search for a Planet: The Google Cluster Architecture" (IEEE Micro 2003) describes document partitioning with replication and explains why term partitioning was rejected: the write path and the load skew.

Dean and Barroso, "The Tail at Scale" (CACM 2013) is the source of the fan-out arithmetic and of hedged and backup requests, which are the standard mitigations here.

Elasticsearch's shard-sizing guidance recommends a target of tens of gigabytes per shard rather than a fixed count, and its documentation on dfs_query_then_fetch explains the IDF problem and the extra round trip it costs.

Google's tiered index serving is described in the search-architecture literature: serving most queries from a small high-quality tier and falling through only when necessary, which is the technique that makes the fan-out tractable.

Lucene's MMapDirectory and its preload option are the mechanism behind warm-up, and the Elasticsearch documentation on index.store.preload exists specifically because cold page cache after a restart is a known production problem.

The Kubernetes readiness probe is the general form of "do not serve until warm", and the distinction it draws between liveness and readiness is exactly the distinction that matters here: the process is alive and is not ready.

The debate

The case for many small shards: faster per-shard queries, faster recovery when a node fails, and finer capacity granularity so you can add one node rather than one twentieth of the fleet.

The case for fewer large shards: the tail. Every additional shard multiplies the chance that a query waits on a slow one, and at 50 shards with a 1 percent slow rate almost 40 percent of queries are affected. Fewer shards also means better IDF statistics and less merge coordination.

The case for term partitioning: a query touches only the shards holding its terms, so the fan-out is bounded by query length rather than by corpus size, which is a genuinely attractive property.

My position: document partitioning, 20 to 50 GB per shard, and spend the effort on reducing the effective fan-out rather than on optimising per-shard latency.

Document partitioning because term partitioning makes indexing a scatter operation and creates irreducible hot shards on common terms, which is why no production search system uses it despite the appealing query-side property.

The shard size target rather than a shard count, because a fixed count is wrong at every scale except the one it was chosen for, and 20 to 50 GB is where recovery time, query latency and merge cost are all tolerable.

The decision I would defend hardest is prioritising routing over per-shard tuning. In the worked example, tiered routing took the effective fan-out from 50 shards to 2, which took the probability of a query hitting a slow shard from 39.5 percent to 2 percent. No amount of query optimisation, cache tuning or hardware achieves that, because the problem is not per-shard speed, it is the maximum over many shards. Teams reliably optimise the shard and not the fan-out.

The operational rule I would state without hedging: a shard must not serve traffic until it is warm, and the health check is where that is enforced. A cold shard in a broadcast makes every query slow, timeouts trigger retries, retries saturate the healthy shards, and a routine rolling restart becomes a fleet outage. That is one line in a readiness probe and it is the difference between a slow rollout and an incident.

Where I would push back on the premise: check whether it needs sharding at all. Two terabytes forces it, and a great many "billion-scale" problems are 50 million documents that fit on one node once the vectors are quantised, and sharding them buys a tail problem for nothing.

Follow-up Q&A

"Document or term partitioning?" Document, and I would say why term partitioning loses despite its attractive query property. With term partitioning a query touches only the shards holding its terms, which sounds ideal, and the write path ruins it: a document with two hundred distinct terms touches many shards on every index operation. And the shard holding a common term serves a huge share of traffic, which is irreducible skew. Document partitioning keeps writes local and its cost is the broadcast, which you attack with routing.

"How do you choose the shard count?" By target shard size, not by count: 20 to 50 gigabytes each, and let the count fall out. Two terabytes gives 50 shards. A fixed count is wrong at every scale except the one it was picked for. And the count matters mainly through the tail: at 50 shards with each 1 percent likely to be slow, nearly 40 percent of queries hit a slow one, so shard count is a latency decision as much as a capacity one.

"That fan-out arithmetic sounds bad. What do you do about it?" Reduce the effective fan-out, which is worth more than any per-shard optimisation. Two techniques. Partition-aware routing: shard by a field most queries carry, like locale or tenant, so a query touches a subset. And tiered routing: put the best few percent of documents in a small fast tier, query it first, and fall through only when it yields too few good results. In the case I worked, those took the typical query from 50 shards to 2, which took the chance of hitting a slow shard from 39.5 percent to 2.

"What breaks when you shard by locale?" IDF. BM25 computes inverse document frequency from per-shard counts, and locale partitioning is maximally non-random, so a term common in English shards and rare in Japanese ones scores wildly differently across them, and one locale's results dominate the merge for no relevance reason. With random assignment this is a small effect; with any deliberate partitioning it is severe. The fix is broadcasting approximate global term statistics hourly, which is cheap and accurate enough, rather than paying dfs_query_then_fetch's extra round trip on every query.

"A shard restarts. What happens?" Its page cache is empty, so every postings read is a disk seek and queries against it go from about 15 milliseconds to 400 or worse. In a broadcast that makes every query slow, because the query waits on the slowest shard. Then client timeouts fire, clients retry, the retries load the already-slow fleet, other shards saturate, and a routine restart becomes an outage.

"So how do you prevent that?" Five things, and the first matters most. The health check must fail while the shard is cold, so the load balancer does not route to it: a process that has started is not a process that can serve, and that one line converts a fleet outage into a slow rollout. Then warm deliberately by replaying a few thousand recent production queries before marking it ready. Then restart one shard at a time with a delay, because a rolling restart with no delay is fifty cold shards in sequence. Then hedge at the p95 so a cold replica loses the race to a warm one. And preload the mmapped index files.

"How do you rebuild a sharded index?" Build all shards offline in parallel, since serially at six hours per shard it would be three hundred hours. Verify recall per shard on a held-out set against the current index, and a shard that regressed does not ship. Then swap the alias per shard, gradually, watching quality between shards, which gives a signal before full exposure. And keep the old shards for the rollback window. The constraint that introduces is that old and new must be query-compatible during the transition, so the embedding model cannot change in the same operation.

"When is sharding the wrong answer?" When it fits on one node, which is more often than people assume. Fifty million documents with int8-quantised vectors is well under a hundred gigabytes, so it fits, and sharding it buys a tail problem and scatter-gather coordination for no benefit. I would check the storage arithmetic before designing a distributed index, because the replicated-single-node design is dramatically simpler and it is available more often than teams expect.

Common misconceptions

"More shards is faster." Each shard is faster and the query is not, because the query waits for the slowest of them. Shard count is a tail-latency decision.

"Term partitioning gives smaller fan-out, so it should win." Its write path is a scatter and common terms create irreducible hot shards, which is why no production system uses it.

"IDF skew is a theoretical concern." With random assignment, mostly. With locale or category partitioning, which is exactly what routing introduces, it is severe.

"A restarted process is ready." It is alive. A cold shard in a broadcast makes every query slow and triggers a retry cascade.

"Optimise the shard." Optimise the fan-out. Going from 50 shards to 2 does more than any per-shard tuning can.

Interview delivery note

Give the two partitioning strategies and say why one won, because knowing both is the signal: "Two options: document partitioning, where each shard indexes a subset of documents and you broadcast; or term partitioning, where each shard holds full postings for a subset of terms and you route by query terms. Term partitioning has the better query property and essentially nobody uses it, because a document with two hundred distinct terms touches many shards on every write, and the shard holding a common term is an irreducible hot spot."

Size by shard rather than count, and connect it to the tail: "I'd target twenty to fifty gigabytes per shard and let the count fall out, so two terabytes is about fifty shards. And shard count is really a latency decision: at fifty shards with each one percent likely to be slow, nearly forty percent of queries hit a slow shard."

Then the move that is the actual answer: "Which is why I'd spend the effort on reducing effective fan-out rather than on per-shard tuning. Partition by locale if most queries carry one, and tier by document quality so the best five percent sits in a small fast tier you query first. In the case I worked that took the typical query from fifty shards to two, so the chance of hitting a slow shard went from forty percent to two. No per-shard optimisation gets you that."

Volunteer the cold-start cascade, because it is where the operational experience shows: "and the thing I'd design for explicitly is cold start. A restarted shard has an empty page cache, so its queries go from fifteen milliseconds to four hundred, and in a broadcast that makes every query slow. Timeouts fire, clients retry, the retries saturate the healthy shards, and a routine rolling restart becomes a fleet outage. The fix is that the health check fails while it's cold, so it doesn't receive traffic. That's one line and it's the difference between a slow rollout and an incident."

The Staff & Team Lead Interview Reference