Query Fanout & Result Merging

## Why a search index is split into shards Eventually one machine can no longer hold the whole **inverted index** (the map from each term to the documents containing it) or answer queries over it fast enough. The usual fix is **document partitioning**. The corpus is divided into **shards**, and each shard is a self-contained index over its own subset of documents. Each shard is copied to several **replicas** for availability and read throughput. Any shard might hold a matching document, so a plain keyword query has to consult all of them. The pattern for serving such a query is **scatter-gather**. ## The scatter phase A **coordinator** (also called a root, broker or aggregator) receives the query and runs these steps: 1. Parse the query once and pick one replica of every shard, choosing by health, load or recent latency. 2. Send the query to all of those replicas **in parallel**, with `k` (the number of results the caller wants) and a deadline. 3. Each shard evaluates the query against only its own documents, scores the matches and keeps its **local top-k** in a bounded heap. A shard does not return every match. A common term can match millions of documents, and shipping them all would saturate the network and the coordinator. Each shard returns at most `k` compact entries: a document ID, a score and any sort values the query needs. ## The gather phase: merging partial top-k lists The coordinator now holds `S` lists of at most `k` entries each, where `S` is the number of shards. There are two common ways to merge them into one global top-k: - **Bounded min-heap** of size `k`. Push every candidate and evict the weakest whenever the heap grows past `k`. This costs about `S·k·log k`. - **k-way merge.** Each list arrives already sorted, so the coordinator can keep a heap of the `S` list heads and pop only `k` times. This costs about `k·log S`. With 20 shards and `k = 10`, the coordinator sees at most 200 candidates and keeps 10. ```pseudocode function search(query, k): lists = parallel for shard in shards: shard.topK(query, k) heap = empty min-heap ordered by (score, docId) for list in lists: for hit in list: heap.push(hit) if heap.size > k: heap.popMin() return heap.drainDescending() ``` ## Why the local top-k is enough The merge is **exact**, not an approximation, as long as scores from different shards are comparable. Take any document `d` in the true global top 10, and suppose `d` were missing from its own shard's top 10. Then 10 documents on that shard would outrank it. Those same 10 documents also outrank it globally, so `d` could not be in the global top 10, which contradicts the starting point. Every global winner is therefore in some shard's list. How engines make scores comparable across shards is a ranking question and is outside this mechanism. ## Query-then-fetch: two phases instead of one Only `k` of the `S·k` candidates survive the merge, so having every shard send full documents would waste most of the bytes. Many engines therefore serve a query in two phases: 1. **Query phase.** Shards return only IDs, scores and sort values, and the coordinator merges them. 2. **Fetch phase.** The coordinator asks only the shards that own the final `k` winners for stored fields, snippets or highlights. The cost is one extra round trip. The gain is that the heavy payload is fetched for `k` documents instead of `S·k`. When `k` is tiny and documents are small, a single phase can be cheaper. ## What grows with the number of shards | Cost | How it scales | |---|---| | Sub-requests per query | one per shard (`S`) | | Candidates merged | up to `S·k` | | End-to-end latency | set by the slowest shard it waits for, not the average | | Coordinator memory | proportional to `S·k` | More shards keep each shard small and fast, but every query pays for all of them, and the chance that at least one shard is slow rises with `S`. Some systems place documents by a routing key, such as a tenant ID. When a query names that key, the coordinator can skip every shard that cannot hold a match, and the fan-out drops to one shard.