Warming up the neural circuits...
By the end of this chapter you will:
Your database is the one component you can't just "add more of." When it buckles, everything buckles. Replication and sharding are how you keep a database standing when the load would crush a single machine.
A city library has grown so popular that its single building can't keep up. The waiting line snakes around the block. There are two problems: (1) too many people trying to read books, and (2) too many people trying to check books out.
Replication (read scaling): The library opens satellite branches across the city, each with an identical copy of every book. When someone wants to read, they go to the nearest branch — 90% faster. But when someone wants to check out a book, they still have to go to the main branch, because only the main branch has the authority to lend. Every morning, a truck delivers newly acquired books to all branches.
Sharding (write scaling): The library splits its collection: fiction goes to Branch A, non-fiction to Branch B, reference to Branch C. Now checkouts are distributed — you can check out fiction at Branch A, non-fiction at Branch B. But if you want a fiction book AND a non-fiction book, you have to visit two branches.
This is the tension at the heart of database scaling. Replication gives you read throughput but doesn't help with writes. Sharding helps with writes but makes queries that span multiple shards expensive or impossible. And many teams rush to shard too early, burdening themselves with complexity they didn't need.
Database replication means maintaining copies of the same data on multiple nodes. The fundamental challenge: how do you keep copies in sync without killing performance?
Primary-Replica (Single Leader):
One node is the primary (accepts all writes). All other nodes are replicas (accept reads, replicate writes from the primary). This is the most common pattern — used by PostgreSQL streaming replication, MySQL replication, MongoDB replica sets.
┌─────────────┐
┌──────────►│ Replica 1 │◄──── Read queries
│ └─────────────┘
│
┌───────┴───────┐
│ Primary │◄──── All writes + some reads
└───────┬───────┘
│
│ ┌─────────────┐
└──────────►│ Replica 2 │◄──── Read queries
└─────────────┘Synchronous vs Asynchronous Replication:
-- PostgreSQL streaming replication configuration
-- On primary (postgresql.conf):
wal_level = replica
max_wal_senders = 5
wal_keep_size = 1024 -- MB of WAL to retain for lagging replicas
synchronous_commit = remote_apply -- Wait for replica to apply WAL
synchronous_standby_names = 'ANY 1 (replica1, replica2, replica3)'
-- On replica (postgresql.conf):
primary_conninfo =
Even with synchronous replication to one replica, other replicas will lag. A read from an async replica might return data that's seconds (or minutes, during heavy write loads) old. Your application must handle this: either route stale-sensitive reads to the primary, or design UX that tolerates eventual consistency (like showing "processing" for 30 seconds after a write).
Multi-Primary (Multi-Leader):
Multiple primaries accept writes, and changes replicate between them. This is complex — if two primaries update the same row simultaneously, you have a conflict.
Multi-primary is useful for: (1) multi-region where users write to their local region, (2) offline-first apps where clients are effectively primaries, (3) collaborative editing (Google Docs uses Operational Transformation, a of multi-primary).
Leaderless Replication (Dynamo-style):
No primary. Any replica can accept writes. Writes go to multiple replicas. Reads go to multiple replicas and reconcile. Cassandra, DynamoDB, and Riak use this model. Conflict resolution uses: last-write-wins (simple but loses data), version vectors (tracks causality), or CRDTs (Conflict-free Replicated Data Types — merge automatically).
Sharding (horizontal partitioning) splits a dataset across multiple independent database instances. Each shard holds a subset of the data. The shard key determines which shard holds which data.
The shard key is the most important decision you'll make:
A bad shard key creates hot shards (one shard gets 90% of traffic), forces expensive cross-shard queries, and is nearly impossible to change later.
// Good shard key: user_id for a multi-tenant SaaS
// All data for one customer lives on one shard
// Most queries are scoped to one customer → single shard
function getShard(userId: string, totalShards: number): number {
return hash(userId) % totalShards;
}
Consistent hashing for shard allocation:
Just like sharding, consistent hashing minimizes data movement when adding or removing shards. Instead of hash(key) % N, use a hash ring. When you add shard S_new, only data in its hash range moves to it.
// Consistent hashing for database shards
class ShardRouter {
private ring: ConsistentHashRing;
private shardConnections: Map<string, Pool>;
async query(sql: string, params: any[], shardKey
| Strategy | How it works | Pros | Cons | Who uses it |
|---|---|---|---|---|
| Range-based | Shard 0: users 1-1M, Shard 1: users 1M-2M | Simple to reason about, efficient range scans | Hot shards if keys are sequential (timestamps) | HBase, Bigtable |
| Hash-based | hash(key) % N | Even distribution across shards | Adding shards requires rehashing all data | Redis Cluster, DynamoDB |
| Directory-based | Lookup table maps keys → shards | Flexible, can move individual keys | Lookup table becomes bottleneck | Pinterest, Uber (early) |
| Entity-group | Related entities (user + their orders) on same shard | No cross-shard joins for related data | Uneven shard sizes if some users are huge |
The moment you shard, queries that span shards become expensive:
-- Before sharding: simple
SELECT COUNT(*) FROM orders WHERE status = 'pending';
-- After sharding by user_id: must query all shards
-- Shard 0: SELECT COUNT(*) FROM orders WHERE status = 'pending';
-- Shard 1: SELECT COUNT(*) FROM orders WHERE status = 'pending';
-- ... and sum the results in application code
-- JOINs across shards are even worse:
Techniques for handling cross-shard queries:
Vitess is the sharding that powers YouTube's MySQL infrastructure. Key concepts:
A well-tuned PostgreSQL instance on modern hardware handles 10,000-50,000 writes/second and billions of rows. Before sharding, exhaust these options: read replicas, connection pooling, query optimization, proper indexing, partitioning within a single instance, and archiving cold data. Most teams shard 2-3 years too early.
Instagram's database architecture is one of the most studied sharding implementations. With over 2 billion users and billions of photos, a single database was impossible.
Their approach:
The takeaway: Instagram's sharding works because they chose a shard key that matches their access pattern. Most queries hit one shard. Cross-shard operations are limited to infrequent writes and asynchronous processes.
| Mistake | Why it's wrong | What to do instead |
|---|---|---|
| Sharding before you need to | Sharding adds massive operational complexity: resharding, cross-shard queries, distributed transactions. Most apps never reach the scale that requires it | Exhaust vertical scaling, read replicas, query optimization, and partitioning first. Shard when your write throughput exceeds what one well-tuned instance can handle |
| Choosing a shard key based on even data distribution alone | Even distribution is useless if every query hits all shards. A perfectly balanced system that's always doing scatter-gather is worse than a slightly imbalanced one with single-shard queries | Choose the shard key that keeps 95%+ of queries on a single shard. Even distribution is secondary |
| Using auto-increment IDs across shards | Each shard's auto-increment starts at 1 — you get duplicate IDs. Merging data becomes a nightmare | Use UUIDs, ULIDs, or Snowflake-style IDs that are unique across shards. Or use a global ID service |
| Async replication with no lag monitoring | Replication lag of 30 seconds means users see stale data for 30 seconds after writes. "I just posted a photo but it's not showing up" | Monitor replication lag. Alert if lag exceeds your application's tolerance. Route lag-sensitive reads to the primary |
| Assuming read replicas solve write throughput | Read replicas help read throughput. They do nothing for write throughput. If your bottleneck is writes (common for logging, IoT, financial systems), replicas are irrelevant | Shard or use a write-optimized database (Cassandra, ScyllaDB, TimescaleDB for time-series) |
| Not planning for resharding | You will need to add shards. If your sharding scheme can't accommodate this without downtime, you've built a trap | Use consistent hashing or a directory-based approach. Plan and test the resharding process before you need it |
| Running cross-shard transactions without distributed support | BEGIN; UPDATE shard_1...; UPDATE shard_2...; COMMIT; — shard 1 commits, shard 2 fails. Data is now inconsistent | Avoid cross-shard transactions. If unavoidable, use two-phase commit (expensive) or the Saga pattern (compensating transactions) |
SELECT EXTRACT(EPOCH FROM (NOW() - pg_last_xact_replay_timestamp())) on PostgreSQL replicas. Set alerts at 5 seconds (warning) and 30 seconds (critical). Lag usually spikes during bulk writes or vacuum operations.synchronous_standby_names = 'ANY 1 (...)' not 'FIRST 1 (...)'. With ANY 1, any one synchronous replica confirms the write. With FIRST 1, writes wait for the first replica in the list — if it's down, writes block even if other replicas are available.SELECT * FROM users WHERE created_at BETWEEN '2024-01-01' AND '2024-01-31' hits every shard. If you need range queries, consider range-based sharding or maintain a separate time-series .max(shard_latency) not sum(shard_latency).Replication topology design: Design a PostgreSQL replication topology for a read-heavy social media app with 10,000 QPS reads and 500 QPS writes. How many replicas? Synchronous or async? Which queries go where? Draw the topology and justify each decision.
Shard key evaluation: Given a food delivery app with tables: users, restaurants, orders, order_items, deliveries, reviews — evaluate user_id vs restaurant_id vs order_id as the shard key. For each, list which queries are single-shard and which are cross-shard.
Implement a shard router: Write a class ShardRouter that uses consistent hashing to route queries to shards. It should: (a) accept SQL + shard key, (b) route to the correct shard, (c) support a scatterGather method for cross-shard COUNT/SUM queries, (d) handle shard connection failures gracefully by retrying on a different shard if the query wasn't shard-specific.
Simulate replication lag: Write a script that: (a) writes 10,000 rows to a primary, (b) measures replication lag on a replica every 100ms, (c) plots lag over time, (d) identifies the maximum lag and time-to-converge. Then add a slow query on the primary (e.g., pg_sleep(5) inside a transaction) and observe how it affects replication.
Design an online resharding process: Design a step-by-step process to add a new shard to a 4-shard cluster without downtime. Cover: how data moves, how to ensure consistency during the move, how to switch traffic, and how to rollback if something goes wrong. Reference Vitess's MoveTables workflow as inspiration.
Implement a CQRS system with PostgreSQL logical replication: Set up PostgreSQL logical replication to stream changes from the primary to a separate "read model" database with a different schema optimized for queries. Write the consumer that transforms WAL events into read-model updates. Handle schema evolution (new columns, new tables) without breaking the replication stream.
Q: What's the difference between database replication and sharding? A: Replication creates copies of the same data on multiple nodes. It primarily improves read throughput and provides high availability — if the primary fails, a replica can be promoted. Sharding splits the dataset into distinct subsets, each on a different node. It primarily improves write throughput — writes are distributed across shards. Replication is about copies of the same data; sharding is about splitting data.
Q: What is replication lag, and why does it matter? A: Replication lag is the delay between a write being committed on the primary and that write becoming visible on replicas. It matters because: (1) users who wrote data might not see it when reading from a replica ("I updated my profile but it shows the old one"), (2) business logic that reads from replicas might make decisions based on stale data, (3) monitoring dashboards may show outdated metrics. Lag is typically milliseconds in a single region but can spike during bulk operations or network issues.
Q: Why is the shard key the most important decision in sharding? A: The shard key determines which queries hit one shard vs all shards. A good key keeps 95%+ of queries on a single shard, enabling the system to scale linearly. A bad key (like timestamp for a social feed) concentrates all recent traffic on one "hot shard," defeating the purpose. Changing the shard key later requires migrating all data — a massive, risky operation. The shard key should match your dominant access pattern, not your data distribution.
Q: Describe the tradeoffs between synchronous and asynchronous replication. When would you use each? A: Synchronous replication guarantees the replica has the data before the write returns to the client. This means zero data loss on primary failure — the replica can take over with all committed data. The cost is write latency (primary + replica + network) and the risk that writes block if the synchronous replica goes down. Use synchronous for: financial transactions, critical configuration data, systems where data loss is unacceptable. Asynchronous replication writes locally, returns immediately, and ships data to replicas in the background. This gives fast writes but risks data loss if the primary crashes before replication completes. Use asynchronous for: social media likes, analytics, logs, user preferences — data where losing the last 1-2 seconds of writes is acceptable. The most common production pattern: synchronous to one replica (durability), asynchronous to the (read throughput).
Q: How would you handle a celebrity user problem in a sharded database? A: The "celebrity problem" is when one shard key (e.g., a user with 100M followers) generates 1000× the traffic of an average key, creating a hot shard. Solutions: (1) Denormalized feed cache: Pre-generate the celebrity's content feed and cache it in Redis/. Followers read from the cache, not from the celebrity's shard. (2) Separate shard: Give the celebrity their own dedicated shard with more resources. The shard router maps that specific user_id to the dedicated shard. (3) Lazy materialization: Instead of reading the celebrity's data on every follower request, push the celebrity's new content to follower timelines asynchronously (fan-out on write). (4) Read-only replicas for hot shards: Add extra read replicas specifically for hot shards, giving them more read capacity. The common thread: move reads off the hot shard through caching, pre-computation, or dedicated infrastructure.
Q: You need to add a new shard to a running system with 500M users sharded by user_id using hash(user_id) % 4. What happens, and how do you handle it?
A: With modulo-based sharding (hash(user_id) % 4 → hash(user_id) % 5), ~80% of data needs to move to a different shard. This is essentially a full data migration — unacceptable for a live system. The better approach: use consistent hashing from the start. With consistent hashing, adding shard 5 only moves ~20% of data (the portion of the ring assigned to the new shard). If you're already on modulo hashing and need to add a shard: (1) Set up the new shard with replication from existing shards. (2) Write a migration script that copies data to the correct shard based on the new hashing scheme. (3) During migration, dual-write: write to both the old shard (modulo 4) and the new shard (consistent hashing). (4) Backfill historical data. (5) Verify consistency between old and new. (6) Switch reads to the new scheme. (7) Switch writes to the new scheme. (8) Decommission old sharding. This is a months-long project requiring careful planning. This is why you should use consistent hashing from day one of sharding.
Database scaling is the highest-stakes engineering challenge because the database is the last component you want to fail. Replication (primary-replica) gives you read scalability and high availability — synchronous for zero data loss, asynchronous for performance, and you probably need both in a tiered setup. Sharding splits your data across independent instances to scale writes, but the shard key is a decision you'll live with for years — it must match your dominant access pattern. Consistent hashing minimizes data movement when your shard topology changes. Cross-shard queries are expensive; design your schema so most queries hit one shard. Instagram's user_id sharding and Vitess's transparent resharding are the production patterns worth studying. Before you shard, exhaust every other option — most teams shard years too early and pay the complexity tax for scale they'll never reach.
What is the primary benefit of database replication? A) Faster writes B) Improved read throughput and high availability C) Smaller database size D) Automatic query optimization
Which shard key is likely to create a "hot shard" problem? A) user_id with hash-based sharding B) created_at timestamp with range-based sharding (all recent data on one shard) C) A randomly generated UUID D) A composite key of user_id + order_id
What does synchronous_commit = remote_apply mean in PostgreSQL replication?
A) Writes are applied locally and sent later B) The primary waits for the replica to receive, write to disk, AND apply the WAL before confirming the write C) Replication is disabled D) The replica becomes the primary
With consistent hashing and 150 virtual nodes per server, adding a 5th server to a 4-server cluster moves approximately what percentage of keys? A) 75% B) ~20% (1/5 of the ring) C) 0% D) 100%
Why did Instagram choose user_id as their shard key? A) It provides the most even data distribution B) Most queries are scoped to a single user — 95%+ of queries hit exactly one shard C) It's the simplest to implement D) It was required by their
When should you implement database sharding? A) As soon as you launch your product B) When you have more than 1,000 users C) When write throughput exceeds what a single well-tuned database instance can handle, after exhausting vertical scaling, read replicas, and query optimization D) Always — it's the standard architecture
What happens to auto-increment IDs when you shard a database without using UUIDs or a global ID service? A) Everything works fine — each shard auto-increments independently B) Duplicate IDs across shards, making data merging impossible C) IDs automatically become globally unique D) Only the first shard generates IDs