Distributed Storage Engines and Data Partitioning

3.1 Storage Engine Internals: B-Trees and LSM-Trees

A storage engine trades three quantities against each other, and no design escapes the trade. Read amplification is the bytes read from the device per byte returned. Write amplification is the bytes written per byte the user wrote. Space amplification is the bytes stored per byte of live data. The RUM conjecture states the rule formally: a structure optimizes at most two of read, update, and memory overhead.

The B+tree write path

A B+tree stores keys in internal nodes and key-value pairs in leaves, and it links the leaves so that a range scan walks them in order. Every leaf sits at the same depth. The unit of I/O is the page, which is 8 kibibytes in PostgreSQL and 16 kibibytes in InnoDB.

An update finds the leaf, modifies it in place, and marks the page dirty. The page reaches the device later during a checkpoint. Before any of that, the change goes to the write-ahead log, because the log record must be durable before the page moves.

That path produces the write amplification number that surprises people. An update of one 100 byte row dirties an 8 kibibyte page, and the page gets written whole. Under full-page logging, which PostgreSQL uses after each checkpoint to protect against torn pages, the same update also writes a full page image into the log. One small update therefore costs 16 kibibytes of device traffic, which is 160 times the logical write.

An insert into a full node splits it, which allocates a new node, moves half the entries, and pushes a separator key into the parent. Splits can cascade to the root, and that is the only way the tree gets taller. Deletes leave underfilled nodes, and most engines do not merge them, because merges are expensive and rarely necessary.

Concurrency uses latch crabbing. A thread holds a latch on a node, takes a latch on the child, and releases the parent as soon as it knows the child cannot split. B-link trees remove most of the coupling by adding a right-link pointer and a high key, so a reader that arrives during a split follows the link instead of holding a latch. PostgreSQL uses a B-link tree for this reason.

The LSM write path

A log-structured merge tree turns random writes into sequential ones. A write goes to the write-ahead log for durability and to an in-memory table, which is usually a skip list or a concurrent B-tree. When the memtable fills, it becomes immutable, a new one takes over, and a background thread flushes the old one to an immutable sorted file.

  write --> [ WAL, sequential append ]   durability
        --> [ active memtable ]          skip list, in memory
                    | full
                    v
            [ immutable memtable ] --flush--> sequential write
                                                  |
  L0  [sst][sst][sst][sst]   overlapping ranges, up to 4 files
       |  compaction trigger: L0 file count
       v
  L1  [sst][sst][sst] ....   non-overlapping, 256 MB budget
       |  10x size ratio
       v
  L2  [sst][sst][sst] ....   2.5 GB budget
       |  10x
       v
  L3  ....                   25 GB, and so on

  Point read order: memtable, immutable memtable, every L0 file,
  then exactly one file per level below. A Bloom filter per file
  skips the files that cannot hold the key.

A sorted string table, or SSTable, holds sorted key-value pairs plus an index block, a Bloom filter block, and a footer. The file is immutable, which is what makes the design safe: a reader never sees a partially written file, and no locking is needed between readers and the flusher.

Compaction strategies

Leveled compaction keeps each level sorted and non-overlapping, and merges a file from level N into the overlapping files of level N plus 1. Each key exists roughly once per level, so space amplification stays near 1.1. Each key gets rewritten once per level it passes through, so write amplification runs 10 to 30.

Tiered compaction, which Cassandra calls size-tiered, collects several files of similar size and merges them into one larger file. A key moves fewer times, so write amplification falls to 4 to 10. Several versions of a key coexist, so space amplification rises to 2 or 3, and a read consults more files.

Key-value separation, which the WiscKey design introduced and RocksDB ships as BlobDB, stores large values in a separate log and keeps only pointers in the tree. Compaction then moves pointers rather than values, so write amplification falls sharply. The costs are one extra read to fetch the value and a separate garbage collection process for the value log.

Bloom filters are what make LSM point reads viable. Each file carries a filter over its keys, and a read tests the filter before reading the index. At 10 bits per key with a 1 percent false positive rate, a read that misses skips almost every file and costs one device read. Without filters, a point read touches every level.

The write-ahead log

The rule is in the name: the log record reaches stable storage before the modified page does. Recovery replays the log and reaches a consistent state.

ARIES is the design that almost every relational engine follows. Write-ahead logging gives durability. Repeating history during redo means that recovery first restores the exact state at the crash, including uncommitted changes, and then undoes the uncommitted ones. Logging the undo actions themselves makes recovery idempotent, so a crash during recovery is survivable.

Log records are physical, logical, or both. A physical record names a page, an offset, and the bytes, which is idempotent and large. A logical record names an operation, which is small and not idempotent. ARIES uses physiological logging, logical within a page and physical between pages, with a log sequence number on every page so that redo applies each record exactly once.

The cost of durability is one fsync per commit.

Group commit is why the last column differs. The engine collects commit records from many transactions, writes them in one device operation, and completes all of them together. Per-transaction latency stays at one fsync, and throughput rises by the batch factor.

The consumer SSD row is a data loss trap. Without power loss protection, the drive acknowledges a write when it lands in a volatile cache, and a power cut loses it. The database did everything correctly and the data is gone. Any machine holding a primary copy needs a drive with power loss protection or a replication factor that tolerates the loss.

3.2 Data Partitioning and Sharding

Partitioning splits data across independent groups so that each group holds a fraction. It is the answer when one machine cannot hold the data, cannot serve the write rate, or cannot meet a latency target from one location.

The partition key decides everything

Three properties matter. The key must distribute load evenly, so that no partition becomes hot. It must keep together the data that queries read together, so that most queries touch one partition. And it must appear in almost every query, because a query without it fans out to every partition.

Hash partitioning distributes well and destroys range queries. Range partitioning preserves range queries and creates a hot spot when the range is a timestamp, because every write lands on the newest partition. A composite key gives both: hash the high-order component and range-order the low-order one, which is exactly what a partition key plus sort key provides.

The last row is usually right for this domain, and the failure column is the point. Global analytics must not run against the transactional store. Stream changes to a column store through change data capture and let an analytical engine answer them.

Consistent hashing and virtual nodes

A modulo hash over the node count remaps almost every key when the count changes, which empties every cache and moves nearly all the data. Consistent hashing fixes that. Nodes and keys both map onto a ring, and a key belongs to the first node clockwise from it. Adding a node takes over only the arc between it and its predecessor.

Plain consistent hashing balances badly. With N nodes placed randomly on a ring, the largest arc is several times the average, so one node holds several times the data. Virtual nodes fix it: each physical node claims many points on the ring, typically 128 to 256. The law of large numbers then flattens the distribution, and removing a node spreads its load across many others rather than dumping it on one neighbor.

  WITHOUT VNODES: 4 nodes, 4 ring points
    0 ......... A ........................ B ... C ................ D ... 2^64
                |<---- B owns this huge arc ---->|
    B holds ~4x the average. B's failure dumps all of it on C.

  WITH VNODES: 4 nodes, 256 points each
    0 .A.C.B.D.A.B.D.C.A.D.B.C.D.A.C.B.A.C.D.B.A.B.C.D. ...... 2^64
    Each node owns ~256 small arcs. Ownership is even to within a
    few percent. B's failure splits across A, C and D in proportion
    to how often each follows a B point on the ring.

  Rebalancing moves only the arcs that change hands: roughly 1/N of
  the data, rather than the (N-1)/N that a modulo hash would move.

Two variants improve on the ring. Rendezvous hashing, also called highest random weight, computes a hash of the key combined with each node identifier and picks the maximum. It needs no ring structure and gives the same minimal disruption property. Jump consistent hash maps a key to one of N buckets with no memory at all, in a few lines of arithmetic, and it only supports adding or removing buckets at the end of the range.

Pre-splitting and rebalancing

Resharding is the operation that teams underestimate. Moving from 8 shards to 16 remaps half the keys under a naive hash, which means moving half the data while serving traffic.

Pre-splitting removes the problem at design time. Create many more logical partitions than physical nodes, for example 1,024 logical partitions across 8 nodes. Growing the cluster then reassigns whole logical partitions to new nodes with no rehashing and no key movement within a partition. Kafka, Elasticsearch, and several distributed databases all work this way, and the logical partition count becomes the number that is hard to change later rather than the node count.

Dynamic splitting takes the other approach. A partition that exceeds a size or load threshold splits in two, and adjacent small partitions merge. Range-based systems such as HBase, Bigtable, and CockroachDB do this automatically, and it handles skew that no static scheme anticipates. The cost is that partition boundaries move at run time, so the routing layer must track them.

Routing a query

Three designs exist for getting a request to the right partition.

A routing tier sits in front and holds the partition map. Clients are simple and the tier is one more hop and one more thing to operate. A partition-aware client fetches the map and routes directly, which removes the hop and couples every client to the map format. Or any node accepts any request and forwards it, which is simplest for the client and doubles the internal traffic.

The map itself needs a consistent source. Most systems keep it in a small strongly consistent store, which is a consensus group, and cache it in the routing layer with a version number. A request that arrives at the wrong node gets a redirect carrying the current version, so a stale client corrects itself within one round trip.

Cross-partition queries are where the design shows. A query that touches every partition costs the latency of the slowest one. Tail latency amplification prices it: with 100 partitions at a 100 millisecond p99, 63 percent of fan-out queries hit at least one slow partition. Scatter-gather works at small partition counts and degrades sharply as the count grows. Push predicates and aggregation into each partition so that only small results travel, and cap the fan-out with a secondary index or a denormalized read model.

3.3 Multi-Leader and Active-Active Replication

Single-leader replication gives a simple ordering story and one write bottleneck. Multi-leader replication accepts writes at several places, which buys local write latency and offline operation, and which creates conflicts that something must resolve.

Quorum arithmetic

A leaderless system stores each item on N replicas, writes to W of them, and reads from R. When W plus R exceeds N, every read quorum overlaps every write quorum, so a read observes the latest completed write.

The arithmetic is simple and the implications are not. With N equal to 3 and W and R both 2, a write survives one node failure and a read still sees it. Setting W to 1 makes writes fast and loses data when that node dies before replicating. Setting R to 1 makes reads fast and returns stale values. Setting W equal to N makes writes unavailable when any node is down.

Quorum overlap guarantees only that a read sees the value. It does not repair the replicas that missed the write, and it does not order concurrent writes. Two more mechanisms handle those.

Anti-entropy with Merkle trees

Read repair fixes divergence that a read happens to discover: the coordinator notices that replicas disagree and writes the newest value back. It only fixes data that someone reads, so cold data drifts forever.

Hinted handoff covers a short outage. A coordinator that cannot reach a replica stores the write locally with a hint naming the intended replica, and delivers it when the replica returns. It bounds the divergence window and does not help after a long outage, because hint storage is finite.

Full reconciliation needs to compare two large key ranges cheaply, and a Merkle tree does that. Each leaf hashes a small key range, each internal node hashes its children, and the root hashes everything. Two replicas compare roots. Equal roots mean the ranges are identical and the comparison is over. Different roots mean they descend into the differing subtree, and the number of comparisons is logarithmic in the range size rather than linear.

  replica A                              replica B
        root: 8f3a                             root: 8f3a  -> equal, stop
       /            \                         /            \
    a1: 4c2         a2: bb7                a1: 4c2       a2: bb7

  After a write to B that A missed:
        root: 8f3a                             root: 21dd  -> differ
       /            \                         /            \
    a1: 4c2       a2: bb7                  a1: 4c2       a2: 9e0
       equal ^        differ ^                             /      \
                                                     b1: 33f    b2: 71c
                                                        equal      differ
                                                                 /     \
                                                            keys 900-999

  Two replicas holding 10 million keys with one divergent key exchange
  about 24 hashes, not 10 million keys.

Last-write-wins and what it costs

The simplest conflict resolution keeps the write with the highest timestamp. It is easy to implement and it silently loses data.

The loss happens inside the clock skew window. Two writes arrive within the skew, the one with the larger timestamp wins, and the larger timestamp belongs to the machine with the faster clock rather than to the later write. The losing write returned success to its client and then vanished. No error is raised and no metric moves.

Last-write-wins is acceptable for data where losing a concurrent update is harmless: a cache entry, a presence indicator, a last-seen timestamp. It is unacceptable for anything a user typed. Cassandra makes it the default, which is why its documentation carries warnings about it, and why teams that treat a Cassandra table as a general-purpose store discover the behavior during an incident.

The alternative is to keep both versions as siblings and let the application merge, which is what Riak and Dynamo-style systems offer. That pushes the decision to the layer that understands the data, and it requires the application to handle a read that returns two values.

Conflict-free replicated data types

A CRDT is a type whose merge function is commutative, associative, and idempotent. Replicas that receive the same set of updates in any order, with any duplication, converge to the same state. No coordination is needed and no conflict resolution is required, because the merge is total.

Two families exist. State-based, or convergent, types send the whole state and merge with a join operation. Operation-based, or commutative, types send operations and require the delivery layer to deliver each exactly once, in causal order.

CRDTs power collaborative editors, offline-first mobile applications, and multi-region caches. The limit is expressiveness. Any invariant that spans objects, such as "the total never goes below zero", cannot be a CRDT, because two replicas can each decrement legally and the merged result breaks the rule. Where such an invariant exists, coordination is unavoidable.

The second limit is metadata growth. An OR-Set carries a tag per live element, and a sequence CRDT carries position identifiers that accumulate with edit history. A long-lived collaborative document can hold more metadata than content. Production systems run periodic garbage collection that requires agreement on a stable point, which reintroduces a small amount of coordination.

3.4 Multi-Version Concurrency Control at Scale

MVCC keeps several versions of each row so that a reader sees a consistent snapshot without blocking a writer. Removing the read lock is the single most important idea in modern database concurrency.

Snapshots and visibility

A transaction gets a snapshot at its start, which is a rule for deciding which versions it can see. Each row version carries the identifier of the transaction that created it and the one that deleted it. Visibility is a function of those identifiers and the snapshot.

PostgreSQL forms a snapshot from three values: the lowest still-running transaction identifier, the next identifier to be assigned, and the list of in-progress identifiers between them. A version is visible if its creating transaction committed before the snapshot and its deleting transaction did not.

Where the old versions live drives the operational behavior, and the engines differ.

PostgreSQL stores every version in the table itself. An update writes a new row version and marks the old one dead. The consequence is table bloat: dead rows occupy space until vacuum reclaims it. Autovacuum runs continuously, and a table with a high update rate and an under-configured autovacuum grows without bound.

InnoDB stores the current version in the clustered index and old versions in a separate undo log. A reader that needs an old version walks the undo chain backward. The table stays compact, and a long-running read transaction makes the undo log grow and every read in that transaction slower.

The anomalies that snapshot isolation permits

Snapshot isolation prevents dirty reads, non-repeatable reads, and phantoms. It permits write skew, and write skew is the anomaly that surprises everyone.

Two transactions read an overlapping set, each makes a decision based on what it read, and each writes a disjoint set. Neither write conflicts, both commit, and the combined result breaks an invariant that each transaction alone preserved.

-- Invariant: at least one doctor stays on call for each shift.
-- Alice and Bob run this concurrently under snapshot isolation.

BEGIN;
SELECT count(*) FROM doctors
 WHERE on_call = true AND shift_id = 42;      -- both read 2

-- Each transaction concludes that removing itself is safe.
UPDATE doctors SET on_call = false
 WHERE name = 'alice' AND shift_id = 42;      -- Bob updates 'bob'
COMMIT;

-- Both commit. Zero doctors on call. No row conflicted, so no engine
-- that only detects write-write conflicts will catch it.
--
-- Fixes, in order of preference:
--   SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;   -- SSI detects it
--   SELECT ... FOR UPDATE on the rows the decision reads
--   a materialized conflict row that both transactions update
--   a database constraint that expresses the invariant directly

Serializable snapshot isolation fixes it without locks. It runs snapshot isolation and tracks read-write dependencies between concurrent transactions. When it detects the dangerous structure that can produce a non-serializable outcome, it aborts one transaction. PostgreSQL implements this for its SERIALIZABLE level. The cost is aborts under contention, so the application must retry, and the retry loop is not optional.

Garbage collection of old versions

Old versions must be removed, and the rule is the same everywhere: a version can go once no possible reader needs it. The cleanup horizon is therefore set by the oldest active snapshot in the system.

That single fact produces the most common MVCC failure in production. One transaction left open for hours holds the horizon back. In PostgreSQL, vacuum cannot remove any row version newer than that snapshot, so bloat grows across every table in the database, not only the one the transaction touched. In InnoDB the undo log grows for the same reason, and every read in the long transaction walks a longer chain.

In a distributed store the horizon becomes a cluster-wide value. Every node must know the oldest snapshot anywhere before it can collect, which means a periodic exchange of the minimum active timestamp. One node with a stuck transaction blocks collection on every node.

Version storage in distributed engines

A distributed MVCC engine has an extra problem: the snapshot timestamp must be meaningful across nodes. Three approaches are in production.

A centralized timestamp oracle hands out monotonically increasing timestamps. It is simple and correct, and it is a single point that every transaction contacts. TiDB uses a placement driver for this, and batching keeps the cost low.

Hybrid logical clocks let each node generate timestamps locally while preserving causality. CockroachDB uses this, and it carries an uncertainty interval: a read that encounters a value inside its uncertainty window must restart with a higher timestamp, which shows up as retry traffic under contention.

TrueTime, in Spanner, makes the uncertainty explicit and waits it out before releasing locks. That buys external consistency at the cost of a few milliseconds per read-write transaction, and it requires atomic clocks and GPS receivers in every datacenter.

Read-only transactions are where the design pays off. All three schemes let a read-only transaction pick a timestamp in the past and read a consistent snapshot from any replica, with no locks, no coordination, and no effect on writers. A large fraction of production query load is read-only, so this is the optimization that makes distributed SQL viable.

Sources and further reading

Alex Petrov, "Database Internals", for engine mechanics. Andy Pavlo's CMU 15-445 and 15-721 lecture series. C. Mohan and others, "ARIES", ACM TODS 1992. Lu and others, "WiscKey", FAST 2016. Michael Cahill, Uwe Rohm, and Alan Fekete, "Serializable Isolation for Snapshot Databases", SIGMOD 2008. Giuseppe DeCandia and others, "Dynamo", SOSP 2007, for consistent hashing, hinted handoff, and Merkle tree anti-entropy. David Karger and others, "Consistent Hashing and Random Trees", STOC 1997. Marc Shapiro and others, "A comprehensive study of Convergent and Commutative Replicated Data Types". Martin Kleppmann's writing on CRDTs and on the hazards of last-write-wins.