Information Retrieval · Search Operations

Shard Routing and Replica Consistency in Distributed Search

How a search index is split across many machines, how a query fans out to shards and merges their results, and how replicas stay consistent with their primary shards.

A single machine can index only as much data as fits in its memory and disk, and can serve only as many queries as its CPU and I/O allow. Distributed search breaks the index into shards, each an independent index on one machine, and replicates each shard for throughput and fault tolerance. The mechanisms that make this work, routing, scatter-gather, and replica synchronization, are where the latency and consistency guarantees of a search system are ultimately decided.

Shards as independent indexes

A shard is a complete, self-contained inverted index over a subset of the documents. The assignment of a document to a shard is decided by a routing key, typically the document id hashed or ranged against the shard count, so each document lives on exactly one primary shard. Because each shard is an independent index, it runs the entire local stack described earlier in this cluster: its own segments and merges, its own refresh interval, and its own write-ahead log.

The consequence is that the global index has no global statistics. A term’s document frequency is computed per shard, not per corpus, and scores that depend on corpus-wide statistics are approximate, because each shard normalizes against its own slice. This is the central trade of sharding: it buys horizontal scalability at the cost of making every global statistic local.

The granularity of the split matters more than the count. If the routing key is coarse, such as a document’s language or tenant, then queries that span many keys touch many shards and pay the fan-out, while queries that span one key touch one shard and are cheap. If the key is a uniform hash of the id, then every query touches every shard equally, and the shard count becomes a pure parallelism lever. The choice is between data-locality routing, which makes some queries cheap at the cost of others, and uniform routing, which makes every query pay the full fan-out but none more than another.

Scatter-gather query execution

A query is sent to every shard in parallel, a fan-out called scatter-gather, and each shard returns its top K documents. The coordinator merges the per-shard results into a single global top K. The merge is a k-way selection over already-sorted lists, so the coordinator does not need to re-score; it needs only the rank of each returned document’s score, which each shard supplies.

The latency of the query is the latency of the slowest shard, plus the merge. This is why a distributed system’s tail latency is dominated by its stragglers: one shard with a cold cache or a merge in progress delays the entire query, because the coordinator must wait for every shard before it can return a correct global top K. The standard mitigations are speculative retries to a replica when a shard is slow, and giving each shard a larger K than needed so that merging tolerates some ranking error at the tail.

The larger-K trick addresses a specific flaw in naive top-K merge. If each shard returns exactly the global K, then the true global top K requires every shard to return its own top K correctly, because a document that is globally rank five might be rank one on its shard. Returning more than K from each shard, a common value being two to ten times K, gives the coordinator slack to absorb the case where one shard’s ranking is slightly off, at the cost of sending more data over the network and doing a larger merge.

Replicas and their consistency

Each primary shard has one or more replicas, full copies that serve read traffic and stand in if the primary fails. Writes go to the primary, which then forwards them to the replicas. The consistency question is how many replicas must acknowledge a write before it is considered successful, and how a read decides whether to go to the primary or a replica.

The write path offers a quorum choice: a write that waits for the primary and all replicas is strongly consistent but slow; a write that acknowledges after the primary alone is fast but leaves replicas briefly stale. Between these, a write that waits for a majority of the shard’s copies trades some latency for the guarantee that at least one acknowledged copy survives any single failure. Reads face the same choice in reverse: reading from the primary sees the freshest data, while reading from a replica can return a document that has since been updated or deleted, because replication is asynchronous and a replica lags its primary by the propagation delay.

The propagation delay is bounded by the primary’s write rate and the network. A replica catches up by replaying the primary’s write stream, and if the primary is under heavy ingest, the replica’s replay queue grows, stretching the lag from milliseconds to seconds or longer. The lag is directly observable as a read-your-writes violation: a client writes a document, reads it back from a replica before the replica has received it, and sees it missing. Systems that need read-your-writes therefore route the write and the subsequent read to the same node, or force a refresh-and-replicate on the write path, accepting the latency to close the window.

Primary election and the split-brain hazard

When a primary fails, a replica is promoted to primary. The hazard is split brain: two nodes both believing they are the primary for the same shard, accepting writes independently, and diverging. The standard defense is a quorum-based election where a majority of the shard’s nodes must agree on the new primary, which guarantees at most one primary per shard as long as the cluster can form a majority. The same quorum logic that governs write acknowledgment also governs who may be promoted, so the two mechanisms share one invariant: no write is acknowledged unless a majority of copies has it, and no node becomes primary without a majority’s blessing.

The cost of this guarantee is availability. A cluster that cannot form a majority, because too many nodes are down or partitioned, must refuse writes and possibly reads rather than risk divergence. This is the consistency-availability trade made explicit: the system chooses to be correct over available, and the quorum is the number that encodes how much partition it can tolerate before making that choice.

Failure modes at the boundary

Distributed search fails in characteristic ways. A hot shard, one that receives a disproportionate share of queries or documents, becomes the system’s bottleneck even when the cluster as a whole is idle, because sharding on the hash of the id does not spread query load, only data. A replica that falls far behind its primary serves stale results indefinitely, because there is no global mechanism forcing it to catch up; it catches up only as fast as the primary can forward its writes. And a routing key that is skewed, such as a hash that maps many documents to one shard, produces a permanently imbalanced index that no amount of hardware fixes.

The shard and replica layer is where a search system stops being a single well-tuned process and becomes a distributed protocol. The inverted index, the merge, the refresh, and the log all remain per-shard concerns; what the distribution layer adds is routing, fan-out, and the quorum discipline that keeps every copy of the index from silently diverging from the others.

References

  1. Dynamo: Amazon's Highly Available Key-value Store — DeCandia et al., SOSP 2007
  2. Elasticsearch documentation — Shards and replicas — Elastic
  3. Introduction to Information Retrieval — Manning, Raghavan & Schütze, Cambridge University Press