One machine from many

The Elasticsearch essays · January 2026

An index in the wild is cut into pieces and spread across many computers, yet a search must still feel like it asked one machine. This essay is about the arithmetic of that illusion: how one query fans out and merges back, why scores and counts can quietly lie in the merge, which copy of the data answers, and why page 1,000 costs a thousand times more than page one.

1 · One search, many machines

An index in the wild is sharded: cut into pieces, each a fully working index of its slice, spread across machines. Any node can receive your query and becomes its coordinator: it broadcasts to one copy of every shard, each shard ranks its own documents and returns only its few best candidates, just IDs and scores, and the coordinator merges them and fetches full text for the winners only.

The figure below holds one search, frozen in time. The timeline slider is time: drag it slowly through the request's life, backwards too. Then change the shard count and drag through it again:

timeline

shards3

Scrub through: fan-out → each shard scans and ranks its slice in parallel → candidates travel back → merge → fetch the winners.
Now the experiment: set shards to 1 and scrub the scan phase, then set it to 6 and scrub again. The scan shrinks because each shard works a smaller slice at the same time; the merge grows because more candidates come back. Same 24 documents, same final top-3, very different work.

The elegance is in what doesn't move: a query over a billion documents ships a few hundred bytes each way. The heavy work happens where the data lives, in parallel. Add shards and the same search gets faster, at the price of a slightly busier merge, a trade you just measured yourself.

2 · The rarity problem

There is a subtle bug hiding in "each shard ranks its own slice." BM25 weighs terms by rarity, and rarity is computed from the documents a shard can see. If one shard happens to hold most of the documents mentioning kubernetes, that term looks common there and exotic on its neighbor, and identical documents get different scores depending on where they landed. Usually the shards are big enough that the statistics agree; when they don't, one round-trip fixes it:

DFS pre-phase: agree on global term statistics first

Two identical documents, one per shard. Without the pre-phase, shard B thinks "kubernetes" is rare (it only sees 2 mentions among its docs) and inflates its document's score: an artifact of placement, not relevance. Turn on the pre-phase (dfs_query_then_fetch): the shards first pool their term statistics, score with the same global rarity, and the tie is honest again. It costs a round-trip, which is why it's reserved for small or skewed indices.

3 · Choosing the fastest copy

Every shard exists in copies, so every fan-out faces a choice: which replica answers? Round-robin is fair and blind; it keeps sending queries to a node that is busy garbage-collecting. Elasticsearch instead scores each copy by its recent response times, queue depth, and service time, and sends traffic where the wait is shortest: adaptive replica selection. Both strategies below face the same three nodes, one of which degrades on and off:

–

slow queries · round-robin

–

slow queries · adaptive

time

node B degradation

Scrub time forward. Node B periodically slows (garbage collection, a noisy neighbor, a merge storm). Round-robin keeps feeding it one query in three, so whenever B struggles, a third of all searches are slow, and those are exactly the requests your p99 is made of. The adaptive router notices B's response times within a few requests and routes around it. It does not forget B forever: as its information goes stale, B looks attractive again and gets an occasional probe (the dots), which is how the router learns that B has recovered. Raise the degradation: the count of slow queries barely moves for the adaptive router.

4 · The deep-paging cliff

Distributed reading has a trap built into its arithmetic. To serve results 10 through 20, the coordinator cannot ask each shard for "your share": it must fetch each shard's entire top 20, because any shard might own all ten winners. Now ask for page 1,000:

–

entries the coordinator sorts

with search_after, any page

page number (10 results per page)

Each bar is what one of three shards must produce; the coordinator sorts the union to find the ten you asked for and throws the rest away. The cost grows with the page number itself, which is why Elasticsearch caps from + size at 10,000 by default. The escape is search_after: remember the sort values of the last result you saw (inside a point-in-time, so the pages stay consistent), and ask each shard only for what comes after them, ten per shard, forever, at page one prices.

5 · The wrong top ten

Aggregations fan out too, and here the merge can quietly lie. Ask for the top 3 error codes: each shard sends its local top 3, the coordinator sums them. But a term that is fourth place everywhere is never sent by anyone, and a term that is merely hot on one shard looks like a global champion. This dataset is rigged to break it:

shard_size · how many terms each shard sends3

At shard_size 3 the merged list crowns auth_fail, and cache_miss, the true champion at 222, does not appear at all: it is fourth on every shard, so no shard ever mentions it. The blue bar is what the coordinator counted, the gray bar the truth. Widen shard_size and watch the truth assemble. Elasticsearch defaults shard_size to 1.5 × size + 10, and the response carries doc_count_error_upper_bound, computed as in the last line: an unsent term can have no more, on each shard, than the smallest count that shard did send. It is the honest confession of how wrong the counts could be.

The partial-then-merge pattern you have now seen three times, in scores, in pages and in counts, reappears inside ES|QL's compute engine; what happens when these machines fail, elections, replicas, rebalancing, is The cluster.