第09课：LSM树

索引结构 第9课 / 共25课

📖 课程概述

LSM树(Log-Structured Merge-Tree)是写优化索引结构的代表，被LevelDB、RocksDB、Cassandra、HBase等广泛采用。与B+树的就地更新不同，LSM树将所有写入追加到内存表，然后批量刷写到磁盘，通过后台合并保持有序。本课深入实现LSM树的核心机制。

📊 LSM树架构

写放大、读放大、空间放大

💻 C语言实现：LSM树

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <time.h>

#define MAX_KEY    64
#define MAX_VAL    255
#define MEMTABLE_MAX 50
#define SST_MAX    200
#define MAX_LEVELS 4

typedef struct {
    char key[MAX_KEY];
    char value[MAX_VAL];
    uint8_t deleted;  // 墓碑标记
    uint64_t timestamp;
} Entry;

// MemTable (简化为有序数组)
typedef struct {
    Entry entries[MEMTABLE_MAX];
    int   count;
    int   immutable;   // 是否为不可变MemTable
} MemTable;

// SSTable
typedef struct {
    Entry entries[SST_MAX];
    int   count;
    int   level;
    int   sst_id;
    char  filename[64];
} SSTable;

// LSM树
typedef struct {
    MemTable* memtable;
    MemTable* immutable_memtable;
    SSTable*  levels[MAX_LEVELS][SST_MAX];
    int       level_counts[MAX_LEVELS];
    int       next_sst_id;
    uint64_t  timestamp;
    int       compactions;
    int       reads;
    int       writes;
} LSMTree;

LSMTree* lsm_create() {
    LSMTree* lsm = calloc(1, sizeof(LSMTree));
    lsm->memtable = calloc(1, sizeof(MemTable));
    lsm->timestamp = 1;
    printf("[LSM] 创建LSM树，MemTable容量=%d\n", MEMTABLE_MAX);
    return lsm;
}

// 向MemTable插入
void memtable_insert(MemTable* mt, const char* key, const char* val,
                     uint8_t deleted, uint64_t ts) {
    // 找插入位置(保持有序)
    int pos = 0;
    while (pos < mt->count && strcmp(mt->entries[pos].key, key) < 0) pos++;
    // 如果key已存在，更新
    if (pos < mt->count && strcmp(mt->entries[pos].key, key) == 0) {
        mt->entries[pos] = (Entry){{0}, {0}, deleted, ts};
        strcpy(mt->entries[pos].key, key);
        if (!deleted) strcpy(mt->entries[pos].value, val);
        return;
    }
    // 移动后续元素
    memmove(&mt->entries[pos+1], &mt->entries[pos],
            (mt->count - pos) * sizeof(Entry));
    Entry* e = &mt->entries[pos];
    strcpy(e->key, key);
    if (!deleted) strcpy(e->value, val);
    e->deleted = deleted;
    e->timestamp = ts;
    mt->count++;
}

// Flush MemTable到L0 SSTable
void lsm_flush(LSMTree* lsm) {
    if (lsm->level_counts[0] >= SST_MAX) {
        printf("  [LSM] L0 SSTable已满，需要Compaction!\n");
        return;
    }
    SSTable* sst = calloc(1, sizeof(SSTable));
    sst->level = 0;
    sst->sst_id = lsm->next_sst_id++;
    snprintf(sst->filename, sizeof(sst->filename), "sst_L0_%d.dat", sst->sst_id);
    // 复制数据
    memcpy(sst->entries, lsm->immutable_memtable->entries,
           lsm->immutable_memtable->count * sizeof(Entry));
    sst->count = lsm->immutable_memtable->count;
    lsm->levels[0][lsm->level_counts[0]++] = sst;
    printf("  [LSM] Flush: MemTable → SSTable L0_%d (%d条记录)\n",
           sst->sst_id, sst->count);
    // 释放Immutable Memtable
    free(lsm->immutable_memtable);
    lsm->immutable_memtable = NULL;
}

// Compaction: L_i → L_{i+1}
void lsm_compact(LSMTree* lsm, int level) {
    if (level >= MAX_LEVELS - 1) return;
    if (lsm->level_counts[level] == 0) return;

    printf("  [LSM] Compaction: L%d → L%d\n", level, level + 1);
    lsm->compactions++;

    // 合并当前层和下一层的所有SSTable
    Entry merged[SST_MAX * 2];
    int merged_count = 0;

    // 收集当前层
    for (int i = 0; i < lsm->level_counts[level]; i++) {
        SSTable* sst = lsm->levels[level][i];
        for (int j = 0; j < sst->count; j++) {
            if (merged_count < SST_MAX * 2)
                merged[merged_count++] = sst->entries[j];
        }
        free(sst);
    }
    lsm->level_counts[level] = 0;

    // 收集下一层
    for (int i = 0; i < lsm->level_counts[level + 1]; i++) {
        SSTable* sst = lsm->levels[level + 1][i];
        for (int j = 0; j < sst->count; j++) {
            if (merged_count < SST_MAX * 2)
                merged[merged_count++] = sst->entries[j];
        }
        free(sst);
    }
    lsm->level_counts[level + 1] = 0;

    // 排序(简化: 按key排序，同key保留timestamp大的)
    for (int i = 0; i < merged_count - 1; i++)
        for (int j = i + 1; j < merged_count; j++)
            if (strcmp(merged[i].key, merged[j].key) > 0) {
                Entry tmp = merged[i]; merged[i] = merged[j]; merged[j] = tmp;
            }

    // 去重(同key保留最新版本)
    Entry deduped[SST_MAX * 2];
    int deduped_count = 0;
    for (int i = 0; i < merged_count; i++) {
        if (i + 1 < merged_count && strcmp(merged[i].key, merged[i+1].key) == 0) {
            // 保留timestamp更大的
            if (merged[i].timestamp >= merged[i+1].timestamp)
                deduped[deduped_count++] = merged[i];
            else
                deduped[deduped_count++] = merged[i+1];
            i++; // 跳过下一个
        } else {
            deduped[deduped_count++] = merged[i];
        }
    }

    // 分配到新的SSTable
    int per_sst = SST_MAX / 2;
    for (int i = 0; i < deduped_count; i += per_sst) {
        if (lsm->level_counts[level + 1] >= SST_MAX) break;
        SSTable* new_sst = calloc(1, sizeof(SSTable));
        new_sst->level = level + 1;
        new_sst->sst_id = lsm->next_sst_id++;
        int cnt = 0;
        for (int j = i; j < deduped_count && j < i + per_sst; j++) {
            if (!deduped[j].deleted) {  // 过滤墓碑
                new_sst->entries[cnt++] = deduped[j];
            }
        }
        new_sst->count = cnt;
        if (cnt > 0)
            lsm->levels[level + 1][lsm->level_counts[level + 1]++] = new_sst;
        else
            free(new_sst);
    }
    printf("  [LSM] Compaction完成: L%d有%d个SSTable\n",
           level + 1, lsm->level_counts[level + 1]);
}

// LSM插入
void lsm_put(LSMTree* lsm, const char* key, const char* val) {
    lsm->writes++;
    memtable_insert(lsm->memtable, key, val, 0, lsm->timestamp++);

    // MemTable满？
    if (lsm->memtable->count >= MEMTABLE_MAX) {
        lsm->memtable->immutable = 1;
        lsm->immutable_memtable = lsm->memtable;
        lsm->memtable = calloc(1, sizeof(MemTable));
        // Flush
        lsm_flush(lsm);
        // L0太多？触发Compaction
        if (lsm->level_counts[0] >= 4) {
            lsm_compact(lsm, 0);
        }
    }
}

// LSM删除(墓碑)
void lsm_delete(LSMTree* lsm, const char* key) {
    lsm->writes++;
    memtable_insert(lsm->memtable, key, "", 1, lsm->timestamp++);
    if (lsm->memtable->count >= MEMTABLE_MAX) {
        lsm->memtable->immutable = 1;
        lsm->immutable_memtable = lsm->memtable;
        lsm->memtable = calloc(1, sizeof(MemTable));
        lsm_flush(lsm);
        if (lsm->level_counts[0] >= 4) lsm_compact(lsm, 0);
    }
}

// LSM查找
char* lsm_get(LSMTree* lsm, const char* key) {
    lsm->reads++;
    // 1. 查MemTable
    for (int i = lsm->memtable->count - 1; i >= 0; i--) {
        if (strcmp(lsm->memtable->entries[i].key, key) == 0) {
            if (lsm->memtable->entries[i].deleted) {
                printf("  [LSM] %s: 已删除(MemTable)\n", key);
                return NULL;
            }
            printf("  [LSM] %s: 命中MemTable → %s\n", key, lsm->memtable->entries[i].value);
            return lsm->memtable->entries[i].value;
        }
    }
    // 2. 查Immutable MemTable
    if (lsm->immutable_memtable) {
        for (int i = lsm->immutable_memtable->count - 1; i >= 0; i--) {
            if (strcmp(lsm->immutable_memtable->entries[i].key, key) == 0) {
                if (lsm->immutable_memtable->entries[i].deleted) return NULL;
                printf("  [LSM] %s: 命中Immutable → %s\n", key,
                       lsm->immutable_memtable->entries[i].value);
                return lsm->immutable_memtable->entries[i].value;
            }
        }
    }
    // 3. 查SSTable (L0→L1→L2→L3)
    for (int level = 0; level < MAX_LEVELS; level++) {
        for (int i = lsm->level_counts[level] - 1; i >= 0; i--) {
            SSTable* sst = lsm->levels[level][i];
            for (int j = sst->count - 1; j >= 0; j--) {
                if (strcmp(sst->entries[j].key, key) == 0) {
                    if (sst->entries[j].deleted) {
                        printf("  [LSM] %s: 已删除(L%d_SST%d)\n", key, level, sst->sst_id);
                        return NULL;
                    }
                    printf("  [LSM] %s: 命中L%d_SST%d → %s\n",
                           key, level, sst->sst_id, sst->entries[j].value);
                    return sst->entries[j].value;
                }
            }
        }
    }
    printf("  [LSM] %s: 不存在\n", key);
    return NULL;
}

void lsm_stats(LSMTree* lsm) {
    printf("\n=== LSM树统计 ===\n");
    printf("MemTable: %d条  Immutable: %s\n",
           lsm->memtable->count,
           lsm->immutable_memtable ? "有" : "无");
    int total = lsm->memtable->count;
    if (lsm->immutable_memtable) total += lsm->immutable_memtable->count;
    for (int level = 0; level < MAX_LEVELS; level++) {
        int level_records = 0;
        for (int i = 0; i < lsm->level_counts[level]; i++)
            level_records += lsm->levels[level][i]->count;
        total += level_records;
        printf("L%d: %d SSTable, %d条记录\n",
               level, lsm->level_counts[level], level_records);
    }
    printf("总记录: %d  写入: %d  读取: %d  Compaction: %d\n",
           total, lsm->writes, lsm->reads, lsm->compactions);
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   LSM树实现                          ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    LSMTree* lsm = lsm_create();
    const char* names[] = {"alice","bob","charlie","diana","eve","frank",
        "grace","henry","ivy","jack","kate","leo","mia","noah","olivia"};

    // 大量写入
    printf("--- 批量写入 ---\n");
    for (int i = 0; i < 15; i++) {
        char val[32];
        snprintf(val, sizeof(val), "city_%d", i);
        printf("Put %s→%s\n", names[i], val);
        lsm_put(lsm, names[i], val);
    }
    // 更多写入触发Compaction
    for (int i = 15; i < 80; i++) {
        char key[32], val[32];
        snprintf(key, sizeof(key), "key_%03d", i);
        snprintf(val, sizeof(val), "val_%03d", i);
        lsm_put(lsm, key, val);
    }

    lsm_stats(lsm);

    // 查找测试
    printf("\n--- 查找 ---\n");
    lsm_get(lsm, "alice");
    lsm_get(lsm, "frank");
    lsm_get(lsm, "key_050");

    // 删除测试
    printf("\n--- 删除 ---\n");
    lsm_delete(lsm, "bob");
    lsm_get(lsm, "bob");

    // 更新测试
    printf("\n--- 更新 ---\n");
    lsm_put(lsm, "alice", "NewYork");
    lsm_get(lsm, "alice");

    lsm_stats(lsm);
    printf("\n✅ LSM树实现运行完成\n");
    return 0;
}

🐍 Python实现：带Bloom Filter的LSM树

"""
LSM树 + Bloom Filter
Bloom Filter减少无效SSTable查找
"""
import hashlib, struct, random
from bitarray import bitarray

class BloomFilter:
    def __init__(self, capacity=10000, error_rate=0.01):
        self.size = int(-capacity * (error_rate ** (-1)) / (0.693 ** 2)) + 1
        self.num_hashes = int(self.size / capacity * 0.693) + 1
        self.bits = bitarray(self.size)
        self.bits.setall(0)

    def _hashes(self, key):
        h1 = int(hashlib.md5(key.encode()).hexdigest(), 16)
        h2 = int(hashlib.sha1(key.encode()).hexdigest(), 16)
        for i in range(self.num_hashes):
            yield (h1 + i * h2) % self.size

    def add(self, key):
        for pos in self._hashes(key):
            self.bits[pos] = 1

    def might_contain(self, key):
        return all(self.bits[pos] for pos in self._hashes(key))

class SSTable:
    def __init__(self, level, sst_id):
        self.level = level
        self.sst_id = sst_id
        self.data = {}  # key → (value, deleted, timestamp)
        self.bloom = BloomFilter()
        self.min_key = None
        self.max_key = None

    def add(self, key, value, deleted, ts):
        self.data[key] = (value, deleted, ts)
        self.bloom.add(key)
        if self.min_key is None or key < self.min_key: self.min_key = key
        if self.max_key is None or key > self.max_key: self.max_key = key

    def get(self, key):
        if not self.bloom.might_contain(key): return None, False  # Bloom过滤
        if key in self.data:
            v, d, ts = self.data[key]
            if d: return None, True  # 墓碑
            return v, False
        return None, False

class LSMTreeWithBloom:
    def __init__(self, memtable_size=30):
        self.memtable = {}
        self.immutable = None
        self.levels = [[] for _ in range(4)]
        self.memtable_size = memtable_size
        self.next_sst = 0
        self.ts = 1
        self.bloom_checks = 0
        self.bloom_avoids = 0

    def put(self, key, value):
        self.memtable[key] = (value, False, self.ts)
        self.ts += 1
        if len(self.memtable) >= self.memtable_size:
            self._flush()

    def delete(self, key):
        self.memtable[key] = ("", True, self.ts)
        self.ts += 1
        if len(self.memtable) >= self.memtable_size:
            self._flush()

    def _flush(self):
        sst = SSTable(0, self.next_sst)
        self.next_sst += 1
        for k, (v, d, ts) in sorted(self.memtable.items()):
            sst.add(k, v, d, ts)
        self.levels[0].append(sst)
        self.memtable = {}
        print(f"  [LSM] Flush → L0_SST{sst.sst_id} ({len(sst.data)} keys)")
        if len(self.levels[0]) >= 4:
            self._compact(0)

    def _compact(self, level):
        print(f"  [LSM] Compaction L{level} → L{level+1}")
        merged = {}
        for sst in self.levels[level]:
            for k, (v, d, ts) in sst.data.items():
                if k not in merged or ts > merged[k][1]:
                    merged[k] = ((v, d), ts)
        for sst in self.levels[level + 1]:
            for k, (v, d, ts) in sst.data.items():
                if k not in merged or ts > merged[k][1]:
                    merged[k] = ((v, d), ts)
        new_sst = SSTable(level + 1, self.next_sst)
        self.next_sst += 1
        for k in sorted(merged.keys()):
            (v, d), ts = merged[k]
            if not d: new_sst.add(k, v, d, ts)  # 清除墓碑
        self.levels[level] = []
        self.levels[level + 1] = [new_sst]

    def get(self, key):
        # MemTable
        if key in self.memtable:
            v, d, ts = self.memtable[key]
            return None if d else v
        # SSTable levels
        for level in range(4):
            for sst in reversed(self.levels[level]):
                self.bloom_checks += 1
                val, is_tomb = sst.get(key)
                if is_tomb: return None
                if val is not None: return val
                else: self.bloom_avoids += 1
        return None

# 测试
lsm = LSMTreeWithBloom(memtable_size=20)
for i in range(100):
    lsm.put(f"key_{i:03d}", f"val_{i}")
print(f"\nBloom Filter统计: 检查{lsm.bloom_checks}, 跳过{lsm.bloom_avoids}")

# 查找
for key in ["key_010", "key_050", "key_999"]:
    val = lsm.get(key)
    print(f"  {key} → {val}")

print("✅ LSM + Bloom Filter实现完成")

🔑 关键概念总结