Scatter-gather
Asking every shard and merging the answers, and the tail latency it brings.
01The intuition
The problem it solves
A search engine holds a billion documents, far too many for one machine. So the index is split into 100 shards, each holding ten million documents on its own server. A query for "monsoon forecast" could match documents on any shard. No single machine can answer it alone.
The simple option is to ask the shards one after another: shard 1, wait, shard 2, wait, and so on. If each takes 20 ms, the answer takes 2 seconds. Users leave after about 300 ms. The query must be sent to all shards at the same time and the partial answers combined.
Build it one piece at a time
Step 1: a coordinator. One server receives the query and does no searching itself. It sends the query to every shard in parallel and waits.
Step 2: each shard returns only its best few. Each shard runs the query over its ten million documents and returns its top 10 by score, not every match. That is 100 shards × 10 results = 1,000 candidates arriving at the coordinator, each a document id and a score.
| Shard | Time | Top results (id: score) |
|---|---|---|
| 1 | 18 ms | d41: 9.2, d77: 8.1, … |
| 2 | 22 ms | d13: 9.5, d90: 7.0, … |
| … | ||
| 100 | 19 ms | d58: 8.9, … |
Step 3: merge. The coordinator merges the 100 sorted lists into one and keeps the global top 10, a k-way merge with a heap. Scores must be comparable across shards, so shards use the same scoring formula with shared global statistics (how rare each word is overall), refreshed periodically.
Step 4: the slowest shard sets the latency. The coordinator waits for all 100. If 99 answer in 20 ms and one is doing garbage collection and takes 400 ms, the user waits 400 ms. Scatter-gather makes the tail of every shard the tail of every query.
Step 5: tame the tail. Three tools. Replicas: keep two copies of each shard and send the query to the one that is less loaded. Hedged requests: after, say, 30 ms with no answer from a shard, send the same query to its replica and take whichever answers first. A deadline: at 100 ms, return the best results from the shards that answered and mark the response partial, because a slightly worse answer now beats a perfect one later.
Why it works, with the numbers
Parallel fan-out makes latency the maximum of the shard times rather than the sum, but maximums are cruel. If each shard answers within 20 ms 99 percent of the time and is slow 1 percent of the time, the chance that all 100 are fast is
0.99100 ≈ 0.37, so 63% of queries hit at least one slow shard
Hedging after 30 ms changes that: a slow shard is covered by its replica, and the probability both copies are slow at once is 0.01 × 0.01, so nearly every query finishes in about 50 ms. The cost is extra load, since hedged queries are sent twice; limiting hedges to the slowest 5 percent of requests keeps the overhead near 5 percent. The guarantee is "top k by score across all shards" when every shard answers, and a clearly labelled partial result otherwise. The honest failure modes: a coordinator that fans out to 100 shards and merges 1,000 results per query is itself a hot spot, so there are many coordinators behind a load balancer; and deep pagination (page 100 of results) forces every shard to return 1,000 results, which is why search engines cap how far you can page.
The trade-off you must say out loud
More shards mean more parallelism and a worse tail; fewer, bigger shards mean each query does more work. Say the shard count and the hedge threshold as numbers. The interviewer's follow-up is usually about ranking across shards: how do you make a 9.2 from shard 1 mean the same as a 9.2 from shard 2, which is shared global term statistics, or a second ranking stage over the merged candidates.
Where it shows up in the 100 Google problems
- Design a distributed search engine like Elasticsearch: the coordinator, per-shard top k, replicas, hedged requests and the pagination cap are the core of the design.
- Design Google Search: thousands of index shards queried in parallel, with a strict deadline and partial results rather than a late page.
- Design Twitter search and Gmail search fan out over time-based or per-user shards in the same way.