Skip to content
ArchCrux

Database Internals & Transactions · Staff

Each shard returned its best ten. Why is the global top ten wrong?

Search Returned an Answer. Was It the Right One? · Part 4 of 6 · 6 min read

Read the worked explanation ↓

Interview question. A search service holds one billion documents across twenty shards. Each shard returns its ten highest-scoring hits and a coordinator merges them into a global top ten. Users sometimes see unexpectedly different rankings after a reshard, and paging to the next screen occasionally repeats or skips documents. Is collecting ten candidates per shard fundamentally incorrect? Design a distributed query path that explains both problems.

This question hides two different claims. One is about finding a global top ten from partitions. The other is about whether scores and snapshots mean the same thing across those partitions. Conflating them produces the wrong fix.

If all shards assign documents a single globally comparable score and each shard returns its exact local top ten by that score, taking ten from each shard is sufficient to find the exact global top ten. Any document ranked eleventh locally has at least ten documents on that same shard with scores no worse than its own, so it cannot outrank the global top ten, given a deterministic global tie-breaker. That proof matters. Shard-local top ten is not inherently wrong.

The real system can still be wrong because local scores may not be globally comparable, candidate retrieval may be approximate, and successive page requests may observe different index views.

Each shard contributes scored local candidates. A coordinator merges them, while shard statistics and refreshes affect score comparisons and paging consistency.
Each shard contributes scored local candidates. A coordinator merges them, while shard statistics and refreshes affect score comparisons and paging consistency.
Think first

Design a billion-document search API with twenty shards, 100 QPS, and five pages of stable results. Demonstrate why exact local top-K suffices under globally comparable scores, then select scoring mode, fetch strategy, failure policy and pagination mechanics.

Take a few minutes to form an approach, then compare it with the worked explanation.

Reveal the worked explanation

Start with the scoring population

Take the rare term quasar. Shard A stores 100 documents and contains the term in 2. Shard B stores 100,000 documents and contains it in 2. BM25's rarity component, inverse document frequency, depends on document counts and term frequencies. Under a search mode using shard-local statistics, the same term receives a different scarcity weight on A and B. The scores coming back from those shards are not necessarily based on the same corpus-wide assumptions.

Elasticsearch commonly uses query_then_fetch, whose scoring can use local shard statistics. It also supports dfs_query_then_fetch, which performs an extra distributed frequency collection step before scoring. The latter can reduce differences caused by shard-local term statistics but costs additional work and a network round trip. It is not automatically the right default for every query. Measure whether the difference matters.

Now distinguish query and fetch phases. Each shard evaluates the query over its local search view and returns enough candidate identifiers, scores and sort information for the requested page. The coordinator merges those local lists. Only after that does it fetch the needed stored fields or _source for winners, subject to the API and selected features. This limits fetching but retains shard fanout.

At twenty shards, asking for ten from each yields at most 200 local candidate entries before merge. That arithmetic alone isn't frightening. The costly part is that twenty shards must participate and the query completes only after enough responses arrive or the engine applies its partial-result policy. A slow shard can dominate tail latency, and broad queries may perform substantial per-shard filtering and scoring even when the response contains ten hits.

Why resharding changes ranking

Moving documents among shards can change local term statistics, segment layout and the distribution of scoring work. If ranking uses local statistics, the same query and identical documents can yield different relative scores after placement changes. Replica variations and refreshing index views can also change intermediate search state. A correct distributed design states which scoring contract is expected. Is approximate relevance acceptable? Is stability across a short session more important than perfect corpus-wide statistics?

Do not oversell global term statistics either. A global BM25 score is still only a lexical scoring model. It does not repair a poorly designed query, irrelevant products or missing candidates. The previous episode on relevance showed how to measure those faults separately.

Suppose we replace exact lexical retrieval with approximate vector search. Each shard might not return its true local top ten in vector-distance order because an approximate nearest neighbor structure explores only part of the graph or cluster. The exact top-ten proof no longer applies to missing candidates. Increasing per-shard candidate breadth or using a global reranker can improve recall, at added latency. That's a different failure mode from scoring with local BM25 statistics.

Pagination is a separate consistency problem

A shopper sees page one at 12:00:00 and page two at 12:00:03. In between, several shards refresh. A product is added ahead of the previous rank boundary and another document changes score. If the application uses from and size against each new current view, the rank positions have shifted. A document may appear twice, and another may be skipped. Deep from + size also requires collecting and sorting increasing numbers of hits from each shard, which becomes expensive.

For stable deep pagination in Elasticsearch, prefer search_after with a point-in-time search context when that behavior fits the product. Keep the query and sort criteria unchanged, use the returned sort values including a deterministic tie-breaker, pass the latest PIT ID as documented, and close the PIT after use. A PIT preserves a search view for a bounded period. PIT retention consumes resources and does not freeze business truth.

If the interface intentionally shows live trending results, a consistent snapshot across multiple pages may not be the desired product. You could refresh from page one instead. The contract must be explicit.

What would I test?

I would create a fixture with known scores and distribute it unevenly across shards. Verify the exact top-K merge proof under a shared score and tie-breaker. Then vary local document-frequency distributions to detect ranking changes. Measure query_then_fetch against DFS mode, shard p99, coordinator CPU, partial shard failures, and fetching cost. Finally, add writes between page requests and test search_after with and without PIT.

The engineering lesson is that distributed top-K correctness depends on comparable scores, genuine local candidates and a defined search snapshot. Candidate counts cannot prove correctness. A strong interview answer proves the simple case first, then identifies exactly which production assumption breaks that proof.

Go deeper

For deterministic sorting with equal scores, include an explicit stable tie-breaker or use the PIT-specific tie-breaker mechanism. Term statistics collected by DFS improve cross-shard score comparability for supported queries, but global ranking can still be affected by analyzer differences, index updates, and scoring-model limitations. For aggregations and approximate nearest neighbor retrieval, shard-local candidate limits have different mathematical implications than exact top-K under a common score. Do not generalize the exact lexical ranking proof to all distributed search operations.

Check your reasoning

  • Prove local exact top K is sufficient under comparable scores and total tie-breaking.
  • Explain per-shard BM25 term statistics and DFS alternative.
  • Identify ANN candidate recall as a distinct break of exactness.
  • Budget shard fanout, tail latency and query/fetch work.
  • Use PIT, search_after and stable sort criteria to avoid cross-page gaps.

Common mistake: For exact comparable scores the coordinator always needs more than K from every shard.

Distributed search rankings shift after resharding and pages repeat documents. Which investigation addresses the assumptions behind exact local top-K merging?

Read the answer explanation

Explain score comparability and use DFS or PIT where justified — Exact local K is mathematically enough under one globally comparable ranking, but shard statistics and changing views can invalidate production assumptions.

Apply it elsewhere

A vector-search product divides embeddings across regions. Each region returns approximate local nearest neighbors, and the central service re-ranks them. Explain why the lexical exact-top-K proof does not guarantee vector recall, and design a recall-versus-latency experiment.

Compare your approach

Prove or refute exact local top-K rigorously, distinguish shard-local BM25 scoring from global statistics, account for query/fetch fanout, separate ANN candidate recall, and use a consistent PIT plus deterministic tie-breaking for stable pagination.

One more constraint

If every shard computes the same exact global score and provides its exact local top K with a total deterministic ordering, must a global top-K result use more than K documents from a single shard?

Quick revision

Exact local top-K from every shard is enough to recover exact global top-K when scores and tie-breaking are globally comparable. Search can still rank differently because shard-local BM25 statistics produce noncomparable score conditions, ANN candidate retrieval is approximate, or refreshes change the search view. Elasticsearch offers DFS scoring at extra cost. Coordinate shard candidates, then fetch winners. For stable deep paging, combine search_after, a point-in-time view and deterministic sort values, and budget slow-shard fanout.

References