Distributed search: query then fetch

One machine cannot hold 50 million products and answer in 50 ms, so a search index is split into primary shards. Each shard is a complete, small index on its own machine. A query has to visit every shard, then merge the results. Elasticsearch does this in two phases: query, then fetch.

The big idea

Picture a school sports day. Each class picks its own top three runners and reports them. The organizer merges the reports into one top three, then looks up the athlete details for only those three.

The picture breaks in one place. Each class times runners with its own stopwatch. Scores are only comparable if everyone uses the same rules. Search has the same issue, covered below under IDF.

Walking through it

Say a query asks for the top 10 results, and the index has three shards.

First, the query. Each shard scores its own documents and returns only the IDs and scores of its top hits. Here that is 30 entries in total.

Then, the fetch. The coordinator keeps the global top 10. It then fetches the full documents only for those 10, here 4 from shard 1 and 6 from shard 3. Shard 2's hits did not make the cut, so it sends nothing more.

Fetching full documents for all 30 candidates would waste work, which is why the two phases exist.

What it costs

• Latency follows the slowest shard. More shards mean more parallel work, but also more chances for one slow shard to delay the whole query.
• Scores can disagree across shards. Each shard computes IDF from its own document counts by default. With small or skewed shards, the same document can score differently depending on where it lives. Large, evenly routed shards smooth this out.

Sizing and routing

Elastic's guidance is shards of roughly 10 to 50 GB. For 50 million products at about 2 KB each, that is about 100 GB, so 3 to 5 primary shards with one replica each is a sensible start.

A document goes to the shard chosen by hash(routing) % number_of_primary_shards. Because of that modulo, the primary shard count is fixed at index creation. Changing it means a reindex.

Common mistakes

• Deep pagination. from: 10000 makes every shard produce its top 10,010 results. Use search_after instead.
• Too many tiny shards or a few giant ones. Tiny shards waste memory, and giant shards are slow to recover and rebalance.
• Comparing scores across differently sized shards. Expect small differences in ranking until shards are large.
• Forgetting the tail. A query that hits one slow shard is slow for the user, even if the other shards are fast.

Check yourself

1. Using the diagram, how many entries does the coordinator merge to return a top 10 from three shards?

Answer: 30 entries, the top 10 from each shard, merged down to the best 10.

2. Why can't you change the number of primary shards on a live index?

Answer: Documents are routed by a hash modulo the shard count, so changing the count would send most documents to different shards.

Key takeaways

• A query scatters to every shard and gathers a merged top list.
• The query phase returns IDs and scores; the fetch phase returns full documents for winners only.
• Latency tracks the slowest shard.
• Choose the primary shard count up front, and keep shards in the 10 to 50 GB range.

Further learning

Next, see how the index stays in sync with the database in the search indexing pipeline.