第04课：堆文件与索引文件

存储引擎 第4课 / 共25课

📖 课程概述

数据库表在磁盘上的物理组织方式主要有两种：堆文件(Heap File)和索引组织表(Index-Organized Table)。堆文件将记录无序存储，通过索引加速查找；索引组织表按主键排序存储，数据即索引。本课深入探讨两种方案的实现与取舍。

🗄️ 堆文件 vs 索引组织表

对比分析

💻 C语言实现：堆文件管理器

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

#define PAGE_SIZE   4096
#define MAX_PAGES   64
#define MAX_RECORDS 100
#define MAX_DATA    200

// 元组标识符(TID)
typedef struct {
    uint32_t page_id;
    uint16_t slot_id;
} TID;

// 堆记录
typedef struct {
    TID     tid;
    uint16_t length;
    char     data[MAX_DATA];
    uint8_t  deleted;
} HeapRecord;

// 堆页面
typedef struct {
    uint32_t    page_id;
    uint32_t    next_page;  // 链表下一页
    uint16_t    num_records;
    uint16_t    free_space;
    HeapRecord  records[MAX_RECORDS];
} HeapPage;

// 堆文件
typedef struct {
    HeapPage* pages[MAX_PAGES];
    int       num_pages;
    uint32_t  next_page_id;
} HeapFile;

// 索引条目
typedef struct {
    char     key[64];
    TID      tid;
} IndexEntry;

// 简单哈希索引
#define INDEX_BUCKETS 256
typedef struct {
    IndexEntry entries[INDEX_BUCKETS][8];
    int        counts[INDEX_BUCKETS];
} HashIndex;

// ========== 堆文件操作 ==========
HeapFile* heap_file_create() {
    HeapFile* hf = calloc(1, sizeof(HeapFile));
    hf->next_page_id = 0;
    printf("[Heap] 创建堆文件\n");
    return hf;
}

HeapPage* heap_page_create(HeapFile* hf) {
    if (hf->num_pages >= MAX_PAGES) return NULL;
    HeapPage* hp = calloc(1, sizeof(HeapPage));
    hp->page_id = hf->next_page_id++;
    hp->next_page = UINT32_MAX;
    hp->free_space = PAGE_SIZE - sizeof(HeapPage) + MAX_DATA * MAX_RECORDS;
    hf->pages[hf->num_pages++] = hp;
    if (hf->num_pages > 1) {
        hf->pages[hf->num_pages - 2]->next_page = hp->page_id;
    }
    printf("[Heap] 创建堆页 %u\n", hp->page_id);
    return hp;
}

TID heap_insert(HeapFile* hf, const char* data) {
    // 找有空间的最后一页
    HeapPage* page = NULL;
    if (hf->num_pages > 0) {
        page = hf->pages[hf->num_pages - 1];
    }
    if (!page || page->num_records >= MAX_RECORDS) {
        page = heap_page_create(hf);
        if (!page) {
            TID empty = {UINT32_MAX, UINT16_MAX};
            return empty;
        }
    }
    TID tid = {page->page_id, page->num_records};
    HeapRecord* rec = &page->records[page->num_records];
    rec->tid = tid;
    rec->length = strlen(data);
    strncpy(rec->data, data, MAX_DATA - 1);
    rec->deleted = 0;
    page->num_records++;
    page->free_space -= rec->length;
    return tid;
}

HeapRecord* heap_get(HeapFile* hf, TID tid) {
    for (int i = 0; i < hf->num_pages; i++) {
        if (hf->pages[i]->page_id == tid.page_id) {
            if (tid.slot_id < hf->pages[i]->num_records) {
                return &hf->pages[i]->records[tid.slot_id];
            }
        }
    }
    return NULL;
}

void heap_delete(HeapFile* hf, TID tid) {
    HeapRecord* rec = heap_get(hf, tid);
    if (rec) {
        rec->deleted = 1;
        printf("[Heap] 删除 TID=(%u,%u)\n", tid.page_id, tid.slot_id);
    }
}

// 全表扫描
void heap_scan(HeapFile* hf) {
    printf("[Heap] 全表扫描:\n");
    int count = 0;
    for (int i = 0; i < hf->num_pages; i++) {
        HeapPage* page = hf->pages[i];
        for (int j = 0; j < page->num_records; j++) {
            if (!page->records[j].deleted) {
                printf("  TID=(%u,%u): %s\n",
                       page->records[j].tid.page_id,
                       page->records[j].tid.slot_id,
                       page->records[j].data);
                count++;
            }
        }
    }
    printf("[Heap] 扫描完成: %d 条活跃记录\n", count);
}

// ========== 哈希索引 ==========
HashIndex* hash_index_create() {
    HashIndex* hi = calloc(1, sizeof(HashIndex));
    printf("[Index] 创建哈希索引\n");
    return hi;
}

uint32_t hash_func(const char* key) {
    uint32_t h = 5381;
    while (*key) h = h * 33 + (unsigned char)*key++;
    return h % INDEX_BUCKETS;
}

void hash_index_insert(HashIndex* hi, const char* key, TID tid) {
    uint32_t bucket = hash_func(key);
    if (hi->counts[bucket] >= 8) {
        printf("[Index] 桶 %u 已满!\n", bucket);
        return;
    }
    IndexEntry* e = &hi->entries[bucket][hi->counts[bucket]];
    strncpy(e->key, key, 63);
    e->tid = tid;
    hi->counts[bucket]++;
}

TID* hash_index_lookup(HashIndex* hi, const char* key) {
    uint32_t bucket = hash_func(key);
    for (int i = 0; i < hi->counts[bucket]; i++) {
        if (strcmp(hi->entries[bucket][i].key, key) == 0) {
            printf("[Index] 命中: key=%s → TID=(%u,%u)\n",
                   key, hi->entries[bucket][i].tid.page_id,
                   hi->entries[bucket][i].tid.slot_id);
            return &hi->entries[bucket][i].tid;
        }
    }
    printf("[Index] 未找到: key=%s\n", key);
    return NULL;
}

// ========== 主函数 ==========
int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   堆文件 & 索引文件管理器            ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    HeapFile*  hf = heap_file_create();
    HashIndex* hi = hash_index_create();

    // 插入数据
    printf("\n--- 插入数据 ---\n");
    struct { const char* key; const char* data; } rows[] = {
        {"alice",   "name=Alice,age=30,city=Beijing"},
        {"bob",     "name=Bob,age=25,city=Shanghai"},
        {"charlie", "name=Charlie,age=35,city=Shenzhen"},
        {"diana",   "name=Diana,age=28,city=Hangzhou"},
        {"eve",     "name=Eve,age=32,city=Chengdu"},
        {"frank",   "name=Frank,age=27,city=Wuhan"},
        {"grace",   "name=Grace,age=29,city=Nanjing"},
        {"henry",   "name=Henry,age=33,city=Xi'an"},
    };
    for (int i = 0; i < 8; i++) {
        TID tid = heap_insert(hf, rows[i].data);
        hash_index_insert(hi, rows[i].key, tid);
        printf("[Insert] key=%s → TID=(%u,%u)\n",
               rows[i].key, tid.page_id, tid.slot_id);
    }

    // 全表扫描
    printf("\n--- 全表扫描 ---\n");
    heap_scan(hf);

    // 索引查找
    printf("\n--- 索引查找 ---\n");
    const char* queries[] = {"alice", "diana", "xyz", "henry"};
    for (int i = 0; i < 4; i++) {
        TID* tid = hash_index_lookup(hi, queries[i]);
        if (tid) {
            HeapRecord* rec = heap_get(hf, *tid);
            if (rec) printf("  → 数据: %s\n", rec->data);
        }
    }

    // 删除+查找
    printf("\n--- 删除+查找测试 ---\n");
    TID* diana_tid = hash_index_lookup(hi, "diana");
    if (diana_tid) heap_delete(hf, *diana_tid);
    // 再次查找
    diana_tid = hash_index_lookup(hi, "diana");
    if (diana_tid) {
        HeapRecord* rec = heap_get(hf, *diana_tid);
        if (rec && rec->deleted) printf("  记录已删除\n");
    }

    printf("\n--- 统计 ---\n");
    printf("堆页数: %d\n", hf->num_pages);
    for (int i = 0; i < hf->num_pages; i++) {
        printf("页 %u: %u 条记录\n",
               hf->pages[i]->page_id, hf->pages[i]->num_records);
    }

    printf("\n✅ 堆文件管理器运行完成\n");
    return 0;
}

🐍 Python实现：索引组织表模拟

"""
索引组织表(IOT)模拟 - MySQL/InnoDB风格
数据按主键B+树组织，二级索引存储主键值
"""
from dataclasses import dataclass
from typing import List, Optional, Dict, Any

@dataclass
class Row:
    pk: int
    fields: Dict[str, Any]

class BPlusTreeNode:
    """B+树节点(简化版)"""
    def __init__(self, is_leaf=True, order=4):
        self.is_leaf = is_leaf
        self.order = order
        self.keys: List[int] = []
        self.values: List = []  # 叶子节点存行数据，内部节点存子节点
        self.next_leaf: Optional['BPlusTreeNode'] = None
        self.prev_leaf: Optional['BPlusTreeNode'] = None

class IndexOrganizedTable:
    """索引组织表 - InnoDB风格"""
    def __init__(self, order=4):
        self.root = BPlusTreeNode(is_leaf=True, order=order)
        self.order = order
        self.row_count = 0
        self.secondary_indexes: Dict[str, Dict] = {}

    def insert(self, pk: int, **fields):
        """按主键插入，数据存在B+树叶子节点"""
        row = Row(pk=pk, fields=fields)
        self._insert_to_tree(self.root, pk, row)
        self.row_count += 1
        # 更新二级索引
        for idx_name, idx in self.secondary_indexes.items():
            if idx_name in fields:
                idx[fields[idx_name]] = pk  # 存主键值!

    def _insert_to_tree(self, node, key, value):
        if node.is_leaf:
            # 找插入位置
            pos = 0
            while pos < len(node.keys) and node.keys[pos] < key:
                pos += 1
            node.keys.insert(pos, key)
            node.values.insert(pos, value)
            # 检查是否需要分裂
            if len(node.keys) >= node.order:
                self._split_leaf(node)
        else:
            # 内部节点：找子节点
            pos = 0
            while pos < len(node.keys) and node.keys[pos] <= key:
                pos += 1
            self._insert_to_tree(node.values[pos], key, value)

    def _split_leaf(self, node):
        mid = len(node.keys) // 2
        new_node = BPlusTreeNode(is_leaf=True, order=node.order)
        new_node.keys = node.keys[mid:]
        new_node.values = node.values[mid:]
        new_node.next_leaf = node.next_leaf
        new_node.prev_leaf = node
        if node.next_leaf:
            node.next_leaf.prev_leaf = new_node
        node.keys = node.keys[:mid]
        node.values = node.values[:mid]
        node.next_leaf = new_node

    def find_by_pk(self, pk: int) -> Optional[Row]:
        """主键查找 - 直接在B+树中定位"""
        return self._find_in_tree(self.root, pk)

    def _find_in_tree(self, node, key) -> Optional[Row]:
        if node.is_leaf:
            for i, k in enumerate(node.keys):
                if k == key:
                    return node.values[i]
            return None
        pos = 0
        while pos < len(node.keys) and node.keys[pos] <= key:
            pos += 1
        return self._find_in_tree(node.values[pos], key)

    def range_scan(self, start: int, end: int) -> List[Row]:
        """主键范围扫描 - 叶子链表顺序读"""
        results = []
        node = self._find_leaf(self.root, start)
        while node:
            for i, k in enumerate(node.keys):
                if k > end:
                    return results
                if start <= k <= end:
                    results.append(node.values[i])
            node = node.next_leaf
        return results

    def _find_leaf(self, node, key):
        if node.is_leaf:
            return node
        pos = 0
        while pos < len(node.keys) and node.keys[pos] <= key:
            pos += 1
        return self._find_leaf(node.values[pos], key)

    def create_secondary_index(self, column: str):
        """创建二级索引 - 存主键值而非TID"""
        self.secondary_indexes[column] = {}
        # 扫描重建
        self._rebuild_secondary_index(column)

    def _rebuild_secondary_index(self, column: str):
        idx = self.secondary_indexes[column]
        idx.clear()
        node = self._leftmost_leaf(self.root)
        while node:
            for row in node.values:
                if column in row.fields:
                    idx[row.fields[column]] = row.pk
            node = node.next_leaf

    def _leftmost_leaf(self, node):
        if node.is_leaf:
            return node
        return self._leftmost_leaf(node.values[0])

    def find_by_secondary(self, column: str, value) -> Optional[Row]:
        """二级索引查找 - 需要回表!"""
        idx = self.secondary_indexes.get(column)
        if not idx:
            return None
        pk = idx.get(value)
        if pk is None:
            return None
        # 回表：用主键值再查B+树
        print(f"  [回表] 二级索引 {column}={value} → PK={pk} → 查主键索引")
        return self.find_by_pk(pk)

# ========== 演示 ==========
iot = IndexOrganizedTable(order=4)

# 插入数据
for i in range(10):
    iot.insert(i+1, name=f"user_{i+1}", age=20+i, city=f"city_{i%3}")

print("=== 索引组织表 ===")
print(f"行数: {iot.row_count}")

# 主键查找
print("\n--- 主键查找 ---")
row = iot.find_by_pk(5)
print(f"PK=5: {row.fields if row else 'Not found'}")

# 范围扫描
print("\n--- 范围扫描 PK [3,7] ---")
for r in iot.range_scan(3, 7):
    print(f"  PK={r.pk}: {r.fields}")

# 二级索引
iot.create_secondary_index("city")
print("\n--- 二级索引查找 ---")
row = iot.find_by_secondary("city", "city_1")
if row:
    print(f"  city=city_1 → PK={row.pk}: {row.fields}")

print("\n✅ 索引组织表模拟完成")

🔑 关键概念总结