LSM Tree (Log-Structured Merge Tree)

LSM Tree Architecture

Key Components

Write Path Flow

Read Path Flow

Compaction Process

Compaction Strategies

Complexity Analysis

Amplification Factors

Python Implementation

MemTable Class

class MemTable:
    """In-memory buffer for recent writes

    Writes go here first for fast O(1) insertion.
    When full, it's flushed to disk as an immutable SSTable.
    """

    def __init__(self, max_size: int = 100):
        self.data: Dict[str, Any] = {}
        self.max_size = max_size

    def put(self, key: str, value: Any) -> None:
        """Insert or update a key-value pair

        Time Complexity: O(1)
        """
        self.data[key] = value

    def get(self, key: str) -> Optional[Any]:
        """Retrieve a value by key

        Time Complexity: O(1)
        """
        return self.data.get(key)

    def delete(self, key: str) -> None:
        """Mark a key as deleted (tombstone)

        LSM trees use tombstones instead of immediate deletion
        """
        self.data[key] = None  # Tombstone marker

    def is_full(self) -> bool:
        """Check if memtable has reached capacity"""
        return len(self.data) >= self.max_size

SSTable Class

class SSTable:
    """Sorted String Table - immutable sorted data structure

    - Stored on disk in real implementation
    - Sorted by key for efficient binary search
    - Never modified after creation (immutable)
    """

    def __init__(self, data: List[Tuple[str, Any]], level: int = 0):
        # Store as sorted list of (key, value) tuples
        self.data = sorted(data)
        self.keys = [item[0] for item in self.data]
        self.level = level

    def get(self, key: str) -> Optional[Any]:
        """Binary search for key

        Time Complexity: O(log n)
        """
        idx = bisect.bisect_left(self.keys, key)
        if idx < len(self.keys) and self.keys[idx] == key:
            return self.data[idx][1]
        return None

    def scan_range(self, start_key: str, end_key: str) -> List[Tuple[str, Any]]:
        """Scan a range of keys

        Time Complexity: O(log n + k) where k is number of results
        """
        start_idx = bisect.bisect_left(self.keys, start_key)
        end_idx = bisect.bisect_right(self.keys, end_key)
        return self.data[start_idx:end_idx]

LSM Tree Main Class

class LSMTree:
    """Simplified LSM Tree implementation

    Architecture:
    1. Writes go to MemTable (in-memory)
    2. When MemTable is full, flush to L0 SSTable
    3. Background compaction merges SSTables to reduce read amplification
    4. Reads check MemTable first, then SSTables (newest to oldest)
    """

    def __init__(self, memtable_size: int = 100, compaction_threshold: int = 4):
        self.memtable = MemTable(max_size=memtable_size)
        self.immutable_memtable = None
        self.sstables: List[List[SSTable]] = [[] for _ in range(7)]  # 7 levels
        self.compaction_threshold = compaction_threshold

    def put(self, key: str, value: Any) -> None:
        """Write operation - always goes to memtable first

        Time Complexity: O(1) amortized
        """
        self.memtable.put(key, value)

        if self.memtable.is_full():
            self._flush_memtable()
            self._maybe_compact()

    def get(self, key: str) -> Optional[Any]:
        """Read operation - check memtable first, then SSTables

        Time Complexity: O(k * log n) where k is number of SSTables
        """
        # 1. Check active memtable (most recent data)
        value = self.memtable.get(key)
        if value is not None:
            return value

        # 2. Check immutable memtable if exists
        if self.immutable_memtable:
            value = self.immutable_memtable.get(key)
            if value is not None:
                return value

        # 3. Check SSTables level by level (newer levels first)
        for level in range(len(self.sstables)):
            for sstable in reversed(self.sstables[level]):
                value = sstable.get(key)
                if value is not None:
                    return None if value is None else value  # Handle tombstones

        return None

    def _flush_memtable(self) -> None:
        """Flush memtable to L0 SSTable"""
        if len(self.memtable.data) > 0:
            sstable = SSTable(self.memtable.get_sorted_items(), level=0)
            self.sstables[0].append(sstable)
            self.memtable.clear()

    def _compact_level(self, level: int) -> None:
        """Compact SSTables at a given level

        Merges all SSTables at this level into next level
        Removes duplicates and tombstones
        """
        merged_data = {}
        for sstable in self.sstables[level]:
            for key, value in sstable.data:
                merged_data[key] = value  # Newer values overwrite

        # Remove tombstones
        merged_data = {k: v for k, v in merged_data.items() if v is not None}

        if merged_data:
            new_sstable = SSTable(list(merged_data.items()), level=level + 1)
            self.sstables[level + 1].append(new_sstable)

        self.sstables[level].clear()

LSM Tree vs Other Data Structures

Real-World Usage

Interview Tips