LSM Tree ও SSTable — write-optimized-এর বিপ্লব

import heapq, random, time
from bisect import bisect_left

TOMBSTONE = object()              # delete-marker — value-র জায়গায় বসে
MEM_LIMIT = 4_000                 # এত entry হলেই flush


class MiniLSM:
    """memtable + immutable SSTable + k-way newest-wins compaction"""

    def __init__(self):
        self.memtable, self.tables = {}, []   # tables newest-first
        self.flushes = self.compactions = 0
        self.entries_written = 0               # WAF-এর numerator

    def put(self, key, val):
        self.memtable[key] = val
        if len(self.memtable) >= MEM_LIMIT:
            self.flush()

    def delete(self, key):
        self.put(key, TOMBSTONE)               # delete-ও একটা write!

    def flush(self):
        if not self.memtable:
            return
        table = sorted(self.memtable.items())  # immutable run — আর বদলায় না
        self.tables.insert(0, table)
        self.entries_written += len(table)
        self.memtable, self.flushes = {}, self.flushes + 1

    def get(self, key):
        if key in self.memtable:               # freshest source
            v = self.memtable[key]
            return None if v is TOMBSTONE else v
        for table in self.tables:              # নতুন → পুরনো
            i = bisect_left(table, (key,))
            if i < len(table) and table[i][0] == key:
                v = table[i][1]
                return None if v is TOMBSTONE else v   # tombstone = নিশ্চিত-absent
        return None

    def _rows(self, tbl, rank):
        """এক table-কে (key, rank, value)-stream-এ — rank ছোট মানে newest"""
        for k, v in tbl:
            yield k, rank, v

    def compact(self):
        """সব run এক merge-এ: newest-wins; tombstone নিচতলায় ফেলে দেয়"""
        self.flush()
        streams = [self._rows(t, r) for r, t in enumerate(self.tables)]
        merged, last_key = [], None
        for key, _rank, v in heapq.merge(*streams):
            if key == last_key:
                continue                       # (k, rank)-sort → newest আগে
            last_key = key
            if v is not TOMBSTONE:
                merged.append((key, v))        # পুরনো version + tombstone বাদ
        self.tables = [merged]
        self.entries_written += len(merged)
        self.compactions += 1


if __name__ == '__main__':
    random.seed(7)
    KEYSPACE = 30_000
    db, ref, deleted = MiniLSM(), {}, set()
    t0 = time.perf_counter()
    for i in range(100_000):
        k, v = f'key:{random.randrange(KEYSPACE)}', f'profile-{i}'
        db.put(k, v); ref[k] = v
    dt = time.perf_counter() - t0
    print(f'write burst : 100,000 puts -> {db.flushes} SSTables '
          f'in {dt:.2f}s ({100_000 / dt:,.0f} ops/s)')
    probe = [f'key:{random.randrange(KEYSPACE)}' for _ in range(2_000)]

    def read_stats():
        lat = []
        for k in probe:
            t0 = time.perf_counter(); db.get(k)
            lat.append((time.perf_counter() - t0) * 1e6)
        lat.sort()
        return lat[len(lat) // 2], lat[int(len(lat) * .99)]

    n_before = len(db.tables)
    med_b, p99_b = read_stats()
    for k in random.sample(sorted(ref), 500):   # 500 delete → tombstones
        db.delete(k); deleted.add(k); ref.pop(k)
    db.compact()                                # newest-wins full merge
    med_a, p99_a = read_stats()
    live = len(db.tables[0])
    assert all(db.get(k) == ref[k] for k in random.sample(sorted(ref), 300))
    assert all(db.get(k) is None for k in list(deleted)[:50])
    print(f'uncompacted : {n_before} tables | p50={med_b:.1f}µs p99={p99_b:.1f}µs')
    print(f'compacted   : 1 table      | p50={med_a:.1f}µs p99={p99_a:.1f}µs')
    print(f'WAF = {db.entries_written / live:.1f}x  '
          f'({db.entries_written:,} written vs {live:,} live)')

প্রত্যাশিত output (timing machine-ভেদে):

write burst : 100,000 puts -> 23 SSTables in 0.21s (468,000 ops/s)
uncompacted : 23 tables | p50=7.9µs p99=45.6µs
compacted   : 1 table      | p50=2.4µs p99=3.4µs
WAF = 4.3x  (122,127 written vs 28,388 live)

তিনটা পাঠ: write throughput বিশাল (সব memory-append); compaction-এর পরে p99 ১৩× কমে গেছে — read-amplification কাটা; কিন্তু WAF ৪.৩× — প্রতিটা live byte গড়ে ৪.৩ বার disk-এ গেছে। এই ত্রিভুজটাই আজকের লেসনের হৃদয়।

নিজে বাড়ান: (১) per-table bloom filter — absent-key probe-এ table-skip দেখুন; (২) size-tiered mode — সম-মাপের run pairwise merge, leveled-এর সাথে WAF/p99 তুলনা; (৩) WAL append-log + crash-recovery (replay করে memtable পুনর্গঠন)।