Storage Engines:
B-Tree vs LSM

btree_insert(root, key, value):
  # 1. Traverse root → leaf (O(log n) random disk reads)
  node = root
  path = []
  while not node.is_leaf:
    path.append(node)
    child_idx = find_child(node, key)        # binary search within 4KB page
    node = read_page_from_disk(node.children[child_idx])  # RANDOM DISK SEEK!

  # 2. Insert into leaf page in-place
  insert_into_page(node, key, value)
  write_page_to_disk(node)               # RANDOM DISK WRITE

  # 3. Split node if full → propagate upward (cascade splits possible)
  if node.is_full():
    split_and_propagate(node, path)

# Performance reality for a 1TB database (4KB pages, fan-out 500):
# Tree height ≈ log_500(1TB / 4KB) ≈ 4 levels
# Each level = 1 random disk read
# HDD:  4 × 10ms = 40ms write latency
# NVMe: 4 × 0.1ms = 0.4ms write latency

class LSMTree:
    def __init__(self):
        self.wal = WriteAheadLog()      # Sequential write — crash recovery
        self.memtable = SkipList()       # In-memory sorted structure
        self.sstables = []               # Immutable sorted disk files
        self.bloom_filters = {}          # One Bloom filter per SSTable

    def put(self, key: str, value: str):
        self.wal.append(key, value)      # Step 1: WAL — sequential, 0.1ms
        self.memtable.set(key, value)    # Step 2: MemTable — pure RAM, 0.01ms
        if self.memtable.size() > MEMTABLE_MAX:
            self._flush_to_sstable()     # Step 3: flush to L0 SSTable (background)

    def get(self, key: str) -> Optional[str]:
        if val := self.memtable.get(key):  # Check MemTable first (newest)
            return val
        for sstable in reversed(self.sstables):
            if not self.bloom_filters[sstable.id].might_contain(key):
                continue               # Bloom says NO — skip disk read entirely
            if val := sstable.get(key):
                return val
        return None

    def _flush_to_sstable(self):
        # Write MemTable to disk as sorted, immutable SSTable file
        sstable = SSTable.from_sorted_iter(self.memtable.items())
        self.bloom_filters[sstable.id] = BloomFilter(sstable.keys)
        self.sstables.append(sstable)
        self.memtable = SkipList()
        self._maybe_compact()            # Step 4: compaction in background

from bitarray import bitarray
import mmh3   # MurmurHash3 — fast non-cryptographic hash
import math

class BloomFilter:
    def __init__(self, capacity=1_000_000, error_rate=0.01):
        # Optimal bit array size: m = -n*ln(p) / (ln(2)^2)
        self.size = int(-capacity * math.log(error_rate) / (math.log(2)**2))
        self.bits = bitarray(self.size)
        self.bits.setall(0)
        # Optimal hash count: k = (m/n) * ln(2)
        self.num_hashes = int(self.size / capacity * math.log(2))
        # At capacity=1M, error_rate=1%: size≈9.6Mb bits, num_hashes=7

    def add(self, item: str):
        for seed in range(self.num_hashes):
            pos = mmh3.hash(item, seed) % self.size
            self.bits[pos] = 1

    def might_contain(self, item: str) -> bool:
        # ALL k bits must be 1. Any single 0 → definite miss
        return all(
            self.bits[mmh3.hash(item, seed) % self.size]
            for seed in range(self.num_hashes)
        )

# In LSM tree: one BloomFilter per SSTable, kept in memory
bf = BloomFilter(capacity=1_000_000, error_rate=0.01)
bf.add("user:42")
bf.might_contain("user:42")   # True  — might exist, do disk read
bf.might_contain("user:99")   # False — definitely not here, skip!