Distributed retrieval and sharding
Split the index across machines and query them all: scatter-gather. Each shard returns its own top k and a coordinator merges them. Two things decide whether it works: how you shard, and the fact that your latency is now the slowest shard's, not the average.
Overview
Shards and replicas are different things
A shard holds a slice of the corpus. Sharding handles data that will not fit — memory or index build time — and every query must visit every shard.
A replica holds a complete copy of a shard. Replication handles query volume and failure, and a query goes to one replica of each shard.
They solve different problems and are often confused. If the index fits on one machine and you are simply serving too many queries, you need replicas and no sharding at all — which is much simpler, because there is nothing to merge.
Parameters
Visualisation
—Readout
What to watch
- Every shard is queried; the coordinator merges local top-k lists.
- Random sharding spreads relevant documents evenly — that is the point.
- Latency is the slowest shard, not the average.
Distributed retrieval and sharding: A Practical Guide
How do you scale retrieval past one machine? What are the trade-offs between sharding strategies?
Shard randomly
The instinct is to shard by topic or tenant so a query only touches one shard. For a multi-tenant system where every query is scoped to one tenant, that is right — it is really many small indexes.
For a general corpus it is a mistake. If all the finance documents live on shard 2, a finance query's true top 50 are all there, and taking only the local top 5 discards forty-five of them. Random sharding spreads relevant documents evenly so a modest per-shard k captures nearly all of them.
The fix if you must shard semantically is to over-fetch from every shard, which costs the bandwidth and latency the routing was meant to save.
Tail latency, and the merge
Scatter-gather waits for the slowest shard, so p99 response time is roughly the p99 of any shard. Add shards and the chance one is slow rises — the classic result is that latency gets worse as you scale out, not better.
Mitigations: hedged requests (ask two replicas, take the first answer), a deadline after which a late shard is dropped and the degraded result served, and keeping shard counts modest.
The merge itself needs care. Scores must be comparable across shards — fine for cosine similarity, not fine for BM25, whose IDF depends on corpus statistics that differ per shard unless they are shared globally. That is a real bug in hand-rolled distributed lexical search.
Things to try
- Switch to semantic sharding. Recall collapses: every relevant document is on one shard, whose local top-k discards the rest.
- With semantic sharding, raise per-shard k until recall recovers. You have paid back the bandwidth the topic routing was meant to save.
- Raise shards to 40 with random sharding. Recall is fine and p99 latency climbs, because scatter-gather waits for whichever shard was slow.
What to remember
Scatter-gather sends the query to every shard, takes a local top-k from each and merges. Shard randomly rather than by topic, or the relevant documents concentrate in one shard and its local k throws most of them away. Latency becomes the slowest shard's, not the average, so p99 gets worse with every shard added. Shards solve data that will not fit; replicas solve query volume.