Key-value store
A distributed key-value store is put(key, value) and get(key) over thousands of machines that never stop taking writes. Two layers make it: a storage engine per node that turns random writes into sequential ones (write-ahead log, sorted memtable, immutable SSTables merged by compaction, Bloom filters to skip files on reads), and a distribution layer that places each key on N replicas around a consistent-hash ring and lets each request choose R and W. With R + W > N reads overlap the latest acknowledged write; with smaller quorums you get lower latency and higher availability, and repair (read repair, hinted handoff, Merkle-tree anti-entropy) makes replicas converge. Writes are O(1) appends; reads are O(log n) per file checked.
The problem
Section titled “The problem”Shopping carts, session state, user preferences, feature flags, message metadata: huge volumes of small records, always looked up by key, never joined. Amazon built Dynamo (2007) because its cart had to accept an “add to cart” even while servers and links were failing. Cassandra, Riak, ScyllaDB and DynamoDB’s ancestors followed.
The goals pull against each other:
| Goal | What it means | Pulls toward |
|---|---|---|
| Always writable | Accept writes during node failures and partitions | Leaderless replication, sloppy quorums, conflicts resolved later |
| Read your data back | A read returns the latest acknowledged write | Larger quorums, a leader per key, consensus |
| Scale linearly | Twice the nodes, twice the throughput and storage | Partitioning with no central coordinator |
| Predictable latency | p99 in single-digit ms | Sequential I/O, no read-before-write, wait only for the fastest replicas |
The CAP theorem puts a name on the first two: during a network partition you choose between availability and consistency. Dynamo-style stores choose availability and let each request tune how much consistency it wants.
Requirements and scope
Section titled “Requirements and scope”| In scope (typical) | Ask about | |
|---|---|---|
| Functional | put, get, delete by key; values up to ~1 MB; optional TTL; per-request consistency level. |
Range scans by key prefix? Conditional writes (compare-and-set)? Secondary indexes? |
| Scale | 100k writes/s, 300k reads/s, 50 TB logical data, growing. | Read:write ratio? Record size distribution? Growth per year? |
| Latency | p99 under 10 ms for reads and writes inside a region. | Is cross-region replication needed, and synchronous or async? |
| Consistency | Tunable per request: eventual by default, quorum when asked. | Is read-your-writes required? Can two clients write the same key concurrently, and what should win? |
| Durability / availability | No acknowledged write lost if one node dies; writes accepted with a node or rack down. | How many simultaneous failures? Rack and zone awareness? |
| Out of scope | Multi-key transactions, joins, a query language. |
Back of the envelope
Section titled “Back of the envelope”Storage multiplies by N. Disk bandwidth multiplies by N and by the engine’s write amplification: compaction rewrites each byte many times. Watch the per-node disk bandwidth row.
What to say out loud: “150 TB with replication, roughly 75 nodes. The interesting number is disk write bandwidth: every logical write lands on three nodes and compaction rewrites it about ten times, so I’ll pick an engine and compaction strategy for writes, and leave headroom on disk for compaction to run.”
API and data model
Section titled “API and data model”put(key, value, context?, consistency = QUORUM) -> ok | timeoutget(key, consistency = QUORUM) -> [ (value, context) ] -- usually one; siblings on conflictdelete(key, context?) -> ok -- writes a tombstone
consistency ∈ { ONE, QUORUM, ALL, LOCAL_QUORUM } -- picks R or W for this request
Stored per replica, per key: key bytes; hashed to a ring position (token) value bytes, or TOMBSTONE version write timestamp (µs) -- last-writer-wins or a vector clock -- [(node, counter), ...] ttl optional expiryTwo decisions hide in that sketch:
getreturns a list. With vector clocks, two writes that didn’t see each other are both kept as siblings, and the client merges them (Dynamo’s cart unions the items). Thecontextthe client passes back on the nextputsays which versions it has seen. With last-writer-wins the list always has one element, and the losing write disappears silently.- Deletes are writes. You can’t just remove a key from one replica: another replica that missed the delete would “resurrect” it during repair. A delete writes a tombstone that must survive until every replica has seen it, then compaction drops it (Cassandra’s
gc_grace_seconds, default 10 days).
High-level design
Section titled “High-level design”Step through the flows. There’s no leader and no master: every node runs the same code, and any node can coordinate any request.
| Component | Role | Why it’s there |
|---|---|---|
| Smart client / driver | Knows the ring; sends a request straight to a replica when it can. | Saves a hop; spreads coordination over the cluster. |
| Coordinator | Any node. Finds the N replicas for the key, fans out, waits for R or W replies, resolves versions. | Leaderless: no single node whose failure stops writes to a key. |
| Replicas | Each runs a local storage engine: WAL + memtable + SSTables. | Durable, write-optimised storage per node. |
| Gossip + failure detector | Every second, each node swaps membership and ring state with a few peers; a phi-accrual detector marks nodes as suspect. | Membership without a central registry. |
| Hinted handoff | Stores writes for an unreachable replica and replays them on recovery. | Short outages don’t need a full repair. |
| Anti-entropy repair | Background Merkle-tree comparison per token range. | Catches everything else: long outages, dropped hints, bit rot. |
Deep dive: the storage engine on one node
Section titled “Deep dive: the storage engine on one node”A node must absorb tens of thousands of writes per second on its disks. Updating a B-tree in place means a random read and a random write per update. An LSM tree (log-structured merge tree) never updates in place:
- Append to the write-ahead log (WAL, Cassandra’s commit log), so a crash loses nothing that was acknowledged.
- Insert into the memtable, a sorted in-memory structure (a skip list or balanced tree; see Sorted containers).
- Flush a full memtable as an immutable, sorted file: an SSTable (sorted string table), with a sparse index and a Bloom filter. Then truncate the WAL.
- Compact in the background: merge SSTables, keep the newest version of each key, drop overwritten values and expired tombstones.
Drive it by hand. Put a few keys, delete one, read a key that’s in several files, then turn automatic flushing off and read a key that doesn’t exist.
The read path is the price of cheap writes. Newer data shadows older data, so a read checks from newest to oldest and stops at the first version it finds:
get(key): if key in memtable: return memtable[key] -- may be a tombstone: "not found" for table in L0 tables, newest first: -- L0 files can overlap each other if not table.bloom.might_contain(key): continue -- no disk I/O at all if found(table, key): return it -- index lookup + one block read for level in L1, L2, ...: -- deeper levels: non-overlapping ranges table <- the one file whose key range covers key -- binary search over file ranges if table.bloom.might_contain(key) and found(table, key): return it return not foundA Bloom filter with about 10 bits per key and 7 hash functions has a false-positive rate near 1%, so a read for a missing key touches the disk about once per hundred files it would otherwise check. The two compaction families trade the three amplifications differently:
| Size-tiered (STCS) | Leveled (LCS) | |
|---|---|---|
| How | Merge a few SSTables of similar size into one bigger one | Each level is ~10× the previous; a file is merged into the overlapping files of the next level |
| Write amplification | Low (about one rewrite per tier, a handful in total) | Higher (≈10× per level, 10-30× total) |
| Read amplification | Higher: many overlapping files | Low: about one file per level |
| Space amplification | Up to 2× during a big merge | ≈10% |
| Fits | Write-heavy, append-mostly | Read-heavy, many updates of the same keys |
Cassandra also has time-window compaction (for TTL’d time series, where whole files expire together). B-trees (InnoDB, Postgres) are the other side of the trade: one copy of each key, good reads and range scans, more random writes.
Deep dive: replication and quorums
Section titled “Deep dive: replication and quorums”Each key lives on N replicas. Each request picks how many must answer: W for a write, R for a read. If R + W > N, any read quorum shares at least one replica with any write quorum, so a read sees the latest acknowledged write. Tune it below, then take replicas down.
| Setting (N = 3) | Read sees latest ack? | Survives 1 down? | Latency | Use |
|---|---|---|---|---|
| W = 3, R = 1 | Yes | Reads yes, writes no | Writes wait for the slowest replica | Read-heavy, rarely written |
| W = 2, R = 2 | Yes | Yes | 2nd fastest replica | Common default (Cassandra QUORUM) |
| W = 1, R = 1 | Not guaranteed | Yes | Fastest replica | Highest availability; stale reads for milliseconds |
| W = 1, R = 3 | Yes | Writes yes, reads no | Reads wait for the slowest | Rare |
The coordinator’s job, as a building block:
coordinate_write(key, value, W): replicas <- preference_list(key, N) -- next N distinct nodes on the ring version <- now_micros() or context.increment(me) send (key, value, version) to every replica in parallel for any replica that is unreachable: store a hint for it on a healthy node wait until W acks or timeout -> ok | timeout -- a timeout is NOT "failed": some replicas may have it
coordinate_read(key, R): ask R replicas (one full value, the rest a digest) if digests all agree: return the value fetch full values, newest <- resolve(versions) -- LWW: max timestamp; vector clocks: keep all concurrent send newest to every replica that was behind -- read repair, in the background return newestDeep dive: partitioning, membership and conflicts
Section titled “Deep dive: partitioning, membership and conflicts”Partitioning. Keys are hashed onto a ring, and each node owns positions (tokens) on it. With one token per node, adding a node takes load from just one neighbour and node sizes vary widely. Virtual nodes give each physical node many tokens (Cassandra 4.0 defaults to 16, earlier versions 256), so load spreads evenly and a new node takes a little from everyone. The replicas of a key are the next N distinct physical nodes clockwise, skipping other vnodes of the same machine and, with rack awareness, the same rack. The load balancing page has a ring to play with.
Membership. Nodes gossip ring state every second. A phi-accrual failure detector turns heartbeat gaps into a suspicion level instead of a yes/no, so a slow GC pause isn’t treated like a dead machine. Adding a node streams the token ranges it now owns from the current replicas; the ring change itself is a small piece of metadata.
Conflicts. Two clients write the same key during a partition. What survives?
| Last-writer-wins (LWW) | Vector clocks | |
|---|---|---|
| Version | One timestamp per write | A list of (node, counter) pairs |
| Concurrent writes | Highest timestamp wins; the other is silently lost | Detected: neither descends from the other, both kept as siblings |
| Who merges | Nobody | The client, on the next read (or a CRDT does it automatically) |
| Clock dependence | Yes: a node with a fast clock wins every race | None (logical clocks) |
| Used by | Cassandra (per cell), DynamoDB global tables | Dynamo, Riak (dotted version vectors) |
The comparison at the heart of vector clocks:
descends(a, b): -- a has seen everything b has for each (node, count) in b: if a[node] < count: return false return true
resolve(a, b): if descends(a, b): keep a else if descends(b, a): keep b else: keep both as siblings -- concurrent; the application mergesDynamo truncates a clock to about ten entries, dropping the oldest, which can rarely turn a true descendant into a false conflict. The versioning page covers version numbers and history more generally.
Failure modes
Section titled “Failure modes”| What breaks | What you see | Design response |
|---|---|---|
| A replica is down for minutes | Writes still succeed at W; that replica falls behind | Hinted handoff; read repair on hot keys |
| A replica is down for days | Hints expire (Cassandra: 3 h window) | Run anti-entropy repair; replace the node and stream its ranges |
| Node crash | Unflushed memtable lost | Replay the WAL on restart; with periodic fsync (Cassandra’s default: every 10 s) the other replicas cover the gap |
| Clock skew between nodes | LWW drops the newer write | NTP with alerts; client-supplied timestamps; vector clocks where loss matters |
| Deleted data reappears | A replica missed the tombstone and repair copied the old value back | Repair every node within gc_grace_seconds; never drop tombstones earlier |
| Compaction falls behind | Read latency climbs; disk fills with SSTables | Throttle writes, add nodes, keep ≥ 50% free disk, pick leveled for read-heavy tables |
| Hot partition | One replica set at 100% while others idle | Better key design (add a bucket suffix); cache in front; split the hot key |
| Large partitions / wide rows | GC pauses, slow repairs | Cap values and partition sizes; bucket by time |
What it costs
Section titled “What it costs”| Operation | Time | Space / I/O |
|---|---|---|
| Write on one replica | O(1) WAL append + O(log m) memtable insert | One sequential append now; ≈ write-amp × bytes later through compaction |
| Point read on one replica | O(log m) memtable + per checked SSTable O(log n) index search + one block read | Bloom filters skip ≈99% of irrelevant files; leveled: ≈1 file per level |
| Bloom filter | O(k) hashes per check | ≈10 bits per key for ≈1% false positives |
| Quorum write / read | latency of the W-th / R-th fastest of N replicas | N copies of every byte |
| Locate replicas | O(log T) binary search over T ring tokens | ring of nodes × vnodes tokens, on every node |
| Merkle repair, d differing leaves of L | O(d log L) hash comparisons | one tree per token range per replica |
| Compaction | O(bytes merged), sequential | needs free disk ≈ size of the files being merged |
In the interview
Section titled “In the interview”A good 45-minute arc:
- Clarify (5 min). Access pattern, value sizes, scale, consistency needs, and above all the concurrent-write question.
- Single node (10 min). WAL, memtable, SSTables, compaction, Bloom filters. Say why it beats in-place updates for writes, and what reads pay.
- Distribute (10 min). Consistent hashing with vnodes, replicas as the next N physical nodes, a coordinator, N/R/W.
- Failures (10 min). Hinted handoff, read repair, Merkle anti-entropy, tombstones, gossip and failure detection.
- Conflicts and stretch (5-10 min). LWW vs vector clocks; multi-region (
LOCAL_QUORUMper region, async between regions); compare-and-set needs consensus.
Follow-ups to expect: “How do you add a node without downtime?” It takes over some token ranges; stream those ranges from current replicas, then switch reads. “How would you support range scans?” Order-preserving partitioning (by key range, like Bigtable) instead of hashing, at the cost of hot spots on sequential keys. “What if you need strong consistency?” A consensus group per range (Raft), so each range has one leader that orders writes. “How do you test it?” Fault injection (kill nodes, partition the network, skew clocks) with a history checker such as Jepsen that looks for lost or stale reads.
Check yourself
Section titled “Check yourself”Go deeper
Section titled “Go deeper”- DeCandia et al., Dynamo: Amazon’s Highly Available Key-value Store (SOSP 2007): consistent hashing with virtual nodes, sloppy quorums, hinted handoff, vector clocks, Merkle-tree anti-entropy.
- Chang et al., Bigtable: A Distributed Storage System for Structured Data (OSDI 2006): memtables, SSTables, compactions and Bloom filters, the storage engine Cassandra borrowed.
- Apache Cassandra docs: Dynamo: how a production system implements the ring, vnodes, replication, tunable consistency and repair.
- RocksDB wiki: Leveled Compaction: levels, size ratios and the write/read/space amplification trade.
Concept pages to drill: Distributed thinking · Sorted containers · Hash map · Versioning · Snapshots · Load balancing · TTL and expiry