Distributed retrieval and sharding

Shard randomly

The instinct is to shard by topic or tenant so a query only touches one shard. For a multi-tenant system where every query is scoped to one tenant, that is right — it is really many small indexes.

For a general corpus it is a mistake. If all the finance documents live on shard 2, a finance query's true top 50 are all there, and taking only the local top 5 discards forty-five of them. Random sharding spreads relevant documents evenly so a modest per-shard k captures nearly all of them.

The fix if you must shard semantically is to over-fetch from every shard, which costs the bandwidth and latency the routing was meant to save.

Tail latency, and the merge

Scatter-gather waits for the slowest shard, so p99 response time is roughly the p99 of any shard. Add shards and the chance one is slow rises — the classic result is that latency gets worse as you scale out, not better.

Mitigations: hedged requests (ask two replicas, take the first answer), a deadline after which a late shard is dropped and the degraded result served, and keeping shard counts modest.

The merge itself needs care. Scores must be comparable across shards — fine for cosine similarity, not fine for BM25, whose IDF depends on corpus statistics that differ per shard unless they are shared globally. That is a real bug in hand-rolled distributed lexical search.

The merge, and the scores that do not survive it

Merging local top-k lists assumes the scores are comparable across shards, and that assumption quietly fails for lexical retrieval.

BM25's idf term depends on corpus statistics — document frequency and average document length — which differ per shard. A term that is rare on shard 1 and common on shard 3 gets different weights, so a merged ranking is comparing numbers computed on different scales. The fix is global statistics: compute df and avgdl across the whole corpus and distribute them, which is an extra coordination step people discover only after the rankings look wrong.

Dense retrieval escapes this. Cosine similarity between a query and a document vector involves no corpus statistics at all, so shard-local scores are directly comparable. It is one of the underrated operational advantages of vector search.

Replication, failure and the degraded answer

Sharding splits data; replication copies it. They solve different problems, and conflating them is the most common design error here: if the index fits on one machine and you are simply serving too many queries, you want replicas and no sharding at all, and there is then nothing to merge.

Once shards exist, so does partial failure. Scatter-gather waits for everyone, so a single slow or dead shard degrades every query. The practical answers are a deadline — serve what returned in time and mark the result degraded — and hedged requests, asking two replicas and taking whichever answers first, which trades a few percent extra load for a much better tail.

Decide explicitly whether a degraded answer is acceptable. For search it usually is. For a RAG answer that will be presented as authoritative, silently dropping a shard means silently dropping evidence, and the user has no way to know.

When one machine is not enough

A vector index has to fit in memory to be fast. That gives a hard ceiling per machine:

memory ≈ vectors × dimensions × bytes × index overhead

For 100 million 768-dimensional vectors at 4 bytes each, that is 307GB before the HNSW graph's 1.5–2× overhead — so 450–600GB. Beyond a single machine.

Three separate reasons drive distribution, and they want different architectures:

Capacity. The index does not fit. Split it — sharding.

Throughput. Too many queries per second. Copy it — replication.

Availability. A machine failure must not take the system down. Replication again.

Most production systems need both: shards for capacity, replicas of each shard for throughput and resilience.

Sharding, and how to split

Random sharding distributes evenly and requires querying every shard for every request, then merging. Simple, predictable, and the cost grows with shard count.

Metadata sharding is usually the best choice when the data has a natural partition. Shard by tenant, and each query touches one shard. Shard by language, and a query in French touches the French shard. The saving is proportional to the selectivity.

Semantic sharding clusters similar vectors together so a query only needs the nearest clusters. Attractive in principle, and it risks recall loss when the answer sits in an unqueried shard, and it needs rebalancing as the distribution shifts.

Time-based sharding suits corpora where recency matters: query the recent shards first, and older ones only if needed. It also makes retention easy — drop an old shard.

Scatter-gather, and the detail that breaks it

With random or semantic sharding, a query fans out and the results are merged:

1. Send the query to all (or the selected) shards.
2. Each shard returns its local top k.
3. Merge, sort by score, take the global top k.

The correctness condition is easy to get wrong: each shard must return its own top k, not top k/n. If the global top 10 all happen to live in one shard, asking each of five shards for 2 results loses 8 of them.

The latency condition is different and equally important: the query takes as long as the slowest shard. With ten shards, the chance that at least one is having a slow moment is ten times higher than with one — so tail latency degrades as shard count grows. This is the standard fan-out problem, and the standard mitigations are hedged requests (send to a replica after a short delay and take whichever answers first) and a timeout with partial results.

Tail latency, and the scores that do not survive the merge

Scatter-gather looks simple until you measure it. Two things bite: a query is only as fast as the slowest shard it waited for, and the per-shard scores you merge are often not comparable. Both are arithmetic, and both are here.

import numpy as np

rng = np.random.default_rng(6)

print("SCATTER-GATHER: send the query to every shard, take the top k from")
print("each, merge, and return the overall top k. the correctness of that")
print("depends on one assumption -- that a document's score means the")
print("same thing on every shard.")
print()

print("PROBLEM 1 -- TAIL LATENCY. a query waits for ALL shards, so its")
print("latency is the MAXIMUM of theirs, not the mean:")
N = 20000
def shard_latency(n):
    # a realistic shape: mostly fast, occasionally slow
    base = rng.gamma(shape=4.0, scale=6.0, size=(N, n))
    return base

print("%-12s %14s %14s %14s %14s"
      % ("shards", "mean of max", "p50", "p95", "p99"))
for k in (1, 2, 4, 8, 16, 64):
    lat = shard_latency(k).max(axis=1)
    print("%-12d %14.1f %14.1f %14.1f %14.1f"
          % (k, lat.mean(), np.percentile(lat, 50),
             np.percentile(lat, 95), np.percentile(lat, 99)))
one = shard_latency(1).max(axis=1)
many = shard_latency(64).max(axis=1)
print("   each shard is IDENTICAL in every row -- same distribution, same")
print("   hardware. only the count changed, and the p50 rose from %.1f to"
      % np.percentile(one, 50))
print("   %.1f ms." % np.percentile(many, 50))
print("   the reason is that a rare slow response stops being rare once")
print("   you need all of them: if a single shard exceeds its p99 one time")
print("   in a hundred, then with 64 shards the chance that NONE of them")
print("   does is 0.99^64 = %.2f, so about %.0f%% of queries hit a p99"
      % (0.99 ** 64, 100 * (1 - 0.99 ** 64)))
print("   response from somewhere.")
print("   this is why adding shards to make each one faster can make the")
print("   SERVICE slower, and why it is the p99 of a shard that matters")
print("   rather than its mean.")
print()
print("   the standard mitigations, and what each gives up:")
print("%-30s %s" % ("hedged requests", "duplicate work for a shorter tail"))
print("%-30s %s" % ("a deadline, partial results", "recall, silently"))
print("%-30s %s" % ("fewer, larger shards", "per-shard latency"))
print("   the middle one deserves care. returning results from 62 of 64")
print("   shards looks like a success and is a %.1f%% loss of the corpus"
      % (100 * 2 / 64.0))
print("   on that query, with nothing in the response to say so.")
print()

print("PROBLEM 2 -- THE MERGE. distance scores merge fine, because a")
print("cosine similarity means the same thing everywhere. these do not:")
print()
print("   BM25 scores depend on CORPUS STATISTICS -- specifically idf,")
print("   which is computed from the documents a shard happens to hold:")
SHARDS = {
    "shard A (10,000 docs)": (10000, 40),
    "shard B (10,000 docs)": (10000, 3000),
}
print("%-28s %12s %12s %14s"
      % ("", "N on shard", "df on shard", "idf"))
for name, (n, df) in SHARDS.items():
    print("%-28s %12s %12s %14.4f"
          % (name, "{:,}".format(n), "{:,}".format(df),
             np.log((n - df + 0.5) / (df + 0.5) + 1.0)))
gn = sum(n for n, _ in SHARDS.values())
gdf = sum(df for _, df in SHARDS.values())
print("%-28s %12s %12s %14.4f"
      % ("the true global values", "{:,}".format(gn), "{:,}".format(gdf),
         np.log((gn - gdf + 0.5) / (gdf + 0.5) + 1.0)))
print("   the same term is 'rare' on shard A and 'common' on shard B, so")
print("   a document on A gets a much larger score for containing it --")
print("   not because it is a better match, but because of who its")
print("   neighbours are.")
print("   merge those scores and the ranking is decided by shard")
print("   assignment. that is why the article says to shard RANDOMLY: it")
print("   does not fix the problem, it makes the per-shard statistics")
print("   converge to the global ones, so the error is small and unbiased")
print("   instead of large and systematic.")
print("   sharding by date or by customer -- the two most natural choices")
print("   -- is exactly what you must not do.")
print()

print("HOW WRONG DOES IT GET? one document, scored on both shards:")
TF = 3.0
for name, (n, df) in SHARDS.items():
    idf = np.log((n - df + 0.5) / (df + 0.5) + 1.0)
    print("   %-28s score %.4f" % (name, idf * TF))
gidf = np.log((gn - gdf + 0.5) / (gdf + 0.5) + 1.0)
print("   %-28s score %.4f" % ("what it should have been", gidf * TF))
a = np.log((10000 - 40 + 0.5) / (40 + 0.5) + 1.0) * TF
b = np.log((10000 - 3000 + 0.5) / (3000 + 0.5) + 1.0) * TF
print("   an identical document scores %.1fx higher on one shard than the"
      % (a / b))
print("   other. no merge that only sees the numbers can undo that.")
print("   the fixes are to distribute global statistics to every shard, or")
print("   to re-score the merged candidates centrally. both cost a round")
print("   trip, which is why many systems simply accept the error and")
print("   shard randomly so that it stays small.")
print()

print("REPLICAS ARE A DIFFERENT AXIS, and conflating them is common:")
print("%-19s %-30s %s" % ("", "shards", "replicas"))
for row in (("splits", "the DATA", "the TRAFFIC"),
            ("more of them", "more capacity", "more throughput"),
            ("one fails", "you lose that data", "you lose nothing"),
            ("query goes to", "ALL of them", "ONE of them"),
            ("effect on latency", "raises the tail", "no effect")):
    print("%-19s %-30s %s" % row)
print("   a query touches every shard and exactly one replica of each, so")
print("   replicas do nothing for the tail-latency problem above -- they")
print("   protect availability and add throughput, and a slow query stays")
print("   slow.")
print("   the arithmetic to size a cluster follows from those two rows:")
CORPUS, PER_SHARD, QPS, PER_REPLICA = 400e6, 25e6, 4000, 900
sh = int(np.ceil(CORPUS / PER_SHARD))
rp = int(np.ceil(QPS / PER_REPLICA))
print("      %s vectors at %s per shard  ->  %d shards"
      % ("{:,.0f}".format(CORPUS), "{:,.0f}".format(PER_SHARD), sh))
print("      %s QPS at %s per replica    ->  %d replicas"
      % ("{:,}".format(QPS), "{:,}".format(PER_REPLICA), rp))
print("      total machines = %d x %d = %d" % (sh, rp, sh * rp))
print("   and every one of those %d shards is another chance to be the"
      % sh)
print("   slow one on any given query.")

Things to try

1. Switch to semantic sharding. Recall collapses: every relevant document is on one shard, whose local top-k discards the rest.
2. With semantic sharding, raise per-shard k until recall recovers. You have paid back the bandwidth the topic routing was meant to save.
3. Raise shards to 40 with random sharding. Recall is fine and p99 latency climbs, because scatter-gather waits for whichever shard was slow.

What to remember

Scatter-gather sends the query to every shard, takes a local top-k from each and merges. Shard randomly rather than by topic, or the relevant documents concentrate in one shard and its local k throws most of them away. Latency becomes the slowest shard's, not the average, so p99 gets worse with every shard added. Shards solve data that will not fit; replicas solve query volume.

Replication

Copying each shard serves more queries and survives failures.

That last row matters for retrieval specifically. A document indexed a moment ago may be present on one replica and not another, so two identical queries can return different results. For most retrieval that is acceptable; where it is not — a user searching for something they just uploaded — the usual fix is to route that user's reads to the replica that accepted the write, or to accept the lag and say so in the interface.

The routing choice is between load balancing (any replica, best throughput) and consistent routing (the same replica for a session, better cache behaviour and freshness). Consistent routing also improves cache hit rates, which is a meaningful secondary benefit.

The metadata and filtering problem

Distribution interacts badly with filtering unless it is planned for.

A selective filter across random shards means every shard is queried and most return nothing. The work is done and thrown away. Sharding by the filtered attribute instead makes the filter a routing decision.

Permissions must be enforced per shard, inside each shard's search, not merged afterwards. Filtering after the gather step means unauthorised content has already crossed the boundary and has already occupied result slots.

Global statistics break. BM25's IDF term depends on how many documents in the corpus contain a term, and each shard only knows its own counts. With random sharding the local approximation is usually close enough; with metadata sharding it can be badly skewed, and correcting it requires distributing global term statistics.

That last point is a real and under-discussed source of ranking inconsistency in sharded hybrid search.

Operational realities

Rebalancing. Adding a shard means moving data. Consistent hashing limits how much moves; semantic sharding may require re-clustering the whole index.

Rebuilds. Changing the embedding model invalidates everything. Build into a new set of shards and switch atomically rather than migrating in place.

Monitoring per shard. Aggregate latency hides a single slow shard. Track per-shard latency, size and error rate.

Deletions. Tombstones accumulate per shard, so schedule compactions.

Cost. Memory-resident indexes are the expensive part. Reducing dimensions — 1,536 to 384 — is a fourfold saving and often a small accuracy cost, and it is usually a better first move than adding machines.

That last point is worth trying before distributing at all: quantisation and dimension reduction frequently keep a corpus on one machine, which removes every problem above.

Questions people ask

When do I need to shard? When the index no longer fits in one machine's memory after quantisation and dimension reduction. Try those first.

How many shards? As few as fit. Each additional shard adds fan-out cost and tail latency.

Should each shard return k or k/n? k. Returning k/n loses results when relevance is concentrated.

Does sharding hurt recall? Random sharding with per-shard top-k does not. Semantic sharding can, if the answer is in an unqueried shard.

How do I handle multi-tenancy? Shard by tenant where volumes allow — it gives natural isolation and cheap filtering. Very many small tenants are better served by one index with a tenant field.

What about BM25 across shards? IDF becomes shard-local. Acceptable with random sharding; distribute global term statistics if the skew matters.

Recap in one screen

• Shard for capacity, replicate for throughput and availability — most systems need both.
• Shard by metadata where a natural partition exists, so a query touches one shard instead of all of them.
• In scatter-gather, each shard must return its own top k, and latency is set by the slowest shard.
• Enforce permission filters inside each shard's search, never after the merge.
• Try quantisation and fewer dimensions first — staying on one machine avoids every problem here.