System Design - Database Sharding & Replication

When a single database node runs out of CPU, memory, or disk capacity, vertical scaling (buying a bigger server) quickly hits hardware ceilings and exponential cost curves. Horizontal scaling—dividing data and traffic across multiple independent database servers—becomes the only path forward.

The two foundational pillars of database horizontal scaling are Replication (copying the same data across multiple nodes to scale read throughput and ensure high availability) and Sharding (partitioning data subsets across different machines to scale write throughput and storage capacity).

Replication Strategies

Replication creates redundant copies of data across multiple database instances to protect against hardware failures and distribute read queries.

1. Primary-Replica (Leader-Follower) Replication

The most common relational database pattern (MySQL, PostgreSQL):

Writes: All write queries (INSERT, UPDATE, DELETE) are directed to a single Primary (Leader) node.
Replication Stream: The Primary records changes into its Write-Ahead Log (WAL/binlog) and streams updates to one or more Replica (Follower) nodes.
Reads: Read-only queries can be distributed across all Replicas using a read-write splitting proxy (e.g., ProxySQL, AWS Aurora Reader endpoints).
Synchronous vs. Asynchronous Replication:
- Asynchronous Replication (Default): The Primary commits locally and immediately returns success to the client before replicas confirm receipt. Low write latency, but risks data loss if the primary crashes before replicas catch up.
- Synchronous Replication: The Primary waits until at least one replica has acknowledged writing the transaction. Guarantees zero data loss, but increases write latency and stalls if replicas become unreachable.
- Semi-Synchronous: The primary commits when at least one synchronous replica confirms receipt to the relay log, while remaining replicas sync asynchronously.

2. Multi-Leader (Master-Master) Replication

Multiple database instances act as primaries capable of accepting writes:

Common in multi-datacenter deployments to reduce write latency for geographically distributed users.
Major Challenge: Conflict Resolution. If two users update the same record concurrently in US-East and EU-West, conflict resolution strategies must be applied:
- Last Write Wins (LWW): Relies on timestamps; prone to clock skew data loss.
- Conflict-Free Replicated Data Types (CRDTs): Mathematically sound data structures that converge automatically without coordination.
- Operational Transformation / Application Logic: Merges conflicts via custom business rules.

3. Leaderless Replication (Dynamo-Style)

Used in distributed NoSQL stores like Amazon DynamoDB, Apache Cassandra, and ScyllaDB:

Clients can write to and read from any node in the cluster.
Quorum Consistency ($R + W > N$):
- $N$ = Replication factor (e.g., 3 copies).
- $W$ = Number of nodes that must acknowledge a write for it to succeed (e.g., 2).
- $R$ = Number of nodes that must respond to a read query (e.g., 2).
- If $W + R > N$, at least one node in the read set is guaranteed to have the most recent write.
- Node disagreements are resolved via Read Repair (client writes the newer version back to stale nodes) and background Anti-Entropy (using Merkle Trees to sync divergent replicas).

Database Sharding (Horizontal Partitioning)

Sharding splits a large dataset into smaller, independent subsets called shards, hosted across separate database servers. Each shard contains a distinct partition of the data, while maintaining the same schema.

Sharding Architecture Models

Range-Based Sharding:
- Shards are partitioned based on contiguous ranges of a key (e.g., User IDs 1–1,000,000 on Shard 1, 1,000,001–2,000,000 on Shard 2; or dates by month).
- Pros: Simple; efficient for range scans (e.g., querying orders placed between March 1 and March 15).
- Cons: Prone to Hotspotting. If traffic is skewed toward recent dates or high IDs, a single shard absorbs 100% of the write load while older shards sit idle.
Hash-Based Sharding:
- A cryptographic hash function (e.g., MurmurHash3, MD5) is applied to the shard key: $$\text{Shard ID} = \text{hash}(\text{Shard Key}) \pmod N$$
- Pros: Uniformly distributes reads and writes across all $N$ shards, eliminating hotspots.
- Cons: Range queries are catastrophic because adjacent records are scattered across different physical machines, requiring a "scatter-gather" query across all nodes.
- Re-sharding Penalty: If the cluster expands from $N$ to $N+1$ nodes, almost every key's modulus changes, triggering a massive, cluster-wide data migration.
Consistent Hashing (The Industry Standard):
- Maps both database nodes and shard keys onto a circular $2^{32} - 1$ hash ring.
- Keys are assigned to the first node encountered moving clockwise along the ring.
- Virtual Nodes (Vnodes): Each physical node is mapped to multiple points (e.g., 100–200 virtual tokens) across the ring to ensure balanced distribution.
- Advantage: Adding or removing a node only migrates a fraction ($\frac{1}{N}$) of the keys from neighboring nodes, without requiring a global cluster reshuffle.
Directory-Based (Lookup) Sharding:
- A centralized, highly cached lookup service maps each Shard Key to its corresponding physical Shard ID.
- Pros: Complete flexibility to dynamically move individual accounts or high-volume enterprise tenants to dedicated hardware without altering mathematical hash functions.
- Cons: The lookup table itself is a critical single point of failure and adds an extra network hop to every query.

Choosing the Optimal Shard Key

The choice of Shard Key is irreversible in production without a complete database migration. Follow these evaluation criteria:

High Cardinality: The key must have millions of distinct values (e.g., user_id, uuid). Low cardinality fields (such as country_code or gender) will create massive, un-splittable data skew.
Even Traffic Distribution: Avoid keys that correlate with time, celebrity influence, or seasonal events.
Query Colocation: Choose a key that groups related tables onto the same physical shard to preserve local table joins. For an e-commerce platform, sharding both Users, Orders, and OrderItems by user_id allows order lookups and transactions to run locally within a single shard without cross-node network coordination.

The Complications of Sharding

Sharding is not free. It introduces significant distributed systems complexity:

1. Cross-Shard Joins & Aggregations

Relational databases cannot natively perform efficient SQL JOIN operations across separate physical database instances:

Denormalization: Duplicate reference data (e.g., product names or user profiles) directly into child tables so queries can be satisfied from a single shard.
Application-Level Joins: The backend queries Shard A, extracts IDs, then queries Shard B in parallel, stitching the JSON objects together in application memory.
Global Secondary Indexes (GSI): Asynchronous streaming pipelines (via Kafka/Debezium) build separate search tables indexed by alternate keys (e.g., searching users by email when the primary shard key is user_id).

2. Cross-Shard Transactions (2PC vs. Sagas)

When a business transaction spans multiple shards (e.g., transferring money from User A on Shard 1 to User B on Shard 2):

Two-Phase Commit (2PC): A coordinator asks all participants to prepare (Phase 1), then commits (Phase 2). While 2PC guarantees strict ACID consistency, it is a blocking protocol. If the coordinator or any participant hangs, database locks are held indefinitely, destroying system availability.
Saga Pattern (Eventual Consistency): Deconstructs the transaction into a sequence of local transactions coordinated via message queues. If a step fails, the saga triggers compensating transactions (undo actions) in reverse order. Sagas are preferred for scalable microservices architectures.

3. Re-sharding & Live Migrations

When data growth exceeds capacity and you must scale from 16 to 32 shards:

Spin up the new database instances.
Update the routing proxy with the new partition map.
Use a Dual-Write approach:
- Application writes to both old and new shards.
- Background backfill workers migrate historical records.
- Verify checksums between old and new stores.
- Cut over reads to the new shards, then decommission the old instances.

Sharding Architecture Diagram

                                  [ Client Applications ]
                                             │
                                             ▼
                                   [ Sharding Proxy / Routers ]
                                (Vitess / Citus / Envoy / ProxySQL)
                                             │
                 ┌───────────────────────────┼───────────────────────────┐
                 │                           │                           │
                 ▼                           ▼                           ▼
        [ Shard 1 (Primary) ]       [ Shard 2 (Primary) ]       [ Shard 3 (Primary) ]
           (Keys: 0x00..0x54)          (Keys: 0x55..0xAA)          (Keys: 0xAB..0xFF)
                 │                           │                           │
          ┌──────┴──────┐             ┌──────┴──────┐             ┌──────┴──────┐
          ▼             ▼             ▼             ▼             ▼             ▼
      [Replica]     [Replica]     [Replica]     [Replica]     [Replica]     [Replica]
      (Read-Only)   (Read-Only)   (Read-Only)   (Read-Only)   (Read-Only)   (Read-Only)

Summary Checklist for System Design Interviews

Start with Read Replicas first: Never jump straight to sharding. If read traffic is the bottleneck, primary-replica replication and aggressive caching (Redis/CDN) will solve 90% of scaling problems with far less engineering overhead.
Shard only when writes or data size exceed single-node limits: Shard when your dataset exceeds terabytes and cannot fit on large cloud NVMe drives, or when write QPS exceeds 10,000–20,000 updates/second.
Explicitly state your Shard Key and justify it: Explain why the key avoids hotspots and how it colocates critical entity queries.
Address cross-shard query solutions: Propose denormalization, global secondary indexes, or asynchronous event streams to handle secondary lookups.