Sharding & the Distributed Index

Overview

Query Serving assumes the index is already spread across shards and glosses over how that split happens. This page is that decision on its own: how the corpus gets partitioned, how replicas keep a shard failure from becoming a data-loss or availability event, and what changes when a shard is added. Worth treating separately, since interviewers frequently push on it once the scatter-gather pattern has been introduced.

API Design

Sharding has no client-facing endpoint of its own — it’s the internal routing layer the broker (from Query Serving) depends on:

ShardRouter.shardsFor(query) -> [shard_1, shard_2, ..., shard_N]
ShardRouter.replicasFor(shard_id) -> [replica_a, replica_b]

shardsFor is trivial under document partitioning (every query goes to every shard, since any shard might hold a matching document) — the more interesting question this page answers is why that’s the right default over the alternative.

Database Schema

shard_map        -- shard_id -> assigned doc_id hash range
replica_map       -- shard_id -> [replica node addresses]
shard_health      -- shard_id -> last_heartbeat_ts, status

The shard_health table (or equivalent in a coordination service like ZooKeeper/etcd) is what the broker consults to decide whether a shard is healthy enough to include in scatter-gather, or should be skipped in favor of returning a partial result quickly.

Basic Approach — Single Index Node

How it works

The entire inverted index lives on one machine, serving all queries directly.

Client ──▶ Broker (pass-through) ──▶ Single Index Node

Tradeoffs

• Pro: Nothing to route, merge, or rebalance — as simple as it gets.
• Con: Caps out the moment the index no longer fits in one machine’s memory, or query volume exceeds what one machine can serve.
• Con: A single point of failure — that one machine going down takes search availability with it entirely.

Scaled Approach — Document-Partitioned Sharding

How it works

Split the document corpus across N shards by hashing doc_id (or a similar key), so each shard holds a complete, independent inverted index over its own subset of documents. A query has to be sent to every shard, since any shard might contain a matching document — this is exactly the scatter-gather pattern from Query Serving.

Doc Corpus ──▶ hash(doc_id) % N ──▶ [Shard 1 | Shard 2 | ... | Shard N]
Query ──▶ Broker ──▶ scatter to all N shards ──▶ gather

Tradeoffs

• Pro: Both index size and query throughput now scale with shard count instead of being capped by one machine.
• Pro: Straightforward to reason about — each shard is a smaller version of the same single-node index from the basic tier.
• Con: Every query touches every shard, so total query cost scales with shard count even though only a few shards may actually hold a relevant document.
• Con: A hot document (one that legitimately matches an enormous number of queries) still lives on exactly one shard, which can become a hotspot.

The alternative — term-partitioned sharding, where each shard owns a subset of terms rather than documents — is worth naming even though it’s not the default: it lets a query for a rare term touch only one shard, but common terms create severe load imbalance across shards and merging a single term’s postings list across shards is awkward. Document partitioning’s more even load distribution is why it’s the standard choice in practice.

Advanced Approach — Replication, Rebalancing, and Failure Handling

How it works

Each shard is replicated across multiple nodes (e.g., 3 copies), so the broker can route a query to any healthy replica and a single node failure never takes a shard offline. Shard assignment uses consistent hashing instead of plain hash(doc_id) % N, so adding a new shard only moves a fraction of documents instead of reshuffling nearly everything. A shard health service tracks replica liveness so the broker can skip a dead replica and, if all replicas for one shard are unreachable, return a partial: true result from the remaining shards rather than failing the whole query.

Shard 1 ──▶ [Replica A | Replica B | Replica C]
Broker ──▶ picks any healthy replica per shard ──▶ scatter ──▶ gather
Shard Health Service ──▶ marks unhealthy replicas, triggers rebalancing

Tradeoffs

• Pro: Replication turns “one machine dies” from a data-loss/availability event into a non-event — the broker just routes around it.
• Pro: Consistent hashing bounds how much data moves when the cluster is resized, which matters enormously once shard counts are large and rebalancing happens routinely.
• Con: Every replica needs its own copy of the index kept in sync, which multiplies both storage cost and the write-side work needed to keep them consistent as documents get re-indexed.
• Con: More moving parts — a shard health/coordination service is now required infrastructure, not an afterthought, and it needs its own availability story.

Tech Choices

• Document partitioning — the default sharding strategy; even load distribution and simple query routing (every query touches every shard).
• Consistent hashing — bounds data movement when shards are added or removed, versus naive modulo hashing which reshuffles almost everything.
• Replication (N=3 typical) — the standard redundancy factor balancing availability against storage/write cost.
• A coordination service (ZooKeeper/etcd) — tracks shard-to-replica mapping and health, so the broker always routes to a live node.

How to Vocalize This in an Interview

Motivate sharding the same way it’s motivated everywhere else in this guide: name the limit of a single machine first (“the index doesn’t fit, or one node can’t serve this QPS”) before introducing shards as the fix. When asked “document or term partitioning,” name both and explain document partitioning’s load-balancing advantage explicitly rather than asserting it’s just the standard choice — that’s the answer that sounds reasoned instead of memorized.