← DBMS
Advanced Data Patterns & Internals

Distributed Database Internals — LSM Trees, Spanner & Dynamo-Style Systems

Watch: "How Databases Actually Store Your Data (B-Trees vs LSM Trees Explained)" — ByteMonk

Beyond "SQL versus NoSQL" there are three internal design choices that explain most of how modern databases behave: how a single node stores data (B-tree pages updated in place, or a log-structured merge tree that appends and merges later), how a cluster orders transactions across machines (a globally synchronized clock as in Google Spanner, or no global order at all), and how it stays available when nodes are unreachable (leaderless, Dynamo-style replication with quorums and version reconciliation). Interviewers probe these because each is a clean example of a trade-off: write throughput against read cost, strict consistency against latency, availability against conflicts you must resolve. For a Backend or AI engineer this is what lets you say why Cassandra or RocksDB handles heavy writes well, why Spanner-like systems pay commit latency, and why a Dynamo-style store hands you conflicting versions. This page builds on Distributed Databases, CAP and PACELC, B+ trees and WAL.

Deep Dive: The LSM Write Path vs the B-tree Write Path

B-TREE (e.g. PostgreSQL heap+index, InnoDB)         LSM TREE (e.g. RocksDB, LevelDB, Cassandra)
write(k,v)                                           write(k,v)
  1. append to WAL (sequential)                        1. append to WAL (sequential)
  2. find the target page (random read)                2. insert into MEMTABLE (in-memory sorted map)
  3. modify the page in the buffer pool                3. return  (no random disk I/O on the write path)
  4. later: write the dirty page back IN PLACE            ...
     (random write; page may split)                    when memtable is full:
                                                         4. freeze it, flush as an immutable SORTED FILE
read(k): walk root -> leaf, usually one path             (SSTable), sequential write
         page reads, often a few                       read(k): memtable, then newest sorted file, then older...
                                                                (bloom filter per file skips most misses)
                                                       background: COMPACTION merges files, drops old versions

Both engines write a sequential log first for durability, so the difference is what happens after. A B-tree updates the right page where it lives, which needs a random read to find the page and, later, a random page write, and it rewrites a whole page (for example 8 KB in PostgreSQL, 16 KB in InnoDB) even if one row changed. An LSM turns every write, including updates and deletes, into an append: the memtable absorbs it in memory and a flush writes one large sorted file sequentially. That makes ingest cheap, and that is why LSM engines are favoured for write-heavy or time-series-like workloads. The bill arrives on reads (a key may live in several files, so reads consult more places) and in background work (compaction), which is the next section.

Deep Dive: Compaction, Tombstones and the Three Amplifications

Three amplification factors (you can improve two at the expense of the third):
  WRITE amplification : bytes physically written / bytes the app wrote   (compaction rewrites data)
  READ  amplification : files/pages consulted per lookup                 (more sorted runs -> more probes)
  SPACE amplification : bytes on disk / live data bytes                  (stale versions and tombstones linger)

Size-tiered compaction (e.g. Cassandra STCS)      Leveled compaction (e.g. RocksDB/LevelDB style)
  merge several similar-size files into one         L0 -> L1 -> L2 -> ... each level ~fanout x bigger,
  lower write amp, higher read + space amp          files in a level have non-overlapping key ranges
  (needs spare space for a big merge)               fewer files per read, but data is rewritten as it
                                                    moves down levels: higher write amp

Delete = write a TOMBSTONE (a marker), not an erase:
  run0 (new): c -> DELETE        run2 (old): c -> 1
  a read of c must see the tombstone first; the old value is physically removed only when
  compaction merges both files (and, in distributed systems, after replicas have seen the tombstone)

Compaction is the deferred cost of cheap writes. Because files are immutable, an overwrite or delete cannot change an old file; it adds a newer version that shadows the old one, and compaction later merges files (sorted merge, newest version wins) and physically discards shadowed data. Each merge rewrites bytes that were already on disk, so a byte can be written several times over its life: that is write amplification, and on flash it also consumes device endurance. The strategy tunes the trade: leveled compaction keeps reads and space tight but rewrites more, size-tiered rewrites less but tolerates more overlapping files. Bloom filters, one per file, are the standard read-side mitigation: a "definitely not here" answer skips the file, at the cost of memory and occasional false positives. Compaction that falls behind is a classic production incident: flushes outrun merges, read latency climbs, and the engine throttles writes.

Deep Dive: A Toy LSM, Run for Real

# Real output of the script in "Try It Yourself" below (python3):
After 8 writes (a overwritten, c deleted, f in memtable):
  memtable: {'c': '<DEL>', 'f': 3}
  run0: [('a', 2), ('b', 2), ('d', 2)]
  run1: [('a', 1), ('c', 1), ('e', 1)]
  get('a') -> 2  (sorted runs probed: 1)
  get('c') -> None  (sorted runs probed: 0)
  get('e') -> 1  (sorted runs probed: 2)
  get('f') -> 3  (sorted runs probed: 0)
  get('zzz') -> None  (sorted runs probed: 2)
  user_writes=8, entries written to 'disk' so far=6
After one more write (memtable flushed):
  memtable: {}
  run0: [('c', '<DEL>'), ('f', 3), ('g', 3)]
  run1: [('a', 2), ('b', 2), ('d', 2)]
  run2: [('a', 1), ('c', 1), ('e', 1)]
After compaction (newest wins, tombstone dropped, one run):
  memtable: {}
  run0: [('a', 2), ('b', 2), ('d', 2), ('e', 1), ('f', 3), ('g', 3)]
  get('a') -> 2  (sorted runs probed: 1)
  get('c') -> None  (sorted runs probed: 1)
  get('e') -> 1  (sorted runs probed: 1)
  user_writes=9, entries written to 'disk'=15, write amplification ~ 1.67x

Trace it: keys a, c, e flush into the older run, then b, a, d flush into a newer run, so a now exists in both, and the read of a finds the newer value (2) after probing one run. The delete of c is a tombstone in the memtable, so the read stops there and returns nothing without touching disk. A miss for zzz probes both runs, which is exactly the read amplification a bloom filter is there to avoid. After compaction the three runs become one, the shadowed a=1 and the deleted c are gone, and reads probe one run. The "1.67x" figure is only this toy's ratio of entries written to disk to user writes; it is illustrative and not a benchmark, and real engines add a WAL and multiple levels. Note also that the toy drops tombstones on a full compaction, which is only safe when no older data can still exist elsewhere; in a distributed store, dropping a tombstone too early can let a deleted value reappear from a replica that never saw the delete.

Deep Dive: Spanner, TrueTime and External Consistency

Goal: if transaction T1 commits BEFORE T2 starts (real time), then T1's timestamp < T2's timestamp
      (external consistency = linearizability for transactions; "strict serializability").

TrueTime API (as described in the 2012 Spanner paper):
   TT.now() returns an INTERVAL [earliest, latest]  -- the true time is guaranteed inside it
   the interval width (uncertainty) is small, on the order of a few milliseconds per the paper,
   kept small by GPS receivers + atomic clocks in each datacenter

Commit of a write transaction (simplified):
   1. Acquire locks; if it spans several shards, run two-phase commit across the Paxos groups
   2. choose commit timestamp s >= TT.now().latest
   3. COMMIT WAIT: do not make the commit visible / acknowledge until TT.after(s) is true
      i.e. wait out the clock uncertainty (about the interval width)
   4. release locks, acknowledge

Result: any transaction that starts after this ack gets a later timestamp, so timestamp order
matches real-time order without a central timestamp server.

The hard part of a distributed database with lock-free consistent reads is agreeing on an order between transactions on different machines whose clocks disagree. Spanner's answer is to bound the disagreement explicitly: rather than pretending clocks are exact, TrueTime reports an uncertainty interval, and the system waits out that uncertainty before exposing a commit, so a later transaction can never be assigned an earlier timestamp. That waiting is the cost, paid on every write transaction, and it is why the design invests in tight hardware clocks: the smaller the uncertainty, the shorter the commit wait. Data is split into ranges, each replicated across zones with Paxos, and cross-shard transactions use two-phase commit on top. Timestamps also allow read-only transactions and snapshot reads at a chosen time without taking locks. Treat these details as those of the published paper: the current managed service has evolved, and other systems that call themselves "Spanner-like" (for example CockroachDB, which uses hybrid logical clocks and no special clock hardware) generally give weaker real-time ordering guarantees than the original design, so check each system's documentation instead of assuming equivalence.

Deep Dive: Dynamo-Style Leaderless Replication

Hash ring with virtual nodes; each key is stored on N consecutive nodes (preference list), e.g. N=3
   key K --hash--> position on ring --> replicas on nodes A, B, C

   write: coordinator sends to all N replicas, waits for W acks
   read : coordinator asks replicas,    waits for R responses, returns the newest/all conflicting versions

   R + W > N  -> read and write sets overlap (only under strict quorums; see caveats)
   typical example N=3, W=2, R=2      N=3, W=1, R=1 : fast, but reads can be stale

Repair mechanisms for replicas that missed writes:
   read repair      : reader notices a stale replica and pushes the newer version to it
   hinted handoff   : if a target replica is down, another node holds the write and hands it back later
                      (a SLOPPY quorum: W acks may come from nodes outside the key's preference list,
                       which raises write availability but breaks the R + W > N overlap guarantee)
   anti-entropy     : replicas compare Merkle trees (hash trees over key ranges) to find and fix differences

A leaderless design has no single node whose failure requires an election: any replica can accept a write, so the system stays writable through node failures and partitions, which is the availability-favouring corner of the CAP trade-off. The cost is that two clients can write the same key through different coordinators at the same time and nobody is in charge of ordering them, so consistency becomes tunable and eventual. Choosing R and W trades latency and availability against staleness: lower values are faster and more available, values with R + W > N give overlapping quorums but only under strict quorums with no sloppy quorum, no concurrent writes racing, and no failed-write side effects, so it is not a linearizability guarantee. The original Amazon Dynamo paper (2007) describes this design; Cassandra and Riak were inspired by it, though each differs in detail, and the managed AWS DynamoDB service is a different system whose internals should not be assumed to match the paper.

Deep Dive: Vector Clocks and Conflicting Writes

Each version carries a vector clock: a set of (node, counter) pairs.
   V1 dominates V2 if every counter in V1 is >= the one in V2 (and at least one is greater)
   -> V2 is an ancestor: safe to discard.   Otherwise the two are CONCURRENT: keep both (siblings).

(the example shape from the Dynamo paper)
  write handled by node X:            D1  [X:1]
  update, handled by X again:         D2  [X:2]
  partition: update via node Y:       D3  [X:2, Y:1]      (descends from D2)
             update via node Z:       D4  [X:2, Z:1]      (descends from D2)

  compare D3 vs D4:  neither dominates -> CONFLICT. A reader gets BOTH.
  application (or user) merges them and writes back through X:   D5  [X:3, Y:1, Z:1]
                                                                  (dominates D3 and D4, conflict resolved)

Alternative: last-write-wins (LWW) using a timestamp
  concurrent writes A(t=100ms) and B(t=101ms): B wins, A is silently dropped;
  if B's clock is ahead of a later, truly newer write, that newer write can be lost.

Wall-clock time cannot tell you whether write B was made knowing about write A, so a vector clock tracks causality instead: it records which versions each new version descended from. If one dominates the other, the older one is obsolete; if neither does, the writes were concurrent and the system cannot pick a correct winner on its own. Then the trade-off is explicit: last-write-wins is simple and needs no client work, but it silently loses concurrent updates and is exposed to clock skew (Cassandra's default resolution is timestamp based, and it is common in practice); keeping siblings for the application to merge is safe but pushes complexity onto the client, and vector clocks can grow (implementations prune them, which can reintroduce false conflicts). CRDTs are the structured middle ground: data types (counters, sets, registers) whose merge function is defined so that replicas converge without hand-written conflict resolution.

Key Pointers

How It Works

  1. A write is appended to the write-ahead log (for durability across crashes) and inserted into the in-memory memtable; the client is acknowledged without touching the sorted files.
  2. When the memtable reaches its size limit it is frozen and flushed as an immutable, sorted file, and a fresh memtable takes over; over time the number of files grows.
  3. A read checks the memtable, then files from newest to oldest, using per-file bloom filters and indexes to skip files, and returns the first (newest) version found or a tombstone meaning "deleted".
  4. Background compaction merges files by key, keeps only the newest version of each key, drops tombstones and old versions once it is safe, and writes new files; the strategy decides which files are merged and when.
  5. In a distributed store, each key is placed on several nodes (by consistent hashing in Dynamo-style systems, by range and Paxos group in Spanner); each node runs its own storage engine like the above.
  6. Cross-node consistency is then decided by the coordination scheme: majority-replicated groups with timestamped commits (Spanner-style) or configurable quorums with repair and version reconciliation (Dynamo-style).

Try It Yourself

# Save as lsm_toy.py and run:  python3 lsm_toy.py
# Toy LSM: memtable + immutable sorted runs (run0 = newest) + tombstones + full compaction.
TOMB = object()

class LSM:
    def __init__(self, memtable_limit=3):
        self.mem = {}
        self.limit = memtable_limit
        self.runs = []                 # each run: sorted list of (key, value); index 0 = newest
        self.bytes_written = 0         # entries written to "disk" (flushes + compactions)
        self.user_writes = 0

    def put(self, k, v):
        self.user_writes += 1
        self.mem[k] = v                # (a real engine also appends to a WAL first)
        if len(self.mem) >= self.limit:
            self.flush()

    def delete(self, k):
        self.put(k, TOMB)

    def flush(self):
        run = sorted(self.mem.items())
        self.runs.insert(0, run)
        self.bytes_written += len(run)
        self.mem = {}

    def get(self, k):
        probes = 0
        if k in self.mem:
            v = self.mem[k]
            return (None if v is TOMB else v), probes
        for run in self.runs:          # newest to oldest
            probes += 1
            for kk, vv in run:         # real engines: bloom filter + binary search / index
                if kk == k:
                    return (None if vv is TOMB else vv), probes
        return None, probes

    def compact(self):
        merged = {}
        for run in reversed(self.runs):    # oldest first so newer overwrites
            for k, v in run:
                merged[k] = v
        out = sorted((k, v) for k, v in merged.items() if v is not TOMB)  # drop tombstones (safe: full compaction)
        self.bytes_written += len(out)
        self.runs = [out] if out else []

    def show(self, title):
        print(title)
        print("  memtable:", {k: ('<DEL>' if v is TOMB else v) for k, v in self.mem.items()})
        for i, r in enumerate(self.runs):
            print(f"  run{i}:", [(k, '<DEL>' if v is TOMB else v) for k, v in r])

db = LSM(memtable_limit=3)
for k, v in [("a",1),("c",1),("e",1)]: db.put(k, v)           # flush -> run0
for k, v in [("b",2),("a",2),("d",2)]: db.put(k, v)           # flush -> run0, old run1
db.delete("c"); db.put("f", 3)
db.show("After 8 writes (a overwritten, c deleted, f in memtable):")
for key in ["a", "c", "e", "f", "zzz"]:
    val, probes = db.get(key)
    print(f"  get({key!r}) -> {val}  (sorted runs probed: {probes})")
print(f"  user_writes={db.user_writes}, entries written to 'disk' so far={db.bytes_written}")
db.put("g", 3)   # third entry in memtable {c:DEL, f, g}? triggers flush
db.show("After one more write (memtable flushed):")
db.compact()
db.show("After compaction (newest wins, tombstone dropped, one run):")
for key in ["a", "c", "e"]:
    val, probes = db.get(key)
    print(f"  get({key!r}) -> {val}  (sorted runs probed: {probes})")
print(f"  user_writes={db.user_writes}, entries written to 'disk'={db.bytes_written}, "
      f"write amplification ~ {db.bytes_written/db.user_writes:.2f}x")

# Exercises:
#  1. Set memtable_limit=2 and watch the number of runs and the amplification grow; call compact() every 3 flushes.
#  2. Make compact() keep the tombstone instead of dropping it and observe when get("c") still works.
#  3. Add a simple bloom filter per run (a set of hashes is enough for a toy) and count probes skipped.
#  4. Write a 3-replica toy where two replicas accept concurrent writes to the same key and compare
#     last-write-wins with a vector-clock version that keeps siblings.
# Real engines you can try locally: sqlite (B-tree) vs RocksDB via `pip install rocksdict` (not run here).

Interview Q&A

Q: Compare the write path of an LSM tree with a B-tree. Why are LSM engines better for write-heavy workloads?

A: Both write a sequential log first. A B-tree then finds the target page and updates it in place, which means random reads and, later, random writes of whole pages, and page splits. An LSM inserts into an in-memory sorted structure and flushes big sorted files sequentially, so ingest avoids random I/O; the trade-off is that reads may consult several files and background compaction consumes I/O and CPU. Neither is better in general: B-trees tend to give more predictable read latency and are the default in most relational engines, while LSM engines fit write-heavy, append-like or time-series workloads.

Q: What is compaction, and what is write amplification?

A: Compaction merges immutable sorted files so that shadowed versions and tombstones can be discarded and the number of files stays bounded. Because it rewrites data that was already on disk, one user write can cause several physical writes over its life, which is write amplification. Leveled compaction generally lowers read and space amplification but raises write amplification, while size-tiered does the reverse; you choose based on whether the workload is read-heavy, write-heavy, or space-constrained, and you monitor for compaction falling behind.

Q: What is external consistency in Spanner and how does TrueTime provide it?

A: External consistency means that if one transaction commits before another begins in real time, the first gets the smaller commit timestamp, so the database order matches what clients observe. TrueTime exposes clock uncertainty as an interval, and a committing transaction picks a timestamp no earlier than the interval's upper bound and waits until that time has definitely passed before acknowledging ("commit wait"). The trade-off is added commit latency of about the clock uncertainty, plus the cost of specialized clock infrastructure, so this is not something you get for free from ordinary NTP-synced servers. These are the details from the published paper, and specifics of the current service may differ.

Q: How does a Dynamo-style leaderless database stay available, and what does it give up?

A: Any replica among the N owners of a key can accept a write, so there is no leader whose loss requires an election, and the store keeps accepting writes as long as W replicas (or, with sloppy quorums, W nodes) are reachable. What it gives up is a single ordering of writes: concurrent writes can conflict, reads can be stale if R + W is not greater than N, and even with overlap the guarantee is not full linearizability. It uses read repair, hinted handoff and Merkle-tree anti-entropy to converge replicas, and requires the application to accept or resolve conflicts.

Q: What are vector clocks, and why are they preferred over timestamps for detecting conflicts?

A: A vector clock records, per node, how many updates a version has seen, so comparing two clocks tells you whether one version descends from the other or whether they were concurrent. Timestamps only give a total order that depends on clock synchronization, so with last-write-wins a concurrent update is silently discarded and clock skew can make an older write beat a newer one. The trade-off is that vector clocks detect conflicts but do not resolve them: someone must merge the siblings, and the clocks themselves can grow and need pruning.

Follow-up: You delete a row in an LSM-based, replicated store, run compaction, and the row comes back later. What happened?

A: A likely cause is that the tombstone was garbage-collected before every replica had received it. A replica that missed the delete still held the old value, and when it was repaired or read it spread the value again because no tombstone remained to shadow it. The fix is to keep tombstones for a grace period longer than the maximum time a replica can be down (and run repairs within that period), which costs extra space and slower reads over tombstone-heavy ranges.

Follow-up: Why not just use NTP timestamps and skip TrueTime?

A: NTP's error is typically much larger and, importantly, not bounded with a guarantee, so a system that assumes clocks are close can assign inconsistent order and violate external consistency. TrueTime's contribution is a bounded uncertainty that the database can wait out, which is safe by construction as long as the bound holds. Systems without such bounded clocks generally choose weaker guarantees or use logical or hybrid logical clocks and accept restrictions such as retries when the uncertainty matters.

Follow-up: With N=3, W=2, R=2 can a read still return stale data in a Dynamo-style store?

A: Yes, in several cases. If sloppy quorum or hinted handoff let writes land on nodes outside the key's owners, the read quorum may not include them; if a write failed part-way but reached one replica, a later read may or may not see it; and concurrent writes may leave siblings or be resolved by last-write-wins. The R + W > N overlap only guarantees that a read set intersects a write set under strict quorum assumptions, not that the result is the latest by any global order.

System Design Tie-in

When a design interview asks for a write-heavy store (activity logs, time-series metrics, message history, IoT ingestion) an LSM-based engine is a reasonable default answer, and you should be able to name what you are accepting: compaction I/O, read amplification handled by bloom filters and caches, tombstone-heavy delete patterns, and the need to monitor compaction backlog. For globally distributed data that needs strong transactional guarantees (payments, inventory, ledgers), a Spanner-style system (Spanner itself, or a distributed SQL database with a documented consistency model) is the direction; be clear that you are paying cross-zone commit latency and should keep such transactions short and co-located where possible, and say which consistency guarantee the chosen product actually documents. For always-writable, massively scalable workloads that tolerate staleness (shopping carts, sessions, user preference data), Dynamo-style systems fit, but the design must state a conflict policy (last-write-wins, siblings with application merge, or CRDT), the per-request R and W choices, and the repair process. Tie the choice back to the CAP and PACELC conversation: it is the same trade-off, viewed from the inside of the engine.

Trade-offs

LSM trees trade read speed, space efficiency and predictable latency (compaction stalls) for cheap writes; a B-tree is usually the better choice for read-heavy, point-and-range workloads with moderate writes. Spanner-style external consistency trades commit latency, infrastructure cost and operational lock-in for a simple mental model in which the database behaves as though transactions ran one at a time in real-time order; it is the wrong choice when eventual consistency is acceptable and low write latency matters more. Dynamo-style leaderless replication trades consistency and simplicity for availability and horizontal scale: you inherit conflicts, stale reads, tombstone and repair operations, and a data model that fits key-based access far better than joins or multi-key transactions. In every case the honest answer to "which is best" is "which cost can this workload absorb", and the answer should come with the numbers from your own load tests and not from a blog post.