第21课：简易KV存储

实战项目 第21课 / 共25课

📖 课程概述

本课开始实战项目阶段，首先实现一个简易的KV(Key-Value)存储引擎。KV存储是数据库最简单的形式，但它包含了存储引擎的核心要素：内存索引、磁盘持久化、WAL日志和崩溃恢复。我们将用C语言实现一个LevelDB风格的KV存储。

🏗️ KV存储架构

💻 C语言实现：MiniKV存储引擎

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

#define MAX_KEY       64
#define MAX_VAL       255
#define MEMTABLE_MAX  100
#define SST_MAX_RECS  500
#define MAX_SSTABLES  16
#define WAL_MAX       4096

// ===== KV条目 =====
typedef struct {
    char     key[MAX_KEY];
    char     value[MAX_VAL];
    uint8_t  deleted;    // 墓碑标记
    uint64_t timestamp;
} KVEntry;

// ===== WAL =====
typedef struct {
    KVEntry entries[WAL_MAX];
    int     count;
    uint64_t next_seq;
} WAL;

void wal_init(WAL* wal) { memset(wal, 0, sizeof(WAL)); wal->next_seq = 1; }

uint64_t wal_append(WAL* wal, const char* key, const char* val, uint8_t deleted) {
    KVEntry* e = &wal->entries[wal->count++];
    strncpy(e->key, key, MAX_KEY - 1);
    if (!deleted) strncpy(e->value, val, MAX_VAL - 1);
    e->deleted = deleted;
    e->timestamp = wal->next_seq++;
    return e->timestamp;
}

// ===== MemTable (有序数组,简化版) =====
typedef struct {
    KVEntry entries[MEMTABLE_MAX];
    int     count;
    uint64_t max_seq;
} MemTable;

void memtable_init(MemTable* mt) { memset(mt, 0, sizeof(MemTable)); }

int mt_find(MemTable* mt, const char* key) {
    int lo = 0, hi = mt->count - 1;
    while (lo <= hi) {
        int mid = (lo + hi) / 2;
        int cmp = strcmp(mt->entries[mid].key, key);
        if (cmp == 0) return mid;
        if (cmp < 0) lo = mid + 1;
        else hi = mid - 1;
    }
    return -1;
}

void mt_insert(MemTable* mt, const char* key, const char* val,
               uint8_t deleted, uint64_t seq) {
    // 查找位置
    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].deleted = deleted;
        if (!deleted) strncpy(mt->entries[pos].value, val, MAX_VAL - 1);
        mt->entries[pos].timestamp = seq;
        return;
    }

    // 插入新条目
    assert(mt->count < MEMTABLE_MAX);
    memmove(&mt->entries[pos + 1], &mt->entries[pos],
            (mt->count - pos) * sizeof(KVEntry));
    strncpy(mt->entries[pos].key, key, MAX_KEY - 1);
    if (!deleted) strncpy(mt->entries[pos].value, val, MAX_VAL - 1);
    mt->entries[pos].deleted = deleted;
    mt->entries[pos].timestamp = seq;
    mt->count++;
    if (seq > mt->max_seq) mt->max_seq = seq;
}

// ===== SSTable =====
typedef struct {
    KVEntry entries[SST_MAX_RECS];
    int     count;
    int     level;
    int     sst_id;
    int     min_key_idx;  // 最小key的索引
    int     max_key_idx;  // 最大key的索引
} SSTable;

SSTable* sst_create(int level, int id) {
    SSTable* sst = calloc(1, sizeof(SSTable));
    sst->level = level;
    sst->sst_id = id;
    return sst;
}

// ===== MiniKV引擎 =====
typedef struct {
    WAL       wal;
    MemTable  memtable;
    MemTable  immutable;   // 正在flush的memtable
    int       has_immutable;
    SSTable*  sstables[4][MAX_SSTABLES];  // 4层SSTable
    int       sst_counts[4];
    int       next_sst_id;
    uint64_t  next_seq;
    int       total_puts;
    int       total_gets;
    int       total_deletes;
    int       cache_hits;
    int       disk_reads;
} MiniKV;

MiniKV* minikv_create() {
    MiniKV* kv = calloc(1, sizeof(MiniKV));
    wal_init(&kv->wal);
    memtable_init(&kv->memtable);
    kv->next_seq = 1;
    kv->next_sst_id = 0;
    printf("[MiniKV] 创建KV存储引擎\n");
    return kv;
}

// Flush MemTable到SSTable
void minikv_flush(MiniKV* kv) {
    if (!kv->has_immutable) return;
    if (kv->sst_counts[0] >= MAX_SSTABLES) {
        printf("  [MiniKV] L0已满，需要Compaction!\n");
        return;
    }

    SSTable* sst = sst_create(0, kv->next_sst_id++);
    for (int i = 0; i < kv->immutable.count; i++) {
        sst->entries[i] = kv->immutable.entries[i];
    }
    sst->count = kv->immutable.count;
    kv->sstables[0][kv->sst_counts[0]++] = sst;

    printf("  [MiniKV] Flush: %d条 → SST_L0_%d\n", sst->count, sst->sst_id);
    kv->has_immutable = 0;
    memtable_init(&kv->immutable);
}

// PUT操作
void minikv_put(MiniKV* kv, const char* key, const char* val) {
    uint64_t seq = wal_append(&kv->wal, key, val, 0);
    mt_insert(&kv->memtable, key, val, 0, seq);
    kv->total_puts++;

    // MemTable满?
    if (kv->memtable.count >= MEMTABLE_MAX) {
        kv->immutable = kv->memtable;
        kv->has_immutable = 1;
        memtable_init(&kv->memtable);
        minikv_flush(kv);
    }
}

// GET操作
const char* minikv_get(MiniKV* kv, const char* key) {
    kv->total_gets++;

    // 1. 查MemTable
    int idx = mt_find(&kv->memtable, key);
    if (idx >= 0) {
        if (kv->memtable.entries[idx].deleted) {
            printf("  [MiniKV] GET %s: 已删除(MemTable)\n", key);
            return NULL;
        }
        kv->cache_hits++;
        return kv->memtable.entries[idx].value;
    }

    // 2. 查Immutable
    if (kv->has_immutable) {
        idx = mt_find(&kv->immutable, key);
        if (idx >= 0) {
            if (kv->immutable.entries[idx].deleted) return NULL;
            kv->cache_hits++;
            return kv->immutable.entries[idx].value;
        }
    }

    // 3. 查SSTable
    for (int level = 0; level < 4; level++) {
        for (int i = kv->sst_counts[level] - 1; i >= 0; i--) {
            SSTable* sst = kv->sstables[level][i];
            kv->disk_reads++;
            for (int j = 0; j < sst->count; j++) {
                if (strcmp(sst->entries[j].key, key) == 0) {
                    if (sst->entries[j].deleted) {
                        printf("  [MiniKV] GET %s: 已删除(L%d_SST%d)\n",
                               key, level, sst->sst_id);
                        return NULL;
                    }
                    return sst->entries[j].value;
                }
            }
        }
    }

    printf("  [MiniKV] GET %s: 不存在\n", key);
    return NULL;
}

// DELETE操作
void minikv_delete(MiniKV* kv, const char* key) {
    uint64_t seq = wal_append(&kv->wal, key, "", 1);
    mt_insert(&kv->memtable, key, "", 1, seq);
    kv->total_deletes++;
}

// SCAN操作(范围扫描)
void minikv_scan(MiniKV* kv, const char* start, const char* end) {
    printf("  [MiniKV] SCAN [%s, %s]:\n", start, end);
    int count = 0;

    // 收集所有活跃KV对(简化: 不做归并)
    KVEntry results[MEMTABLE_MAX * 2 + SST_MAX_RECS * MAX_SSTABLES * 4];
    int result_count = 0;

    // 从MemTable
    for (int i = 0; i < kv->memtable.count; i++) {
        KVEntry* e = &kv->memtable.entries[i];
        if (!e->deleted && strcmp(e->key, start) >= 0 && strcmp(e->key, end) <= 0) {
            results[result_count++] = *e;
        }
    }
    // 从SSTable
    for (int level = 0; level < 4; level++) {
        for (int i = 0; i < kv->sst_counts[level]; i++) {
            SSTable* sst = kv->sstables[level][i];
            for (int j = 0; j < sst->count; j++) {
                if (!sst->entries[j].deleted &&
                    strcmp(sst->entries[j].key, start) >= 0 &&
                    strcmp(sst->entries[j].key, end) <= 0) {
                    results[result_count++] = sst->entries[j];
                }
            }
        }
    }

    // 排序输出
    for (int i = 0; i < result_count - 1; i++) {
        for (int j = i + 1; j < result_count; j++) {
            if (strcmp(results[i].key, results[j].key) > 0) {
                KVEntry tmp = results[i];
                results[i] = results[j];
                results[j] = tmp;
            }
        }
    }

    for (int i = 0; i < result_count; i++) {
        printf("    %s = %s\n", results[i].key, results[i].value);
        count++;
    }
    printf("  共 %d 条\n", count);
}

// 统计
void minikv_stats(MiniKV* kv) {
    printf("\n=== MiniKV 统计 ===\n");
    printf("MemTable: %d条  Immutable: %s\n",
           kv->memtable.count, kv->has_immutable ? "有" : "无");
    for (int level = 0; level < 4; level++) {
        int records = 0;
        for (int i = 0; i < kv->sst_counts[level]; i++)
            records += kv->sstables[level][i]->count;
        printf("L%d: %d SSTable, %d条记录\n",
               level, kv->sst_counts[level], records);
    }
    printf("操作: PUT=%d GET=%d DEL=%d\n",
           kv->total_puts, kv->total_gets, kv->total_deletes);
    printf("缓存命中: %d  磁盘读取: %d\n", kv->cache_hits, kv->disk_reads);
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   MiniKV - 简易KV存储引擎            ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    MiniKV* kv = minikv_create();

    // 写入
    printf("--- PUT操作 ---\n");
    struct { const char* k; const char* v; } data[] = {
        {"user:1", "{name:Alice,age:30}"},
        {"user:2", "{name:Bob,age:25}"},
        {"user:3", "{name:Charlie,age:35}"},
        {"config:db", "{host:localhost,port:5432}"},
        {"config:cache", "{host:redis,port:6379}"},
        {"user:4", "{name:Diana,age:28}"},
        {"user:5", "{name:Eve,age:32}"},
        {"log:1", "{action:login,user:1}"},
        {"log:2", "{action:purchase,user:2}"},
        {"log:3", "{action:logout,user:3}"},
    };
    for (int i = 0; i < 10; i++) {
        minikv_put(kv, data[i].k, data[i].v);
        printf("  PUT %s = %s\n", data[i].k, data[i].v);
    }

    // 大量写入触发Flush
    printf("\n--- 批量写入(触发Flush) ---\n");
    for (int i = 10; i < 150; i++) {
        char key[32], val[64];
        snprintf(key, sizeof(key), "key:%04d", i);
        snprintf(val, sizeof(val), "value_%d_%ld", i, random());
        minikv_put(kv, key, val);
    }

    // 查找
    printf("\n--- GET操作 ---\n");
    const char* queries[] = {"user:1", "config:db", "key:0050", "key:9999"};
    for (int i = 0; i < 4; i++) {
        const char* val = minikv_get(kv, queries[i]);
        if (val) printf("  GET %s → %s\n", queries[i], val);
    }

    // 删除
    printf("\n--- DELETE操作 ---\n");
    minikv_delete(kv, "user:2");
    minikv_get(kv, "user:2");

    // 更新
    printf("\n--- UPDATE操作 ---\n");
    minikv_put(kv, "user:1", "{name:Alice,age:31,city:Beijing}");
    minikv_get(kv, "user:1");

    // 范围扫描
    printf("\n--- SCAN操作 ---\n");
    minikv_scan(kv, "user:1", "user:5");

    minikv_stats(kv);
    printf("\n✅ MiniKV运行完成\n");
    return 0;
}

🐍 Python实现：带Bloom Filter的KV存储

"""
Python KV存储 + Bloom Filter
"""
import hashlib, struct
from collections import OrderedDict

class BloomFilter:
    def __init__(self, capacity=10000, fp_rate=0.01):
        self.size = max(int(-capacity * (fp_rate**-1) / (0.693**2)) + 1, 64)
        self.num_hashes = max(int(self.size / capacity * 0.693) + 1, 1)
        self.bits = [False] * self.size

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

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

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

class SSTableFile:
    def __init__(self, data, sst_id):
        self.sst_id = sst_id
        self.data = dict(data)  # key → (value, deleted, seq)
        self.bloom = BloomFilter()
        for k in self.data: self.bloom.add(k)
        self.keys = sorted(self.data.keys())

    def get(self, key):
        if not self.bloom.might_contain(key): return None, False
        if key in self.data:
            val, deleted, seq = self.data[key]
            if deleted: return None, True  # 墓碑
            return val, False
        return None, False

    def scan(self, start, end):
        return {k: v for k, v in self.data.items()
                if start <= k <= end and not v[1]}

class MiniKVPython:
    def __init__(self, memtable_size=50):
        self.memtable = OrderedDict()
        self.immutable = None
        self.sstables = [[] for _ in range(4)]
        self.memtable_size = memtable_size
        self.next_sst = 0
        self.seq = 1
        self.bloom_avoids = 0
        self.bloom_checks = 0

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

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

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

    def scan(self, start, end):
        results = {}
        for k, (v, d, s) in self.memtable.items():
            if start <= k <= end and not d: results[k] = v
        for level in range(4):
            for sst in self.sstables[level]:
                results.update(sst.scan(start, end))
        return dict(sorted(results.items()))

    def _flush(self):
        sst = SSTableFile(self.memtable, self.next_sst)
        self.next_sst += 1
        self.sstables[0].append(sst)
        self.immutable = None
        self.memtable = OrderedDict()
        print(f"  [Flush] → SST_{sst.sst_id} ({len(sst.data)} keys)")

    def stats(self):
        total = len(self.memtable)
        for level in range(4):
            cnt = sum(len(s.data) for s in self.sstables[level])
            total += cnt
            print(f"  L{level}: {len(self.sstables[level])} SST, {cnt} keys")
        print(f"  Total: {total} keys, Bloom avoids: {self.bloom_avoids}/{self.bloom_checks}")

# 测试
kv = MiniKVPython(memtable_size=30)
for i in range(200):
    kv.put(f"key:{i:04d}", f"value_{i}")

print("\n--- GET ---")
for k in ["key:0001", "key:0050", "key:0199", "key:9999"]:
    v = kv.get(k)
    print(f"  {k} → {v}")

print("\n--- SCAN ---")
results = kv.scan("key:0010", "key:0020")
for k, v in results.items():
    print(f"  {k} = {v}")

print("\n--- Stats ---")
kv.stats()
print("\n✅ Python MiniKV完成")

🔑 关键概念总结