LSM Tree：Cassandra・RocksDB・BigTable の書込み性能を支えるストレージエンジン

class LSMTree:    def __init__(self, mem_limit=4):        self.memtable = {}        # in-memory dict (本物は Skip List)        self.sstables = []        # [{k: v}, ...] (新しいものが末尾)        self.mem_limit = mem_limit
    def put(self, k, v):        '''書込み: MemTable に追加、フルなら flush'''        self.memtable[k] = v        if len(self.memtable) >= self.mem_limit:            self._flush()
    def _flush(self):        '''MemTable → SSTable'''        sstable = dict(sorted(self.memtable.items()))        self.sstables.append(sstable)        self.memtable = {}
    def get(self, k):        '''読込み: MemTable → 新しい SSTable → 古い SSTable の順'''        if k in self.memtable:            return self.memtable[k]        for sst in reversed(self.sstables):            if k in sst:                return sst[k]        return None
    def delete(self, k):        '''削除は tombstone を書く (物理削除は Compaction 時)'''        self.put(k, None)  # None = tombstone
    def compact(self):        '''複数 SSTable をマージ + tombstone を物理削除'''        merged = {}        for sst in self.sstables:            for k, v in sst.items():                if v is None:                    merged.pop(k, None)  # tombstone は削除                else:                    merged[k] = v        self.sstables = [merged]
# 動作確認lsm = LSMTree(mem_limit=4)for i in range(10):    lsm.put(f'k{i}', f'v{i}')
print(f'SSTable 数: {len(lsm.sstables)}')print(f'MemTable サイズ: {len(lsm.memtable)}')print(f'k3 = {lsm.get("k3")}')
# 削除lsm.delete('k3')print(f'\n削除後 k3 = {lsm.get("k3")}')
# Compactionlsm.compact()print(f'Compaction 後 SSTable 数: {len(lsm.sstables)}')print(f'Compaction 後 k3 = {lsm.get("k3")}')

from typing import Optionalimport hashlib, math
class BloomedSSTable:    def __init__(self, data: dict, p=0.01):        self.data = data        self.bloom = self._build_bloom(list(data.keys()), p)
    def _build_bloom(self, keys, p):        n = max(len(keys), 1)        m = int(-n * math.log(p) / (math.log(2) ** 2))        bits = bytearray(math.ceil(m / 8))        k_h = max(int(m / n * math.log(2)), 1)
        def idx(item):            b = item.encode()            h1 = int.from_bytes(hashlib.sha256(b).digest()[:8], 'big')            h2 = int.from_bytes(hashlib.blake2b(b, digest_size=8).digest(), 'big')            return [(h1 + i * h2) % m for i in range(k_h)]
        for k in keys:            for i in idx(k):                bits[i // 8] |= (1 << (i % 8))
        self._idx_func = idx        self._bits = bits        return None
    def __contains__(self, k):        '''Bloom 判定: No なら確実、Yes なら次に data を見る'''        return all(self._bits[i//8] & (1 << (i%8)) for i in self._idx_func(k))
    def get(self, k) -> Optional[str]:        if k not in self:            return None  # Bloom が No なら data を見ない        return self.data.get(k)