"""A log-structured merge tree, small enough to read in one sitting. The write path only ever *appends*. A write lands in an in-memory `MemTable`; when that fills, it is flushed whole to an immutable sorted segment file (an `SSTable`) and a fresh memtable takes over. Segments are never edited afterwards -- only merged into new ones and deleted. put/delete -> MemTable (RAM) --flush--> seg0000.sst seg0001.sst ... '---- compact ----' Two consequences fall out of "append only", and they are the whole toy: 1. A **delete is a write**. There is no record to erase -- the key may sit in older segments this code cannot reach cheaply -- so a delete appends a *tombstone*: an entry saying "as of here, gone". 2. A **read is a search backwards through time**. Newest segment first, stopping at the first entry found. A tombstone found first wins. That makes compaction -- merging segments to reclaim space -- a *correctness* problem, not housekeeping. See `LSMTree.compact`. Deterministic by construction: no clock anywhere (flush triggers on an entry count, segment names come from a counter), no randomness. """ import os SET = "set" DEL = "del" class MemTable: """The write buffer, and the only unsorted thing here. Sorting is deferred to flush time, so a write is O(1) and the sort is paid once per segment rather than once per key. """ def __init__(self): self.entries = {} # key -> (op, value) def put(self, key, value): self.entries[key] = (SET, value) def delete(self, key): # Not `del self.entries[key]`. Dropping the key here would hide it # only from *this* memtable; older segments on disk would still # answer the next get(). The tombstone out-votes them by being newer. self.entries[key] = (DEL, "") def get(self, key): return self.entries.get(key) def __len__(self): return len(self.entries) class SSTable: """An immutable segment file: one `keyopvalue` line per key, written in key order and never modified afterwards. Sorted order is what makes merging cheap and lookups short-circuitable, and it costs nothing: the file is written once from an already-complete memtable. """ def __init__(self, path): self.path = path self.name = os.path.basename(path) @classmethod def write(cls, path, entries): with open(path, "w") as f: for key in sorted(entries): op, value = entries[key] f.write(f"{key}\t{op}\t{value}\n") return cls(path) def items(self): """Every (key, (op, value)) pair, in key order.""" with open(self.path) as f: for line in f: key, op, value = line.rstrip("\n").split("\t") yield key, (op, value) def lookup(self, key): """Linear scan. Returns (entry_or_None, lines_read); the line count is there so the demo can *count* read amplification instead of asserting it. The `k > key` break is the payoff of sorted order: a miss costs half a segment on average, not a whole one. """ lines = 0 for k, entry in self.items(): lines += 1 if k == key: return entry, lines if k > key: break return None, lines def size(self): return os.path.getsize(self.path) class LSMTree: """put / delete / get / flush / compact over a directory of segments.""" def __init__(self, dirname, flush_threshold=2): self.dir = dirname self.flush_threshold = flush_threshold self.memtable = MemTable() self.segments = [] # oldest first, newest last self.next_id = 0 # monotonic counter, NOT a timestamp (see ยง8) self.last_get_segments = 0 self.last_get_lines = 0 os.makedirs(dirname, exist_ok=True) # -- writes --------------------------------------------------------- def put(self, key, value): self.memtable.put(key, value) self._maybe_flush() def delete(self, key): self.memtable.delete(key) self._maybe_flush() def _maybe_flush(self): # An entry count, not a byte size and emphatically not a timer. if len(self.memtable) >= self.flush_threshold: self.flush() def flush(self): """Freeze the memtable into a new newest segment.""" if not len(self.memtable): return None seg = SSTable.write(self._new_path(), self.memtable.entries) self.segments.append(seg) self.memtable = MemTable() return seg def _new_path(self): path = os.path.join(self.dir, f"seg{self.next_id:04d}.sst") self.next_id += 1 return path # -- reads ---------------------------------------------------------- def get(self, key): """Newest to oldest; the first entry found wins, tombstone or not.""" self.last_get_segments = 0 self.last_get_lines = 0 entry = self.memtable.get(key) if entry is None: for seg in reversed(self.segments): self.last_get_segments += 1 entry, lines = seg.lookup(key) self.last_get_lines += lines if entry is not None: break # A tombstone found first means deleted; running off the end of the # oldest segment means the same -- so a miss is the costliest read. if entry is None or entry[0] == DEL: return None return entry[1] # -- compaction ----------------------------------------------------- def compact(self, n, drop_tombstones): """Merge the newest `n` segments into a single new segment. `drop_tombstones` is a DELIBERATELY UNSAFE parameter, and it is the lesson of this toy rather than a bug in it. Dropping tombstones is the obvious way to reclaim space. But a tombstone is the *only* evidence that a key was deleted: throw it away while an older, un-merged segment still holds the original value, and the next get() walks past the merged segment, finds that value, and returns it. The key comes back from the dead. It is safe only when the merge covers every segment that could hold the key -- here, when n == len(self.segments). Cassandra spells the same rule `gc_grace_seconds`; breaking it yields zombie rows. """ assert 2 <= n <= len(self.segments) victims = self.segments[-n:] merged = {} for seg in victims: # oldest first, so newer entries overwrite for key, entry in seg.items(): merged[key] = entry if drop_tombstones: merged = {k: e for k, e in merged.items() if e[0] != DEL} new = SSTable.write(self._new_path(), merged) for seg in victims: os.remove(seg.path) self.segments[-n:] = [new] return new def store_bytes(self): return sum(seg.size() for seg in self.segments)