Replication and sharding are the two standard answers to "one database is not enough", and the most common interview mistake is treating them as the same thing. They solve different problems. Replication keeps copies of the same data on several nodes, which buys availability, durability and read throughput. Sharding splits the data so each node owns a different slice, which buys capacity and write throughput. Interviewers probe this topic because the follow-up questions expose real engineering judgement: what does an asynchronous replica do to a user who just wrote something, how do you pick a shard key that does not create a hot shard, what happens to your data when you add a node, and how do you search a vector index that no longer fits on one machine. For a Backend or AI engineer these are daily concerns, since vector databases, feature stores and chat-history stores all combine both techniques. The mechanics are covered in Replication (DBMS) and Partitioning & Sharding (DBMS); this page adds the AI-system angle and does not repeat those pages.
Replication means keeping several copies of the same data on different machines instead of just one. If one machine goes down, catches fire, or simply gets too busy answering reads, the others still have the exact same data and can keep serving requests. It's the database version of not keeping your only copy of an important document in a single drawer.
Sharding means splitting the data itself into separate pieces and putting each piece on a different machine, so no single machine has to hold everything or handle every write. A company with a billion users might put users A through M on one machine and N through Z on another, rather than cramming all billion onto one overloaded server.
Think of a large hospital's medical records. Replication is like photocopying every patient's chart and keeping copies at the front desk and in a backup archive, so a fire in one room doesn't destroy the only record. Sharding is like organizing charts by patient last name across several filing cabinets, A through F in one cabinet, G through M in another, so no single cabinet becomes impossible to search through or overflow with paper.
REPLICATION (same data, many copies) SHARDING (different data, many nodes)
writes writes for key K -> shard = f(K)
| |
[primary] ---log---> [replica 1] +------+------+------+
| ---log---> [replica 2] [shard 0] [shard 1] [shard 2]
reads can go to any replica each holds ~1/3 of the keys and ~1/3 of the writes
Every node holds ALL the data. No node holds all the data.
Scales reads, gives failover. Scales storage, writes and (usually) reads.
Does NOT scale writes or total size. Adds routing, cross-shard queries, resharding pain.
Typical production shape: S shards x R replicas each -> S x R nodes
e.g. 6 shards x 3 replicas = 18 nodes; each shard has its own primary (or leader) and two followersThe distinction becomes concrete when you ask which resource is the bottleneck. If reads are the problem and the dataset fits on one machine, add replicas: every replica serves a share of the read traffic. If writes saturate the primary, or the dataset no longer fits in RAM or on disk, more replicas do nothing, because every replica still applies every write and stores every byte. Only splitting the data helps. Replication and sharding compose: shards give you scale, and each shard is replicated for availability, so losing one machine loses neither data nor a slice of the keyspace. Also keep the vocabulary straight: partitioning usually means splitting a table inside one database instance (or one logical database), sharding means spreading partitions across separate machines.
Async replication (the common default for read replicas)
t0 client --write--> PRIMARY (commit, ack to client immediately)
t1 client --read---> REPLICA (log entry has not arrived yet) -> returns the OLD value
t2 PRIMARY ---log---> REPLICA (applied) -> a later read is correct
lag = t2 - t0 (milliseconds normally; seconds or minutes under load, big transactions, or network trouble)
Ack policy durability on primary loss write latency if a replica is down
--------------------------------------------------------------------------------------------------------------------
async recent acked writes can be lost lowest no impact
semi-sync (1 replica must ack) survives one node loss + one round trip falls back or blocks, config-dependent
sync (all replicas must ack) strongest + slowest replica writes can stallAsynchronous replication makes reads scale but introduces replication lag, and lag leaks into product behaviour as three classic anomalies: a user does not see what they just wrote (read-your-writes), a user sees newer data and then older data on a refresh (monotonic reads), and a reply shows up before the message it answers (consistent prefix). The fixes are session-level rather than global: read from the primary for a short window after a write, or for data the user is known to have just edited; carry a version or log-position token from the write and only accept a replica that has reached it (falling back to the primary otherwise); or pin a session to one replica, which fixes monotonic reads but not read-your-writes if that replica is the slow one. The CAP and consistency page has a runnable simulation of exactly these three policies.
Durability is the other half. With async replication, a primary crash can lose the last few acknowledged commits, and a promoted replica may be missing them, so failover has a data-loss window (RPO greater than zero) unless you pay for synchronous or semi-synchronous acknowledgement. The trade-off is direct: each additional synchronous acknowledgement adds a network round trip to every write and ties write availability to replica health. Systems built on consensus (Raft or Paxos groups, see Consensus & Coordination) make the majority-acknowledged version the default. For AI workloads this matters most for the small, valuable stores (metadata, ledgers, model registry) and matters little for rebuildable data such as a vector index that can be re-derived from the source documents.
Candidate keys for an AI platform, and what each one does:
key what is co-located good for risk
----------------------------------------------------------------------------------------------------------------
user_id one user's data on one shard chat history, per-user features one heavy user = one hot shard
tenant_id one customer's data together multi-tenant RAG, isolation giant tenant dominates a shard
doc_id (hash) chunks spread evenly vector index, bulk ingestion every search must fan out
conversation_id one conversation together ordered chat log, single writer needs conversation lookups by user
created_at / time time-ordered ranges range scans, TTL/retention all new writes hit the newest shard
Rule of thumb: the key should have high cardinality, spread load evenly for the ACTUAL access pattern,
and make the most common query touch one shard.
Hot key problem (illustrative): one document or one tenant gets 30% of all traffic.
hashing spreads MANY keys well, but cannot split ONE key: that key's shard runs hot no matter how good the hash.
Mitigations: replicate the hot key across shards/replicas, cache it in front, split it (key#0..key#7 salted),
or give large tenants a dedicated shard.A shard key is a bet about the future access pattern, and it is expensive to change. Sharding by tenant_id makes each tenant's queries single-shard and makes isolation and per-tenant deletion easy, but a tenant that is 100 times bigger than the median overloads its shard while the others sit idle. Sharding by a hash of doc_id spreads storage evenly but forces every similarity search to visit every shard. Time-based keys are the classic mistake for write-heavy data: all new writes land on the newest shard while old shards idle, so use time only for partitioning within a shard (retention, cold storage), not as the distribution key. Uneven load has three separate causes worth naming in an interview: skewed data volume, skewed traffic (hot keys), and skewed queries (some queries fan out, others do not).
Naive: shard = hash(key) mod N. Grow from N=4 to N=5 -> almost every key changes shard.
Ring: place shards (with many virtual nodes) on a hash ring; a key belongs to the next shard clockwise.
Adding a shard takes a slice from the existing shards, so only ~1/(N+1) of keys move.
Slots: fix a large number of logical slots up front (e.g. Redis Cluster uses 16384 hash slots);
key -> slot is permanent, and slot -> node is a table you edit. Resharding = moving whole slots.
Growing 4 -> 5 shards, 20,000 keys (measured by the program in "Try It Yourself"):
mod-N ~80% of keys must move (only keys where h mod 4 == h mod 5 stay: ~20%)
consistent hashing ~17% move, all to the NEW shard (the theoretical minimum is 1/5 = 20%; the new shard got a bit less than its fair share here)
Cost of a move: copy the data, keep serving reads and writes while it is copied, cut over atomically,
then delete the old copy. For a vector index this also means rebuilding or transferring graph segments.With mod-N, adding one node changes the modulus for every key, so nearly the whole dataset is remapped: a cache would suffer a near-total miss storm, and a database would have to shuffle almost every row. Consistent hashing bounds the movement to the share the new node takes over, and virtual nodes (many points per shard on the ring) keep the load even; the measurements below show a single ring point per shard leaving one shard with over three times the average load, while around 200 points per shard is close to even. The practical alternative used by many systems is a fixed number of logical partitions or slots, much larger than the node count, with a mapping table from slot to node; rebalancing then means moving whole slots, and the key-to-slot function never changes. Kafka is a good related example: increasing a topic's partition count changes the key-to-partition mapping for keyed data and breaks per-key ordering guarantees across the change, and partitions cannot be reduced, so partition counts are chosen with headroom. Elasticsearch-style systems similarly fix the primary shard count at index creation, and changing it generally means splitting, shrinking or reindexing, so the shard count has to be planned.
Live resharding is a procedure, not a switch: create the new placement, copy existing data in the background, capture concurrent changes (a write-ahead log, dual writes, or change data capture, see Change Data Capture), verify counts and checksums, flip routing, and only then retire the old copy. The failure modes are inconsistent routing tables across routers, double-counting during the transition, and copy traffic starving production traffic, so throttle it.
Why a vector index shards (ILLUSTRATIVE arithmetic, assumptions stated):
100,000,000 vectors x 768 dims x 4 bytes (float32) = 307.2 GB of raw vectors
ANN graph/quantizer overhead depends on the index type (HNSW links, PQ codebooks); assume +30% -> ~400 GB
if a node gives ~48 GB of RAM to the index -> 400 / 48 = 8.3 -> at least 9 shards
3 replicas per shard for availability/QPS -> at least 27 nodes. Replicas did nothing for the capacity problem.
(int8 scalar quantization would cut the raw vectors ~4x, at some recall cost; measure it on your data)
Scatter-gather search over doc-partitioned shards:
query -> router --+--> shard 0: local top-k --+
+--> shard 1: local top-k --+--> merge, keep global top-k --> rerank / prompt
+--> shard 2: local top-k --+
latency of the query = latency of the SLOWEST shard (tail latency amplification)
Tenant-partitioned alternative: route the query to the one shard that owns the tenant -> no fan-out,
but a big tenant can overload its shard and a small one wastes a whole shard's capacity.Two things go wrong when a vector index is sharded by document, and both are measurable. First, each shard must return at least k candidates for a global top-k to be exact; returning fewer per shard (to save network) silently lowers recall, and with approximate search each shard's own recall compounds into the merged result, so per-shard settings (efSearch, nprobe, or their equivalents, depending on the engine) need tuning against a recall target rather than left at defaults. Second, tail latency grows with fan-out: if each shard is slow 2% of the time, the chance that at least one of 64 shards is slow on a given query is 1 - 0.9864, about 73%, so a wide fan-out makes rare slowness common. Standard mitigations are hedged requests (send to a second replica after a short delay), per-shard timeouts with partial results, fewer and bigger shards, and replicas to absorb load. See Vector Databases and Approximate Nearest Neighbor Search for the index side.
Feature stores have the opposite access pattern: most lookups are point reads by entity id (a user, an item), so sharding by entity id makes a single-entity lookup single-shard and very fast, while a batch lookup of 500 items fans out. Popular entities (a viral product, a celebrity account) create hot keys, which are usually handled by an in-process or edge cache and by replicating the hottest keys rather than by re-sharding. Sharding also has to respect freshness: streaming updates for one entity should go through one partition to keep per-entity ordering. See Feature Stores for the pipeline side and Load Balancing for request-level distribution.
The first program compares how many keys move when a fourth-to-fifth shard is added under mod-N versus a consistent-hash ring, shows how the number of virtual nodes affects balance, and demonstrates that a single hot key skews load however good the hash is. It uses MD5 for a stable hash because Python's built-in hash() is salted per process. The hot-key traffic figures are constructed for the demo, not measurements of a real system.
import hashlib, bisect
from collections import Counter
def h(s): # stable 64-bit hash (Python's built-in hash() is salted per process)
return int.from_bytes(hashlib.md5(s.encode()).digest()[:8], "big")
class Ring:
def __init__(self, nodes, vnodes=100):
self.vnodes = vnodes
self.points = [] # sorted list of (hash, node)
for n in nodes: self.add(n)
def add(self, node):
for i in range(self.vnodes):
bisect.insort(self.points, (h(f"{node}#{i}"), node))
def lookup(self, key):
i = bisect.bisect(self.points, (h(key), "")) % len(self.points)
return self.points[i][1]
keys = [f"doc:{i}" for i in range(20000)]
nodes4 = [f"shard{i}" for i in range(4)]
# --- mod-N: 4 shards -> 5 shards
before = {k: h(k) % 4 for k in keys}
after = {k: h(k) % 5 for k in keys}
moved_mod = sum(before[k] != after[k] for k in keys)
# --- consistent hashing: 4 shards -> 5 shards
r = Ring(nodes4)
b = {k: r.lookup(k) for k in keys}
r.add("shard4")
a = {k: r.lookup(k) for k in keys}
moved_ring = sum(b[k] != a[k] for k in keys)
to_new = sum(1 for k in keys if b[k] != a[k] and a[k] == "shard4")
print(f"keys: {len(keys)} ideal minimum to move when going 4 -> 5 shards: {len(keys)//5} (20.0%)")
print(f"mod-N moved: {moved_mod:5d} ({100*moved_mod/len(keys):.1f}%)")
print(f"consistent hashing moved: {moved_ring:5d} ({100*moved_ring/len(keys):.1f}%) all to shard4: {to_new == moved_ring}")
# --- balance: vnodes matter
for vn in (1, 10, 200):
rr = Ring(nodes4, vnodes=vn)
c = Counter(rr.lookup(k) for k in keys)
loads = sorted(c.values())
print(f"vnodes={vn:3d} keys per shard: {[c[n] for n in nodes4]} max/avg = {max(loads)/(len(keys)/4):.2f}")
# --- hot key: one celebrity key is 30% of traffic, hashing cannot spread it
traffic = Counter()
r2 = Ring(nodes4, vnodes=200)
for i, k in enumerate(keys[:10000]):
traffic[r2.lookup(k)] += 1
traffic[r2.lookup("doc:hot")] += 4300 # one key gets 30% of all requests
tot = sum(traffic.values())
print("with a hot key :", {n: f"{100*traffic[n]/tot:.0f}%" for n in nodes4}, "->", "hot key lands on", r2.lookup("doc:hot"))
# Real output from python3:
# keys: 20000 ideal minimum to move when going 4 -> 5 shards: 4000 (20.0%)
# mod-N moved: 15919 (79.6%)
# consistent hashing moved: 3356 (16.8%) all to shard4: True
# vnodes= 1 keys per shard: [525, 16877, 996, 1602] max/avg = 3.38
# vnodes= 10 keys per shard: [2885, 7923, 3772, 5420] max/avg = 1.58
# vnodes=200 keys per shard: [5356, 4522, 4847, 5275] max/avg = 1.07
# with a hot key : {'shard0': '19%', 'shard1': '46%', 'shard2': '17%', 'shard3': '19%'} -> hot key lands on shard1The second program shards a small synthetic vector corpus by id, runs scatter-gather nearest-neighbour search with exact per-shard search, and shows how the number of candidates each shard returns affects the recall of the merged answer. It then computes how the chance of hitting at least one slow shard grows with fan-out; the 5 ms and 50 ms shard times and the 2% slow rate are illustrative assumptions, and the theory column is 1 - 0.98n.
import random, heapq, math
rnd = random.Random(11)
DIM, N, SHARDS, K = 8, 4000, 4, 10
vecs = [[rnd.gauss(0, 1) for _ in range(DIM)] for _ in range(N)]
def dist(a, b): return math.dist(a, b)
# hash-partition by id: each shard holds a slice of the corpus
shards = [[] for _ in range(SHARDS)]
for i, v in enumerate(vecs): shards[i % SHARDS].append((i, v))
def shard_topk(s, q, k): return heapq.nsmallest(k, ((dist(q, v), i) for i, v in shards[s]))
def global_exact(q, k): return heapq.nsmallest(k, ((dist(q, v), i) for i, v in enumerate(vecs)))
queries = [[rnd.gauss(0, 1) for _ in range(DIM)] for _ in range(200)]
for per_shard_k in (10, 3, 1):
hit = 0
for q in queries:
truth = {i for _, i in global_exact(q, K)}
merged = heapq.nsmallest(K, (r for s in range(SHARDS) for r in shard_topk(s, q, per_shard_k)))
hit += len(truth & {i for _, i in merged})
print(f"each shard returns top-{per_shard_k:2d} -> merged top-{K} recall vs exact: {hit/(K*len(queries)):.3f}")
# tail latency of fan-out: query time = max over shards. Each shard: 5 ms typical, 2% of calls take 50 ms (illustrative).
def shard_ms(): return 50.0 if rnd.random() < 0.02 else 5.0
for n in (1, 4, 16, 64):
slow = sum(max(shard_ms() for _ in range(n)) > 5.0 for _ in range(20000)) / 20000
print(f"fan-out to {n:2d} shards: fraction of queries that hit a slow shard = {slow:.3f} (theory {1-0.98**n:.3f})")
# Real output from python3:
# each shard returns top-10 -> merged top-10 recall vs exact: 1.000
# each shard returns top- 3 -> merged top-10 recall vs exact: 0.881
# each shard returns top- 1 -> merged top-10 recall vs exact: 0.378
# fan-out to 1 shards: fraction of queries that hit a slow shard = 0.021 (theory 0.020)
# fan-out to 4 shards: fraction of queries that hit a slow shard = 0.076 (theory 0.078)
# fan-out to 16 shards: fraction of queries that hit a slow shard = 0.273 (theory 0.276)
# fan-out to 64 shards: fraction of queries that hit a slow shard = 0.723 (theory 0.726)Both are toy models: no network, no rebalancing procedure, brute-force search instead of an ANN index. A real deployment would express the same ideas through configuration such as a shard count and replica count on the index or cluster (check your engine's documentation for what can be changed after creation), which is why the planning arithmetic above matters more than any particular flag.
Q: What is the difference between replication and sharding, and when do you need each?
A: Replication copies the same data to several nodes, so it scales reads and gives failover and durability, but each node still stores everything and applies every write. Sharding divides the data so each node holds only a slice, which scales storage and writes at the price of routing, cross-shard queries and resharding. Use replicas when reads are the bottleneck or you need availability; add shards when a single node cannot hold the data or absorb the write rate; in production you normally do both.
Q: Why is hash(key) mod N a poor sharding function for a system that will grow?
A: Changing N changes the result for most keys; going from 4 to 5 shards remaps about 80% of them, versus a theoretical minimum of about 20%. That means a huge data migration, or for a cache a miss storm that hits the database. Consistent hashing with virtual nodes, or a fixed set of logical slots that are reassigned as whole units, limits movement to what the new node takes over.
Q: How would you shard a multi-tenant RAG service's vector store?
A: Start from the access pattern: nearly every query is scoped to one tenant, so tenant-based routing makes searches single-shard, simplifies isolation and makes per-tenant deletion straightforward. The risk is skew, so I would put large tenants on dedicated shards or sub-shard them by document hash (accepting fan-out for those tenants only), and pack many small tenants together. If tenants are all similar and queries span the whole corpus, hashing by document id spreads load better but every search fans out.
Q: Users report that a document they just uploaded is not found immediately. Where would you look?
A: There are several places a lag can hide. The ingestion pipeline may still be chunking or embedding; the vector engine may batch new vectors and only make them searchable after a refresh interval; a replica serving the query may be behind its primary; and a cache in front may hold an older answer. I would trace one document id through each stage with timestamps to find the largest delay, then decide whether to shorten it, route "just uploaded" queries to a path that includes fresh data, or set the user's expectation with an explicit indexing status.
Q: How does sharding affect the latency of a vector search?
A: When the index is partitioned by document, a query fans out to all shards and completes when the slowest one answers, so the tail gets worse as shards are added even if the average shard is fast. If each shard is slow 2% of the time, 16 shards produce a slow query roughly 27% of the time, and 64 shards about 73%. Mitigations are fewer larger shards, hedged requests to replicas, per-shard deadlines that return partial results, and routing by tenant so queries hit one shard.
Q: More replicas or more shards for a vector database that is running out of QPS?
A: If the index still fits in memory on a node and the load is read throughput, add replicas: each one holds the full index and serves a share of the queries, and it also improves availability. If the index no longer fits, or ingestion writes are the limit, replicas do not help and you need more shards. Note that in a fan-out design every shard still handles every query, so throughput does not scale with shard count the way it does with replicas (each shard searches a smaller index, but per-query overhead and the wait for the slowest shard remain), which is why many systems scale QPS with replicas and capacity with shards.
Follow-up: A single tenant now generates 40% of traffic and its shard is saturated. What do you do?
A: Short term, cache that tenant's hot queries and add read replicas for its shard. Medium term, split the tenant across several shards (sub-shard by document or by hash of a salted key) and route its queries as a scatter-gather among just those shards, or move it to a dedicated cluster. The cost is more routing complexity and higher latency variance for that tenant, and it must be done live with the copy, catch-up and cut-over procedure rather than by a downtime window.
Follow-up: The primary of a shard fails and an asynchronous replica is promoted. What can go wrong?
A: Writes the old primary acknowledged but had not shipped are missing on the new primary, so data is lost or, when the old primary returns, conflicting history appears (split brain). Guard against it with fencing (a monotonically increasing epoch so the old primary cannot accept writes), reconciling or discarding the old primary's tail, and, where the data is valuable, semi-synchronous or consensus-based replication so that acknowledged writes exist on a majority. It is a durability versus write-latency decision.
Follow-up: An operation needs to update data on two different shards atomically. Options?
A: Avoid it by choosing the shard key so related data co-locates. If unavoidable, use two-phase commit (see Two-Phase Commit), which blocks on coordinator failure and adds round trips, or a saga of local transactions with compensating actions plus idempotent steps, which is eventually consistent and needs careful failure handling. I would prefer the saga for user-facing workflows and reserve two-phase commit for narrow cases where inconsistency is unacceptable.
In a design interview, replication and sharding appear as a capacity calculation followed by a topology. For a retrieval service, assume (illustratively) 100 million chunks of 768-dimension float32 embeddings: about 300 GB raw, plausibly 400 GB with index overhead, so with roughly 48 GB usable per node you need at least nine shards, and with three replicas per shard for availability and read QPS, about 27 nodes; quantizing vectors changes that arithmetic, so state the assumption and offer to measure recall. Write the sharding decision next to the access pattern: tenant-scoped queries argue for routing by tenant, corpus-wide search argues for hash by document with scatter-gather and a stated p99 fan-out budget. For state around the model, chat history is typically sharded by conversation or user id with a primary per shard and asynchronous replicas, with read-your-writes handled by reading the primary for a short window; the feature store shards by entity id with replicas and a cache for hot entities; the metadata, ledger and model registry stay small enough for one replicated primary or a consensus-backed store. Mention how you will grow: a fixed number of logical partitions or slots (or a ring with virtual nodes) so that adding a node moves a bounded fraction of the data, plus per-shard dashboards for size, QPS, replication lag and tail latency so hot shards are found before users find them. Finish with the failure story: what happens when a replica lags, when a primary fails (fencing, data-loss window), and when one shard is slow (hedging and partial results).
Replication is the cheap, reversible step: it adds availability and read capacity with modest complexity, but it introduces staleness and a data-loss window under asynchronous replication, and it does not raise the ceiling on write rate or dataset size. Sharding is the expensive, hard-to-reverse step: it removes the single-node ceiling, but it adds routing, cross-shard queries and transactions, hot-shard risk, tail-latency amplification for fan-out reads, and a resharding procedure you must operate live. It is the wrong choice when the data fits comfortably on one node with room to grow (a vertically scaled, well-indexed database with replicas is simpler and often cheaper), and it is a mistake to choose the shard key by convenience: a key that concentrates traffic or forces every query to fan out gives you the costs of sharding without the benefits. Plan for growth with more logical partitions than nodes, measure per-shard skew from day one, and delay sharding until the numbers, not intuition, show a single node is the limit.