Key-Value Store
Design a distributed key-value store like DynamoDB or Cassandra: from a single-node hash map to a consistent-hashing ring with replication, quorum reads, and tunable consistency.
What is a key-value store?
A key-value store maps arbitrary keys to arbitrary values with three operations: put, get, and delete. The apparent simplicity is deceptive: making those operations work across many nodes, survive node crashes, and meet a latency target requires explicit choices about partitioning, replication, storage, and conflict resolution.
This question is useful because it strips away application-layer complexity and exposes distributed-systems fundamentals: consistent hashing, replication, quorum reads and writes, LSM-tree storage, and the consistency trade-offs described by the CAP theorem.
TL;DR
Start with a single-node store that writes a WAL before updating an in-memory map. Scale it by partitioning keys across a consistent-hashing ring made of virtual nodes, then replicate each key to an illustrative group of three nodes. A coordinator fans out reads and writes, waits for configurable quorums, and repairs stale replicas in the background.
Use an LSM tree on each storage node so the write path is append-heavy and crash-recoverable. Carry version metadata with every value: HLC-based last-write-wins is simple for data where losing a concurrent update is acceptable; vector clocks expose concurrent versions as siblings when both updates must survive. State the consistency contract precisely: quorum overlap is not, by itself, a universal guarantee of linearizability.
Scope and assumptions
The following is an illustrative single-region interview baseline. The figures are capacity-planning inputs for the exercise, not product limits or guarantees:
- About 10 TB of logical data, 100,000 writes/second, and 1 million reads/second at peak, with reads roughly 10 times more frequent than writes.
- Key-level
put,get, anddeleteoperations only. Values are bounded in size, and range scans, secondary indexes, and cross-key transactions are out of scope. - Replication factor
N=3across independent failure domains. The default example usesW=2andR=2; the service documents whether that means quorum consistency or a stronger linearizable contract. - A single region is the baseline. Cross-region replication, disaster recovery objectives, and workload-specific TTL policies are follow-up choices.
- The p99 latency target is sub-10ms for the coordinator path in the stated deployment; storage, network distance, and quorum settings determine whether that target is attainable.
Functional Requirements
Core Requirements
put(key, value): write or overwrite a value for a key.get(key): retrieve the value for a key.delete(key): remove a key-value pair.- Horizontal scaling across many nodes with no downtime for node additions or removals.
Below the Line (out of scope)
- Range scans and secondary indexes
- Full SQL semantics
- Cross-key transactions (ACID)
The hardest part in scope: Distributing data consistently across N nodes while surviving failures is the single hardest problem here. Everything else in this article is in service of that one constraint.
Range scans and secondary indexes are below the line because they require sorted on-disk structures or separate index tables. To add them, layer a sorted SSTable-based scan path on top of the LSM tree and build secondary indexes as separate key spaces that map index values back to primary keys.
Full SQL semantics are below the line because they require a query planner, JOIN support, and transactions spanning multiple keys. An existing relational engine like CockroachDB or YugabyteDB is better suited here than a custom key-value layer.
Cross-key ACID transactions are below the line because they require distributed locking or a multi-version concurrency control (MVCC) layer. To add limited transaction support, use optimistic locking: read a set of keys with their vector clock versions, write all or none with a conditional check that aborts if any version changed.
Non-Functional Requirements
Core Requirements
- Availability: 99.99% uptime. The store favors availability over consistency (eventual consistency by default, tunable to strong).
- Latency: Sub-10ms p99 for both
getandput. - Scale: 10 TB total data, 100K writes/sec, 1M reads/sec at peak.
- Durability: Data survives individual node failures and restarts.
Below the Line
- Cross-region active-active replication
- Point-in-time recovery from before an application bug
- Per-key TTL-based eviction (can be layered on top)
Read/write ratio: Reads outpace writes 10:1 at peak. That ratio drives two decisions: use LSM trees on each node (optimized for write throughput, sequential I/O) and place a tiered read cache above the node tier. Almost every design decision in this article traces back to surviving 1M reads/sec without starving the 100K writes/sec.
Use the sub-10ms p99 target as the forcing constraint. A design that adds avoidable synchronous hops or forces random disk seeks on the hot path needs a clear justification.
30-second answer / outline
- Put a coordinator layer in front of storage nodes and route keys with a consistent-hashing ring made of virtual nodes.
- Store each key on
N=3replicas. Fan out requests in parallel and return after the configured read or write quorum; run read repair and hinted handoff for lagging replicas. - Use a WAL, MemTable, SSTables, Bloom filters, and background compaction on every storage node.
- Return version metadata with reads. Use HLC/LWW where the workload accepts a winner, and vector clocks plus sibling reconciliation where concurrent updates must remain visible.
- Explain the failure contract: quorum settings, node failure detection, degraded reads/writes, recovery, and the difference between quorum consistency and linearizability.
5-minute explanation
Separate the control plane from the request path. A membership service distributes a versioned ring and node health; coordinators use the local ring to identify the primary and replica set. The coordinator sends a write to all replicas, waits for W durable acknowledgements, and lets the remaining replica catch up. A read queries R replicas, selects the newest non-dominated version, and schedules repair when one replica is stale.
The storage node is durable but write-friendly: append the operation to a WAL, insert it into a MemTable, flush immutable SSTables, and compact them in the background. Bloom filters avoid most disk reads for absent keys. Deletes are tombstones, not immediate physical removals, so an old replica cannot resurrect deleted data during repair.
The correctness discussion is workload-dependent. R + W > N guarantees quorum overlap under the usual single-version assumptions, but linearizable reads and writes require a stronger coordination protocol or an explicitly single-leader/consensus-backed path. Vector clocks make causal concurrency visible; HLC/LWW is cheaper when a deterministic winner is acceptable.
45-minute interview approach
This is a time-boxed interview plan, not a promise that the article can or should be read in 45 minutes.
- 0β5 minutes β Clarify the contract: Confirm value size, read/write mix, latency target, TTLs, single-key versus transactional operations, and whether eventual, quorum, or linearizable consistency is required.
- 5β10 minutes β Establish scale: Use the illustrative data, throughput, replication, and failure-domain assumptions. Separate the client-facing coordinator path from node-local storage.
- 10β15 minutes β Define APIs and invariants: Show
put,get,delete, version metadata, idempotency, tombstones, and the invariant that a successful write has the required durable acknowledgements. - 15β22 minutes β Draw partitioning: Explain modulo hashing first, then the vnode ring, replica selection, ring-version propagation, rebalancing, and hot-key mitigation.
- 22β30 minutes β Draw replication: Walk through
N,R,W, failure detection, hinted handoff, read repair, and what the client sees during a partial outage. - 30β37 minutes β Deep dive on storage and conflicts: Compare hash-map snapshots, B-trees, and LSM trees; then choose HLC/LWW or vector clocks based on the data type.
- 37β42 minutes β Reliability, security, and operations: Cover WAL and fsync policy, compaction pressure, encryption, authentication, quotas, node replacement, repair lag, and backup/restore.
- 42β45 minutes β Trade-offs and close: Compare smart clients with coordinators, quorum with leader-based consistency, and LWW with sibling reconciliation. Recap the bottleneck and invite follow-up questions.
Core Entities
- KeyValueEntry: The stored record containing the key, value bytes, version (vector clock or timestamp), and an optional TTL.
- Node: A physical or virtual machine in the cluster, owning a slice of the key space and running an LSM storage engine locally.
- VNode (Virtual Node): A logical partition token assigned to a physical node. Each physical node owns multiple vnodes, distributing load more evenly across the ring.
- ReplicationGroup: The set of N nodes (typically 3) responsible for a given key, selected by walking clockwise on the consistent hashing ring from the key's hash position.
The full schema for KeyValueEntry includes the vector clock and tombstone flag for deletes; the conflict-resolution deep dive covers those fields in detail.
API Design
Start with one endpoint per functional requirement, then note where routing transparency matters.
FR 1 - put:
PUT /keys/{key}
Body: { "value": "<bytes>", "ttl_seconds": 3600 }
Response: 200 OK
FR 2 - get:
GET /keys/{key}
Response: { "key": "mykey", "value": "<bytes>", "version": "1:3,2:1" }
FR 3 - delete:
DELETE /keys/{key}
Response: 204 No Content
Use PUT for writes, not POST. PUT has idempotent upsert semantics: repeated calls with the same key and value are safe and produce the same result. POST implies creating a new resource each time, which is wrong for a key-value store where the key is the identity.
The version field in the get response is a serialized vector clock. The client needs it to perform conditional updates and for the store to detect concurrent write conflicts. Without a version, two clients retrieving the same key simultaneously have no way to know their writes are causally concurrent.
Routing is transparent to the client. A coordinator node (or a smart client library, in the Dynamo style) resolves which physical nodes own the key and fans out requests. Clients always talk to the same coordinator endpoint, not directly to storage nodes. Routing via consistent hashing is covered in the deep dives.
High-Level Design
Critical flows
- Write: The coordinator hashes the key, selects the replica group, sends the versioned value to replicas in parallel, waits for
Wdurable acknowledgements, and returns success. - Read: The coordinator queries
Rreplicas, chooses the newest non-dominated version, returns siblings when required, and repairs stale replicas asynchronously. - Membership and failure: A versioned ring is disseminated to coordinators; suspected nodes are avoided, hinted handoff stores temporary replicas, and repair restores the intended replica set after recovery.
1. Single-node store
The simplest store that satisfies the put/get/delete requirements: a single server with an in-memory hash map, backed by an append-only write-ahead log (WAL) for durability.
Components:
- Client: Sends HTTP requests for put/get/delete.
- Server: Holds the in-memory hash map. On every write, appends the operation to the WAL before acknowledging.
- WAL (Write-Ahead Log): An append-only log on disk. On crash recovery, the server replays the WAL to rebuild the in-memory state.
- Disk: Stores the WAL file. The hash map itself lives in RAM.
Request walkthrough (put):
- Client sends
PUT /keys/session:abc123with the value. - Server appends
{op: SET, key, value, timestamp}to the WAL on disk. - Server updates the in-memory hash map:
map["session:abc123"] = value. - Server returns
200 OK.
Request walkthrough (get):
- Client sends
GET /keys/session:abc123. - Server reads from the in-memory hash map: O(1) lookup.
- Server returns the value.
This works perfectly for a small dataset. Two things break at scale: the entire dataset must fit in RAM, and one node dying takes the whole store offline. Both are solved by distributing data across nodes.
Start with the single-node design before distribution. It makes the first failure boundaries explicit: the dataset must fit the node's storage, and a node failure takes the service offline.
2. Partitioning data across nodes (consistent hashing)
One server can hold maybe a few hundred GB in RAM before cost becomes untenable. With 10 TB of data, you need to spread keys across many nodes.
The naive approach: modulo hashing. Assign a key to node i using node_id = hash(key) % N. This is simple and fast. The problem surfaces the moment you add or remove a node. When N changes from 10 to 11, every key whose hash(key) % N maps to a different bucket must move to a new node. On average that is (N-1)/N of all keys; adding one node to a 10-node cluster triggers migration of roughly 90% of all keys. That is catastrophic for a production system.
Continue Reading with Premium
Unlock this article and every other in-depth system design guide on the platform with SDEpedia Premium.