Key-Value Store

Assume a regional cluster storing 5 PB of logical user data with an average logical record size of 4 KB including key, value, metadata, version, checksum, and TTL fields. Use replication factor 3 across failure domains. The workload averages 1M writes per second and 2M reads per second, with a 5x peak multiplier.

The storage engine is LSM-based, so physical capacity must include replicas, WALs, indexes, Bloom filters, tombstones, and compaction headroom. Assume 8 TB of safe usable data per storage node after reserving local disk for operating system, logs, and emergency free space.

• Records: 5 PB divided by 4 KB per record is about 1.25 trillion records.
• Replication: with N = 3, 5 PB logical becomes 15 PB replicated before LSM overhead.
• LSM overhead: compaction, tombstones, Bloom filters, partition indexes, and WAL reserve commonly require 1.7x to 2.0x headroom, so 15 PB becomes 25.5 to 30 PB physical capacity.
• Node count: 30 PB divided by 8 TB safe usable per node is about 3,750 nodes. Add spare capacity for repairs, drains, and skew, so plan for roughly 4,000 to 4,500 storage nodes.
• Write bandwidth: 1M writes per second times 4 KB is 4 GB per second logical. With 3 replicas, the cluster writes about 12 GB per second before protocol overhead, compaction rewrite, and repair traffic.
• Peak writes: a 5x multiplier turns 1M average writes per second into 5M peak writes per second.
• Reads: 2M average reads per second and 10M peak reads per second require coordinators, network, and replica thread pools sized independently from compaction and repair work.
• Virtual nodes: with 4,000 nodes and 256 virtual nodes per physical node, the ring has about 1,024,000 token ranges, which gives fine-grained balancing and manageable streaming units.

• The primary serving lookup is table_name plus key_hash plus item_key inside the owning token range.
• Token ring metadata is indexed by table_name plus token_start so coordinators can find the replica preference list for a key hash.
• Membership is indexed by node_id and also queried by status for operator workflows.
• Repair and hint logs are indexed by target_node_id plus created_at to replay missed mutations in order and expire old hints.
• Avoid general secondary indexes in the core KV store. If needed, build them as separate asynchronous projection tables with their own partitioning and repair model.

A logical item belongs to exactly one token range and is stored on N replica nodes selected from the ring preference list. Each physical node owns many virtual nodes, which allows fine-grained rebalancing. Membership state feeds ring placement, ring placement feeds coordinator routing, and repair logs reconcile mutations that were missed because a replica was down or partitioned.

The closest production systems are Dynamo-style and Cassandra-style stores. Dynamo emphasizes consistent hashing, sloppy quorums, hinted handoff, vector clocks, and application-assisted conflict resolution. Cassandra emphasizes wide-column storage, tunable consistency, gossip, hinted handoff, read repair, and an LSM-tree engine.

If the product needs strict serializable transactions, FoundationDB, Spanner, or a consensus-backed database may be a better fit. If the product mostly stores large immutable blobs, object storage plus a smaller metadata store is often cheaper than putting blobs directly into the KV engine.

Consistent hashing maps the output of a hash function onto a logical ring. A key is assigned to the first token range at or after its hash, and replicas are selected by continuing around the ring while respecting failure-domain rules. When nodes join or leave, only nearby token ranges need to move instead of reshuffling the entire dataset.

Virtual nodes improve balance. Instead of giving each physical node one large range, assign it hundreds of smaller ranges. A powerful node can own more virtual nodes, a weaker node can own fewer, and rebalancing can stream small ranges gradually. This also smooths random skew in token assignment.

The tradeoff is metadata and operational complexity. More virtual nodes mean more ring entries, more streams during repair and bootstrap, and more small compaction histories. A practical design keeps enough vnodes for balance but not so many that membership changes become noisy.

With replication factor N, every key has N preferred replicas. A write consistency W means the coordinator waits for W acknowledgements. A read consistency R means it consults R replicas. If R + W > N, then a successful read and successful write overlap on at least one replica, which reduces stale reads when there is no unresolved concurrent write.

For N = 3, common choices are W = 2 and R = 2 for quorum, W = 1 and R = 1 for lowest latency, or W = 3 and R = 1 for write durability with fast reads. The right choice is per workload. Shopping carts may prefer availability and mergeable conflicts. Payment state should not rely on this alone and may need a transactional system.

Quorums are not magic. Network partitions, timeouts, hinted writes, clock skew, sloppy quorum, and concurrent updates can still produce conflicts or stale reads. A strong candidate explains both the usefulness and limits of R + W > N.

In an always-writable distributed store, two clients can update the same key through different coordinators while replicas are partitioned. If neither update causally follows the other, the system must preserve or resolve the conflict.

Vector clocks track causal history by keeping counters for writers or replica actors. If one clock dominates another, the dominated version is older and can be discarded. If neither dominates, the versions are concurrent siblings. The safest approach returns siblings to the application for a semantic merge, such as combining shopping cart items.

Last-write-wins stores a timestamp and picks the largest timestamp. It is operationally simple and works for cache-like or idempotent state, but it can silently lose a valid concurrent update. If using LWW, use server-assigned hybrid logical clocks where possible and make the data loss tradeoff explicit.

The write path is optimized for sequential IO. A replica appends the mutation to the WAL, applies it to an in-memory memtable, and acknowledges after durable logging. When the memtable reaches a threshold, it is flushed to an immutable SSTable sorted by key. Reads check the memtable, then recent SSTables, aided by Bloom filters and sparse indexes.

Compaction merges SSTables, drops overwritten values, purges expired tombstones after the repair safety window, and reduces the number of files a read must check. Size-tiered compaction improves write throughput but can increase space amplification. Leveled compaction improves read latency but writes more data during compaction.

At petabyte scale, compaction is often the hidden bottleneck. The design needs backpressure, compaction debt metrics, per-tenant throttles, and enough spare disk to survive a node rebuild while compaction is behind.

Hinted handoff handles short outages. If replica C is down and a write reaches replicas A and B, the coordinator can store a hint for C. When C returns, the hint is replayed so C catches up. Hints should expire because a long-dead node may be too stale and should be rebuilt from streaming repair instead.

Read repair handles hot data opportunistically. When a quorum read discovers that one replica has an older version or missing tombstone, the coordinator sends the correct version back to the stale replica asynchronously. This converges keys that are frequently read.

Merkle-tree anti-entropy handles cold data. Replicas build hash trees over token ranges. If roots differ, workers descend the tree to find mismatching subranges and stream only those rows. This avoids comparing every key over the network while still proving that replicas converge.

Consistent hashing balances keys, not traffic. A single celebrity key, tenant, or time-bucketed key prefix can overload one replica set even if storage bytes are balanced. Mitigations include better key design, write sharding, adaptive key splitting, hot-key caching, per-tenant throttling, and moving hot virtual nodes to stronger hardware.

Rebalancing must be controlled. When a node joins, it receives token ranges and streams data from existing replicas. If too many nodes bootstrap or repair at once, streaming competes with customer traffic and compaction. Rate-limit streams, preserve spare capacity, and avoid ring churn during incidents.

Gossip-based failure detection is eventually consistent. Marking a node down too quickly causes unnecessary hinted handoff and replica churn; marking it down too slowly increases tail latency. The failure detector should incorporate recent heartbeat variance, network conditions, and operator override states.

Alternative one is a leader-based sharded database. It is easier to reason about for single-key linearizability, but leader failover and cross-region writes can reduce availability or increase latency.

Alternative two is a consensus-backed distributed SQL or transactional KV system. It provides stronger guarantees and transactions, but every write typically pays consensus latency and throughput is lower for write-heavy, globally distributed workloads.

Alternative three is object storage plus a metadata database. This is better for large blobs and cheap durability, but it does not provide low-latency fine-grained updates for trillions of small keys.

Do not use a Dynamo or Cassandra-style key-value store when the core requirement is ad hoc querying, joins, strict cross-key transactions, global serializability, or large blob streaming. It is also a poor fit when the application cannot tolerate stale reads or cannot define a safe conflict resolution policy.