Design a Key-Value Store

Adapted from: Alex Xu, System Design Interview – An Insider's Guide (2020), Chapter 6 — rewritten and condensed for teaching.

A key-value store (or key-value database) is a non-relational database that stores each value under a unique key. You can only reach a value through its key. Keys can be plain text (last_logged_in_at) or hashed values (253DDEC4). Shorter keys perform better. Values can be strings, lists, objects, or anything else. Stores such as Amazon Dynamo, Memcached, and Redis usually treat the value as an opaque blob.

Key	Value
145	john
147	bob
160	julia

We'll design a store that supports two operations:

put(key, value)   // store value under key
get(key)          // return the value stored under key

This lesson pulls together several earlier ones: consistent hashing for partitioning, the CAP trade-off, quorums, and storage-engine internals. It is based mainly on three real systems: Dynamo, Cassandra, and Bigtable.

What you'll be able to do after this lesson:

• Explain CAP with concrete partition scenarios, and choose CP or AP for a use case.
• Partition and replicate data on a hash ring, and tune consistency with N, W, and R quorums.
• Detect and resolve conflicting writes with vector clocks.
• Handle failures with gossip, sloppy quorums, hinted handoff, and Merkle-tree anti-entropy.
• Trace the write path (commit log → memtable → SSTable) and the read path (memory → Bloom filter → SSTable).

---

Step 1: Understand the Problem and Establish Scope

No design is perfect. Each one balances read speed, write speed, and memory, and it also has to choose between consistency and availability. Our store should have:

• Small entries: each key-value pair under 10 KB.
• Big data: the ability to store a very large data set.
• High availability: fast responses, even during failures.
• High scalability: support for a large data set.
• Automatic scaling: servers are added and removed automatically based on traffic.
• Tunable consistency.
• Low latency.

---

Single-Server Key-Value Store

On one machine this is easy: keep all pairs in an in-memory hash table. Memory is fast but limited, so two optimizations stretch it further:

• Compress the data.
• Keep frequently used data in memory and the rest on disk.

Even so, one server fills up quickly. Big data needs a distributed key-value store, also called a distributed hash table, which spreads pairs across many servers.

---

The CAP Theorem

The CAP theorem says a distributed system cannot provide more than two of these three guarantees at the same time:

• Consistency (C): every client sees the same data at the same time, whichever node it connects to.
• Availability (A): every client that requests data gets a response, even if some nodes are down.
• Partition tolerance (P): the system keeps working despite a partition, meaning a communication break between nodes.

flowchart TB
  C((Consistency)) --- CP[CP systems] --- P((Partition tolerance))
  P --- AP[AP systems] --- A((Availability))
  A --- CA["CA systems<br/>(not practical)"] --- C

Key-value stores are classified by which two they keep:

• CP: consistent and partition-tolerant, at the cost of availability.
• AP: available and partition-tolerant, at the cost of consistency.
• CA: consistent and available, but not partition-tolerant. Networks do fail, so every real distributed system must tolerate partitions. CA cannot exist in practice.

CAP in a concrete scenario

Suppose data is replicated on three nodes, n1, n2, and n3.

• Ideal world (no partitions): writes to n1 replicate to n2 and n3 automatically. You get both consistency and availability.
• Real world: n3 loses contact with n1 and n2. Writes to n1 or n2 can't reach n3. Writes that reached n3 but haven't propagated leave n1 and n2 stale.

flowchart LR
  n1((n1)) <--> n2((n2))
  n1 -. "partition ✗" .- n3((n3))
  n2 -. "partition ✗" .- n3

Now you must choose:

• Choose consistency (CP): block writes to n1 and n2 until the partition heals, so the three nodes can't diverge. The system becomes unavailable for those writes. Banks typically choose this, because showing a wrong balance is worse than returning an error.
• Choose availability (AP): keep accepting reads and writes, even though some reads may return stale data. When the partition heals, data syncs to n3.

Choosing the right CAP guarantees for your use case is a key early decision, and a good one to discuss with your interviewer. The Linearizability lesson goes further and explains why CAP's choice only arises during a partition.

Check: CAP

Quiz

Consistency vs. availability

1 / 2

Why can't a real distributed key-value store be a "CA" system?

Pick an answer to see if you are right.

---

System Components

We'll build the store from these pieces: data partition, data replication, consistency, inconsistency resolution, handling failures, the architecture, and the write and read paths.

Data partition

The full data set can't fit on one server, so it is split into partitions. That raises two challenges:

1. Distribute data evenly across servers.
2. Minimize data movement when nodes join or leave.

Consistent hashing solves both. Servers s0–s7 sit on a hash ring, and a key is stored on the first server clockwise from its hash. On top of that, it gives us:

• Automatic scaling: servers can be added and removed based on load, and only a small slice of keys moves.
• Heterogeneity: a server's number of virtual nodes can be proportional to its capacity, so bigger machines take more of the ring.

Data replication

For availability and reliability, each key is replicated asynchronously to N servers, where N is configurable. To choose them, map the key onto the ring, then walk clockwise and pick the first N servers. With N = 3, a key landing just before s1 is stored on s1, s2, and s3.

flowchart LR
  K((key0)) -- clockwise --> S1[s1] --> S2[s2] --> S3[s3]
  S1 -.- R["replicas: s1, s2, s3 (N = 3)"]

Two refinements:

• Unique physical servers. With virtual nodes, the next N positions might belong to fewer than N real machines. Skip any virtual node whose physical server has already been chosen.
• Spread across data centers. Nodes in the same data center often fail together (power, network, natural disaster), so place replicas in different data centers connected by fast links.

Practice: build the preference list

The list of servers responsible for a key is often called its preference list. Implement preference_list(ring, key_pos, n):

• ring is a list of (position, server) virtual nodes in any order.
• Start at the first virtual node with position >= key_pos, wrapping to the lowest position if there is none.
• Walk clockwise and collect servers, skipping any already chosen, until you have n.

Input: line 1 is position:server pairs, and line 2 is key_pos n.

Consistency: quorum consensus

Replicas must stay in sync. Quorum consensus gives tunable guarantees for both reads and writes:

• N: the number of replicas.
• W: the write quorum. A write succeeds once W replicas acknowledge it.
• R: the read quorum. A read succeeds once it has responses from R replicas.

sequenceDiagram
  participant C as Client
  participant CO as Coordinator
  participant S0 as s0
  participant S1 as s1
  participant S2 as s2
  C->>CO: put(key, val)
  CO->>S0: write
  CO->>S1: write
  CO->>S2: write
  S1-->>CO: ack
  Note over CO: W = 1 satisfied, so reply now
  CO-->>C: success
  S0-->>CO: ack (late)
  S2-->>CO: ack (late)

W = 1 does not mean the data is written to only one server. The data still goes to all N replicas (s0, s1, s2). W = 1 means the coordinator, the node that proxies between the client and the replicas, waits for just one acknowledgment before reporting success.

The trade-off is latency vs. consistency. Small W or R returns fast, because the coordinator only waits for the quickest replica. Larger values give stronger consistency but wait on slower replicas.

The overlap rule: if W + R > N, every read quorum overlaps every write quorum in at least one node. That node has the latest write, so reads are strongly consistent.

Configuration	Effect
R = 1, W = N	Optimized for fast reads
W = 1, R = N	Optimized for fast writes
W + R > N (typically N = 3, W = R = 2)	Strong consistency guaranteed
W + R ≤ N	Strong consistency not guaranteed

Practice: analyze a quorum configuration

Implement analyze(n, w, r) that returns a dict with:

• strong: whether W + R > N.
• write_tolerates: how many replicas can be down while writes still reach W acknowledgments.
• read_tolerates: the same for reads.

Look at the read-optimized case (N=3, W=3, R=1). Reads are fast and survive two failures, but a single down replica blocks every write. Tuning W and R is really deciding which operation gets to stay available.

Consistency models

A consistency model defines how fresh the data a reader sees must be:

• Strong consistency: every read returns the result of the most recent write. Clients never see stale data.
• Weak consistency: later reads may not see the latest value.
• Eventual consistency: a form of weak consistency. Given enough time without new writes, every replica converges.

Strong consistency usually means a replica refuses new reads and writes until all replicas agree on the current write. That blocking is a bad fit for a highly available store. Dynamo and Cassandra use eventual consistency, and so does our design. With concurrent writes, eventual consistency lets conflicting values into the system, and clients reconcile them when they read. Versioning makes that possible.

Check: quorums and consistency

Quiz

Tuning N, W, R

1 / 3

With N = 3 and W = 1, a write is acknowledged by s1 first. Where is the data stored?

Pick an answer to see if you are right.

---

Inconsistency Resolution: Versioning and Vector Clocks

Replication buys availability but lets replicas diverge. Versioning treats every modification as a new immutable version. Vector clocks tell us how versions relate to each other.

How conflicts arise. Replicas n1 and n2 both hold name = "john". Server 1 sets it to "johnSanFrancisco" while, at the same time, server 2 sets it to "johnNewYork". Both versions descend from the original, so the original can be discarded, but nothing tells us which of the two new versions should win.

flowchart TB
  O["name = john<br/>(original)"] --> V1["v1: johnSanFrancisco<br/>(via server 1)"]
  O --> V2["v2: johnNewYork<br/>(via server 2)"]
  V1 -. conflict .- V2

Vector clocks

A vector clock is a set of [server, counter] pairs attached to a data item: D([S1, v1], [S2, v2], …, [Sn, vn]). When a server Si writes D, it:

• increments vi if an [Si, vi] entry exists, or
• creates [Si, 1] if it doesn't.

Walk-through:

Step	What happens	Handled by	Vector clock
1	Client writes D1	Sx	D1([Sx, 1])
2	Client reads D1, writes D2 (overwrites D1)	Sx	D2([Sx, 2])
3	Client reads D2, writes D3	Sy	D3([Sx, 2], [Sy, 1])
4	Another client reads D2, writes D4	Sz	D4([Sx, 2], [Sz, 1])
5	A client reads D3 and D4, sees a conflict, merges them, and writes D5	Sx	D5([Sx, 3], [Sy, 1], [Sz, 1])

flowchart LR
  D1["D1 [Sx,1]"] --> D2["D2 [Sx,2]"]
  D2 --> D3["D3 [Sx,2][Sy,1]"]
  D2 --> D4["D4 [Sx,2][Sz,1]"]
  D3 --> D5["D5 [Sx,3][Sy,1][Sz,1]<br/>(reconciled)"]
  D4 --> D5

Comparing clocks. A missing entry counts as 0.

• X is an ancestor of Y (no conflict) if every counter in X is ≤ the matching counter in Y. Example: D([s0, 1], [s1, 1]) is an ancestor of D([s0, 1], [s1, 2]).
• X and Y are siblings (in conflict) if each has some counter greater than the other's. Example: D([s0, 1], [s1, 2]) and D([s0, 2], [s1, 1]).

(The book's wording, "Y has a counter less than X's," states only one direction. That alone could also mean X descends from Y. A real conflict needs it in both directions.)

Downsides:

1. Client complexity. Clients must implement conflict resolution.
2. Clock growth. The [server: version] list can get long. A common fix is to cap its length and drop the oldest pairs. That can make ancestry inaccurate, but the Dynamo paper reports Amazon never hit the problem in production, so it is an acceptable compromise for most systems.

Practice: vector clocks

Implement two functions. Clocks are dicts such as {'Sx': 2, 'Sy': 1}.

• increment(clock, server) returns a new clock with server's counter incremented, or set to 1 if it is missing. Don't modify the input.
• compare(a, b) returns 'equal', 'ancestor' (a happened before b), 'descendant' (b happened before a), or 'conflict'.

The wrapper first replays the walk-through and prints D5. Then it compares each pair of clocks from stdin. Each line has the form Sx=2,Sy=1 | Sx=2,Sz=1.

For more on detecting concurrent writes, see the version-vector section of Leaderless Replication.

---

Handling Failures

At scale, failures are not just inevitable but routine. First you detect them, then you resolve them.

Failure detection: gossip

One server saying another is down isn't enough evidence. It usually takes at least two independent sources to mark a server down.

All-to-all multicasting, where every node heartbeats every other node, is simple. But it costs O(n²) messages, which is inefficient for large clusters. A decentralized gossip protocol scales better:

• Each node keeps a membership list of member IDs and heartbeat counters.
• Each node periodically increments its own heartbeat.
• Each node periodically sends heartbeats to a few random nodes, which pass them on to others.
• Nodes update their membership lists as heartbeats arrive.
• A member whose heartbeat hasn't increased for a set period is considered offline.

Member ID	Heartbeat counter	Last updated
0	10232	12:00:01
1	10224	12:00:10
2	9908	11:58:02 ← stale
3	10237	12:00:20
4	10234	12:00:34

In this example, s0 notices that s2's counter hasn't moved for a long time. s0 gossips s2's entry to random peers. Once others confirm that s2 has gone quiet, s2 is marked down, and that news spreads through the cluster.

Temporary failures: sloppy quorum and hinted handoff

With a strict quorum, a down replica can block reads and writes. A sloppy quorum improves availability: the coordinator uses the first W healthy servers for writes and the first R healthy servers for reads, walking the ring and skipping offline servers.

The healthy stand-in keeps the data along with a hint about where it belongs. When the failed server comes back, the stand-in pushes the data back to it. This is called hinted handoff.

flowchart LR
  K((key)) --> S1[s1] --> S2["s2 (down ✗)"] --> S3[s3] --> S4[s4]
  S3 -. "holds s2's writes + hint; hands off when s2 recovers" .-> S2

With s2 down, s3 handles its reads and writes temporarily, then hands the data back when s2 recovers.

Permanent failures: anti-entropy with Merkle trees

If a replica is gone for good, or has drifted badly, an anti-entropy protocol compares replicas and brings each one up to the newest version. Comparing every key would move too much data. A Merkle tree narrows the comparison to just the parts that differ.

A Merkle (hash) tree is a tree in which each leaf holds the hash of a block of data, and each internal node holds the hash of its children. Building one over keys 1–12:

1. Divide the key space into buckets: 4 buckets of 3 keys, [1–3] [4–6] [7–9] [10–12]. Buckets keep the tree's depth bounded.
2. Hash each key in each bucket with a uniform hash function.
3. Hash each bucket into a single leaf node.
4. Build upward: each parent is the hash of its children, all the way to the root.

flowchart TB
  ROOT["root = h(h12 + h34)"] --> H12["h12 = h(b1 + b2)"]
  ROOT --> H34["h34 = h(b3 + b4)"]
  H12 --> B1["b1: keys 1–3"]
  H12 --> B2["b2: keys 4–6"]
  H34 --> B3["b3: keys 7–9"]
  H34 --> B4["b4: keys 10–12"]

Comparing two replicas: compare their root hashes. If they match, the replicas are identical and you're done. If not, compare the children, and recurse only into subtrees whose hashes differ. You end up at exactly the buckets that are out of sync, and you sync only those.

The work is proportional to the differences between the replicas, not to their total size. Real systems use large buckets. For example, one million buckets for one billion keys gives about 1,000 keys per bucket.

Practice: find out-of-sync buckets

build_tree is provided. It returns the tree as a list of levels, root first, so tree[0] == [root_hash] and tree[-1] is the list of bucket hashes. The child indexes of node i on one level are 2i and 2i + 1 on the next level.

Implement diff_trees(t1, t2), which returns (differing_bucket_indexes, comparisons):

• Start at the root and compare hashes, counting every comparison.
• If two nodes are equal, stop there.
• If they differ at a leaf, record its bucket index. Otherwise, recurse into the left child, then the right child.

Input: line 1 is num_keys num_buckets, and lines 2 and 3 are each replica's values as key=value pairs.

Notice the comparison counts. Identical replicas cost one comparison. A single bad key costs about 2 × depth + 1 comparisons, whether the replicas hold 12 keys or 12 billion.

Data center outages

Power failures, network outages, and natural disasters can take out a whole data center. Replicate data across data centers, so that users can still reach their data through the others when one goes dark.

Check: failure handling

Quiz

Detect, tolerate, repair

1 / 3

Why does gossip scale better than all-to-all heartbeats for failure detection?

Pick an answer to see if you are right.

---

System Architecture

flowchart LR
  C[Client] -- "get(key) / put(key, value)" --> CO["Coordinator<br/>(any node)"]
  subgraph Ring["Hash ring (consistent hashing)"]
    N0((n0)) --- N1((n1)) --- N2((n2)) --- N3((n3)) --- N4((n4)) --- N5((n5)) --- N0
  end
  CO --> N1
  CO --> N2
  CO --> N3

• Clients use a simple API: get(key) and put(key, value).
• A coordinator node proxies between the client and the store.
• Nodes sit on a consistent-hashing ring.
• The system is fully decentralized, so adding and removing nodes can be automatic.
• Data is replicated on multiple nodes.
• There is no single point of failure. Every node has the same responsibilities.

Because the design is decentralized, every node runs the full set of functions: the client API, failure detection, conflict resolution, failure repair, replication, and the storage engine.

---

Write Path

The write and read paths follow Cassandra's architecture.

flowchart LR
  W[Write request] --> CL["1. Append to commit log<br/>(disk, durable)"]
  CL --> MC["2. Write to memory cache<br/>(memtable)"]
  MC -- "3. full / threshold reached" --> SS["Flush to SSTable<br/>(sorted, on disk)"]

1. Persist the write to a commit log file, so that a crash doesn't lose it.
2. Save it in the memory cache.
3. When the memory cache is full or reaches a threshold, flush it to an SSTable on disk. A sorted-string table is a sorted list of <key, value> pairs.

---

Read Path

If the key is in memory, return it straight from the memory cache.

If not, it must come from disk, but which SSTable holds it? A Bloom filter answers this quickly. It is a compact probabilistic structure that can say "definitely not here" or "maybe here".

flowchart LR
  R[Read request] --> M{"1. In memory<br/>cache?"}
  M -- yes --> RES[Return result]
  M -- no --> BF["2–3. Bloom filters:<br/>which SSTables might have it?"]
  BF --> SS["4. Read candidate SSTables"]
  SS --> RES

1. Check whether the data is in memory. If not, go to step 2.
2. Check the Bloom filter.
3. The Bloom filter identifies which SSTables might contain the key.
4. Those SSTables return the data.
5. The result goes back to the client.

The Hash Indexes and LSM-Trees lesson covers memtables, SSTables, compaction, and Bloom filters in depth.

Check: architecture and storage paths

Quiz

Write and read paths

1 / 3

Order the write path (top = first):

OrderDrag the handle to rearrange — top should be first.

1. ⋮⋮1When the memtable reaches its threshold, flush it to an SSTable on disk
2. ⋮⋮2Insert it into the in-memory cache (memtable)
3. ⋮⋮3Append the write to the commit log

Drag items into the correct order (or use ↑↓), then check.

---

Summary

Goal / problem	Technique
Ability to store big data	Consistent hashing to spread load across servers
High availability reads	Data replication; multi-data-center setup
Highly available writes	Versioning and conflict resolution with vector clocks
Data set partition	Consistent hashing
Incremental scalability	Consistent hashing
Heterogeneity	Consistent hashing (virtual nodes proportional to capacity)
Tunable consistency	Quorum consensus (N, W, R)
Handling temporary failures	Sloppy quorum and hinted handoff
Handling permanent failures	Merkle trees (anti-entropy)
Handling data center outages	Cross-data-center replication

Previous lesson: Design Consistent Hashing.

Next lesson: Design a Unique ID Generator — multi-master auto-increment, UUIDs, ticket servers, and Twitter's Snowflake 64-bit layout.

---

Reference Materials

1. Amazon DynamoDB
2. memcached
3. Redis
4. Dynamo: Amazon's Highly Available Key-value Store
5. Apache Cassandra
6. Bigtable: A Distributed Storage System for Structured Data
7. Merkle tree — Wikipedia
8. Cassandra architecture
9. SSTable and Log Structured Storage: LevelDB
10. Bloom filter — Wikipedia