第05课：存储引擎架构

存储引擎 第5课 / 共25课

📖 课程概述

存储引擎是数据库内核中直接管理数据存取的组件。不同存储引擎提供了不同的存储模型、索引策略和事务支持。本课整合前4课的知识，构建一个完整的存储引擎架构，并对比分析InnoDB、MyISAM、RocksDB等主流引擎的设计选择。

🏗️ 存储引擎架构

可插拔存储引擎

MySQL的可插拔存储引擎架构允许为每张表选择不同的存储引擎。统一接口让上层SQL处理无需关心底层实现。

💻 C语言实现：可插拔存储引擎框架

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

#define MAX_KEY    64
#define MAX_VAL    255
#define MAX_RECORDS 1000

// ============ 通用记录 ============
typedef struct {
    char key[MAX_KEY];
    char value[MAX_VAL];
    uint32_t txn_id;
    uint8_t  deleted;
} Record;

// ============ 存储引擎接口 ============
typedef struct StorageEngine StorageEngine;

struct StorageEngine {
    const char* name;
    void*       state;  // 引擎私有数据

    // 核心API
    int  (*open)(StorageEngine* self, const char* table);
    int  (*close)(StorageEngine* self);
    int  (*insert)(StorageEngine* self, const char* key, const char* val);
    char* (*read)(StorageEngine* self, const char* key);
    int  (*update)(StorageEngine* self, const char* key, const char* val);
    int  (*delete)(StorageEngine* self, const char* key);
    int  (*scan)(StorageEngine* self, void(*callback)(Record*));
    void (*stats)(StorageEngine* self);
};

// ============ Heap引擎 (MyISAM风格) ============
typedef struct {
    Record records[MAX_RECORDS];
    int    count;
    int    read_count;
    int    write_count;
} HeapEngineState;

int heap_open(StorageEngine* self, const char* table) {
    printf("[HeapEngine] 打开表: %s\n", table);
    return 0;
}
int heap_close(StorageEngine* self) {
    printf("[HeapEngine] 关闭\n");
    return 0;
}
int heap_insert(StorageEngine* self, const char* key, const char* val) {
    HeapEngineState* s = (HeapEngineState*)self->state;
    if (s->count >= MAX_RECORDS) return -1;
    Record* r = &s->records[s->count++];
    strncpy(r->key, key, MAX_KEY - 1);
    strncpy(r->value, val, MAX_VAL - 1);
    r->deleted = 0;
    s->write_count++;
    return 0;
}
char* heap_read(StorageEngine* self, const char* key) {
    HeapEngineState* s = (HeapEngineState*)self->state;
    s->read_count++;
    for (int i = 0; i < s->count; i++) {
        if (!s->records[i].deleted && strcmp(s->records[i].key, key) == 0)
            return s->records[i].value;
    }
    return NULL;
}
int heap_delete(StorageEngine* self, const char* key) {
    HeapEngineState* s = (HeapEngineState*)self->state;
    for (int i = 0; i < s->count; i++) {
        if (!s->records[i].deleted && strcmp(s->records[i].key, key) == 0) {
            s->records[i].deleted = 1;
            s->write_count++;
            return 0;
        }
    }
    return -1;
}
int heap_update(StorageEngine* self, const char* key, const char* val) {
    HeapEngineState* s = (HeapEngineState*)self->state;
    for (int i = 0; i < s->count; i++) {
        if (!s->records[i].deleted && strcmp(s->records[i].key, key) == 0) {
            strncpy(s->records[i].value, val, MAX_VAL - 1);
            s->write_count++;
            return 0;
        }
    }
    return -1;
}
int heap_scan(StorageEngine* self, void(*cb)(Record*)) {
    HeapEngineState* s = (HeapEngineState*)self->state;
    int count = 0;
    for (int i = 0; i < s->count; i++) {
        if (!s->records[i].deleted) {
            cb(&s->records[i]);
            count++;
        }
    }
    return count;
}
void heap_stats(StorageEngine* self) {
    HeapEngineState* s = (HeapEngineState*)self->state;
    printf("[HeapEngine] 记录: %d  读取: %d  写入: %d\n",
           s->count, s->read_count, s->write_count);
}

StorageEngine* create_heap_engine() {
    StorageEngine* e = calloc(1, sizeof(StorageEngine));
    e->name = "HeapEngine";
    e->state = calloc(1, sizeof(HeapEngineState));
    e->open = heap_open;
    e->close = heap_close;
    e->insert = heap_insert;
    e->read = heap_read;
    e->update = heap_update;
    e->delete = heap_delete;
    e->scan = heap_scan;
    e->stats = heap_stats;
    return e;
}

// ============ Sorted引擎 (InnoDB风格) ============
typedef struct {
    Record records[MAX_RECORDS];  // 按key排序
    int    count;
    int    read_count;
    int    write_count;
    int    page_splits;
} SortedEngineState;

int sorted_insert(StorageEngine* self, const char* key, const char* val) {
    SortedEngineState* s = (SortedEngineState*)self->state;
    if (s->count >= MAX_RECORDS) return -1;
    // 二分查找插入位置
    int lo = 0, hi = s->count;
    while (lo < hi) {
        int mid = (lo + hi) / 2;
        if (strcmp(s->records[mid].key, key) < 0) lo = mid + 1;
        else hi = mid;
    }
    // 检查是否已存在
    if (lo < s->count && strcmp(s->records[lo].key, key) == 0) {
        strncpy(s->records[lo].value, val, MAX_VAL - 1);
        s->write_count++;
        return 0;
    }
    // 移动记录(模拟页分裂开销)
    memmove(&s->records[lo + 1], &s->records[lo],
            (s->count - lo) * sizeof(Record));
    strncpy(s->records[lo].key, key, MAX_KEY - 1);
    strncpy(s->records[lo].value, val, MAX_VAL - 1);
    s->records[lo].deleted = 0;
    s->count++;
    s->write_count++;
    if ((s->count % 50) == 0) s->page_splits++;
    return 0;
}
char* sorted_read(StorageEngine* self, const char* key) {
    SortedEngineState* s = (SortedEngineState*)self->state;
    s->read_count++;
    // 二分查找
    int lo = 0, hi = s->count - 1;
    while (lo <= hi) {
        int mid = (lo + hi) / 2;
        int cmp = strcmp(s->records[mid].key, key);
        if (cmp == 0) return s->records[mid].deleted ? NULL : s->records[mid].value;
        if (cmp < 0) lo = mid + 1;
        else hi = mid - 1;
    }
    return NULL;
}
int sorted_delete(StorageEngine* self, const char* key) {
    SortedEngineState* s = (SortedEngineState*)self->state;
    for (int i = 0; i < s->count; i++) {
        if (strcmp(s->records[i].key, key) == 0) {
            s->records[i].deleted = 1;
            s->write_count++;
            return 0;
        }
    }
    return -1;
}
int sorted_update(StorageEngine* self, const char* key, const char* val) {
    char* existing = sorted_read(self, key);
    if (!existing) return -1;
    SortedEngineState* s = (SortedEngineState*)self->state;
    for (int i = 0; i < s->count; i++) {
        if (!s->records[i].deleted && strcmp(s->records[i].key, key) == 0) {
            strncpy(s->records[i].value, val, MAX_VAL - 1);
            s->write_count++;
            return 0;
        }
    }
    return -1;
}
int sorted_scan(StorageEngine* self, void(*cb)(Record*)) {
    SortedEngineState* s = (SortedEngineState*)self->state;
    int count = 0;
    for (int i = 0; i < s->count; i++) {
        if (!s->records[i].deleted) { cb(&s->records[i]); count++; }
    }
    return count;
}
void sorted_stats(StorageEngine* self) {
    SortedEngineState* s = (SortedEngineState*)self->state;
    printf("[SortedEngine] 记录: %d  读取: %d  写入: %d  页分裂: %d\n",
           s->count, s->read_count, s->write_count, s->page_splits);
}

StorageEngine* create_sorted_engine() {
    StorageEngine* e = calloc(1, sizeof(StorageEngine));
    e->name = "SortedEngine";
    e->state = calloc(1, sizeof(SortedEngineState));
    e->open = heap_open; e->close = heap_close;
    e->insert = sorted_insert; e->read = sorted_read;
    e->update = sorted_update; e->delete = sorted_delete;
    e->scan = sorted_scan; e->stats = sorted_stats;
    return e;
}

// ============ LSM引擎 (RocksDB风格) ============
#define MEMTABLE_MAX 50
#define SST_LEVELS   3

typedef struct {
    Record memtable[MEMTABLE_MAX];  // 内存表
    int    memtable_count;
    Record sst_files[SST_LEVELS][MAX_RECORDS]; // 简化SST
    int    sst_counts[SST_LEVELS];
    int    read_count;
    int    write_count;
    int    compactions;
} LSMEngineState;

int lsm_insert(StorageEngine* self, const char* key, const char* val) {
    LSMEngineState* s = (LSMEngineState*)self->state;
    // 写入memtable
    if (s->memtable_count >= MEMTABLE_MAX) {
        // Flush到SST Level 0
        printf("  [LSM] Memtable满! Flush到SST L0\n");
        memcpy(s->sst_files[0], s->memtable, sizeof(Record) * MEMTABLE_MAX);
        s->sst_counts[0] = MEMTABLE_MAX;
        s->memtable_count = 0;
        s->compactions++;
    }
    Record* r = &s->memtable[s->memtable_count++];
    strncpy(r->key, key, MAX_KEY - 1);
    strncpy(r->value, val, MAX_VAL - 1);
    r->deleted = 0;
    s->write_count++;
    return 0;
}
char* lsm_read(StorageEngine* self, const char* key) {
    LSMEngineState* s = (LSMEngineState*)self->state;
    s->read_count++;
    // 先查memtable
    for (int i = s->memtable_count - 1; i >= 0; i--) {
        if (strcmp(s->memtable[i].key, key) == 0)
            return s->memtable[i].deleted ? NULL : s->memtable[i].value;
    }
    // 查SST (从L0到L2)
    for (int level = 0; level < SST_LEVELS; level++) {
        for (int i = s->sst_counts[level] - 1; i >= 0; i--) {
            if (strcmp(s->sst_files[level][i].key, key) == 0)
                return s->sst_files[level][i].deleted ? NULL
                     : s->sst_files[level][i].value;
        }
    }
    return NULL;
}
void lsm_stats(StorageEngine* self) {
    LSMEngineState* s = (LSMEngineState*)self->state;
    printf("[LSMEngine] Memtable: %d  SST: L0=%d L1=%d L2=%d\n",
           s->memtable_count, s->sst_counts[0],
           s->sst_counts[1], s->sst_counts[2]);
    printf("  读取: %d  写入: %d  Compactions: %d\n",
           s->read_count, s->write_count, s->compactions);
}

StorageEngine* create_lsm_engine() {
    StorageEngine* e = calloc(1, sizeof(StorageEngine));
    e->name = "LSMEngine";
    e->state = calloc(1, sizeof(LSMEngineState));
    e->insert = lsm_insert; e->read = lsm_read;
    e->update = lsm_insert;  // LSM: update就是insert新版本
    e->delete = NULL; e->scan = NULL;
    e->stats = lsm_stats;
    return e;
}

// ============ 通用回调 ============
void print_record(Record* r) {
    printf("  %s = %s\n", r->key, r->value);
}

// ============ Benchmark ============
void benchmark(StorageEngine* engine, int n) {
    printf("\n=== Benchmark: %s (%d ops) ===\n", engine->name, n);
    engine->open(engine, "test_table");

    clock_t start = clock();
    // 写入
    for (int i = 0; i < n; i++) {
        char key[32], val[32];
        snprintf(key, sizeof(key), "key_%04d", i);
        snprintf(val, sizeof(val), "val_%04d", i);
        engine->insert(engine, key, val);
    }
    clock_t write_time = clock();

    // 读取
    int found = 0;
    for (int i = 0; i < n; i++) {
        char key[32];
        snprintf(key, sizeof(key), "key_%04d", i % (n/2));
        if (engine->read(engine, key)) found++;
    }
    clock_t read_time = clock();

    double write_ms = (double)(write_time - start) / CLOCKS_PER_SEC * 1000;
    double read_ms = (double)(read_time - write_time) / CLOCKS_PER_SEC * 1000;
    printf("写入: %.1fms  读取: %.1fms  命中: %d/%d\n",
           write_ms, read_ms, found, n);
    engine->stats(engine);
    engine->close(engine);
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   可插拔存储引擎框架                  ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    StorageEngine* heap_engine = create_heap_engine();
    StorageEngine* sorted_engine = create_sorted_engine();
    StorageEngine* lsm_engine = create_lsm_engine();

    benchmark(heap_engine, 200);
    benchmark(sorted_engine, 200);
    benchmark(lsm_engine, 200);

    // 扫描测试
    printf("\n=== 堆引擎扫描 ===\n");
    heap_engine->scan(heap_engine, print_record);

    printf("\n✅ 存储引擎框架运行完成\n");
    return 0;
}

🐍 Python实现：引擎对比基准测试

"""
存储引擎对比基准测试
"""
import time, random
from abc import ABC, abstractmethod

class EngineBase(ABC):
    @abstractmethod
    def put(self, key, value): pass
    @abstractmethod
    def get(self, key): pass
    @abstractmethod
    def delete(self, key): pass
    @abstractmethod
    def scan_all(self): pass

class HeapEngine(EngineBase):
    """堆文件引擎 - O(n)查找"""
    def __init__(self):
        self.data = {}  # key → value
        self.ops = 0
    def put(self, k, v): self.data[k] = v; self.ops += 1
    def get(self, k): self.ops += 1; return self.data.get(k)
    def delete(self, k): self.ops += 1; self.data.pop(k, None)
    def scan_all(self): return list(self.data.items())

class SortedEngine(EngineBase):
    """排序引擎 - O(log n)查找"""
    def __init__(self):
        self.keys = []
        self.vals = []
        self.ops = 0
    def _bisect(self, key):
        lo, hi = 0, len(self.keys)
        while lo < hi:
            mid = (lo+hi)//2
            if self.keys[mid] < key: lo = mid+1
            else: hi = mid
        return lo
    def put(self, k, v):
        pos = self._bisect(k)
        if pos < len(self.keys) and self.keys[pos] == k:
            self.vals[pos] = v
        else:
            self.keys.insert(pos, k)
            self.vals.insert(pos, v)
        self.ops += 1
    def get(self, k):
        self.ops += 1
        pos = self._bisect(k)
        if pos < len(self.keys) and self.keys[pos] == k:
            return self.vals[pos]
        return None
    def delete(self, k):
        pos = self._bisect(k)
        if pos < len(self.keys) and self.keys[pos] == k:
            self.keys.pop(pos); self.vals.pop(pos)
        self.ops += 1
    def scan_all(self): return list(zip(self.keys, self.vals))

class LSMEngine(EngineBase):
    """LSM引擎 - 写优化"""
    def __init__(self, flush_size=50):
        self.memtable = {}
        self.flush_size = flush_size
        self.sst_levels = [[] for _ in range(3)]
        self.ops = 0
        self.flushes = 0
    def put(self, k, v):
        self.memtable[k] = v
        self.ops += 1
        if len(self.memtable) >= self.flush_size:
            self._flush()
    def _flush(self):
        items = sorted(self.memtable.items())
        self.sst_levels[0] = items + self.sst_levels[0]
        self.memtable.clear()
        self.flushes += 1
    def get(self, k):
        self.ops += 1
        if k in self.memtable: return self.memtable[k]
        for level in self.sst_levels:
            for key, val in level:
                if key == k: return val
        return None
    def delete(self, k): self.put(k, None)
    def scan_all(self):
        result = {}
        for level in reversed(self.sst_levels):
            for k, v in level: result[k] = v
        result.update(self.memtable)
        return [(k,v) for k,v in sorted(result.items()) if v is not None]

def bench(engine: EngineBase, n: int):
    random.seed(42)
    # 写入测试
    t0 = time.perf_counter()
    for i in range(n):
        engine.put(f"key_{i:05d}", f"value_{i}")
    write_t = time.perf_counter() - t0

    # 随机读取
    t0 = time.perf_counter()
    found = sum(1 for i in range(n) if engine.get(f"key_{random.randint(0,n-1):05d}"))
    read_t = time.perf_counter() - t0

    # 范围扫描
    t0 = time.perf_counter()
    items = engine.scan_all()
    scan_t = time.perf_counter() - t0

    return {"write_ms": write_t*1000, "read_ms": read_t*1000,
            "scan_ms": scan_t*1000, "found": found, "total": len(items)}

N = 5000
engines = [("Heap", HeapEngine()), ("Sorted", SortedEngine()), ("LSM", LSMEngine(100))]
print(f"{'引擎':>8} | {'写入ms':>8} | {'读取ms':>8} | {'扫描ms':>8} | {'记录数':>6}")
print("-" * 55)
for name, eng in engines:
    r = bench(eng, N)
    print(f"{name:>8} | {r['write_ms']:>7.1f} | {r['read_ms']:>7.1f} | {r['scan_ms']:>7.1f} | {r['total']:>6}")
print("\n✅ 引擎基准测试完成")

📊 主流引擎对比

🔑 关键概念总结