第01课：数据库内核概述

存储引擎 第1课 / 共25课

📖 课程概述

数据库是现代软件基础设施的核心组件。从Web应用到大数据分析，从金融系统到物联网平台，几乎所有业务系统都依赖数据库来持久化和管理数据。但数据库内部是如何工作的？一条SQL语句从输入到返回结果，经历了怎样的旅程？

🏛️ 数据库内核是什么

数据库内核（Database Kernel）是数据库管理系统的核心软件层，负责数据的存储、检索、更新和事务管理。它位于用户查询和物理硬件之间，提供高效、可靠、并发的数据访问能力。

三层架构详解

🔬 主流数据库内核对比

💻 C语言实现：迷你数据库内核骨架

我们用C语言实现一个最小的数据库内核骨架，展示各组件如何协作：

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

#define PAGE_SIZE 4096
#define MAX_PAGES 1024
#define MAX_TABLES 32
#define MAX_COLUMN_NAME 64
#define MAX_DATA_SIZE 255

// ============ 页式存储 ============
typedef struct {
    uint32_t page_id;
    uint32_t num_records;
    uint32_t free_space;
    uint8_t data[PAGE_SIZE - 12];
} Page;

// ============ 缓冲池 ============
typedef struct {
    Page* pages[MAX_PAGES];
    int   dirty[MAX_PAGES];     // 脏页标记
    int   pin_count[MAX_PAGES]; // 引用计数
    int   lru_counter[MAX_PAGES];
    int   num_pages;
    int   global_lru;
} BufferPool;

BufferPool* buffer_pool_create() {
    BufferPool* pool = calloc(1, sizeof(BufferPool));
    pool->num_pages = 0;
    pool->global_lru = 0;
    printf("[BufferPool] 创建缓冲池，最大 %d 页\n", MAX_PAGES);
    return pool;
}

Page* buffer_pool_fetch(BufferPool* pool, uint32_t page_id) {
    // 简化：直接查找或分配
    if (page_id < (uint32_t)pool->num_pages && pool->pages[page_id]) {
        pool->lru_counter[page_id] = ++pool->global_lru;
        return pool->pages[page_id];
    }
    // 分配新页
    if (pool->num_pages >= MAX_PAGES) {
        printf("[BufferPool] 缓冲池已满，需要淘汰!\n");
        // LRU淘汰
        int min_lru = 0;
        for (int i = 1; i < pool->num_pages; i++) {
            if (pool->pin_count[i] == 0 &&
                pool->lru_counter[i] < pool->lru_counter[min_lru]) {
                min_lru = i;
            }
        }
        if (pool->dirty[min_lru]) {
            printf("[BufferPool] 将脏页 %d 写回磁盘\n", min_lru);
        }
        pool->pages[min_lru]->page_id = page_id;
        pool->pages[min_lru]->num_records = 0;
        pool->pages[min_lru]->free_space = PAGE_SIZE - 12;
        pool->dirty[min_lru] = 0;
        pool->lru_counter[min_lru] = ++pool->global_lru;
        return pool->pages[min_lru];
    }
    Page* page = calloc(1, sizeof(Page));
    page->page_id = page_id;
    page->free_space = PAGE_SIZE - 12;
    pool->pages[pool->num_pages] = page;
    pool->dirty[pool->num_pages] = 0;
    pool->pin_count[pool->num_pages] = 1;
    pool->lru_counter[pool->num_pages] = ++pool->global_lru;
    pool->num_pages++;
    printf("[BufferPool] 分配新页 %u，当前 %d 页\n", page_id, pool->num_pages);
    return page;
}

// ============ WAL日志 ============
typedef struct {
    uint32_t lsn;           // 日志序列号
    uint32_t txn_id;        // 事务ID
    char     operation[16]; // 操作类型
    uint32_t page_id;       // 涉及页
    char     data[MAX_DATA_SIZE];
} WALRecord;

typedef struct {
    WALRecord records[4096];
    int       count;
    uint32_t  next_lsn;
} WALManager;

WALManager* wal_create() {
    WALManager* wal = calloc(1, sizeof(WALManager));
    wal->next_lsn = 1;
    printf("[WAL] 创建WAL管理器\n");
    return wal;
}

void wal_append(WALManager* wal, uint32_t txn_id,
                const char* op, uint32_t page_id, const char* data) {
    WALRecord* rec = &wal->records[wal->count++];
    rec->lsn = wal->next_lsn++;
    rec->txn_id = txn_id;
    strncpy(rec->operation, op, 15);
    rec->page_id = page_id;
    strncpy(rec->data, data, MAX_DATA_SIZE - 1);
    printf("[WAL] LSN=%u TXN=%u OP=%s PAGE=%u DATA=%s\n",
           rec->lsn, rec->txn_id, rec->operation, rec->page_id, rec->data);
}

// ============ 锁管理器 ============
typedef struct {
    uint32_t txn_id;
    uint32_t resource_id;
    char     lock_type[8]; // "S" or "X"
} LockEntry;

typedef struct {
    LockEntry locks[256];
    int       count;
} LockManager;

LockManager* lock_manager_create() {
    LockManager* lm = calloc(1, sizeof(LockManager));
    printf("[Lock] 创建锁管理器\n");
    return lm;
}

int lock_acquire(LockManager* lm, uint32_t txn_id,
                 uint32_t res_id, const char* type) {
    // 简化：检查冲突
    for (int i = 0; i < lm->count; i++) {
        if (lm->locks[i].resource_id == res_id &&
            lm->locks[i].txn_id != txn_id) {
            if (strcmp(type, "X") == 0 ||
                strcmp(lm->locks[i].lock_type, "X") == 0) {
                printf("[Lock] 冲突! TXN %u 请求 %s 锁被 TXN %u 的 %s 锁阻塞\n",
                       txn_id, type, lm->locks[i].txn_id, lm->locks[i].lock_type);
                return -1; // 冲突
            }
        }
    }
    LockEntry* le = &lm->locks[lm->count++];
    le->txn_id = txn_id;
    le->resource_id = res_id;
    strncpy(le->lock_type, type, 7);
    printf("[Lock] TXN %u 获取资源 %u 的 %s 锁\n", txn_id, res_id, type);
    return 0;
}

// ============ 事务管理器 ============
typedef struct {
    uint32_t         next_txn_id;
    WALManager*      wal;
    LockManager*     lock_mgr;
    BufferPool*      buffer_pool;
} TransactionManager;

TransactionManager* txn_manager_create(BufferPool* bp, WALManager* wal, LockManager* lm) {
    TransactionManager* tm = calloc(1, sizeof(TransactionManager));
    tm->next_txn_id = 1;
    tm->wal = wal;
    tm->lock_mgr = lm;
    tm->buffer_pool = bp;
    printf("[TxnMgr] 创建事务管理器\n");
    return tm;
}

uint32_t txn_begin(TransactionManager* tm) {
    uint32_t id = tm->next_txn_id++;
    printf("[TxnMgr] 事务 %u BEGIN\n", id);
    return id;
}

void txn_commit(TransactionManager* tm, uint32_t txn_id) {
    wal_append(tm->wal, txn_id, "COMMIT", 0, "");
    printf("[TxnMgr] 事务 %u COMMIT\n", txn_id);
    // 释放所有锁
    int new_count = 0;
    for (int i = 0; i < tm->lock_mgr->count; i++) {
        if (tm->lock_mgr->locks[i].txn_id != txn_id) {
            tm->lock_mgr->locks[new_count++] = tm->lock_mgr->locks[i];
        }
    }
    tm->lock_mgr->count = new_count;
}

// ============ 查询执行 ============
typedef struct {
    char column[MAX_COLUMN_NAME];
    char value[MAX_DATA_SIZE];
} Record;

// 简化：向页面插入记录
int page_insert_record(Page* page, const char* col, const char* val) {
    if (page->free_space < strlen(col) + strlen(val) + 8) {
        printf("[Page] 页 %u 空间不足!\n", page->page_id);
        return -1;
    }
    // 简化：直接追加到data区
    int offset = PAGE_SIZE - 12 - page->free_space;
    int written = snprintf((char*)page->data + offset,
        page->free_space, "%s=%s;", col, val);
    page->free_space -= written;
    page->num_records++;
    return 0;
}

void execute_insert(TransactionManager* tm, uint32_t txn_id,
                    uint32_t page_id, const char* col, const char* val) {
    // 1. 获取锁
    lock_acquire(tm->lock_mgr, txn_id, page_id, "X");
    // 2. 写WAL
    char data[256];
    snprintf(data, sizeof(data), "INSERT %s=%s", col, val);
    wal_append(tm->wal, txn_id, "INSERT", page_id, data);
    // 3. 修改页面
    Page* page = buffer_pool_fetch(tm->buffer_pool, page_id);
    page_insert_record(page, col, val);
    tm->buffer_pool->dirty[page_id] = 1;
    printf("[Executor] INSERT 执行完成: %s=%s → 页 %u\n", col, val, page_id);
}

// ============ 主函数 ============
int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   MiniDB - 数据库内核骨架 v0.1       ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    // 初始化各组件
    BufferPool*       bp = buffer_pool_create();
    WALManager*       wal = wal_create();
    LockManager*      lm = lock_manager_create();
    TransactionManager* tm = txn_manager_create(bp, wal, lm);

    printf("\n--- 执行事务1: 插入数据 ---\n");
    uint32_t txn1 = txn_begin(tm);
    execute_insert(tm, txn1, 0, "name", "Alice");
    execute_insert(tm, txn1, 0, "age", "30");
    execute_insert(tm, txn1, 0, "city", "Beijing");
    txn_commit(tm, txn1);

    printf("\n--- 执行事务2: 插入更多数据 ---\n");
    uint32_t txn2 = txn_begin(tm);
    execute_insert(tm, txn2, 1, "name", "Bob");
    execute_insert(tm, txn2, 1, "age", "25");
    execute_insert(tm, txn2, 1, "city", "Shanghai");
    txn_commit(tm, txn2);

    printf("\n--- 模拟并发冲突 ---\n");
    uint32_t txn3 = txn_begin(tm);
    uint32_t txn4 = txn_begin(tm);
    execute_insert(tm, txn3, 0, "email", "alice@test.com");
    execute_insert(tm, txn4, 0, "phone", "123456");  // 冲突!
    txn_commit(tm, txn3);
    // txn4 重试
    execute_insert(tm, txn4, 0, "phone", "123456");
    txn_commit(tm, txn4);

    printf("\n--- 统计信息 ---\n");
    printf("缓冲池页数: %d\n", bp->num_pages);
    printf("WAL记录数:  %d\n", wal->count);
    printf("活跃锁数:   %d\n", lm->count);
    for (int i = 0; i < bp->num_pages; i++) {
        printf("页 %d: %u 条记录, 剩余空间 %u 字节, 脏=%d\n",
               i, bp->pages[i]->num_records,
               bp->pages[i]->free_space, bp->dirty[i]);
    }

    printf("\n✅ 数据库内核骨架运行完成\n");
    return 0;
}

编译与运行

$ gcc -o minidb minidb.c -Wall
$ ./minidb
╔══════════════════════════════════════╗
║   MiniDB - 数据库内核骨架 v0.1       ║
╚══════════════════════════════════════╝

[BufferPool] 创建缓冲池，最大 1024 页
[WAL] 创建WAL管理器
[Lock] 创建锁管理器
[TxnMgr] 创建事务管理器

--- 执行事务1: 插入数据 ---
[TxnMgr] 事务 1 BEGIN
[Lock] TXN 1 获取资源 0 的 X 锁
[WAL] LSN=1 TXN=1 OP=INSERT PAGE=0 DATA=INSERT name=Alice
...
✅ 数据库内核骨架运行完成

🐍 Python实现：内核组件交互模拟

"""
数据库内核组件交互模拟
演示查询从解析到存储的完整路径
"""
import time
from dataclasses import dataclass, field
from typing import List, Optional, Dict

@dataclass
class Page:
    """数据页"""
    page_id: int
    records: List[dict] = field(default_factory=list)
    free_space: int = 4096 - 12

    def insert(self, record: dict) -> bool:
        size = sum(len(str(v)) for v in record.values()) + len(record) * 8
        if size > self.free_space:
            return False
        self.records.append(record)
        self.free_space -= size
        return True

@dataclass
class WALRecord:
    lsn: int
    txn_id: int
    operation: str
    page_id: int
    data: str

class WALManager:
    def __init__(self):
        self.records: List[WALRecord] = []
        self.next_lsn = 1

    def append(self, txn_id: int, op: str, page_id: int, data: str):
        rec = WALRecord(self.next_lsn, txn_id, op, page_id, data)
        self.records.append(rec)
        self.next_lsn += 1
        print(f"  [WAL] LSN={rec.lsn} TXN={txn_id} {op} page={page_id}")

class BufferPool:
    def __init__(self, capacity: int = 1024):
        self.capacity = capacity
        self.pages: Dict[int, Page] = {}
        self.dirty: set = set()
        self.access_time: Dict[int, float] = {}

    def fetch(self, page_id: int) -> Page:
        if page_id in self.pages:
            self.access_time[page_id] = time.time()
            print(f"  [Buffer] 命中页 {page_id} (缓存)")
            return self.pages[page_id]
        if len(self.pages) >= self.capacity:
            self._evict()
        page = Page(page_id=page_id)
        self.pages[page_id] = page
        self.access_time[page_id] = time.time()
        print(f"  [Buffer] 加载页 {page_id} (新分配)")
        return page

    def _evict(self):
        oldest = min(self.access_time, key=self.access_time.get)
        if oldest in self.dirty:
            print(f"  [Buffer] 淘汰脏页 {oldest}，写回磁盘")
        else:
            print(f"  [Buffer] 淘汰干净页 {oldest}")
        del self.pages[oldest]
        del self.access_time[oldest]
        self.dirty.discard(oldest)

class SQLParser:
    """简化SQL解析器"""
    def parse(self, sql: str) -> dict:
        sql = sql.strip().rstrip(";")
        tokens = sql.split()
        if not tokens:
            return {"type": "UNKNOWN"}
        cmd = tokens[0].upper()
        if cmd == "INSERT":
            # INSERT INTO table (cols) VALUES (vals)
            table = tokens[2]
            # 简化解析
            col_start = sql.index("(") + 1
            col_end = sql.index(")")
            cols = [c.strip() for c in sql[col_start:col_end].split(",")]
            val_part = sql[sql.index("VALUES") + 6:].strip()
            val_start = val_part.index("(") + 1
            val_end = val_part.index(")")
            vals = [v.strip().strip("'\"") for v in val_part[val_start:val_end].split(",")]
            return {"type": "INSERT", "table": table, "columns": cols, "values": vals}
        elif cmd == "SELECT":
            return {"type": "SELECT", "table": tokens[3] if len(tokens) > 3 else "?"}
        elif cmd == "UPDATE":
            return {"type": "UPDATE", "table": tokens[1]}
        elif cmd == "DELETE":
            return {"type": "DELETE", "table": tokens[2]}
        return {"type": cmd}

class QueryOptimizer:
    """简化查询优化器"""
    def optimize(self, plan: dict) -> dict:
        # 简化：添加执行代价估算
        if plan["type"] == "INSERT":
            plan["cost"] = 1.0  # 单行插入代价低
            plan["method"] = "direct_insert"
        elif plan["type"] == "SELECT":
            plan["cost"] = 10.0
            plan["method"] = "full_scan"  # 默认全表扫描
            plan["index_hint"] = "consider_index_on_filter"
        print(f"  [Optimizer] 优化计划: {plan.get('method','?')} 代价={plan.get('cost',0):.1f}")
        return plan

class StorageEngine:
    """存储引擎"""
    def __init__(self):
        self.buffer_pool = BufferPool(capacity=8)
        self.wal = WALManager()
        self.tables: Dict[str, List[int]] = {}  # table → page_ids
        self.next_page = 0

    def insert(self, txn_id: int, table: str, record: dict):
        # 分配或查找页面
        if table not in self.tables:
            page_id = self.next_page
            self.next_page += 1
            self.tables[table] = [page_id]
        page_id = self.tables[table][-1]
        page = self.buffer_pool.fetch(page_id)

        # WAL先写
        self.wal.append(txn_id, "INSERT", page_id, str(record))

        # 写入页面
        if not page.insert(record):
            # 当前页满，分配新页
            page_id = self.next_page
            self.next_page += 1
            self.tables[table].append(page_id)
            page = self.buffer_pool.fetch(page_id)
            page.insert(record)

        self.buffer_pool.dirty.add(page_id)
        print(f"  [Storage] 插入 {record} → 页 {page_id}")

    def scan(self, table: str) -> List[dict]:
        results = []
        for pid in self.tables.get(table, []):
            page = self.buffer_pool.fetch(pid)
            results.extend(page.records)
        print(f"  [Storage] 扫描表 {table}: {len(results)} 条记录")
        return results

class MiniDB:
    """迷你数据库 - 完整内核"""
    def __init__(self):
        self.storage = StorageEngine()
        self.parser = SQLParser()
        self.optimizer = QueryOptimizer()
        self.next_txn = 1
        self.txn_active: Dict[int, str] = {}

    def begin(self) -> int:
        txn_id = self.next_txn
        self.next_txn += 1
        self.txn_active[txn_id] = "active"
        print(f"[TXN {txn_id}] BEGIN")
        return txn_id

    def commit(self, txn_id: int):
        self.storage.wal.append(txn_id, "COMMIT", 0, "")
        del self.txn_active[txn_id]
        print(f"[TXN {txn_id}] COMMIT ✓")

    def execute(self, sql: str, txn_id: int = None):
        print(f"\n>>> {sql}")
        # 1. 解析
        plan = self.parser.parse(sql)
        print(f"  [Parser] 解析结果: type={plan['type']}")
        # 2. 优化
        plan = self.optimizer.optimize(plan)
        # 3. 执行
        if plan["type"] == "INSERT":
            if txn_id is None:
                txn_id = self.begin()
            record = dict(zip(plan.get("columns", []), plan.get("values", [])))
            self.storage.insert(txn_id, plan["table"], record)
        elif plan["type"] == "SELECT":
            results = self.storage.scan(plan.get("table", "unknown"))
            for r in results:
                print(f"    → {r}")
            return results
        return None

# ========== 运行演示 ==========
db = MiniDB()

# 事务1：创建用户数据
t1 = db.begin()
db.execute("INSERT INTO users (name, age, city) VALUES (Alice, 30, Beijing)", t1)
db.execute("INSERT INTO users (name, age, city) VALUES (Bob, 25, Shanghai)", t1)
db.commit(t1)

# 事务2：更多数据
t2 = db.begin()
db.execute("INSERT INTO users (name, age, city) VALUES (Charlie, 35, Shenzhen)", t2)
db.execute("INSERT INTO users (name, age, city) VALUES (Diana, 28, Hangzhou)", t2)
db.commit(t2)

# 查询
print("\n=== 查询所有用户 ===")
db.execute("SELECT * FROM users")

# 统计
print(f"\n=== 内核统计 ===")
print(f"WAL记录: {len(db.storage.wal.records)}")
print(f"缓冲池页: {len(db.storage.buffer_pool.pages)}")
print(f"脏页数: {len(db.storage.buffer_pool.dirty)}")
print(f"表信息: {db.storage.tables}")
print("✅ Python数据库内核模拟运行完成")

📊 性能分析：各层开销分布

🔑 关键概念总结