Database Sharding & Replication

must-know senior · part of Skills & Topics · Senior SWE Roadmap · related: Consistent Hashing · Caching Strategies · Scalability Fundamentals

Why scale past one machine

A single database server has a ceiling: finite CPU, RAM, disk I/O, and network bandwidth. Two distinct scaling problems show up once traffic outgrows one box, and they have different fixes:

• Too much data / too much write throughput for one machine → fixed by sharding (horizontal partitioning): split the data across multiple machines so each holds a fraction of it.
• Too many reads for one machine to serve, even though it could hold all the data → fixed by replication: copy the same data onto multiple machines so reads can be spread across them.

These are orthogonal and usually combined: a large system shards data across N partitions, and each shard is itself replicated for availability and read scaling. This article covers both, and where they interact (resharding, cross-shard queries).

flowchart TB
    subgraph "Sharding: split data across machines"
        S1[("Shard 1<br/>users A-M")]
        S2[("Shard 2<br/>users N-Z")]
    end
    subgraph "Each shard is replicated"
        S1L[(Shard 1 Leader)] --> S1F1[(Shard 1 Follower)]
        S1L --> S1F2[(Shard 1 Follower)]
        S2L[(Shard 2 Leader)] --> S2F1[(Shard 2 Follower)]
        S2L --> S2F2[(Shard 2 Follower)]
    end

Sharding strategies

Range-based sharding

Assign each shard a contiguous range of the shard key (e.g., shard 1 holds user IDs 1–1,000,000, shard 2 holds 1,000,001–2,000,000).

flowchart LR
    Router{"Router / query planner"} --> S1["Shard 1<br/>key range: A–H"]
    Router --> S2["Shard 2<br/>key range: I–P"]
    Router --> S3["Shard 3<br/>key range: Q–Z"]

• Pros: range queries (“give me all users with IDs between X and Y”) stay within one or a few shards — efficient. Easy to reason about which shard holds what.
• Hot-spot failure example: shard by created_at timestamp or by a monotonically increasing auto-increment ID. All new writes land on whichever shard owns the current tail of the range — every insert hits the same shard, defeating the entire purpose of sharding for write throughput. This is a textbook range-sharding failure and a very common wrong-first-instinct in interviews (e.g., a naive order_id-range shard means all of today’s orders hit one shard).

Hash-based sharding

Apply a hash function to the shard key, then assign the shard by hash(key) % N (or a range of the hash space).

flowchart LR
    Key["shard_key = user_id: 8842"] --> H["hash(8842) = 0x7f3a...<br/>hash % 4 shards"]
    H --> S0[(Shard 0)]
    H --> S1[(Shard 1)]
    H -->|"result = 2"| S2[(Shard 2)]
    H --> S3[(Shard 3)]

• Pros: distributes writes uniformly (a good hash function has no correlation with insertion order, so no single shard absorbs all new writes) — fixes the range-based hot-spot problem directly.
• Hot-spot failure example (a different one): a skewed key distribution, not insertion order — e.g., sharding a multi-tenant system by tenant_id where one tenant is 1000x larger than all others combined. Hashing distributes tenants uniformly across shards, but one shard still ends up doing 1000x the work because it happens to host the giant tenant. Hashing fixes uniform-distribution-over-time hot spots, not inherent-skew-in-the-data hot spots.
• The % N resharding problem: if N (shard count) changes — adding a shard to scale up — hash(key) % N changes for almost every key, meaning almost all data has to move. This is precisely the problem Consistent Hashing was designed to solve: it bounds the fraction of keys that need to move on a resize to roughly 1/N instead of “almost everything.”

Directory-based (lookup-table) sharding

A separate lookup service/table maps each key (or key range) to a specific shard explicitly, instead of computing the mapping algorithmically.

flowchart LR
    App[App Server] -->|"1: lookup(user_id=8842)"| Dir["Directory service<br/>(key → shard mapping)"]
    Dir -->|"2: shard = 3"| App
    App -->|"3: query"| S3[(Shard 3)]

• Pros: maximum flexibility — the mapping can be arbitrary (move one specific hot user to their own dedicated shard, rebalance incrementally by moving individual keys rather than whole ranges), and resharding doesn’t require a formula change, just updating map entries.
• Cons: the directory itself is a new critical-path dependency — every query now has an extra lookup hop, and the directory service becomes a potential single point of failure/bottleneck unless it’s itself made highly available and heavily cached (in practice, the mapping is small enough to cache aggressively on app servers, refreshed on change).
• Used when shard assignment needs to be dynamic and fine-grained — e.g., systems that need to isolate specific hot tenants onto dedicated hardware.

Strategy	Range queries	Write distribution	Resharding cost	Hot-spot risk
Range-based	Efficient (contiguous)	Poor if key correlates with insert order	Moderate (split a range)	High for monotonic keys (e.g., timestamps, auto-increment IDs)
Hash-based	Poor (scattered across shards)	Good, if hash is uniform	High without consistent hashing	High if underlying key distribution is skewed (e.g., one giant tenant)
Directory-based	Depends on underlying scheme	Fully controllable	Low (update mapping, move data incrementally)	Lowest — can manually isolate hot keys

Shard key selection: a worked example

Take a users table for a global product. Two candidate shard keys:

Option A: shard by user_id (hashed)

• Writes distribute evenly (good hash → uniform spread).
• Reads for a single user hit exactly one shard — fast, no fan-out.
• Bad fit for a query like “get all users in the EU for GDPR compliance/data residency” — that query has to fan out to every shard, because EU users are scattered uniformly across all of them.

Option B: shard by region

• Naturally aligns with data residency/compliance requirements (EU user data physically lives on EU-located shards — often a hard legal requirement, not just an optimization).
• Naturally aligns with “get all users in a region” queries — single shard.
• Bad fit for load distribution: regions have wildly different user populations (a US shard and a region=Iceland shard are not remotely the same size) — a classic skew hot spot, and it gets worse over time as regional growth rates diverge unpredictably.
• Cross-region queries (a global leaderboard, a cross-region friend graph) now require fan-out across shards.

	Shard by user_id (hashed)	Shard by region
Write/load distribution	Even	Uneven (regions differ wildly in size)
Single-user lookup	Fast, one shard	Fast, one shard
”All users in region X” query	Fan-out to all shards	Single shard
Data residency / compliance	Not naturally aligned	Naturally aligned
Growth handling	Predictable, rebalance by adding hash ranges	Unpredictable — one region can outgrow its shard

No universally correct answer — this is exactly the kind of tradeoff an interviewer wants argued explicitly. A common resolution in real systems: shard by user_id for the general case, but special-case a separate mechanism (a distinct database, or dedicated shards) for the subset of data that has a hard legal residency requirement — don’t force one shard key to satisfy both load-distribution and compliance goals simultaneously.

Cross-shard queries and joins: the main pain of sharding

Once data is split across shards, any query that needs data from more than one shard gets expensive:

• Joins across shards aren’t a single SQL query anymore — the application (or a query federation layer) has to fetch from each shard separately and join in application code, which is slower and more complex than a single-node join.
• Aggregations (COUNT, SUM, GROUP BY across the whole dataset) require scatter-gather: query every shard, then merge results — latency is bounded by the slowest shard, and a single unavailable shard can block or degrade the whole query.
• Transactions across shards (updating rows on two different shards atomically) require distributed transaction protocols (two-phase commit) or a redesign to avoid needing cross-shard atomicity at all — expensive and rarely worth it operationally.

The standard mitigation is denormalization: duplicate the data you’d otherwise need a join to get, directly onto the shard where it’ll be queried. E.g., instead of joining orders (sharded by order_id) with users (sharded by user_id) to get a user’s display name on an order, store user_display_name directly on the order row. This trades storage and write-time consistency work (updating the duplicate on change) for eliminating the cross-shard join on the (much more frequent) read path — the same fundamental tradeoff as caching, just applied to schema design. The other mitigation is choosing a shard key that keeps commonly-joined data together in the first place (e.g., shard both orders and order_items by the same user_id, so a user’s orders and their line items always co-locate on the same shard).

Resharding and rebalancing cost

Adding or removing shards means moving data — and how expensive that is depends entirely on the sharding strategy chosen above:

• Naive hash (% N): changing N reshuffles almost every key’s target shard — a full data migration, effectively a rewrite of the whole dataset’s placement.
• Consistent hashing: bounds the moved fraction to roughly 1/N of keys when adding the N+1th node — see Consistent Hashing for the full mechanism (a hash ring, virtual nodes to prevent uneven load, and why this specific property is the entire reason the technique exists).
• Directory-based: rebalancing is a deliberate, incremental operation — pick specific keys/ranges to move, update the directory, no formula-driven mass migration at all. Most operationally controllable, at the cost of maintaining the directory service itself.

This is why, in a system-design interview, “how do you add a shard without downtime” is really asking whether you know consistent hashing (or an equivalent directory-based scheme) exists specifically to solve this — a naive % N answer is an immediate red flag at the senior level.

Replication topologies

Single-leader (leader-follower / primary-replica)

All writes go to one leader; the leader replicates changes to one or more followers; reads can be served by the leader or by any follower.

flowchart TD
    App[App Servers] -->|writes| L[(Leader)]
    App -->|reads, optionally| L
    App -->|reads| F1[(Follower 1)]
    App -->|reads| F2[(Follower 2)]
    L -->|replicate| F1
    L -->|replicate| F2

• Pros: simple to reason about — a single, unambiguous order of writes (whatever order they hit the leader), no write-write conflicts possible.
• Cons: the leader is a single point of failure for writes, and a write-throughput ceiling (one machine).
• The default choice for most systems (PostgreSQL streaming replication, MySQL replication, MongoDB replica sets all default to this model).

Multi-leader (multi-master)

More than one node accepts writes; each leader replicates its writes to the others.

flowchart LR
    L1[("Leader: US datacenter")] <-->|bidirectional replication| L2[("Leader: EU datacenter")]
    App1["App Servers - US"] -->|writes| L1
    App2["App Servers - EU"] -->|writes| L2

• Pros: writes can be accepted locally in each region — much lower write latency for geographically distributed users, and no single write bottleneck.
• Cons: write conflicts are now possible — the same row can be updated concurrently on two leaders before either has replicated to the other. Requires an explicit conflict resolution strategy: last-write-wins (simple, silently loses data), version vectors, or application-level merge logic (e.g., CRDTs for specific data types that can merge deterministically).
• Used when write latency across regions genuinely matters more than the operational complexity of conflict resolution — e.g., a multi-region collaborative document, or region-local writes for regulatory/latency reasons.

Leaderless replication

No node is designated leader; a client (or a coordinator) writes to and reads from multiple nodes directly, using quorums to guarantee consistency despite no single source of truth.

flowchart TD
    App[Client] -->|"write to W nodes"| N1[(Node 1)]
    App -->|"write to W nodes"| N2[(Node 2)]
    App -.->|"write may not reach<br/>every replica immediately"| N3[(Node 3)]
    App2["Client - read"] -->|"read from R nodes"| N1
    App2 -->|"read from R nodes"| N2
    Note["Quorum rule: W + R > N<br/>guarantees read and write sets overlap<br/>by at least one up-to-date node"]

• A write is considered successful once acknowledged by W of N replicas; a read queries R replicas and returns the most recent value(s) seen (using version numbers/vector clocks to detect which is newest).
• W + R > N guarantees every read quorum overlaps with every write quorum by at least one node, so a read is mathematically guaranteed to see the most recent write — this is the core correctness argument for leaderless systems and worth being able to state precisely.
• Used by Dynamo-style systems (Cassandra, Riak, DynamoDB’s internal replication) — high availability, tunable consistency (adjust W/R per operation to trade off latency vs consistency), no single leader to fail over.
• See Design a Key-Value Store (DynamoDB-like) for this model applied end-to-end.

Topology	Write scalability	Conflict handling	Failover complexity	Typical use
Single-leader	Limited to one node	None needed (single order)	Moderate (promote a follower)	Default for most relational/NoSQL systems
Multi-leader	Scales writes across regions	Required — explicit conflict resolution	Higher (multiple write points)	Multi-region, latency-sensitive writes
Leaderless	Scales writes across all nodes	Via versioning + quorum reads	Lowest — no leader to fail over	High-availability KV stores (Dynamo-style)

Synchronous vs asynchronous replication

The leader has a choice about when to acknowledge a write back to the client, relative to replication completing.

sequenceDiagram
    participant App
    participant L as Leader
    participant F as Follower

    rect rgb(235, 245, 255)
    Note over App,F: Synchronous replication
    App->>L: write(x=5)
    L->>F: replicate(x=5)
    F-->>L: ack (durably written on follower)
    L-->>App: ack (only after follower confirms)
    end

    rect rgb(255, 245, 235)
    Note over App,F: Asynchronous replication
    App->>L: write(x=5)
    L-->>App: ack (immediately, before replicating)
    L--)F: replicate(x=5) (in background)
    end

• Synchronous: the leader waits for at least one follower to confirm before acknowledging the client. Guarantees the write survives a leader failure (a follower already has it, safe to promote) — but adds the follower’s write latency (and, worse, its availability) to every write; if the follower is slow or down, writes stall.
• Asynchronous: the leader acknowledges immediately, replicates in the background. Fast writes, leader failures don’t block on follower health — but a leader crash before replication completes means the acknowledged write is lost when a follower without it gets promoted. This is a real durability gap, not a theoretical one.
• Semi-synchronous (the common middle ground): wait for acknowledgment from at least one follower (not all), replicate to the rest asynchronously — bounds the worst-case data loss to “whatever wasn’t yet sent to the one synchronous follower” while not paying the latency cost of waiting for every replica.

This is a direct instance of the classic latency vs durability tradeoff, and stating it in exactly those terms is the sharpest way to answer a “sync or async replication?” question.

Replication lag and the read-your-writes problem

Asynchronous (and even semi-synchronous, for the async-tail followers) replication means followers can lag behind the leader by some real amount of time — milliseconds usually, but seconds or more under load or network issues. This causes a very visible, very common bug class:

sequenceDiagram
    participant U as User
    participant L as Leader
    participant F as Follower (lagging)

    U->>L: POST /profile (update bio)
    L-->>U: 200 OK (write succeeded)
    Note over L,F: replication in flight, not yet applied
    U->>F: GET /profile (page reload, routed to follower)
    F-->>U: 200 OK — shows OLD bio
    Note over U: "I just saved this — where did it go?"

The user just wrote something and immediately reads it back on a different connection that happens to be routed to a follower that hasn’t caught up — a jarring, frequently-reported bug in real products.

Fixes:

• Sticky reads to the leader: after a user writes, route their own subsequent reads to the leader for some window (e.g., the rest of the session, or N seconds) — guarantees read-your-writes for that user specifically without forcing all reads to the leader.
• Read-your-writes tokens: the write response includes a token (e.g., the leader’s replication log position/LSN at time of write); the client passes this token on subsequent reads, and a follower either serves the read only if it has replicated past that position, or the request is routed to a follower/leader that has. More precise than a blanket “stick to leader for N seconds” and works correctly even for reads that happen well after the write.
• Monotonic reads guarantee (a related but distinct issue): without care, a user could read from a more-caught-up follower, then on a later request get routed to a less-caught-up one and see data go “backwards in time.” Fixed by consistently routing a given user’s reads to the same replica (e.g., hash the user ID to a specific follower) for the duration of a session.

Failover mechanics

When a leader fails, some follower needs to be promoted — and doing this safely is harder than it looks.

flowchart TD
    Fail["Leader fails / becomes unreachable"] --> Detect["Failure detection<br/>(heartbeat timeout via a monitoring/consensus process)"]
    Detect --> Elect["Election: pick the follower with<br/>the most up-to-date replicated data"]
    Elect --> Promote["Promote chosen follower to new leader"]
    Promote --> Reconfig["Reconfigure other followers<br/>to replicate from new leader"]
    Reconfig --> OldLeader{"Old leader comes back?"}
    OldLeader -->|"still thinks it's leader"| SplitBrain["Split-brain risk:<br/>two nodes both accepting writes"]
    OldLeader -->|"demoted to follower on rejoin"| Safe["Safe — single leader restored"]

• Failure detection: usually a heartbeat/timeout mechanism — but a leader that’s merely slow (network partition, GC pause) looks identical to a dead one from the followers’ perspective, which is the root of split-brain risk: the followers promote a new leader while the old one is still alive and still accepting writes from clients that haven’t heard about the new leader yet.
• Split-brain: two nodes simultaneously believing they’re the leader, both accepting writes, causing the data to diverge in an unreconciled way — one of the most dangerous failure modes in distributed databases, because unlike a simple crash, it produces silently incorrect / conflicting data rather than an obvious outage.
• Why this needs consensus, not just “the first follower to notice”: safely deciding “who is the new leader” among multiple nodes that can’t fully trust their view of each other requires a distributed consensus protocol — Raft or Paxos by name — specifically so that a majority of nodes agree on exactly one new leader, and the old leader (once it realizes a new term/leader exists) steps down rather than continuing to accept writes. This is why systems like etcd (Raft), ZooKeeper (a Paxos-derived protocol, ZAB), and modern distributed databases (CockroachDB, TiDB — Raft per shard) lean on a consensus layer for leader election rather than ad hoc heartbeat-and-promote logic.
• Choosing which follower to promote also matters: promoting the most caught-up follower (least replication lag) minimizes data loss; promoting an arbitrary one can silently lose more committed-looking writes than necessary.

Interview angles

• “How would you scale this database past a single machine?” — lead by distinguishing the two problems (too much data/write throughput → shard; too many reads → replicate), then go deep on whichever the scenario actually needs. Don’t reach for sharding when the real problem is read load that replication alone would solve.
• “Pick a shard key for [scenario] and defend it against hot-spotting.” — walk through a concrete example the way the user_id vs region comparison does above: name the candidate keys, the failure mode of each, and the tradeoff, rather than asserting one is simply “correct.”
• “What happens when you add a new shard?” — naive % N re-hashing (bad, near-total data movement) vs consistent hashing (bounded movement) — see Consistent Hashing. Knowing this by name, unprompted, is a strong signal.
• “A user says their update disappeared right after saving it.” — replication lag / read-your-writes; name the fix (sticky reads to leader, or a replication-position token) rather than just diagnosing the symptom.
• “How do you fail over safely?” — split-brain risk is the crux of the answer; mention that safe leader election needs a majority/quorum decision (name Raft or Paxos) rather than “the first replica that notices promotes itself.”
• “Why not just always join across shards in the application?” — cross-shard joins are the main operational cost of sharding; the fix is denormalization or choosing a shard key that co-locates commonly-joined data, not building a distributed join engine.
• “Sync or async replication?” — state the tradeoff explicitly in terms of durability (can the acknowledged write survive a leader crash?) vs latency/availability (does a slow/dead follower block writes?), and mention semi-synchronous as the practical middle ground.

My Notes