Replication, partitioning and sharding

Two different reasons to spread data

A single database server eventually runs out of something: availability (it is one machine that can fail), read capacity, write capacity, storage, or proximity to users on other continents. There are two complementary ways to spread data over several machines, and it is important not to confuse them. Replication keeps a copy of the same data on several nodes, for availability, read scaling and lower latency near users. Partitioning, called sharding in many products, splits the data so that each node holds a subset, for write throughput and storage beyond one machine. Real systems usually do both: each partition is replicated.

Both have serious consequences for correctness. Replication raises the question of what a reader sees while copies disagree, which is the subject of consistency models, CAP and PACELC. Partitioning raises the question of what happens to queries, indexes and transactions that span partitions. This section explains the mechanisms in enough depth to judge a database's claims and to design a shard key.

Leader-follower replication

The most common scheme is leader-follower replication (also called primary-replica; older documentation says master-slave). One node, the leader, accepts all writes. It records every change in its log (PostgreSQL's write-ahead log, MySQL's binary log) and streams that log to followers, which apply the changes in the same order. Reads can go to the leader or to followers. PostgreSQL streaming replication, MySQL replication, SQL Server availability groups, MongoDB replica sets and most managed cloud databases work this way.

The key design choice is when the leader reports a commit as successful. With synchronous replication it waits until a follower confirms it has the change, so a committed write survives the loss of the leader; the price is that every commit includes a network round trip, and if the synchronous follower is down, writes stall. With asynchronous replication the leader confirms immediately and ships the change afterwards; commits are fast and followers can be far away, but if the leader dies, the last changes that had not reached any follower are lost even though the client was told they were committed. A common compromise, called semi-synchronous in MySQL, waits for one follower and lets the rest be asynchronous. In the bank, losing an acknowledged payment is unacceptable, so at least one synchronous copy (or a consensus-based database) is normal; for the food-delivery menu database, a few seconds of potential loss may be an acceptable price for speed. This is the RPO trade-off from availability and reliability in concrete form.

Replication lag and the anomalies it causes

Asynchronous followers are usually milliseconds behind, but under load, during a large batch update or across a slow network, the replication lag can grow to seconds or minutes. Reading from followers then produces anomalies that users notice. Douglas Terry and colleagues defined the first two guarantees below, read-your-writes and monotonic reads, as session guarantees in 1994; the third matters most when data is also partitioned.

None of these anomalies exists if every read goes to the leader, so the simplest fix is often to send reads that must be fresh to the leader and use followers only for reads that tolerate staleness: product listings, reports, search feeds. Design this deliberately per query instead of routing all reads to replicas by default.

Failover and split brain

Failover replaces a failed leader with a follower. It has three steps: detect that the leader has failed (usually by a timeout), choose the new leader (ideally the most up-to-date follower, often by a consensus protocol such as Raft among the nodes or an external coordinator such as etcd or ZooKeeper), and redirect writes to it. Each step can go wrong. A timeout that is too short fails over during a brief network blip or load spike, which makes things worse; one that is too long extends the outage. With asynchronous replication, writes the old leader had not yet shipped are lost, and if the old leader comes back and still holds them, they conflict with new writes.

The most dangerous failure is split brain: two nodes both believe they are the leader and both accept writes, so the data diverges in ways that may be impossible to merge automatically. It typically follows a network partition in which the old leader is cut off but alive. Protection needs a quorum (a node may lead only while a majority agrees), fencing (the new leader's authority is enforced by an ever-increasing token that storage checks, or the old leader is forcibly powered off, a technique nicknamed STONITH: shoot the other node in the head), and clients that follow the current leader reliably.

GitHub's incident of 21 October 2018 shows how hard this is in practice. A network fault cut the connection between its US East Coast hub and its main East Coast data centre for 43 seconds. Its failover tooling, working as designed, promoted West Coast database primaries. When connectivity returned, both coasts held writes the other lacked, so GitHub could not simply fail back; to protect data integrity it chose to restore and reconcile, and service was degraded for 24 hours and 11 minutes. Automatic failover across regions had turned a 43-second blip into a day-long recovery.

Multi-leader and leaderless replication

In multi-leader replication several nodes accept writes and replicate to each other. It is used for multiple data centres (each region writes locally, so users avoid cross-ocean latency), for clients that work offline (a calendar or notes app is effectively a leader on each device) and for collaborative editing. The unavoidable problem is write conflicts: two leaders change the same record concurrently. Resolutions include last write wins (simple, but silently discards data and depends on clocks), merging values, keeping both and asking a person, or using data types designed to merge automatically (CRDTs, conflict-free replicated data types). The most reliable approach is to avoid conflicts: give each record a home region that alone accepts its writes, such as a customer's home country.

Leaderless replication, popularised by Amazon's Dynamo paper (DeCandia and colleagues, 2007) and used in Apache Cassandra, ScyllaDB and Riak, has no leader at all. The client, or a coordinator node acting for it, sends each write to all N replicas of a key and treats it as successful when W of them acknowledge; it sends each read to replicas and waits for R answers, taking the newest version. If R + W > N, every read set overlaps every successful write set in at least one replica, so a read normally sees the latest acknowledged write. Replicas that missed writes are repaired by read repair (the coordinator writes back the newest version when it sees a stale one) and background anti-entropy that compares replicas. Cassandra exposes these choices per request as consistency levels such as ONE, QUORUM and ALL.

Quorums are often over-trusted. R + W > N does not by itself make the system linearisable: concurrent writes, writes that failed on some replicas but succeeded on others, sloppy quorums (writing to substitute nodes during failures) and last-write-wins with skewed clocks all allow stale or lost data. Chapter 6 works through quorum arithmetic and these edge cases in CAP and PACELC. Leaderless stores are excellent for high write availability across failures and data centres; they are a poor fit for invariants such as never sell the same seat twice unless they offer a lightweight transaction or consensus feature for those operations.

Partitioning: by key range or by hash

Partitioning assigns each record to one partition by its partition key (or shard key), and the aim is to spread data and load evenly. A partition that receives disproportionate load is a hot spot, and an uneven spread is called skew. There are two basic schemes.

Key-range partitioning gives each partition a contiguous range of keys, like volumes of an encyclopaedia. Range scans are efficient because adjacent keys are stored together (all orders for October in one or two partitions), and partitions can be split when they grow, as in Bigtable, HBase, Spanner and MongoDB's range sharding. The danger is a key that increases with time: if the key is a timestamp or a sequential order number, every new write lands on the last partition, which becomes a hot spot while the others sit idle. Hash partitioning applies a hash function to the key and assigns ranges of hash values to partitions, spreading even sequential keys uniformly. The price is that range queries on the key now have to ask every partition. Many stores combine the two with a compound key: Cassandra hashes the partition key to choose the node and sorts rows within the partition by clustering columns, so (customer_id, placed_at) puts one customer's orders together, in time order, on one node.

Consistent hashing and rebalancing

The obvious way to map hashes to nodes, hash(key) mod N, has a fatal flaw: when N changes, almost every key moves. Going from 4 to 5 nodes relocates about 80% of the keys (any key whose remainder differs for 4 and 5), and from 10 to 11 about 91%, which means copying almost the whole data set just to add one machine. Consistent hashing, introduced by David Karger and colleagues in 1997 for web caches, places both nodes and keys on a circular hash space; each key belongs to the next node clockwise. Adding a node takes over only the keys between it and its predecessor, about 1/N of the total, and removing one hands its keys to its neighbour. With few nodes the arcs are uneven, so each physical node is given many virtual nodes (tokens) at different points on the ring, which evens out the load and lets a new node take a little from every existing node. Dynamo and Cassandra use this scheme.

Two other rebalancing strategies are common. With a fixed number of partitions, you create many more partitions than nodes at the start (say 1,024 for 10 nodes) and assign whole partitions to nodes; adding a node moves a few whole partitions and never re-hashes keys. Elasticsearch shards, Couchbase vBuckets and many hand-built sharding schemes use this, and the main decision is choosing a partition count large enough for future growth. With dynamic partitioning, as in HBase and MongoDB, a partition splits when it exceeds a size threshold and the halves can move to different nodes. Whatever the scheme, rebalancing moves large amounts of data over the network while the system is serving traffic, so many teams prefer that a human approves each move, and they throttle it to protect latency.

Requests also need routing to the right partition: either any node accepts a request and forwards it (Cassandra nodes learn the ring by gossip), or a routing tier knows the partition map (MongoDB's mongos), or the client library holds the map. The map itself is usually kept in a coordination service such as etcd or ZooKeeper or in the database's own consensus group.

Hot keys and skew

Hashing spreads keys evenly, not load. If one key is far more active than the rest, its partition is hot whatever the scheme: a celebrity's account in a social network, a national chain's flagship restaurant during a promotion, or a bank's own settlement account that touches every payment. Managed stores make the limits explicit; Amazon DynamoDB, for example, documents a ceiling per partition of 3,000 read units and 1,000 write units a second. Mitigations include caching hot reads, splitting a hot key by appending a small random suffix (writes to restaurant:88#0 to restaurant:88#9, at the cost of reading all ten and combining), moving a very large tenant to dedicated partitions, and aggregating counters in memory before writing them. All of them add complexity, so first measure which keys are actually hot.

Secondary indexes across partitions

Partitioning works smoothly for lookups by the partition key. Queries by anything else need a secondary index, and there are two ways to partition one. A local index (document-partitioned) lives with each partition and covers only that partition's rows. Writes update a single partition, but a query by the indexed attribute must be sent to every partition and the results merged (scatter-gather), so its latency is set by the slowest partition and its cost grows with the number of partitions. A global index (term-partitioned) is itself partitioned by the indexed value, so a query reads one index partition; but a single write may now need to update index partitions on other nodes, which is usually done asynchronously, so the index can briefly lag the data. DynamoDB shows both: a local secondary index shares the table's partition key, while a global secondary index has its own and is eventually consistent.

Common mistakes

Treating replicas as backups. A replica faithfully copies a mistaken DELETE within milliseconds; backups and point-in-time recovery are still needed (Chapter 0). Reading from replicas by default without deciding which queries tolerate lag. Trusting automatic failover that has never been rehearsed, or that can promote a node while the old leader still accepts writes. Sharding too early: a sharded database makes joins, transactions, schema changes and operations harder, and modern single nodes are very large; exhaust indexing, caching, read replicas, archiving and vertical scaling first. Sharding by a monotonically increasing key, which concentrates every insert on one partition. Forgetting secondary access paths, discovering after the split that an important query must now ask every shard. As the rest of this chapter shows, the good alternatives often involve derived data: read models, caches and analytical stores, each built for one kind of query.