How to Build an LSM-Tree Key-Value Store in Python With a Write-Ahead Log, SSTables, Bloom Filters, and Compaction
Build a small log-structured merge-tree store in Python from scratch, then measure Bloom filters, three compaction policies, the zombie-key bug and five kinds of crash, with every number coming from...
What you will build, and why LSM trees exist
Imagine a service that takes thousands of small writes per second: sensor readings, chat messages, job results. Every write has to survive a crash, and every value has to be readable again by its key. The classic way to keep sorted data on disk is a B-tree, which edits pages in place, so each write changes something in the middle of a large file. There is a second design that does the opposite. It never edits a file after writing it. New data is appended to a log, collected in memory, written out as a fresh sorted file, and old files are merged in the background. It is called a log-structured merge-tree, or LSM-tree, and it is the storage design behind Bigtable, LevelDB and RocksDB.
Table Of Content
- What you will build, and why LSM trees exist
- Terms to know before we start
- Prerequisites
- Create the project
- Step 1: Start with the simplest log, and see why reads hurt
- 1.1 A shared helper module
- 1.2 The log-only store
- Step 2: Make writes durable with a write-ahead log
- 2.1 The record format and the writer
- 2.2 Damage the log on purpose
- Step 3: Freeze sorted data into an SSTable
- 3.1 The Bloom filter that every table carries
- 3.2 The file layout and the writer
- 3.3 The table reader
- 3.4 Try one table
- Step 4: Build the store: memtable, manifest, flush and read path
- 4.1 Counters
- 4.2 Merging sorted streams
- 4.3 The store object
- 4.4 Files, recovery and the commit point
- 4.5 The write path and the flush
- 4.6 The read path
- 4.7 Run the read path with no Bloom filters and no compaction
- Step 5: Add Bloom filters and measure them
- Step 6: Compact the tables, and compare three policies
- 6.1 The two policies
- 6.2 The code
- 6.3 Run the same workload under each policy
- Step 7: Deletes, tombstones, and the zombie-key bug
- Step 8: Crash it five ways
- 8.1 A child process that dies without cleaning up
- 8.2 The five experiments
- Step 9: Test it against a dictionary, then break it on purpose
- 9.1 Make sure the tests can fail
- Check that everything works end to end
- Where this stops working, and what to learn next
- Sources and method
The idea comes from a paper by Patrick O’Neil, Edward Cheng, Dieter Gawlick and Elizabeth O’Neil. Its abstract says: “The Log-Structured Merge-tree (LSM-tree) is a disk-based data structure designed to provide low-cost indexing for a file experiencing a high rate of record inserts (and deletes) over an extended period.” The trick is in how it gets there: “The LSM-tree uses an algorithm that defers and batches index changes, cascading the changes from a memory-based component through one or more disk components in an efficient manner reminiscent of merge sort.”
In this tutorial you will build a small LSM-tree key-value store in under 500 lines of standard-library Python, and then you will test it by measuring it. Along the way you will see:
- why a log you can only scan makes every read slower as the data grows (Step 1);
- how a write-ahead log keeps acknowledged writes safe, and what a torn or corrupt log looks like (Step 2);
- how an immutable sorted file with a sparse index and a Bloom filter answers a lookup with one small read (Step 3);
- how a memtable, a manifest and a merge turn those parts into a store with a read path and a crash-recovery path (Step 4);
- how Bloom filters remove most of the disk reads for keys that are not there (Step 5);
- how doing no compaction, tiered compaction and leveled compaction trade write cost, read cost and wasted space (Step 6);
- why deleting a key writes a marker, and how dropping the marker too early brings deleted keys back to life (Step 7);
- what happens to the store after a killed process, a torn log record, and a failure at the worst possible moment (Step 8);
- how to test the whole thing against a plain dictionary, and how to prove that the tests can fail (Step 9).
Every number in this article was printed by the code shown here. The byte counts and operation counts come from seeded workloads and repeat exactly on your machine; the timings depend on your hardware and will differ.
Terms to know before we start
Memtable. A sorted structure in memory that holds the newest writes. Bigtable’s paper describes the split like this: “the recently committed ones are stored in memory in a sorted buffer called a memtable; the older updates are stored in a sequence of SSTables.” LevelDB’s memtable is a skip list, which our skip list tutorial builds from scratch. We will use a Python dictionary and sort it when we write it out, which is simpler and good enough to show the design.
Write-ahead log (WAL). A file where every change is appended before it is applied in memory, so that a crash cannot lose a change the store has already acknowledged. LevelDB’s documentation puts it plainly: “Each update is appended to the current log file.” Our write-ahead log tutorial covers the idea on its own.
SSTable (sorted string table). An immutable file of entries sorted by key. “Immutable” means it is written once and never modified, only replaced.
Tombstone. A marker that says “this key was deleted”. You cannot remove a key from an immutable file, so a delete is written as a new entry. LevelDB’s documentation: “Each entry is either a value for the key, or a deletion marker for the key.”
Compaction. Merging several tables into one new table and deleting the inputs. It keeps the number of tables small and throws away overwritten values.
Bloom filter. A small bit array that answers “definitely not in this table” or “maybe in this table”. It never gives a false “no”. The Bloom filter tutorial explains how it works; here we use one inside every table.
Read, write and space amplification. Three ratios that describe the cost of a design. Read amplification is how much you read from disk to answer one lookup. Write amplification is how many bytes reach the disk for each byte the user stored. Space amplification is how much disk the data occupies compared with the space its live contents need. You will measure all three.
fsync. A request to the operating system to push a file’s bytes to stable storage instead of leaving them in its cache. Python’s documentation says: “Force write of file with filedescriptor fd to disk. On Unix, this calls the native fsync() function; on Windows, the MS _commit() function.”
Prerequisites
- Python 3.10 or newer. The code uses the
X | Nonetype syntax, which needs 3.10. Everything here ran on Python 3.13.14 on Windows 11. - pytest 9.1.1 for the tests in Step 9. The store itself uses only the standard library:
struct,zlib,hashlib,heapq,bisect,json,os. - Comfort with Python classes and generators. You do not need to know anything about databases; each idea is explained when it first appears.
- About 100 MB of free disk space for the lab folders (they used 43 MB here), and a terminal.
Create the project
Make a folder, create a virtual environment, activate it, and install pytest:
mkdir lsm-lab
cd lsm-lab
python -m venv .venv
.venv\Scripts\activate
python -m pip install pytest==9.1.1On macOS and Linux the activation line is source .venv/bin/activate. Every file below goes in the lsm-lab folder and every command runs from there. The finished folder has four library modules (bloom.py, wal.py, sstable.py, lsm.py), a helper module (labkit.py), one script per step, a small crash helper (crash_child.py) and the test file (test_lsm.py).
Step 1: Start with the simplest log, and see why reads hurt
Before building anything clever, build the least you can: a store where every put appends a record to the end of one file, and every get reads the file from the start and keeps the last record it sees for the key (later records overwrite earlier ones). Appending is sequential writing, the pattern disks handle best, so writes cost very little. The question is what it costs to read.
1.1 A shared helper module
All the step scripts share three tiny helpers. key(i) builds a 13-byte key such as b"user:00000042", value(i, version) builds a value that is always exactly 100 bytes, and fresh_dir empties and recreates a folder under work/ so each experiment starts clean. Save this as labkit.py:
"""labkit.py: helpers shared by the step scripts."""
import os
import shutil
WORK = os.path.join(os.path.dirname(os.path.abspath(__file__)), "work")
def key(i: int) -> bytes:
return b"user:%08d" % i
def value(i: int, version: int = 0) -> bytes:
return (b"v%d:%d:" % (version, i)).ljust(100, b".") # always 100 bytes
def fresh_dir(name: str) -> str:
path = os.path.join(WORK, name)
shutil.rmtree(path, ignore_errors=True)
os.makedirs(path)
return path1.2 The log-only store
Save this as step01_log_only.py. It writes 200,000 records over just 5,000 distinct keys, so every key has been overwritten about 40 times, and it times one lookup after 2,000, 20,000 and 200,000 records.
"""step01_log_only.py: the simplest log-structured store, and why reading it hurts."""
import struct
import time
from labkit import fresh_dir, key, value
class LogStore:
"""Every put appends a record. A get scans the whole file and keeps the last match."""
def __init__(self, path):
self.path = path
self.log = open(path, "wb") # one handle, opened once
def put(self, k, v):
self.log.write(struct.pack(">II", len(k), len(v)) + k + v)
def get(self, k):
self.log.flush()
with open(self.path, "rb") as f:
data = f.read()
found, pos = None, 0
while pos < len(data):
klen, vlen = struct.unpack_from(">II", data, pos)
pos += 8
if data[pos:pos + klen] == k:
found = data[pos + klen:pos + klen + vlen] # later records overwrite earlier ones
pos += klen + vlen
return found, len(data)
if __name__ == "__main__":
import os
store = LogStore(os.path.join(fresh_dir("step01"), "log.bin"))
written = 0
print(f"{'records':>9} {'file size':>12} {'bytes read per get':>20} {'time per get':>14}")
for target in (2_000, 20_000, 200_000):
while written < target:
store.put(key(written % 5_000), value(written)) # 5,000 distinct keys, overwritten again and again
written += 1
start = time.perf_counter()
_, read = store.get(key(42))
elapsed = (time.perf_counter() - start) * 1000
print(f"{target:>9,} {os.path.getsize(store.path):>12,} {read:>20,} {elapsed:>11.1f} ms")Run python step01_log_only.py. You should see something like this (the last column depends on your computer; the byte counts will match exactly):
records file size bytes read per get time per get
2,000 242,000 242,000 0.5 ms
20,000 2,420,000 2,420,000 3.2 ms
200,000 24,200,000 24,200,000 29.8 msA record here is 8 bytes of header, a 13-byte key and a 100-byte value, which is 121 bytes. Two thousand records are 242,000 bytes, and that is exactly how much one get reads, because the log has no idea where any key lives. At 200,000 records a single lookup reads 24,200,000 bytes, and the work grows in step with the file. That is the problem the rest of the tutorial solves: keep the cheap appends, and add structure so that a lookup reads a few kilobytes instead of the whole history.
Step 2: Make writes durable with a write-ahead log
The first piece of that structure is a log that exists for safety rather than for reading. A real store keeps its newest data in memory, and memory vanishes when the process dies. So every change is appended to a write-ahead log first. After a crash, the store replays the log and rebuilds what was in memory.
2.1 The record format and the writer
Each record is a 4-byte CRC32 checksum, a 4-byte payload length, and then the payload: an operation byte (put or delete), the key length, the value length, the key and the value. A checksum is a fingerprint of some bytes that changes if any byte changes, which lets us detect damage. CRC32 catches accidental corruption; it is not protection against someone tampering with the file on purpose.
The writer opens the file with buffering=0, which makes Python hand every write() straight to the operating system instead of holding bytes in its own buffer. With sync="always" it then calls os.fsync after every record, so the call to append does not return until the bytes were handed to stable storage. Save this as wal.py:
"""wal.py: an append-only write-ahead log with a CRC32 on every record."""
import os
import struct
import zlib
HEADER = struct.Struct(">II") # CRC32 of the payload, payload length
BODY = struct.Struct(">BII") # operation, key length, value length
OP_PUT, OP_DELETE = 1, 2
def encode(op: int, key: bytes, value: bytes = b"") -> bytes:
payload = BODY.pack(op, len(key), len(value)) + key + value
return HEADER.pack(zlib.crc32(payload), len(payload)) + payload
class WalWriter:
def __init__(self, path: str, sync: str = "always"):
self.path, self.sync = path, sync
self.file = open(path, "ab", buffering=0) # unbuffered: every write() reaches the OS
def append(self, op: int, key: bytes, value: bytes = b"") -> int:
record = encode(op, key, value)
self.file.write(record)
if self.sync == "always":
os.fsync(self.file.fileno()) # ask the OS to put the bytes on stable storage
return len(record)
def close(self) -> None:
self.file.close()
def replay(path: str):
"""Return (records, good_bytes, total_bytes). Stops at the first torn or corrupt record."""
data = b""
if os.path.exists(path):
with open(path, "rb") as f:
data = f.read()
pos, records = 0, []
while pos + HEADER.size <= len(data):
crc, length = HEADER.unpack_from(data, pos)
end = pos + HEADER.size + length
if end > len(data):
break # torn: the payload was cut short
payload = data[pos + HEADER.size:end]
if zlib.crc32(payload) != crc:
break # corrupt: bytes changed after they were written
op, klen, vlen = BODY.unpack_from(payload)
key = payload[BODY.size:BODY.size + klen]
value = payload[BODY.size + klen:BODY.size + klen + vlen]
records.append((op, key, value))
pos = end
return records, pos, len(data)replay reads the whole log and returns three things: the records it could trust, how many bytes of the file they cover, and the total file size. It stops at the first record that is cut short or fails its checksum. It has to stop there, because once a record is damaged you cannot trust its length field, and without the length you cannot tell where the next record begins. The caller will need good_bytes later, so keep an eye on it.
2.2 Damage the log on purpose
Save this as step02_wal.py. It writes five records and then breaks the log in the two ways a crash can: a torn write, where the power fails while a record is half on the disk, and a corrupt record, where a byte changes after it was written.
"""step02_wal.py: write a log, then damage it in the two ways a crash can."""
import os
from labkit import fresh_dir, key, value
from wal import HEADER, OP_PUT, WalWriter, encode, replay
def write_log(path, n=5):
if os.path.exists(path):
os.remove(path)
log = WalWriter(path, sync="always")
for i in range(n):
log.append(OP_PUT, key(i), value(i))
log.close()
path = os.path.join(fresh_dir("step02"), "000001.wal")
record = encode(OP_PUT, key(0), value(0))
print("one record is", len(record), "bytes; header (CRC32, payload length) =", record[:HEADER.size].hex(" "))
write_log(path)
records, good, total = replay(path)
print(f"intact log: {len(records)} records, {good} of {total} bytes usable")
with open(path, "r+b") as f: # a torn write: the power failed 3 bytes before the end
f.truncate(total - 3)
records, good, total = replay(path)
print(f"torn tail: {len(records)} records, {good} of {total} bytes usable -> cut {total - good} bytes")
write_log(path) # a flipped byte inside record 3 (index 2)
one = len(record)
with open(path, "r+b") as f:
f.seek(2 * one + 20)
byte = f.read(1)
f.seek(2 * one + 20)
f.write(bytes([byte[0] ^ 0xFF]))
records, good, total = replay(path)
print(f"corrupt record: {len(records)} records, {good} of {total} bytes usable -> records 4 and 5 are intact but unreachable")Run python step02_wal.py:
one record is 130 bytes; header (CRC32, payload length) = 62 d8 7f 07 00 00 00 7a
intact log: 5 records, 650 of 650 bytes usable
torn tail: 4 records, 520 of 647 bytes usable -> cut 127 bytes
corrupt record: 2 records, 260 of 650 bytes usable -> records 4 and 5 are intact but unreachableRead the three result lines in order. An intact log of five 130-byte records is 650 bytes, and all of it is usable. After cutting 3 bytes off the end, only four records survive: 520 bytes are trustworthy and the remaining 127 bytes are the leftover front of the fifth record, which was never fully written and was therefore never acknowledged. After flipping one byte inside the third record, only two records are usable, and records four and five are perfectly intact but unreachable. That last result is a real cost of the simple rule. We keep it anyway, because it is safe: a store that guessed where the next record starts could replay garbage as if it were data.
Step 3: Freeze sorted data into an SSTable
When the memtable fills up, we write it to disk as a sorted, immutable file. A lookup in a sorted file can jump straight to the right place instead of scanning, as long as the file carries a little map of itself. This step builds the file format in three parts: a Bloom filter, a table writer and a table reader.
3.1 The Bloom filter that every table carries
We put a Bloom filter inside every table file now, because changing a file format later is painful, and measure what it buys in Step 5. With 10 bits per key the filter uses 7 hash positions per key (round(10 * ln 2)). To avoid hashing a key seven times, hash_pair computes two 64-bit hashes once and the filter derives all seven positions from them. Save this as bloom.py:
"""bloom.py: a Bloom filter small enough to live inside an SSTable file."""
import hashlib
import math
import struct
def hash_pair(key: bytes) -> tuple[int, int]:
"""Two 64-bit hashes of a key. Computed once per lookup and reused for every table."""
digest = hashlib.blake2b(key, digest_size=16).digest()
return int.from_bytes(digest[:8], "little"), int.from_bytes(digest[8:], "little") | 1
class Bloom:
def __init__(self, nbits: int, k: int, bits: bytearray | None = None):
self.nbits = max(64, nbits)
self.k = k
self.bits = bits if bits is not None else bytearray((self.nbits + 7) // 8)
@classmethod
def sized_for(cls, n_keys: int, bits_per_key: int = 10) -> "Bloom":
k = max(1, round(bits_per_key * math.log(2))) # 10 bits per key gives k = 7
return cls(n_keys * bits_per_key, k)
def add(self, h1: int, h2: int) -> None:
for i in range(self.k):
pos = (h1 + i * h2) % self.nbits
self.bits[pos >> 3] |= 1 << (pos & 7)
def might_contain(self, h1: int, h2: int) -> bool:
for i in range(self.k):
pos = (h1 + i * h2) % self.nbits
if not self.bits[pos >> 3] & (1 << (pos & 7)):
return False # one clear bit proves the key was never added
return True # every bit is set: present, or a false positive
def to_bytes(self) -> bytes:
return struct.pack(">II", self.nbits, self.k) + bytes(self.bits)
@classmethod
def from_bytes(cls, data: bytes) -> "Bloom":
nbits, k = struct.unpack_from(">II", data)
return cls(nbits, k, bytearray(data[8:8 + (nbits + 7) // 8]))might_contain returns False as soon as it finds one clear bit, because a key that was added would have set every one of its bits. If all bits are set, the key is probably there, but another set of keys may have set the same bits by chance: that is a false positive.
3.2 The file layout and the writer
One table file has four parts:
data section entry, entry, entry, ... sorted by key
entry = key length, value length, key, value
(a value length of -1 marks a tombstone)
meta section sparse index: the first key of every 16th entry and its file offset
the smallest key and the largest key
the Bloom filter bits
footer (36 bytes) meta offset, Bloom offset, entry count, CRC32 of the meta section, magic numberThe index is sparse on purpose. Indexing every entry would make the index about a fifth of the size of the data here (an index entry is 25 bytes, a data entry 121) and would need memory in proportion to the number of keys. With one index entry per 16 entries, a lookup binary-searches the index in memory, jumps to one block of at most 16 entries and scans it. The footer sits at the end because a writer does not know the offsets until everything else is written; a reader starts from the last 36 bytes. Save the first two parts of sstable.py: the constants and parser, and the writer.
"""sstable.py: immutable sorted tables: entries, a sparse index and a Bloom filter in one file."""
import bisect
import os
import struct
import zlib
from bloom import Bloom, hash_pair
ENTRY = struct.Struct(">Ii") # key length, value length (-1 marks a tombstone)
FOOTER = struct.Struct(">QQQI8s") # meta offset, bloom offset, entry count, CRC32 of meta, magic
MAGIC = b"LSMTBL01"
INDEX_EVERY = 16 # one index entry per 16 data entries
MISSING = object() # "this table has no entry for the key" (a tombstone is None)
class CorruptTable(Exception):
pass
def parse_entries(buf):
pos = 0
while pos < len(buf):
klen, vlen = ENTRY.unpack_from(buf, pos)
pos += ENTRY.size
key = bytes(buf[pos:pos + klen])
pos += klen
if vlen < 0:
yield key, None
else:
yield key, bytes(buf[pos:pos + vlen])
pos += vlen# sstable.py (continued)
def write_table(path, items, bits_per_key=10):
"""Write sorted, unique (key, value-or-None) pairs. Returns (entry count, file size)."""
index, hashes, count, first, last = [], [], 0, None, None
with open(path, "wb") as f:
for key, value in items:
if count % INDEX_EVERY == 0:
index.append((key, f.tell()))
f.write(ENTRY.pack(len(key), -1 if value is None else len(value)) + key + (value or b""))
hashes.append(hash_pair(key))
first = key if first is None else first
last = key
count += 1
bloom = Bloom.sized_for(max(count, 1), bits_per_key)
for h1, h2 in hashes:
bloom.add(h1, h2)
meta_off = f.tell()
meta = bytearray(struct.pack(">I", len(index)))
for key, offset in index:
meta += struct.pack(">IQ", len(key), offset) + key
for key in (first or b"", last or b""):
meta += struct.pack(">I", len(key)) + key
bloom_off = meta_off + len(meta)
meta += bloom.to_bytes()
f.write(meta)
f.write(FOOTER.pack(meta_off, bloom_off, count, zlib.crc32(meta), MAGIC))
f.flush()
os.fsync(f.fileno())
return count, f.tell()Two details matter. write_table ends with f.flush() and os.fsync, because a table must be on stable storage before anything is allowed to rely on it. Python’s documentation says to do exactly this for a buffered file: flush first, then fsync the file descriptor. And the entry format stores a value length of -1 for a tombstone, which is how an empty value (length 0) stays different from a deleted key.
3.3 The table reader
Opening a table reads the footer, then the meta section, verifies the checksum, and keeps the index and the Bloom filter in memory. A lookup then does at most one disk read. Add this to sstable.py:
# sstable.py (continued)
class Table:
def __init__(self, path, stats):
self.path, self.stats = path, stats
self.file = open(path, "rb")
self.size = os.fstat(self.file.fileno()).st_size
if self.size < FOOTER.size:
raise CorruptTable(f"{path}: too small to hold a footer")
self.file.seek(self.size - FOOTER.size)
meta_off, bloom_off, self.count, crc, magic = FOOTER.unpack(self.file.read(FOOTER.size))
if magic != MAGIC:
raise CorruptTable(f"{path}: bad magic number")
self.file.seek(meta_off)
meta = self.file.read(self.size - FOOTER.size - meta_off)
if zlib.crc32(meta) != crc:
raise CorruptTable(f"{path}: index or Bloom filter fails its checksum")
self.data_end = meta_off
(n,), pos = struct.unpack_from(">I", meta), 4
self.index_keys, self.index_offsets = [], []
for _ in range(n):
klen, offset = struct.unpack_from(">IQ", meta, pos)
pos += 12
self.index_keys.append(meta[pos:pos + klen])
self.index_offsets.append(offset)
pos += klen
(klen,) = struct.unpack_from(">I", meta, pos)
self.min_key = meta[pos + 4:pos + 4 + klen]
pos += 4 + klen
(klen,) = struct.unpack_from(">I", meta, pos)
self.max_key = meta[pos + 4:pos + 4 + klen]
self.bloom = Bloom.from_bytes(meta[bloom_off - meta_off:])Now the lookup itself, plus the two iterators the store will need: items reads every entry (used by compaction) and scan reads a key range.
# sstable.py (continued)
def get(self, key, hashes, use_bloom=True):
"""Return the value, None for a tombstone, or MISSING when this table knows nothing."""
s = self.stats
s.tables_consulted += 1
if key < self.min_key or key > self.max_key:
s.range_skips += 1
return MISSING
if use_bloom and not self.bloom.might_contain(*hashes):
s.bloom_skips += 1
return MISSING
i = bisect.bisect_right(self.index_keys, key) - 1
end = self.index_offsets[i + 1] if i + 1 < len(self.index_offsets) else self.data_end
self.file.seek(self.index_offsets[i])
block = self.file.read(end - self.index_offsets[i])
s.block_reads += 1
s.bytes_read += len(block)
for k, v in parse_entries(block):
if k == key:
return v
if k > key:
break
s.wasted_block_reads += 1 # we paid for a disk read and the key was not there
return MISSING
def items(self):
self.file.seek(0)
yield from parse_entries(self.file.read(self.data_end))
def scan(self, start=None, end=None):
i = 0 if start is None else max(0, bisect.bisect_right(self.index_keys, start) - 1)
self.file.seek(self.index_offsets[i])
for k, v in parse_entries(self.file.read(self.data_end - self.index_offsets[i])):
if end is not None and k >= end:
return
if start is None or k >= start:
yield k, v
def close(self):
self.file.close()Notice the sentinel MISSING. A table can answer three different things: “here is the value”, “this key is deleted here” (a tombstone, returned as None), and “I know nothing about this key”. If the last two were both None, a store could not tell “keep looking in older tables” from “stop, it is deleted”, and deleted keys would come back. That is the bug waiting in Step 7, and it starts with this one line.
Every outcome of get is counted: tables consulted, skips by key range, skips by Bloom filter, real block reads, bytes read, and “wasted” block reads where the key was not in the block. Those counters are how we will measure every claim later.
3.4 Try one table
Save this as step03_sstable.py. It writes 5,000 sorted entries to one table, looks up a key that exists, looks up keys that do not, and then damages the index to show that the table notices.
"""step03_sstable.py: write one sorted table and look keys up in it."""
import os
from bloom import hash_pair
from labkit import fresh_dir, key, value
from lsm import Stats
from sstable import FOOTER, MISSING, CorruptTable, Table, write_table
path = os.path.join(fresh_dir("step03"), "000001.sst")
count, size = write_table(path, [(key(i), value(i)) for i in range(5000)])
stats = Stats()
table = Table(path, stats)
print(f"{count} entries, file {size:,} bytes: data {table.data_end:,}, "
f"{len(table.index_keys)} index entries, Bloom filter {len(table.bloom.bits):,} bytes (k={table.bloom.k})")
with open(path, "rb") as f:
f.seek(-FOOTER.size, os.SEEK_END)
meta_off, bloom_off, entries, crc, magic = FOOTER.unpack(f.read())
print(f"footer: meta at {meta_off:,}, Bloom filter at {bloom_off:,}, entries {entries}, magic {magic.decode()}")
hit = key(3141)
found = table.get(hit, hash_pair(hit))
print(f"hit: value starts {found[:12]!r}, {stats.block_reads} block read of {stats.bytes_read} bytes "
f"(the data section is {table.data_end:,} bytes)")
absent = key(3141) + b"x" # sorts inside the key range but was never written
for use_bloom in (False, True):
before = (stats.block_reads, stats.bloom_skips, stats.wasted_block_reads)
result = table.get(absent, hash_pair(absent), use_bloom)
after = (stats.block_reads, stats.bloom_skips, stats.wasted_block_reads)
print(f"miss, bloom={str(use_bloom):5}: found={result is not MISSING}, disk reads {after[0] - before[0]}, "
f"skipped by Bloom {after[1] - before[1]}, wasted reads {after[2] - before[2]}")
outside = key(99_999_999)
before = stats.range_skips
table.get(outside, hash_pair(outside))
print(f"miss outside the range: range skips +{stats.range_skips - before}, no disk read")
table.close()
raw = bytearray(open(path, "rb").read())
raw[table.data_end + 5] ^= 0xFF # damage one byte of the index section
open(path, "wb").write(bytes(raw))
try:
Table(path, Stats())
except CorruptTable as exc:
print("damaged index ->", type(exc).__name__ + ":", os.path.basename(str(exc)))Run python step03_sstable.py:
5000 entries, file 619,157 bytes: data 605,000, 313 index entries, Bloom filter 6,250 bytes (k=7)
footer: meta at 605,000, Bloom filter at 612,863, entries 5000, magic LSMTBL01
hit: value starts b'v0:3141:....', 1 block read of 1936 bytes (the data section is 605,000 bytes)
miss, bloom=False: found=False, disk reads 1, skipped by Bloom 0, wasted reads 1
miss, bloom=True : found=False, disk reads 0, skipped by Bloom 1, wasted reads 0
miss outside the range: range skips +1, no disk read
damaged index -> CorruptTable: 000001.sst: index or Bloom filter fails its checksumThe 5,000 entries are 605,000 bytes of data (5,000 times 121). The index holds 313 entries, one per 16 entries, and the Bloom filter is 6,250 bytes (10 bits for each of the 5,000 keys). Looking up key 3141 took one block read of 1,936 bytes (16 entries of 121 bytes), not the whole 605,000-byte data section. A key that sorts inside the table’s range but is absent costs one wasted disk read when the Bloom filter is off, and none when it is on. A key outside the smallest and largest key is rejected by a comparison before the filter is even asked. And a damaged index is refused when the table is opened, because the footer holds a checksum of the meta section. The data section itself carries no checksum in this lab, which is a gap a real format would close.
Step 4: Build the store: memtable, manifest, flush and read path
Now the pieces join. The store accepts writes into a memtable and a log, flushes a full memtable into a new table, and answers reads by checking the memtable and then the tables from newest to oldest. We will build lsm.py in parts. Each part goes into the same file; the first one creates it.
4.1 Counters
The Stats class holds every counter we will report. write_amplification divides all bytes written to disk (log, flushed tables and compacted tables) by the bytes of keys and values the user stored. Start lsm.py with:
"""lsm.py: a small log-structured merge-tree key-value store."""
import heapq
import json
import os
from dataclasses import asdict, dataclass
from bloom import hash_pair
from sstable import MISSING, Table, write_table
from wal import OP_DELETE, OP_PUT, WalWriter, replay
@dataclass
class Stats:
user_bytes: int = 0
wal_bytes: int = 0
flush_bytes: int = 0
compaction_read_bytes: int = 0
compaction_write_bytes: int = 0
tables_consulted: int = 0
range_skips: int = 0
bloom_skips: int = 0
block_reads: int = 0
wasted_block_reads: int = 0
bytes_read: int = 0
flushes: int = 0
compactions: int = 0
def write_amplification(self) -> float:
total = self.wal_bytes + self.flush_bytes + self.compaction_write_bytes
return total / self.user_bytes if self.user_bytes else 0.0
def snapshot(self) -> dict:
return asdict(self)The step scripts also need two helpers that read those counters. lookups runs a list of gets and returns per-lookup averages, and levels_text prints the shape of the store. Add them to the end of labkit.py:
# labkit.py (continued)
def lookups(db, keys):
"""Run db.get for every key and return per-lookup averages of the read counters."""
before = db.stats.snapshot()
for k in keys:
db.get(k)
after = db.stats.snapshot()
names = ("tables_consulted", "range_skips", "bloom_skips", "block_reads", "wasted_block_reads", "bytes_read")
return {name: (after[name] - before[name]) / len(keys) for name in names}
def levels_text(db):
return " ".join(f"L{i}:{runs}x/{size // 1024}KiB" for i, runs, size in db.describe())4.2 Merging sorted streams
Flushes and compactions both need one operation: take several sorted streams of entries and produce one sorted stream where, for a key that appears in several streams, only the newest version survives. heapq.merge does the sorted merge. We tag each entry with the rank of its stream (0 is the newest), so for equal keys the lowest rank comes out first and everything after it is skipped. Add this to lsm.py:
# lsm.py (continued)
def _tag(rank, entries):
for key, value in entries:
yield key, rank, value # rank breaks ties between equal keys: 0 is the newest source
def merge_newest_first(sources, drop_tombstones=False):
"""Merge sorted (key, value) streams. sources[0] is the newest; on equal keys it wins."""
streams = [_tag(rank, source) for rank, source in enumerate(sources)]
last = None
for key, _rank, value in heapq.merge(*streams):
if key == last:
continue # an older version of a key we already emitted
last = key
if value is None and drop_tombstones:
continue
yield key, value
def _id_of(path: str) -> int:
return int(os.path.basename(path).split(".")[0])The helper _tag exists because of a Python trap. If you write the tagging inline, as [((k, rank, v) for k, v in src) for rank, src in enumerate(sources)], each generator looks up rank only when it runs, after the loop has finished, so every stream carries the last rank. Running that exact line on three one-entry sources printed [[(1, 2, 'a')], [(1, 2, 'b')], [(1, 2, 'c')]]: the same rank 2 everywhere, and the “newest wins” rule silently stops working. A function binds its argument when it is called, so the rank is right.
4.3 The store object
The constructor takes a handful of switches. memtable_bytes is how big the memtable may grow before a flush; we use 64 KiB so that a small workload produces many flushes, where LevelDB’s documentation says its log reaches “approximately 4MB by default”. policy picks the compaction strategy ("none", "tiered" or "leveled"), sync chooses whether every log write is fsynced, use_bloom turns the filters on, and tier_runs and level_ratio tune compaction. The last two switches, unsafe_drop_tombstones and truncate_torn_tail, exist only so that we can reproduce two bugs on demand later.
# lsm.py (continued)
class LSM:
def __init__(self, path, *, memtable_bytes=64 * 1024, policy="leveled", sync="always",
use_bloom=True, bits_per_key=10, tier_runs=4, level_ratio=10,
unsafe_drop_tombstones=False, truncate_torn_tail=True):
self.dir, self.memtable_bytes, self.policy, self.sync = path, memtable_bytes, policy, sync
self.use_bloom, self.bits_per_key = use_bloom, bits_per_key
self.tier_runs, self.level_ratio = tier_runs, level_ratio
self.unsafe_drop_tombstones = unsafe_drop_tombstones
self.truncate_torn_tail = truncate_torn_tail
self.stats = Stats()
os.makedirs(path, exist_ok=True)
self._recover()4.4 Files, recovery and the commit point
A store directory holds four kinds of file. MANIFEST is a small JSON file that lists which tables are live at each level and which log is current. NNNNNN.wal is the log for the current memtable. NNNNNN.sst is a table. MANIFEST.tmp exists only while the manifest is being replaced. LevelDB uses the same idea: “A MANIFEST file lists the set of sorted tables that make up each level, the corresponding key ranges, and other important metadata.”
The manifest is the single source of truth. _commit writes the new manifest to a temporary file, fsyncs it, and swaps it in with os.replace. Python’s documentation says: “If successful, the renaming will be an atomic operation (this is a POSIX requirement).” Atomic means a reader sees the old manifest or the new one, never half of each. Any file the manifest does not mention is garbage. When the store opens, _recover deletes such files, loads the tables, replays the current log into a fresh memtable, and cuts off any torn tail from the log. LevelDB’s documentation describes the same cleanup: “It deletes all log files that are not the current log file. It deletes all table files that are not referenced from some level and are not the output of an active compaction.”
# lsm.py (continued)
# ---- files and recovery -------------------------------------------------
def _path(self, ident, ext):
return os.path.join(self.dir, f"{ident:06d}.{ext}")
def _take_id(self):
self.next_id += 1
return self.next_id - 1
def _recover(self):
manifest = os.path.join(self.dir, "MANIFEST")
state = {"next_id": 1, "wal_id": 0, "levels": []}
if os.path.exists(manifest):
with open(manifest) as f:
state = json.load(f)
self.next_id, self.wal_id = state["next_id"], state["wal_id"]
live = {ident for level in state["levels"] for ident in level}
for name in os.listdir(self.dir): # delete what the manifest does not vouch for
path = os.path.join(self.dir, name)
if name == "MANIFEST.tmp":
os.remove(path)
elif name.endswith(".sst") and _id_of(path) not in live:
os.remove(path)
elif name.endswith(".wal") and _id_of(path) != self.wal_id:
os.remove(path)
self.levels = [[Table(self._path(i, "sst"), self.stats) for i in level]
for level in state["levels"]] or [[]]
self.mem, self.mem_bytes = {}, 0
wal_path = self._path(self.wal_id, "wal")
records, good, total = replay(wal_path)
if good < total and self.truncate_torn_tail:
with open(wal_path, "r+b") as f:
f.truncate(good) # cut the torn tail, or new records would sit behind garbage
for op, key, value in records:
self._apply(op, key, value)
self.recovery = {"records": len(records), "discarded_bytes": total - good}
self.wal = WalWriter(wal_path, self.sync)
if not os.path.exists(manifest):
self._commit()
def _commit(self):
"""Atomically publish the list of live files. This is the commit point for flushes and compactions."""
state = {"next_id": self.next_id, "wal_id": self.wal_id,
"levels": [[_id_of(t.path) for t in level] for level in self.levels]}
tmp = os.path.join(self.dir, "MANIFEST.tmp")
with open(tmp, "w") as f:
json.dump(state, f)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, os.path.join(self.dir, "MANIFEST")) # readers see the old file or the new one
def close(self):
self.wal.close()
for level in self.levels:
for table in level:
table.close()Look at the line f.truncate(good). If a log ends in a torn record and you reopen it for appending without cutting the torn bytes off first, every record you write afterwards sits behind garbage, and the next recovery will stop at the garbage and never see them. Step 8 reproduces that bug.
4.5 The write path and the flush
put and delete go through _write, which appends to the log first (“durable first”) and only then changes the memtable (“visible second”). If the process dies between the two, the log has a record the memtable never saw, and recovery will replay it. A delete stores None in the memtable: that is the tombstone.
flush turns the memtable into a table in four moves: write the table and fsync it, create a new empty log, update the manifest, and finally delete the old log. The manifest update is the commit point. A crash before it leaves the old manifest, which still names the old log that holds everything; the new table and the new log are strays that the next start deletes. A crash after it leaves the new manifest, and the old log is the stray. Either way no acknowledged write is lost. Step 8 tests both claims.
# lsm.py (continued)
# ---- writes -------------------------------------------------------------
def put(self, key: bytes, value: bytes):
self._write(OP_PUT, key, value)
def delete(self, key: bytes):
self._write(OP_DELETE, key, b"")
def _write(self, op, key, value):
self.stats.user_bytes += len(key) + len(value)
self.stats.wal_bytes += self.wal.append(op, key, value) # durable first ...
self._apply(op, key, value) # ... visible second
if self.mem_bytes >= self.memtable_bytes:
self.flush()
def _apply(self, op, key, value):
old = self.mem.get(key, MISSING)
if old is not MISSING:
self.mem_bytes -= len(key) + len(old or b"") + 8
self.mem[key] = value if op == OP_PUT else None # None is the tombstone
self.mem_bytes += len(key) + len(value) + 8
def flush(self):
if not self.mem:
return
path = self._path(self._take_id(), "sst")
_, size = write_table(path, sorted(self.mem.items()), self.bits_per_key)
self.stats.flush_bytes += size
self.stats.flushes += 1
old_wal, old_wal_path = self.wal, self.wal.path
self.wal_id = self._take_id()
self.wal = WalWriter(self._path(self.wal_id, "wal"), self.sync)
old_wal.close()
self.levels[0].insert(0, Table(path, self.stats))
self._commit() # the commit point: new table listed, new log current
os.remove(old_wal_path)
self.mem, self.mem_bytes = {}, 0
self._maybe_compact()4.6 The read path
get checks the memtable first, because it holds the newest data. Then it walks the levels from shallow to deep, and inside a level from the newest table to the oldest, and returns the first answer it finds. A None answer means deleted, so it stops there too. scan returns a sorted range by merging the memtable and every table with the same merge function, and hides tombstones. describe reports one row per level.
# lsm.py (continued)
# ---- reads --------------------------------------------------------------
def get(self, key: bytes):
if key in self.mem:
return self.mem[key] # None means "deleted"
hashes = hash_pair(key) if self.use_bloom else None
for level in self.levels: # shallow levels hold newer data
for table in level: # inside a level, newest table first
found = table.get(key, hashes, self.use_bloom)
if found is not MISSING:
return found
return None
def scan(self, start=None, end=None):
mem = sorted((k, v) for k, v in self.mem.items()
if (start is None or k >= start) and (end is None or k < end))
sources = [iter(mem)] + [t.scan(start, end) for level in self.levels for t in level]
for key, value in merge_newest_first(sources):
if value is not None:
yield key, value
def describe(self):
"""One (level, number of tables, bytes) row per level."""
return [(i, len(level), sum(t.size for t in level)) for i, level in enumerate(self.levels)]One thing is missing. flush ends by calling _maybe_compact, which does not exist yet. Add a placeholder at the end of the class so the store runs; Step 6 replaces it with the real thing:
# lsm.py (continued)
def _maybe_compact(self):
pass # Step 6 replaces this4.7 Run the read path with no Bloom filters and no compaction
Save this as step04_read_path.py. It inserts 10,000 keys in random order, which makes every table span the whole key range (the way hashed IDs behave), with compaction off and Bloom filters off. Then it looks up 1,000 keys that exist and 1,000 keys that do not (inside the key range, so the range check cannot help).
"""step04_read_path.py: a store that only flushes. Every table must be asked."""
import random
from labkit import fresh_dir, key, levels_text, lookups, value
from lsm import LSM
ids = list(range(10_000))
random.Random(1).shuffle(ids) # random arrival order, so every table spans the whole key range
db = LSM(fresh_dir("step04"), policy="none", use_bloom=False, sync="none")
for i in ids:
db.put(key(i), value(i))
print("after 10,000 puts:", levels_text(db), f"| memtable holds {len(db.mem)} keys")
sample = random.Random(2).sample(range(10_000), 1_000)
hits = lookups(db, [key(i) for i in sample])
misses = lookups(db, [key(i) + b"x" for i in sample]) # inside the key range, never written
for name, r in (("hit", hits), ("miss", misses)):
print(f"{name:4} per get: {r['tables_consulted']:5.1f} tables asked, {r['block_reads']:5.1f} disk reads")
print("scan user:00000100 .. user:00000105:", [k.decode() for k, _ in db.scan(key(100), key(105))])
db.close()Run python step04_read_path.py:
after 10,000 puts: L0:18x/1181KiB | memtable holds 244 keys
hit per get: 9.4 tables asked, 9.4 disk reads
miss per get: 18.0 tables asked, 18.0 disk reads
scan user:00000100 .. user:00000105: ['user:00000100', 'user:00000101', 'user:00000102', 'user:00000103', 'user:00000104']The 10,000 puts produced 18 tables, plus 244 keys still in the memtable. A lookup for a key that exists asks 9.4 tables on average and reads a block from each of them, because without a filter the only way to learn that a table does not hold the key is to read it. A lookup for a missing key asks all 18. The last line shows scan returning keys 100 to 104 in order, merged from many tables. This is the problem Bigtable’s paper warns about: “Every minor compaction creates a new SSTable. If this behavior continued unchecked, read operations might need to merge updates from an arbitrary number of SSTables.” Steps 5 and 6 fix it from two directions.
Step 5: Add Bloom filters and measure them
The filters are already inside every table; the store only has to consult them. Table.get asks the filter before touching the disk, and the store computes the key’s hash pair once and reuses it for every table. Bigtable’s paper explains why that is worth doing: “A Bloom filter allows us to ask whether an SSTable might contain any data for a specified row/column pair.” Save this as step05_bloom.py. It uses the same 10,000-key store as Step 4 and flips use_bloom at run time, so both measurements read exactly the same files.
"""step05_bloom.py: Bloom filters turn most missing-key lookups into zero disk reads."""
import math
import random
from labkit import fresh_dir, key, levels_text, lookups, value
from lsm import LSM
ids = list(range(10_000))
random.Random(1).shuffle(ids)
db = LSM(fresh_dir("step05"), policy="none", use_bloom=True, sync="none")
for i in ids:
db.put(key(i), value(i))
print("store:", levels_text(db))
sample = random.Random(2).sample(range(10_000), 1_000)
present = [key(i) for i in sample]
absent = [key(i) + b"x" for i in sample] # inside the key range, never written
print(f"{'':16}{'tables asked':>13}{'Bloom skips':>13}{'disk reads':>12}{'wasted reads':>14}")
for use_bloom in (False, True):
db.use_bloom = use_bloom # the filters are stored in every table either way
for label, keys in (("miss", absent), ("hit", present)):
r = lookups(db, keys)
print(f"bloom={str(use_bloom):5} {label:4}{r['tables_consulted']:>13.1f}{r['bloom_skips']:>13.1f}"
f"{r['block_reads']:>12.2f}{r['wasted_block_reads']:>14.2f}")
db.use_bloom = True
r = lookups(db, absent)
measured = r["wasted_block_reads"] / (r["tables_consulted"] - r["range_skips"])
theory = (1 - math.exp(-7 / 10)) ** 7 # k = 7 hash functions, 10 bits per key
print(f"false positive rate: measured {measured:.4f}, theory {theory:.4f}")
table = db.levels[0][0]
print(f"one table: {table.size:,} bytes, Bloom filter {len(table.bloom.bits):,} bytes ({len(table.bloom.bits) / table.size:.1%} of the file)")
db.close()Run python step05_bloom.py:
store: L0:18x/1181KiB
tables asked Bloom skips disk reads wasted reads
bloom=False miss 18.0 0.0 17.97 17.97
bloom=False hit 9.4 0.0 9.37 8.39
bloom=True miss 18.0 17.8 0.17 0.17
bloom=True hit 9.4 8.3 1.06 0.08
false positive rate: measured 0.0096, theory 0.0082
one table: 67,192 bytes, Bloom filter 678 bytes (1.0% of the file)Read the rows for missing keys first. With the filters off, a missing-key lookup read 17.97 blocks on average, all of them wasted. With the filters on, 17.8 of the 18 tables were skipped without touching the disk and the average dropped to 0.17 reads. For keys that exist, reads fell from 9.37 to 1.06: one read to fetch the value (a few keys sit in the memtable and need none), plus about 0.08 reads wasted on false positives.
The measured false-positive rate was 0.0096 against a theoretical 0.0082. The theory is the standard formula (1 - e^(-k/b))^k, where b is bits per key and k the number of hash positions: with b = 10 and k = 7 it gives (1 - 0.4966)^7, about 0.0082. The measurement lands a little higher, which is plausible for a filter that derives all seven positions from two hashes, but I did not investigate the gap further. The price is small: one table’s filter is 678 bytes of a 67,192-byte file, about 1.0 percent. Bigtable’s paper makes the same trade: “For certain applications, a small amount of tablet server memory used for storing Bloom filters drastically reduces the number of disk seeks required for read operations.”
Step 6: Compact the tables, and compare three policies
Filters make each extra table cheap to skip, but they do not make the tables go away, and every overwritten value still sits in an old file. Compaction fixes both: it merges tables into one, keeps only the newest version of each key, and deletes the inputs. The question is when to do it, and that is where policies differ.
6.1 The two policies
Size-tiered (called "tiered" in the code) waits until a level has tier_runs tables (4 by default) and then merges all of them into one table at the next level. A byte is rewritten once per level it passes through. Writes are cheap, but each level can hold several tables, so reads ask several.
Leveled keeps each deep level as one sorted run, with a size limit that grows by level_ratio (10) per level. New tables collect in level 0 (their key ranges overlap, so they are merged together), and when level 0 reaches 4 tables it is merged with level 1. When a level exceeds its limit it is merged into the next one. Reads ask about one table per level, but every merge into a level rewrites that level’s existing bytes along with the new ones. LevelDB’s documentation describes this layout: “Files in the young level may contain overlapping keys. However files in other levels have distinct non-overlapping key ranges.” It also gives the trigger we copied: “When the number of young files exceeds a certain threshold (currently four), all of the young files are merged together with all of the overlapping level-1 files”.
One honest simplification: LevelDB and RocksDB split each level into many files and compact one file at a time, which keeps every compaction small. RocksDB’s documentation says: “Each level (except level 0) is one data sorted run” and then “Inside each level (except level 0), data is range partitioned into multiple SST files”. Our levels are a single table each, so a compaction rewrites a whole level. That makes our write amplification coarser than the real thing, and the comparison between policies is still fair, because both run on the same engine.
6.2 The code
First, delete the placeholder _maybe_compact you added at the end of Step 4. Then add the policy logic. _pick looks at the shape of the levels and returns which levels to merge and where the result goes (or nothing). _maybe_compact keeps asking until there is no work left, because one merge can overfill the next level.
# lsm.py (continued)
# ---- compaction ---------------------------------------------------------
def _level_limit(self, level):
return self.tier_runs * self.memtable_bytes * self.level_ratio ** (level - 1)
def _pick(self):
lv = self.levels
if self.policy == "tiered":
for i, runs in enumerate(lv):
if len(runs) >= self.tier_runs:
return [i], i + 1 # merge every run of level i into one run of level i+1
elif self.policy == "leveled":
if len(lv[0]) >= self.tier_runs:
return [0, 1], 1 # level 0 runs overlap, so they are merged with level 1 together
for i in range(1, len(lv)):
if sum(t.size for t in lv[i]) > self._level_limit(i):
return [i, i + 1], i + 1
return None
def _maybe_compact(self):
while (job := self._pick()) is not None:
self._compact(*job)The merge itself is _compact. Read the order of operations, because it is what makes a crash safe: write the new table (fsynced), then update the manifest (the commit point), and only then close and delete the input tables. Closing first matters on Windows, where a file that is still open cannot be deleted. Also find the line that sets drop: it decides whether tombstones may be thrown away during this merge. We will come back to it in Step 7, because it is the most dangerous line in the file.
# lsm.py (continued)
def _compact(self, sources, target):
while len(self.levels) <= max(max(sources), target):
self.levels.append([])
victims = [t for i in sources for t in self.levels[i]] # newest first
# A tombstone may only be dropped when no older table could still hold the key.
drop = self.unsafe_drop_tombstones or all(not lv for lv in self.levels[max(sources) + 1:])
path = self._path(self._take_id(), "sst")
count, size = write_table(path, merge_newest_first([t.items() for t in victims], drop),
self.bits_per_key)
self.stats.compactions += 1
self.stats.compaction_read_bytes += sum(t.data_end for t in victims)
self.stats.compaction_write_bytes += size
for i in sources:
self.levels[i] = []
if count:
self.levels[target].insert(0, Table(path, self.stats))
else:
os.remove(path) # everything cancelled out
self._commit()
for table in victims:
table.close()
os.remove(table.path)
while len(self.levels) > 1 and not self.levels[-1]:
self.levels.pop()Finally, compact_all merges everything into the bottom level. Bigtable’s paper calls that a major compaction: “A merging compaction that rewrites all SSTables into exactly one SSTable is called a major compaction.” We use it to measure how much space the same live data needs when it is fully packed.
# lsm.py (continued)
def compact_all(self):
"""A major compaction: merge everything into the bottom level, dropping dead entries."""
self.flush()
if any(self.levels):
bottom = len(self.levels) - 1
self._compact(list(range(bottom + 1)), bottom)6.3 Run the same workload under each policy
Save this as step06_compaction.py. It performs 60,000 puts over 20,000 distinct keys (so every key is overwritten about three times), once per policy, and then measures lookups, write amplification and space.
"""step06_compaction.py: the same workload under three compaction policies."""
import random
import time
from labkit import fresh_dir, key, levels_text, lookups, value
from lsm import LSM
OPS, DISTINCT = 60_000, 20_000
print(f"workload: {OPS:,} puts over {DISTINCT:,} distinct keys, 100-byte values, 64 KiB memtable")
print(f"{'policy':8}{'flushes':>8}{'compactions':>12}{'write amp':>10}{'hit: asked/reads':>18}{'miss: asked/reads':>19}{'space amp':>10} runs per level")
latency_lines = []
for policy in ("none", "tiered", "leveled"):
db = LSM(fresh_dir(f"step06_{policy}"), policy=policy, sync="none")
rng = random.Random(42)
started = time.perf_counter()
put_times = []
for n in range(OPS):
t0 = time.perf_counter()
db.put(key(rng.randrange(DISTINCT)), value(0, n))
put_times.append(time.perf_counter() - t0)
seconds = time.perf_counter() - started
put_times.sort()
latency_lines.append(f"put latency, {policy:7}: median {put_times[len(put_times) // 2] * 1e6:5.0f} microseconds, "
f"slowest {put_times[-1] * 1e3:6.1f} ms")
shape = levels_text(db)
stats = db.stats
write_amp, flushes, compactions = stats.write_amplification(), stats.flushes, stats.compactions
sample = random.Random(3).sample(range(DISTINCT), 1_000)
hit = lookups(db, [key(i) for i in sample])
miss = lookups(db, [key(i) + b"x" for i in sample])
disk = sum(size for _, _, size in db.describe())
db.compact_all()
packed = sum(size for _, _, size in db.describe()) # what the same live data needs in one table
print(f"{policy:8}{flushes:>8}{compactions:>12}"
f"{write_amp:>10.1f}"
f"{hit['tables_consulted']:>10.1f}/{hit['block_reads']:<6.2f}"
f"{miss['tables_consulted']:>10.1f}/{miss['block_reads']:<7.2f}"
f"{disk / packed:>9.2f}x {shape} ({seconds:.1f}s)")
db.close()
print()
for line in latency_lines:
print(line)Run python step06_compaction.py. It takes about ten seconds:
workload: 60,000 puts over 20,000 distinct keys, 100-byte values, 64 KiB memtable
policy flushes compactions write amp hit: asked/reads miss: asked/reads space amp runs per level
none 109 0 2.2 33.5/1.25 109.0/0.92 3.11x L0:109x/7152KiB (2.0s)
tiered 109 34 4.3 5.0/0.98 7.0/0.05 1.98x L0:1x/65KiB L1:3x/760KiB L2:2x/1718KiB L3:1x/2003KiB (4.2s)
leveled 109 40 7.0 2.8/0.97 3.0/0.02 1.13x L0:1x/65KiB L1:1x/253KiB L2:1x/2281KiB (4.7s)
put latency, none : median 7 microseconds, slowest 18.1 ms
put latency, tiered : median 12 microseconds, slowest 395.9 ms
put latency, leveled: median 12 microseconds, slowest 408.8 msTake the columns one at a time.
- Write amplification. With no compaction it is 2.2: every byte is written twice, once to the log and once in a flushed table. Tiered compaction raises it to 4.3 and leveled to 7.0, because bytes are rewritten as they move down. This store has only three levels; a bigger one would show a larger number, and the RocksDB documentation puts the usual figure for leveled compaction at “often larger than 10”, so treat 7.0 as the result for this small store, not as a prediction for yours.
- Read cost. With 109 uncompacted tables, a missing key asks all 109 and a key that exists asks 33.5 on average. Tiered compaction brings that to 7 and 5.0, leveled to 3 and 2.8. Disk reads per hit stay close to 1 in every row; the filters see to that. The difference is how many tables each lookup has to ask, and how many filter false positives it risks: 0.92 wasted reads per missing-key lookup with 109 tables, 0.05 with tiered and 0.02 with leveled.
- Space amplification. Without compaction the disk holds 3.11 times what the live data needs, because every overwritten version is still in a file. Tiered brings it to 1.98 and leveled to 1.13.
The shapes show why. Tiered ended with 1, 3, 2 and 1 tables on four levels, so a lookup asks 7 tables. Leveled ended with one table each on three levels. RocksDB’s documentation describes the same trade: “It is generally regarded that the second strategy provides far better write amplification with worse read amplification”, where the second strategy is the tiered one. Its Universal Compaction page opens with the goal: “Universal Compaction Style is a compaction style, targeting the use cases requiring lower write amplification, trading off read amplification and space amplification.”
The last three lines show the price of doing the merge inside put. The median put took microseconds in every policy. The slowest single put took 18.1 ms with no compaction (a flush that ends with an fsync), and 395.9 ms with tiered and 408.8 ms with leveled: one unlucky write paid for a whole merge. Which policy has the worse spike on a given run depends on where the big merges fall, so read these as an order of magnitude. A production engine moves this work off the writer’s path. LevelDB compacts in a background thread and only slows or stops writers when level-0 files pile up; the constants in its source read “Level-0 compaction is started when we hit this many files.” (4), “Soft limit on number of level-0 files. We slow down writes at this point.” (8) and “Maximum number of level-0 files. We stop writes at this point.” (12). Our single-threaded store trades that machinery for readable code.
Step 7: Deletes, tombstones, and the zombie-key bug
A delete cannot remove anything from an immutable file, so it writes a tombstone: a newer entry that says the key is gone. Reads see the tombstone first and stop. The old value is still on disk in an older table, and the tombstone exists to hide it. LevelDB’s documentation: “Deletion markers are kept around to hide obsolete values present in older sorted tables”.
That gives a compaction a hard rule. It may throw a tombstone away only when no older table could still hold a value for the key; otherwise the old value is uncovered and the deleted key comes back. LevelDB’s documentation states the rule: “Compactions drop overwritten values. They also drop deletion markers if there are no higher numbered levels that contain a file whose range overlaps the current key.” (“Higher numbered” means deeper, so older.) The code in LevelDB’s db_impl.cc carries the proof in a comment, which begins with the condition “there is no data in higher levels” and ends “Therefore this deletion marker is obsolete and can be dropped.” Bigtable’s paper says the same thing at the scale of a whole table: “SSTables produced by non-major compactions can contain special deletion entries that suppress deleted data in older SSTables that are still live. A major compaction, on the other hand, produces an SSTable that contains no deletion information or deleted data.”
Our _compact implements the rule in one line: drop is true only when every level deeper than the deepest input is empty. That is stricter than LevelDB’s per-file overlap check, so it keeps some tombstones longer than it needs to, but it errs on the safe side. The constructor switch unsafe_drop_tombstones=True forces drop to be true always, which is the bug a careless implementation has.
Save this as step07_tombstones.py. It loads 3,000 keys, deletes every third one (1,000 deletes), writes 1,000 more keys to trigger compactions, and counts how many deleted keys read back as alive. It runs the store twice: with the unsafe switch and without.
"""step07_tombstones.py: deletes are writes, and compaction must not forget them too early."""
from labkit import fresh_dir, key, levels_text, value
from lsm import LSM
PARAMS = dict(memtable_bytes=16 * 1024, tier_runs=3, level_ratio=3, policy="leveled", sync="none")
DELETED = range(0, 3000, 3) # every third key: 1,000 deletes
def tombstones(db):
return sum(1 for level in db.levels for t in level for _, v in t.items() if v is None)
def report(label, db):
zombies = [i for i in DELETED if db.get(key(i)) is not None]
in_memtable = sum(1 for v in db.mem.values() if v is None)
print(f"{label} {levels_text(db)}")
print(f" tombstones: {tombstones(db)} in tables + {in_memtable} in the memtable; "
f"deleted keys that read back as alive: {len(zombies)} of {len(DELETED)}")
def run(unsafe):
db = LSM(fresh_dir(f"step07_{unsafe}"), unsafe_drop_tombstones=unsafe, **PARAMS)
print(f"--- unsafe_drop_tombstones={unsafe}")
for i in range(3000):
db.put(key(i), value(i))
print("1. loaded 3,000 keys:", levels_text(db))
for i in DELETED:
db.delete(key(i))
report("2. after 1,000 deletes:", db)
for i in range(10_000, 11_000):
db.put(key(i), value(i))
report("3. after 1,000 more writes:", db)
db.close()
run(True)
run(False)Run python step07_tombstones.py:
--- unsafe_drop_tombstones=True
1. loaded 3,000 keys: L0:1x/16KiB L1:0x/0KiB L2:1x/49KiB L3:1x/296KiB
2. after 1,000 deletes: L0:2x/34KiB L1:0x/0KiB L2:1x/49KiB L3:1x/296KiB
tombstones: 735 in tables + 265 in the memtable; deleted keys that read back as alive: 0 of 1000
3. after 1,000 more writes: L0:0x/0KiB L1:0x/0KiB L2:0x/0KiB L3:0x/0KiB L4:1x/466KiB
tombstones: 0 in tables + 0 in the memtable; deleted keys that read back as alive: 952 of 1000
--- unsafe_drop_tombstones=False
1. loaded 3,000 keys: L0:1x/16KiB L1:0x/0KiB L2:1x/49KiB L3:1x/296KiB
2. after 1,000 deletes: L0:2x/34KiB L1:0x/0KiB L2:1x/49KiB L3:1x/296KiB
tombstones: 735 in tables + 265 in the memtable; deleted keys that read back as alive: 0 of 1000
3. after 1,000 more writes: L0:0x/0KiB L1:0x/0KiB L2:0x/0KiB L3:1x/351KiB
tombstones: 0 in tables + 0 in the memtable; deleted keys that read back as alive: 0 of 1000Stage 2 is the same in both stores: right after the deletes, 735 tombstones sit in tables and 265 are still in the memtable, and zero deleted keys read back as alive, because each tombstone hides its old value. Stage 3 is where they part. In the unsafe store, compaction merged the tombstones into level 1 and threw them away while the old values were still in levels 2 and 3. After the cascade finished, 952 of the 1,000 deleted keys read back as alive, with their old values. The safe store kept its tombstones until a merge reached the bottom level, where they were discarded together with the values they hid: zero zombies, zero tombstones left, and one table of 351 KiB.
Why 952 and not 1,000? The other 48 deleted keys came from the last batch of the load, which was still in level 0 when the deletes arrived. The merge that dropped their tombstones also contained their old values and discarded them in the same pass. I checked that with a throwaway script: all 48 had their old values in level 0 or the memtable, and none had one in a deeper level. This kind of bug is nasty because nothing fails at the moment it happens. The data is wrong later, after unrelated compactions, and only for some keys.
Step 8: Crash it five ways
A store that has never been crashed has not been tested. This step uses real process kills and simulated disk damage to check the claims we made about the log, the manifest and the flush order.
8.1 A child process that dies without cleaning up
Save this as crash_child.py. It writes keys with sync="always" and prints each key’s number only after put has returned, which is the moment the store has acknowledged the write. Then, at a chosen count, it calls os._exit(1), which ends the process immediately with no close(), no flush and no cleanup.
"""crash_child.py: write keys, print each index after put() returns, then die without cleaning up."""
import os
import sys
from labkit import key, value
from lsm import LSM
def main(path, crash_after):
db = LSM(path, memtable_bytes=64 * 1024, sync="always")
for i in range(10_000):
db.put(key(i), value(i))
print(i, flush=True) # printed only after put() returned: the write was acknowledged
if i + 1 == crash_after:
os._exit(1) # die now: no close(), no flush(), no cleanup of any kind
main(sys.argv[1], int(sys.argv[2]))8.2 The five experiments
Save this as step08_crash.py. It runs the child at two crash points, then simulates a torn log tail, the same torn tail with recovery that forgets to cut it, a power failure between writing a table and committing the manifest, and finally times the cost of fsync.
"""step08_crash.py: five ways to die, and what the store does about each."""
import os
import subprocess
import sys
import time
from labkit import fresh_dir, key, value
from lsm import LSM
from wal import OP_PUT, encode
HERE = os.path.dirname(os.path.abspath(__file__))
def reopen_check(path, expected):
db = LSM(path, sync="always")
lost = [i for i in expected if db.get(key(i)) != value(i)]
info = (db.recovery, sum(len(level) for level in db.levels))
db.close()
return lost, info
print("1. The process is killed with os._exit (the child prints an index only after put() returned)")
for crash_after in (1_234, 5_000):
path = fresh_dir(f"step08_kill{crash_after}")
run = subprocess.run([sys.executable, os.path.join(HERE, "crash_child.py"), path, str(crash_after)],
capture_output=True, text=True)
acked = [int(line) for line in run.stdout.split()]
lost, (recovery, tables) = reopen_check(path, acked)
print(f" exit code {run.returncode}, {len(acked):,} puts acknowledged -> "
f"{tables} table(s) on disk, {recovery['records']:,} log records replayed, "
f"acknowledged keys lost: {len(lost)}")
print("2. A write is cut in half by the crash (99 acknowledged records + 1 torn one)")
path = fresh_dir("step08_torn")
db = LSM(path, sync="always")
for i in range(99):
db.put(key(i), value(i))
wal_path = db.wal.path
db.close()
with open(wal_path, "ab") as f:
f.write(encode(OP_PUT, key(99), value(99))[:-7]) # the last record is missing its final 7 bytes
lost, (recovery, _) = reopen_check(path, range(99))
print(f" replayed {recovery['records']} records, discarded {recovery['discarded_bytes']} torn bytes, acknowledged keys lost: {len(lost)}")
print("3. Same torn tail, but recovery forgets to cut it off")
path = fresh_dir("step08_nocut")
db = LSM(path, sync="always")
for i in range(99):
db.put(key(i), value(i))
wal_path = db.wal.path
db.close()
with open(wal_path, "ab") as f:
f.write(encode(OP_PUT, key(99), value(99))[:-7])
db = LSM(path, sync="always", truncate_torn_tail=False)
for i in range(1000, 1005):
db.put(key(i), value(i)) # five new, acknowledged writes
db.close()
lost, _ = reopen_check(path, list(range(99)) + list(range(1000, 1005)))
print(f" after the next restart, acknowledged keys lost: {len(lost)} (the 5 new writes sat behind the garbage)")
print("4. The power fails after the new table is written but before the manifest names it")
path = fresh_dir("step08_commit")
db = LSM(path, sync="always")
for i in range(300):
db.put(key(i), value(i))
before = sorted(os.listdir(path))
def power_cut():
raise RuntimeError("power cut")
db._commit = power_cut
try:
db.flush()
except RuntimeError:
pass
db.close() # a dead process releases its files; the OS does the same
during = sorted(os.listdir(path))
lost, _ = reopen_check(path, range(300))
after = sorted(os.listdir(path))
print(" files before the flush: ", before)
print(" files after the crash: ", during)
print(" files after the restart:", after, f"| acknowledged keys lost: {len(lost)}")
print("5. What waiting for the disk costs (1,000 puts, nothing flushed)")
for sync in ("always", "none"):
db = LSM(fresh_dir(f"step08_sync_{sync}"), memtable_bytes=10**9, sync=sync)
started = time.perf_counter()
for i in range(1000):
db.put(key(i), value(i))
seconds = time.perf_counter() - started
db.close()
print(f" sync={sync!r:9} {1000 / seconds:>10,.0f} puts/s ({seconds / 1000 * 1e6:>8,.1f} microseconds per put)")Run python step08_crash.py. It takes a few seconds:
1. The process is killed with os._exit (the child prints an index only after put() returned)
exit code 1, 1,234 puts acknowledged -> 2 table(s) on disk, 150 log records replayed, acknowledged keys lost: 0
exit code 1, 5,000 puts acknowledged -> 2 table(s) on disk, 122 log records replayed, acknowledged keys lost: 0
2. A write is cut in half by the crash (99 acknowledged records + 1 torn one)
replayed 99 records, discarded 123 torn bytes, acknowledged keys lost: 0
3. Same torn tail, but recovery forgets to cut it off
after the next restart, acknowledged keys lost: 5 (the 5 new writes sat behind the garbage)
4. The power fails after the new table is written but before the manifest names it
files before the flush: ['000000.wal', 'MANIFEST']
files after the crash: ['000000.wal', '000001.sst', '000002.wal', 'MANIFEST']
files after the restart: ['000000.wal', 'MANIFEST'] | acknowledged keys lost: 0
5. What waiting for the disk costs (1,000 puts, nothing flushed)
sync='always' 1,084 puts/s ( 922.4 microseconds per put)
sync='none' 51,137 puts/s ( 19.6 microseconds per put)Experiment 1 killed the child after 1,234 and after 5,000 acknowledged puts. The first crash left two tables on disk plus 150 records in the log that had never been flushed; the second left two tables and 122 log records. In both cases the reopened store found every acknowledged key. Notice that the second crash happened with flushes and compactions already behind it, so the manifest, the tables and the log all took part in the recovery.
Experiment 2 appended a record that was cut 7 bytes short, as if the machine died in the middle of writing it. Recovery replayed the 99 complete records, discarded 123 torn bytes, and lost nothing that had been acknowledged, because the cut-off record was never acknowledged.
Experiment 3 is the bug from Step 4. With truncate_torn_tail=False, the store reopened, accepted 5 new writes and acknowledged them, and after the next restart those 5 acknowledged writes were gone. They were in the log, behind the torn bytes, and replay stops at the first damaged record. This is the quiet kind of data loss: the second run looks perfectly healthy.
Experiment 4 makes the power fail at the worst moment of a flush: after the new table and the new log exist, before the manifest names them. The file lists tell the story. Before the flush there was one log and the manifest. After the crash there were two extra files, 000001.sst and 000002.wal. After the restart they were gone, the store was back to the two original files, and none of the 300 acknowledged keys was lost: recovery replayed the old log, exactly as the commit-point rule promised.
Experiment 5 times what waiting for the disk costs. With sync="always" the store managed 1,084 puts per second, which is about 922 microseconds per put; with sync="none" it managed 51,137, about 20 microseconds. That is a factor of 47 on this machine, and your number will differ, but the shape will not: fsync is the price of durability. A common remedy is group commit, where one fsync covers the writes of several callers; this lab does not implement it.
One limit deserves to be said plainly. Killing a process proves that the store survives a crash of the program, because the operating system still holds whatever the program wrote. It does not prove survival of a power failure, where data that only reached the operating system’s cache is lost. The fsync calls are there for that case, and this lab has no way to cut the power and check.
Step 9: Test it against a dictionary, then break it on purpose
The strongest test for a store is a model: run the same random operations against the store and against a plain Python dictionary, and compare. A dictionary is obviously correct, so any difference is a bug in the store. The tests below also pin down every behavior from the earlier steps: recovery, torn tails, orphan files, tombstones, scans, the Bloom filter’s guarantees and the table checksum.
Save the first part as test_lsm.py. It uses small settings (a 2 KiB memtable) so that even short tests flush and compact a lot.
"""test_lsm.py: checks for the LSM store, including a random comparison with a plain dict."""
import os
import random
import pytest
from bloom import Bloom, hash_pair
from lsm import LSM, Stats, merge_newest_first
from sstable import MISSING, CorruptTable, Table, write_table
from wal import OP_PUT, encode, replay
SMALL = dict(memtable_bytes=2048, tier_runs=3, level_ratio=3, sync="none")
def kv(i):
return b"key%06d" % i, b"value-%d" % i
def release_handles(db):
"""Close every file handle, as the operating system does when a process dies."""
db.wal.close()
for level in db.levels:
for table in level:
table.close()
def test_put_get_delete(tmp_path):
db = LSM(str(tmp_path), **SMALL)
db.put(b"a", b"1")
db.put(b"b", b"2")
assert db.get(b"a") == b"1" and db.get(b"missing") is None
db.delete(b"a")
assert db.get(b"a") is None and db.get(b"b") == b"2"
db.close()
def test_empty_value_is_not_a_tombstone(tmp_path):
db = LSM(str(tmp_path), **SMALL)
db.put(b"k", b"")
db.flush()
assert db.get(b"k") == b""
db.close()
def test_newest_version_wins_across_memtable_and_tables(tmp_path):
db = LSM(str(tmp_path), policy="none", **{k: v for k, v in SMALL.items() if k != "policy"})
for version in range(5):
db.put(b"hot", b"v%d" % version)
db.flush()
db.put(b"hot", b"memtable")
assert db.get(b"hot") == b"memtable"
db.mem.clear()
assert db.get(b"hot") == b"v4"
db.close()
def test_reopen_replays_the_log(tmp_path):
db = LSM(str(tmp_path), **SMALL)
for i in range(10):
db.put(*kv(i))
release_handles(db) # no flush, no clean shutdown
db2 = LSM(str(tmp_path), **SMALL)
assert db2.recovery["records"] == 10
assert all(db2.get(kv(i)[0]) == kv(i)[1] for i in range(10))
db2.close()
def test_torn_log_tail_is_discarded_and_cut(tmp_path):
db = LSM(str(tmp_path), **SMALL)
for i in range(5):
db.put(*kv(i))
wal_path = db.wal.path
release_handles(db)
with open(wal_path, "r+b") as f:
f.truncate(os.path.getsize(wal_path) - 3)
db2 = LSM(str(tmp_path), **SMALL)
assert db2.recovery["records"] == 4 and db2.recovery["discarded_bytes"] > 0
assert db2.get(kv(4)[0]) is None and db2.get(kv(3)[0]) == kv(3)[1]
db2.put(*kv(99))
release_handles(db2)
db3 = LSM(str(tmp_path), **SMALL)
assert db3.get(kv(99)[0]) == kv(99)[1] # the new write is not stuck behind garbage
db3.close()
def test_without_the_cut_later_writes_are_lost(tmp_path):
db = LSM(str(tmp_path), **SMALL)
for i in range(5):
db.put(*kv(i))
wal_path = db.wal.path
release_handles(db)
with open(wal_path, "r+b") as f:
f.truncate(os.path.getsize(wal_path) - 3)
db2 = LSM(str(tmp_path), truncate_torn_tail=False, **SMALL)
db2.put(*kv(99))
release_handles(db2)
db3 = LSM(str(tmp_path), **SMALL)
assert db3.get(kv(99)[0]) is None # acknowledged, then lost: this is the bug the cut prevents
db3.close()
def test_corrupt_record_stops_replay(tmp_path):
path = tmp_path / "x.wal"
path.write_bytes(b"".join(encode(OP_PUT, *kv(i)) for i in range(4)))
raw = bytearray(path.read_bytes())
raw[len(raw) // 2] ^= 0xFF
path.write_bytes(bytes(raw))
records, good, total = replay(str(path))
assert 0 < len(records) < 4 and good < total
def test_crash_before_commit_leaves_no_trace(tmp_path):
db = LSM(str(tmp_path), **SMALL)
for i in range(20):
db.put(*kv(i))
before = sorted(os.listdir(tmp_path))
def power_cut():
raise RuntimeError("power cut")
db._commit = power_cut
with pytest.raises(RuntimeError):
db.flush()
release_handles(db)
db2 = LSM(str(tmp_path), **SMALL)
assert all(db2.get(kv(i)[0]) == kv(i)[1] for i in range(20))
assert sorted(os.listdir(tmp_path)) == before # orphan table and orphan log are gone
db2.close()
def test_disk_files_match_the_manifest(tmp_path):
db = LSM(str(tmp_path), **SMALL)
for i in range(400):
db.put(*kv(i % 90))
release_handles(db)
db2 = LSM(str(tmp_path), **SMALL)
on_disk = {n for n in os.listdir(tmp_path) if n.endswith(".sst")}
listed = {os.path.basename(t.path) for level in db2.levels for t in level}
assert on_disk == listed
db2.close()
def churn(db, deleted):
"""Fill the store, push the data deep, delete some keys, then keep writing."""
for i in range(600):
db.put(*kv(i))
for i in deleted:
db.delete(kv(i)[0])
for i in range(1000, 1400):
db.put(*kv(i))
def test_tombstones_survive_compaction(tmp_path):
deleted = list(range(0, 600, 3))
db = LSM(str(tmp_path), **SMALL)
churn(db, deleted)
assert all(db.get(kv(i)[0]) is None for i in deleted)
assert db.get(kv(1)[0]) == kv(1)[1]
db.close()
def test_unsafe_tombstone_dropping_resurrects_keys(tmp_path):
deleted = list(range(0, 600, 3))
db = LSM(str(tmp_path), unsafe_drop_tombstones=True, **SMALL)
churn(db, deleted)
zombies = [i for i in deleted if db.get(kv(i)[0]) is not None]
assert zombies, "the unsafe flag should have brought deleted keys back"
db.close()
def test_scan_is_sorted_and_hides_deleted(tmp_path):
db = LSM(str(tmp_path), **SMALL)
for i in range(200):
db.put(*kv(i))
for i in range(0, 200, 2):
db.delete(kv(i)[0])
got = list(db.scan(kv(10)[0], kv(40)[0]))
assert [k for k, _ in got] == [kv(i)[0] for i in range(11, 40, 2)]
db.close()
def test_merge_prefers_the_newest_source():
newest = [(b"a", b"new"), (b"c", None)]
middle = [(b"a", b"mid"), (b"b", b"mid")]
oldest = [(b"b", b"old"), (b"c", b"old"), (b"d", b"old")]
assert list(merge_newest_first([newest, middle, oldest])) == [
(b"a", b"new"), (b"b", b"mid"), (b"c", None), (b"d", b"old")]
assert list(merge_newest_first([newest, middle, oldest], drop_tombstones=True)) == [
(b"a", b"new"), (b"b", b"mid"), (b"d", b"old")]
def test_bloom_has_no_false_negatives_and_a_small_false_positive_rate():
keys = [b"present-%d" % i for i in range(5000)]
bloom = Bloom.sized_for(len(keys), 10)
for key in keys:
bloom.add(*hash_pair(key))
assert all(bloom.might_contain(*hash_pair(key)) for key in keys)
false_positives = sum(bloom.might_contain(*hash_pair(b"absent-%d" % i)) for i in range(20000))
assert false_positives / 20000 < 0.02
def test_table_detects_a_damaged_index(tmp_path):
path = str(tmp_path / "t.sst")
write_table(path, [kv(i) for i in range(100)])
raw = bytearray(open(path, "rb").read())
raw[-60] ^= 0xFF # a byte inside the index and filter section
open(path, "wb").write(bytes(raw))
with pytest.raises(CorruptTable):
Table(path, Stats())
def test_table_lookup_reads_one_block(tmp_path):
path = str(tmp_path / "t.sst")
write_table(path, [kv(i) for i in range(1000)])
stats = Stats()
table = Table(path, stats)
assert table.get(kv(500)[0], hash_pair(kv(500)[0])) == kv(500)[1]
assert stats.block_reads == 1
assert table.get(b"key9999999", hash_pair(b"key9999999")) is MISSING
table.close()
def test_leveled_keeps_one_run_per_deep_level(tmp_path):
db = LSM(str(tmp_path), policy="leveled", **SMALL)
rng = random.Random(7)
for _ in range(3000):
db.put(*kv(rng.randrange(500)))
assert all(runs <= 1 for level, runs, _ in db.describe() if level >= 1)
db.close()
def test_tiered_bounds_the_runs_per_level(tmp_path):
db = LSM(str(tmp_path), policy="tiered", **SMALL)
rng = random.Random(8)
for _ in range(3000):
db.put(*kv(rng.randrange(500)))
assert all(runs < 3 for _, runs, _ in db.describe())
db.close()Then add the model test. It runs 3,000 random operations per policy and seed: puts, deletes, gets, an occasional close-and-reopen (which exercises recovery), and an occasional full scan compared with the sorted dictionary.
# test_lsm.py (continued)
@pytest.mark.parametrize("policy", ["none", "tiered", "leveled"])
@pytest.mark.parametrize("seed", [1, 2, 3])
def test_behaves_like_a_dict(tmp_path, policy, seed):
rng = random.Random(seed)
db = LSM(str(tmp_path), policy=policy, **SMALL)
model = {}
for step in range(3000):
roll = rng.random()
key = b"k%03d" % rng.randrange(300)
if roll < 0.55:
value = b"v%d" % rng.randrange(10**6)
db.put(key, value)
model[key] = value
elif roll < 0.75:
db.delete(key)
model.pop(key, None)
elif roll < 0.97:
assert db.get(key) == model.get(key), (step, key)
elif roll < 0.99:
db.close()
db = LSM(str(tmp_path), policy=policy, **SMALL) # reopen: the recovery path
else:
assert list(db.scan()) == sorted(model.items())
assert list(db.scan()) == sorted(model.items())
db.close()Run python -m pytest -q:
........................... [100%]
27 passed in 6.40s9.1 Make sure the tests can fail
A suite you have never seen fail proves little. This script copies the library into a temporary folder, applies one deliberate bug at a time, runs the tests and reports how many fail. Save it as step09_mutations.py:
"""step09_mutations.py: break the store on purpose and count how many tests notice."""
import collections
import os
import re
import shutil
import subprocess
import sys
import tempfile
from labkit import WORK
HERE = os.path.dirname(os.path.abspath(__file__))
FILES = ["bloom.py", "wal.py", "sstable.py", "lsm.py", "labkit.py", "test_lsm.py"]
MUTATIONS = [
("tombstones are always dropped", "lsm.py",
"drop = self.unsafe_drop_tombstones or all(not lv for lv in self.levels[max(sources) + 1:])",
"drop = True"),
("the oldest version wins a merge", "lsm.py",
"streams = [_tag(rank, source) for rank, source in enumerate(sources)]",
"streams = [_tag(-rank, source) for rank, source in enumerate(sources)]"),
("recovery keeps the torn log tail", "lsm.py",
"if good < total and self.truncate_torn_tail:", "if False:"),
("recovery keeps orphan tables", "lsm.py",
'elif name.endswith(".sst") and _id_of(path) not in live:', "elif False:"),
("the Bloom filter sets one hash fewer", "bloom.py",
" def add(self, h1: int, h2: int) -> None:\n for i in range(self.k):",
" def add(self, h1: int, h2: int) -> None:\n for i in range(self.k - 1):"),
]
def run_tests(folder):
env = dict(os.environ, COLUMNS="200", PYTHONUTF8="1")
run = subprocess.run([sys.executable, "-m", "pytest", "-q", "-rf", "-p", "no:cacheprovider", "test_lsm.py"],
cwd=folder, capture_output=True, text=True, env=env)
summary = [line for line in run.stdout.splitlines() if re.search(r"\d+ (passed|failed)", line)][-1]
names = collections.Counter(re.findall(r"^FAILED test_lsm\.py::(\w+)", run.stdout, re.M))
return summary.strip("= "), names
print(f"{'mutation':40} result")
for label, filename, old, new in [("none (the real code)", "lsm.py", None, None)] + MUTATIONS:
folder = tempfile.mkdtemp(dir=WORK)
for name in FILES:
shutil.copy(os.path.join(HERE, name), folder)
if old:
path = os.path.join(folder, filename)
text = open(path, newline="").read()
assert text.count(old) == 1, label
open(path, "w", newline="").write(text.replace(old, new))
summary, names = run_tests(folder)
print(f"{label:40} {summary}")
for test, count in names.most_common(3):
print(f"{'':40} {test}" + (f" (x{count})" if count > 1 else ""))
shutil.rmtree(folder, ignore_errors=True)Run python step09_mutations.py. It takes under a minute:
mutation result
none (the real code) 27 passed in 6.54s
tombstones are always dropped 4 failed, 23 passed in 6.47s
test_behaves_like_a_dict (x3)
test_tombstones_survive_compaction
the oldest version wins a merge 12 failed, 15 passed in 2.84s
test_behaves_like_a_dict (x9)
test_tombstones_survive_compaction
test_scan_is_sorted_and_hides_deleted
recovery keeps the torn log tail 1 failed, 26 passed in 7.41s
test_torn_log_tail_is_discarded_and_cut
recovery keeps orphan tables 1 failed, 26 passed in 7.03s
test_crash_before_commit_leaves_no_trace
the Bloom filter sets one hash fewer 14 failed, 13 passed in 2.29s
test_behaves_like_a_dict (x9)
test_empty_value_is_not_a_tombstone
test_newest_version_wins_across_memtable_and_tablesEvery deliberate bug was caught. Tombstones that are always dropped failed the dictionary comparison and the tombstone test, a merge where the oldest version wins failed 12 tests, a recovery that keeps the torn tail failed the torn-tail test, one that keeps orphan tables failed the crash-before-commit test, and a Bloom filter that sets one hash fewer than it checks produced false negatives and failed 14 tests. Notice that the torn-tail and orphan bugs are caught by exactly one test each, so those two tests are carrying real weight. For deeper checks, property-based testing with Hypothesis can generate the random operation sequences and shrink a failing one to a minimal example.
Check that everything works end to end
From the lsm-lab folder, with the virtual environment active, run the nine steps in order and the tests. If every command finishes without a traceback and the numbers in the byte and count columns match this article, your store works:
python step01_log_only.py # 242,000 / 2,420,000 / 24,200,000 bytes read per lookup
python step02_wal.py # 5 records, then 4 (torn), then 2 (corrupt)
python step03_sstable.py # one 1,936-byte block read; the damaged index is refused
python step04_read_path.py # 18 tables, 9.4 and 18.0 tables asked
python step05_bloom.py # missing keys fall from 17.97 to 0.17 disk reads
python step06_compaction.py # write amplification 2.2, 4.3 and 7.0
python step07_tombstones.py # 952 zombies in the unsafe store, 0 in the safe one
python step08_crash.py # 0 acknowledged keys lost, except experiment 3 (5)
python -m pytest -q # 27 passed
python step09_mutations.py # every deliberate bug is caughtIf a script cannot find lsm or labkit, you are running it from a different folder than the one that holds the files. If step06_compaction.py reports far fewer compactions than 34 and 40, check that you deleted the placeholder _maybe_compact from Step 4. If Windows reports a permission error while deleting a table, a file handle is still open: check that _compact closes each input table before it removes it.
Where this stops working, and what to learn next
This store is small on purpose, so here are the limits that matter before you rely on anything like it.
- One thread. Flushes and compactions run inside
put, which is why the slowest write in Step 6 was so slow. LevelDB’s documentation describes flushing and compacting in the background, with the previous memtable still readable while it is written out, and its level-0 constants show how writers are throttled. - Whole-level merges. Our levels are single tables, so a compaction rewrites a whole level. LevelDB and RocksDB partition levels into many small files and compact one at a time, which bounds the work per compaction.
- No snapshots or sequence numbers. LevelDB tags every entry with a sequence number, which lets a reader see a consistent view while writes continue and lets compaction know which tombstones are still needed by an open snapshot. We have none of that.
- Memory. Compaction reads whole tables into memory, and
Table.scanreads to the end of the file. A real reader streams block by block. - Checksums and durability. Only the log and the table’s meta section are checksummed, the data section is not, and we do not fsync the directory after creating or renaming files. I did not test whether that matters here.
- Crash testing. Our kill test covers a dying process, not a power failure, as said in Step 8.
- Delete-heavy workloads. A key range full of tombstones still has to be read and skipped until compaction reaches the bottom level. I did not measure that cost here.
To keep learning, read LevelDB’s implementation notes, which are short and clear, then the Bigtable paper’s sections on tablet serving and compactions, and the RocksDB pages on leveled and universal compaction. Related tutorials on this site fit the pieces you just built: the skip list is the memtable real engines use, the Bloom filter and write-ahead log tutorials go deeper on two of the parts, SQLite’s WAL mode shows a production log in action, and an LRU cache is the usual next layer: a block cache in front of the table reads. You now have a working store, a test suite that pins down its behavior, and a way to measure whether the next idea you add is worth its cost.
Sources and method
- Patrick O’Neil, Edward Cheng, Dieter Gawlick and Elizabeth O’Neil, The Log-Structured Merge-Tree (LSM-Tree): the abstract quoted in the introduction.
- Chang et al., Bigtable: A Distributed Storage System for Structured Data (OSDI 2006): memtable and SSTable definitions, minor, merging and major compactions, deletion entries, and Bloom filters.
- LevelDB’s implementation notes, dbformat.h, db_impl.cc and memtable.h, read on October 20, 2026: the log, manifest and level descriptions, the level-0 constants, the tombstone-dropping comment, and the skip-list memtable.
- RocksDB’s Leveled Compaction and Universal Compaction pages: the write-amplification remarks and the level layout.
- Python’s os module documentation:
os.fsyncandos.replace.
All code ran on Python 3.13.14 on Windows 11 with pytest 9.1.1. Byte counts, table counts, read counts and amplification ratios come from seeded workloads and repeat exactly; timings (Steps 1, 6 and 8) are from one machine and will differ on yours. The store was also rebuilt from this article’s own code blocks in an empty folder and a fresh virtual environment, and every printed result matched. What this tutorial does not show: behavior under a power failure, concurrent access, or datasets larger than a few megabytes.
No Comment! Be the first one.