第10课：多版本索引

索引结构 第10课 / 共25课

📖 课程概述

多版本索引是MVCC(多版本并发控制)的存储基础。数据库需要同时维护数据的多个历史版本，以支持快照读和并发事务。本课探讨多版本数据的存储方式、版本链的管理、以及PostgreSQL和MySQL不同的多版本实现策略。

🔄 多版本数据存储

两种方案对比

💻 C语言实现：多版本存储引擎

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

#define MAX_VERSIONS  64
#define MAX_KEY       64
#define MAX_VAL       255
#define MAX_ROWS      100

// 事务ID类型
typedef uint32_t TxnID;

// 版本链节点(PostgreSQL风格)
typedef struct VersionNode {
    TxnID   xmin;          // 创建该版本的事务ID
    TxnID   xmax;          // 删除/更新该版本的事务ID(0=仍有效)
    char    key[MAX_KEY];
    char    value[MAX_VAL];
    struct VersionNode* next; // 指向新版本
    int     active;        // 是否仍可被某快照看到
} VersionNode;

// 行的多版本链
typedef struct {
    VersionNode* head;      // 最旧版本
    VersionNode* tail;      // 最新版本
    int          num_versions;
} VersionChain;

// 快照(事务可见性判断)
typedef struct {
    TxnID   txn_id;
    TxnID   snap_min;      // 快照最小活跃事务
    TxnID   snap_max;      // 快照最大活跃事务
    TxnID*  active_txns;   // 活跃事务列表
    int     num_active;
} Snapshot;

// MVCC存储引擎
typedef struct {
    VersionChain rows[MAX_ROWS];
    int          num_rows;
    TxnID        next_txn;
    TxnID*       active_txns;
    int          num_active;
    int          total_versions;
    int          dead_versions;
} MVCCEngine;

MVCCEngine* mvcc_create() {
    MVCCEngine* e = calloc(1, sizeof(MVCCEngine));
    e->next_txn = 1;
    printf("[MVCC] 创建多版本存储引擎\n");
    return e;
}

// 创建快照
Snapshot mvcc_snapshot(MVCCEngine* e, TxnID txn_id) {
    Snapshot snap;
    snap.txn_id = txn_id;
    snap.snap_min = e->active_txns[0];
    snap.snap_max = e->active_txns[e->num_active - 1];
    snap.active_txns = e->active_txns;
    snap.num_active = e->num_active;
    return snap;
}

// 可见性判断
int is_visible(VersionNode* v, Snapshot* snap) {
    // 规则1: xmin已提交，且xmin < 当前事务ID
    if (v->xmin > snap->txn_id) return 0;  // 未来事务创建，不可见
    // 规则2: xmin在活跃列表中 → 不可见
    for (int i = 0; i < snap->num_active; i++) {
        if (v->xmin == snap->active_txns[i]) return 0;
    }
    // 规则3: xmax为0 → 仍有效，可见
    if (v->xmax == 0) return 1;
    // 规则4: xmax >= 当前事务ID → 可见(还没被删除)
    if (v->xmax > snap->txn_id) return 1;
    // 规则5: xmax在活跃列表中 → 可见(删除未提交)
    for (int i = 0; i < snap->num_active; i++) {
        if (v->xmax == snap->active_txns[i]) return 1;
    }
    return 0; // 已被删除，不可见
}

// 插入
int mvcc_insert(MVCCEngine* e, TxnID txn, const char* key, const char* val) {
    if (e->num_rows >= MAX_ROWS) return -1;
    VersionChain* chain = &e->rows[e->num_rows++];
    VersionNode* v = calloc(1, sizeof(VersionNode));
    v->xmin = txn;
    v->xmax = 0;
    strcpy(v->key, key);
    strcpy(v->value, val);
    v->active = 1;
    chain->head = chain->tail = v;
    chain->num_versions = 1;
    e->total_versions++;
    printf("  [MVCC] INSERT: %s=%s (txn=%u)\n", key, val, txn);
    return 0;
}

// 更新(创建新版本)
int mvcc_update(MVCCEngine* e, TxnID txn, const char* key, const char* new_val) {
    for (int i = 0; i < e->num_rows; i++) {
        VersionChain* chain = &e->rows[i];
        if (chain->tail && strcmp(chain->tail->key, key) == 0) {
            // 标记旧版本xmax
            chain->tail->xmax = txn;
            chain->tail->active = 0;
            e->dead_versions++;
            // 创建新版本
            VersionNode* v = calloc(1, sizeof(VersionNode));
            v->xmin = txn;
            v->xmax = 0;
            strcpy(v->key, key);
            strcpy(v->value, new_val);
            v->active = 1;
            chain->tail->next = v;
            chain->tail = v;
            chain->num_versions++;
            e->total_versions++;
            printf("  [MVCC] UPDATE: %s=%s→%s (txn=%u)\n", key,
                   chain->tail->next ? chain->tail->next->value : "?", new_val, txn);
            return 0;
        }
    }
    return -1;
}

// 删除(标记xmax)
int mvcc_delete(MVCCEngine* e, TxnID txn, const char* key) {
    for (int i = 0; i < e->num_rows; i++) {
        VersionChain* chain = &e->rows[i];
        if (chain->tail && strcmp(chain->tail->key, key) == 0 && chain->tail->xmax == 0) {
            chain->tail->xmax = txn;
            chain->tail->active = 0;
            e->dead_versions++;
            printf("  [MVCC] DELETE: %s (txn=%u)\n", key, txn);
            return 0;
        }
    }
    return -1;
}

// 快照读
void mvcc_snapshot_read(MVCCEngine* e, Snapshot* snap) {
    printf("  [MVCC] 快照读 (txn=%u):\n", snap->txn_id);
    for (int i = 0; i < e->num_rows; i++) {
        VersionChain* chain = &e->rows[i];
        VersionNode* v = chain->head;
        while (v) {
            if (is_visible(v, snap)) {
                printf("    %s = %s (xmin=%u, xmax=%u)\n",
                       v->key, v->value, v->xmin, v->xmax);
                break;
            }
            v = v->next;
        }
    }
}

// VACUUM: 清理不可见的死版本
int mvcc_vacuum(MVCCEngine* e) {
    int reclaimed = 0;
    for (int i = 0; i < e->num_rows; i++) {
        VersionChain* chain = &e->rows[i];
        VersionNode* prev = NULL;
        VersionNode* v = chain->head;
        while (v && v != chain->tail) {
            // 如果xmin和xmax都已提交，且xmax不为0 → 可清理
            if (v->xmax != 0) {
                VersionNode* dead = v;
                if (prev) prev->next = v->next;
                else chain->head = v->next;
                v = v->next;
                free(dead);
                chain->num_versions--;
                e->dead_versions--;
                e->total_versions--;
                reclaimed++;
                continue;
            }
            prev = v;
            v = v->next;
        }
    }
    printf("  [VACUUM] 回收 %d 个死版本\n", reclaimed);
    return reclaimed;
}

void mvcc_stats(MVCCEngine* e) {
    printf("\n=== MVCC统计 ===\n");
    printf("行数: %d  总版本: %d  死版本: %d\n",
           e->num_rows, e->total_versions, e->dead_versions);
    for (int i = 0; i < e->num_rows; i++) {
        VersionChain* chain = &e->rows[i];
        printf("  行%d: %d个版本", i, chain->num_versions);
        VersionNode* v = chain->head;
        while (v) {
            printf(" [%s=%s xmin=%u xmax=%u]", v->key, v->value, v->xmin, v->xmax);
            v = v->next;
        }
        printf("\n");
    }
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   多版本索引与MVCC存储               ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    MVCCEngine* e = mvcc_create();

    // 事务1: 初始数据
    printf("--- 事务1: 初始插入 ---\n");
    TxnID t1 = e->next_txn++;
    mvcc_insert(e, t1, "alice", "Beijing");
    mvcc_insert(e, t1, "bob", "Shanghai");
    mvcc_insert(e, t1, "charlie", "Shenzhen");

    // 事务2: 更新
    printf("\n--- 事务2: 更新 ---\n");
    TxnID t2 = e->next_txn++;
    TxnID active1[] = {t2};
    e->active_txns = active1; e->num_active = 1;
    mvcc_update(e, t2, "alice", "Hangzhou");
    mvcc_update(e, t2, "bob", "Nanjing");

    // 事务3: 快照读(看到事务1的版本)
    printf("\n--- 事务3: 快照读 ---\n");
    TxnID t3 = e->next_txn++;
    Snapshot snap3 = {t3, t1, t2, active1, 1};
    mvcc_snapshot_read(e, &snap3);

    // 事务4: 快照读(看到事务2的版本)
    printf("\n--- 事务4: 最新快照读 ---\n");
    TxnID t4 = e->next_txn++;
    Snapshot snap4 = {t4, 0, 0, NULL, 0};
    mvcc_snapshot_read(e, &snap4);

    // 事务5: 删除
    printf("\n--- 事务5: 删除 ---\n");
    TxnID t5 = e->next_txn++;
    mvcc_delete(e, t5, "charlie");

    // VACUUM
    printf("\n--- VACUUM ---\n");
    mvcc_stats(e);
    mvcc_vacuum(e);
    mvcc_stats(e);

    printf("\n✅ 多版本存储引擎运行完成\n");
    return 0;
}

🐍 Python实现：Undo Log版本链

"""
MySQL/InnoDB风格的Undo Log版本链
"""
from dataclasses import dataclass, field
from typing import Optional, Dict, List

@dataclass
class UndoLogRecord:
    trx_id: int
    roll_ptr: Optional['UndoLogRecord'] = None
    data: dict = field(default_factory=dict)

@dataclass
class ClusterIndexRecord:
    """聚簇索引中的最新版本"""
    pk: int
    data: dict
    trx_id: int
    roll_ptr: Optional[UndoLogRecord] = None

class UndoLog:
    def __init__(self):
        self.records: List[UndoLogRecord] = []
        self.purge_ptr = 0

    def add(self, trx_id, data, old_roll_ptr=None):
        rec = UndoLogRecord(trx_id=trx_id, roll_ptr=old_roll_ptr, data=data)
        self.records.append(rec)
        return rec

    def purge(self, max_trx_id):
        """清除所有trx_id < max_trx_id的undo记录"""
        before = len(self.records)
        self.records = [r for r in self.records if r.trx_id >= max_trx_id]
        purged = before - len(self.records)
        print(f"  [Purge] 清除 {purged} 条undo记录")
        return purged

class InnoDBTable:
    """模拟InnoDB聚簇索引+Undo Log"""
    def __init__(self):
        self.rows: Dict[int, ClusterIndexRecord] = {}
        self.undo = UndoLog()
        self.next_trx = 1
        self.committed = set()

    def begin(self):
        txn = self.next_trx
        self.next_trx += 1
        return txn

    def commit(self, txn):
        self.committed.add(txn)
        print(f"  [Txn {txn}] COMMIT")
        # 尝试Purge
        if len(self.committed) > 5:
            min_active = min(self.committed)
            self.undo.purge(min_active)

    def insert(self, txn, pk, **fields):
        rec = ClusterIndexRecord(pk=pk, data=fields, trx_id=txn)
        self.rows[pk] = rec
        print(f"  [Txn {txn}] INSERT pk={pk}: {fields}")

    def update(self, txn, pk, **new_fields):
        if pk not in self.rows: return
        old = self.rows[pk]
        # 写Undo Log
        undo_rec = self.undo.add(old.trx_id, old.data.copy(), old.roll_ptr)
        # 更新聚簇索引
        old.data.update(new_fields)
        old.trx_id = txn
        old.roll_ptr = undo_rec
        print(f"  [Txn {txn}] UPDATE pk={pk}: {new_fields}")

    def delete(self, txn, pk):
        if pk not in self.rows: return
        old = self.rows[pk]
        undo_rec = self.undo.add(old.trx_id, old.data.copy(), old.roll_ptr)
        # 标记删除(写undo但不删除聚簇索引)
        old.data['__deleted__'] = True
        old.trx_id = txn
        old.roll_ptr = undo_rec
        print(f"  [Txn {txn}] DELETE pk={pk}")

    def read(self, txn, pk, read_trx=None):
        """读操作，沿着版本链找到可见版本"""
        if pk not in self.rows: return None
        rec = self.rows[pk]
        # 如果最新版本可见
        if read_trx is None or rec.trx_id in self.committed:
            if rec.data.get('__deleted__'): return None
            return rec.data
        # 沿版本链查找
        current = rec.roll_ptr
        while current:
            if current.trx_id in self.committed:
                if current.data.get('__deleted__'): return None
                return current.data
            current = current.roll_ptr
        return None

    def consistent_read(self, read_trx, pk):
        """一致性非锁定读(MVCC)"""
        val = self.read(read_trx, pk, read_trx)
        print(f"  [Read txn={read_trx}] pk={pk} → {val}")
        return val

    def stats(self):
        total_versions = 0
        for rec in self.rows.values():
            v = rec
            while v:
                total_versions += 1
                v = v.roll_ptr if hasattr(v, 'roll_ptr') else None
        # Count undo chain
        undo_count = len(self.undo.records)
        print(f"\n=== InnoDB MVCC统计 ===")
        print(f"行数: {len(self.rows)}  Undo记录: {undo_count}  已提交事务: {len(self.committed)}")

# 测试
table = InnoDBTable()

t1 = table.begin()
table.insert(t1, 1, name="Alice", age=30, city="Beijing")
table.insert(t1, 2, name="Bob", age=25, city="Shanghai")
table.commit(t1)

t2 = table.begin()
table.update(t2, 1, age=31, city="Hangzhou")
table.commit(t2)

t3 = table.begin()
table.update(t3, 1, age=32)
table.delete(t3, 2)
table.commit(t3)

# 读取
print("\n--- 一致性读 ---")
t4 = table.begin()
table.consistent_read(t4, 1)
table.consistent_read(t4, 2)
table.commit(t4)

table.stats()
print("\n✅ Undo Log版本链实现完成")

🔑 关键概念总结