Storage and Retrieval: How a Distributed Key-Value Store Works

A single server key-value store is a hash table in memory. Access is fast. Fit is limited.

Two optimizations help a little. Compress data. Keep only hot data in memory and the rest on disk. Even with both, a large data set fills one server quickly.

That breaks (1). The fix is a distributed hash table. Split data across many servers. Replicate each key on N servers. The rest of this lesson is how to place, copy, read, and repair that data.

Mental picture: think of servers as stops on a clock face. Keys are riders. Each rider walks clockwise to the first stop that will take them.

Consistent hashing solves placement. Place servers on a hash ring. Hash each key onto the same ring. Store the key on the first server clockwise from its hash.

Example from the chapter: s0 through s7 sit on the ring. Key0 hashes between s0 and s1. It is stored on s1.

Two properties matter. Load spreads evenly. Add or remove of a server moves only nearby keys. Virtual nodes improve this. A big server gets more virtual spots on the ring.

CAP theorem frames the tradeoff. Consistency means all clients see the same data. Availability means every request gets a response. Partition tolerance means the system works despite a split.

You can only pick two. In practice splits happen. So real stores pick CP or AP. CA cannot exist in a real distributed system.

Concrete split: replicas n1, n2, n3 hold the same key. Network drops n3. Writes to n1 cannot reach n3. Now choose. CP blocks writes to n1 and n2. That protects (2) but breaks (3). AP keeps accepting reads and writes. That protects (3) but breaks (2) until repair.

Replication means N copies on N distinct servers. Logic: hash the key. Walk clockwise. Pick the first N unique servers. Unique matters when virtual nodes put two spots from one physical server in a row.

Example: N equals 3. Key0 lands before s1. Copies go to s1, s2, s3. For reliability, put replicas in distinct data centers. Connect centers with high speed networks. This protects (5).

Reads and writes go through a coordinator. The coordinator is a node acting as proxy for the client. Define three numbers. N is replica count. W is write quorum. The write succeeds after W acknowledgements. R is read quorum. The read succeeds after R responses.

W equals 1 does not mean one copy. Copies still go to N nodes. It means the coordinator waits for one acknowledgement. That is fast. It risks stale data.

Tuning:

• R equals 1 and W equals N. Fast read.
• W equals 1 and R equals N. Fast write.
• W plus R greater than N. Strong consistency. Overlap node must have latest data. Typical is N equals 3, W equals 2, R equals 2.
• W plus R less than or equal to N. No strong consistency.

Latency versus consistency is the price. W or R greater than 1 waits for the slowest replica in the set. W or R equals 1 returns after any replica answers.

Consistency models in the chapter:

• Strong consistency. Every read sees the latest write.
• Weak consistency. Reads may see old values.
• Eventual consistency. Given time, all replicas converge. Dynamo and Cassandra use this. It keeps (3) high and pushes conflict handling to versioning.

Replication plus eventual consistency creates versions. Example: n1 and n2 both hold name john. Server 1 writes johnSanFrancisco as v1. Server 2 writes johnNewYork as v2 at the same time. Neither saw the other. There is no obvious winner.

Versioning treats each write as a new immutable version. A vector clock is a list of server and counter pairs per data item. Write D as D with entries like Sx comma 1.

Rules: if Si already has an entry, increment it. If not, add Si comma 1.

Walk from the chapter:

• Client writes D1 through Sx. Clock is D1 with Sx comma 1.
• Client reads D1, writes D2 through Sx. Clock is D2 with Sx comma 2. D2 descends from D1.
• Client reads D2, writes D3 through Sy. Clock is D3 with Sx comma 2, Sy comma 1.
• Another client reads D2, writes D4 through Sz. Clock is D4 with Sx comma 2, Sz comma 1.
• Client reads D3 and D4, finds conflict, reconciles, writes D5 through Sx. Clock is D5 with Sx comma 3, Sy comma 1, Sz comma 1.

Detection rule: X is an ancestor of Y if every counter in X is less than or equal to the matching counter in Y. Example: D with s0 comma 1, s1 comma 1 is ancestor of D with s0 comma 1, s1 comma 2. No conflict.

X conflicts with Y if some counter in Y is less than the matching counter in X. Example: D with s0 comma 1, s1 comma 2 conflicts with D with s0 comma 2, s1 comma 1.

Weakness stated honestly: clients must run reconciliation. Clocks can grow. Fix is a length cap that drops oldest pairs. That can blur ancestry. Dynamo in production had not hit this limit per the chapter. Price is client complexity for high availability.

Failure is normal at scale. One report is not enough to mark a node down. All to all multicast works but wastes messages.

Gossip protocol is decentralized detection. Each node keeps a membership list with member ID and heartbeat counter. Each node increments its own counter. Each node sends heartbeats to random nodes. Those nodes spread them further. If a counter stops rising past a timeout, the member is marked offline and the news spreads.

Example: s0 sees member 2 heartbeat stuck at 8066 while others are past 10220. s0 shares that info. Others confirm. s2 is marked down.

Repair depends on duration.

Temporary failure: strict quorum would block. Sloppy quorum relaxes placement. Use first W healthy servers for writes and first R healthy for reads on the ring. Skip offline nodes. Hinted handoff stores the write elsewhere and pushes it back later. Example: s2 is down. s3 takes writes for s2. When s2 returns, s3 hands data back. This protects (3) and later restores (2).

Permanent failure: use anti-entropy. Compare all data and update to newest. A Merkle tree makes this cheap. Each non leaf node holds the hash of its children. Compare roots first. If equal, data matches. If not, compare left child then right child. Sync only bad buckets.

Build with key space 1 to 12:

• Step 1: split into 4 buckets.
• Step 2: hash each key with uniform hashing.
• Step 3: hash each bucket into one node.
• Step 4: hash upward to root.

Sync cost tracks difference size, not total size. That is the win.

Data center outage: power or network or disaster can kill a site. Fix is copies across data centers. Users reach other centers. This protects (5) at the cost of cross center links.

Nodes share the same duties. No single point of failure. Client API is get with key and put with key and value.

Write path, based on Cassandra:

• Persist request in commit log on disk.
• Save data in memory cache.
• When cache is full or hits threshold, flush sorted data to SSTable on disk. SSTable means sorted-string table. It is a sorted list of key value pairs.

This breaks (2) briefly for speed, then converges. Commit log guards against memory loss.

Read path:

• Check memory cache first. If hit, return.
• If miss, check Bloom filter. Bloom filter tells which SSTables might hold the key. It avoids useless disk reads.
• Read candidate SSTables. Return result to client.

System view: clients hit any node. Coordinator routes by ring. Ring gives decentralization. Replication gives durability. Quorums give tunable correctness. Gossip plus handoff plus Merkle gives repair. Storage engine gives speed plus persistence.

# Simulates N replicas, W/R quorum, and vector clock comparison.
# Run with: python3 kv_demo.py

replicas = {"s0": {}, "s1": {}, "s2": {}}
N = 3

def vector_dominates(a, b):
    # Returns True if a >= b on all entries. Ancestor check.
    keys = set(a) | set(b)
    return all(a.get(k, 0) >= b.get(k, 0) for k in keys)

def clocks_conflict(a, b):
    # Conflict if neither dominates. Sibling check.
    return not vector_dominates(a, b) and not vector_dominates(b, a)

def quorum_write(key, value, clock, W, down=set()):
    # Line that matters: only count healthy nodes, sloppy quorum style.
    healthy = [s for s in replicas if s not in down]
    if len(healthy) < W:
        return False, "blocked: not enough healthy nodes"
    for s in healthy[:W]:
        replicas[s][key] = (value, dict(clock))
    return True, f"acked by {healthy[:W]}"

def quorum_read(key, R, down=set()):
    # Line that matters: caller must compare versions, eventual consistency.
    healthy = [s for s in replicas if s not in down]
    results = [replicas[s].get(key) for s in healthy[:R]]
    return [r for r in results if r is not None]

if __name__ == "__main__":
    ok, msg = quorum_write("name", "johnSF", {"Sx": 2, "Sy": 1}, W=2)
    print("write:", ok, msg)
    print("read R=2:", quorum_read("name", R=2))
    # Illustrative clocks only to show conflict logic:
    c1 = {"s0": 1, "s1": 2}
    c2 = {"s0": 2, "s1": 1}
    print("conflict?", clocks_conflict(c1, c2))
    print("ancestor?", vector_dominates({"s0": 1, "s1": 2}, {"s0": 1, "s1": 1}))

# Expected output:
write: True acked by ['s0', 's1']
read R=2: [('johnSF', {'Sx': 2, 'Sy': 1})]
conflict? True
ancestor? True

• Notes on lines that matter: healthy filter implements skipping down nodes.
• W check implements blocking when too few nodes answer.
• Read returns multiple versions so client sees conflict.
• Dominance check implements ancestor test from the chapter.

• Dynamo: uses eventual consistency plus vector clocks plus sloppy quorum plus hinted handoff plus Merkle trees.WHY: shopping cart needs (3) during partitions and can reconcile versions later.
• Cassandra: uses ring plus quorum tuning plus commit log plus memory cache plus SSTable plus Bloom filter read path.WHY: needs scalable writes and tunable balance between speed and (2).
• BigTable: cited as core key-value design influence.WHY: shows sorted persistent storage model behind SSTable style persistence.

• Q: Why does W plus R greater than N give strong consistency.A: Sets must overlap. Overlap node has latest write. Read quorum of size R must hit it.
• Q: When would you set W equals 1.A: When write speed and (3) matter more than (2). Reads may be stale. Repair later.
• Q: Who resolves a vector clock conflict.A: Client. System returns siblings. Client reconciles and writes merged version with new clock.
• Q: Why not compare all keys after a permanent failure.A: Full compare moves full data. Merkle root check prunes equal subtrees. Transfer tracks difference.
• Q: Why gossip instead of all to all multicast.A: Multicast cost grows fast with node count. Gossip spreads through random peers with bounded messages.