第07课：B+树实现

索引结构 第7课 / 共25课

📖 课程概述

B+树是B树最重要的变种，也是数据库索引的事实标准。与B树不同，B+树只在叶子节点存储数据，内部节点只存键，且叶子节点通过链表连接。这些特性使得B+树在范围查询和顺序扫描方面远优于B树。

本课目标：实现完整的B+树，理解叶子链表、顺序扫描和范围查询的优化，对比B树与B+树的性能差异。

🌳 B+树 vs B树

B树: 数据分布在所有节点 [30|数据] / \ [10|数据|20|数据] [40|数据|50|数据] B+树: 数据只在叶子节点，内部节点只做路由 [30] ← 只有键，无数据 / \ [10|20] [40|50] ← 只有键 ↓ ↓ ↓ ↓ [10,D][20,D][40,D][50,D] ← 叶子有数据+链表 ↔ ↔ ↔ ↔ ← 叶子链表连接 B+树优势: 1. 内部节点无数据 → 每页存更多键 → 树更矮 2. 叶子链表 → 范围查询只需遍历链表 3. 查找路径长度一致 → 性能稳定 4. 顺序扫描极快 → 适合OLAP

💻 C语言实现：完整B+树

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

#define ORDER 4
#define MAX_KEYS (ORDER)
#define MIN_KEYS ((ORDER+1)/2)

typedef struct BPlusLeaf {
    int keys[ORDER + 1];           // 多一个用于溢出
    char* values[ORDER + 1];       // 数据指针
    int num_keys;
    struct BPlusLeaf* next;        // 叶子链表
    struct BPlusLeaf* prev;
    int is_leaf;
} BPlusLeaf;

typedef struct BPlusInternal {
    int keys[ORDER + 1];
    void* children[ORDER + 2];     // 子节点指针
    int num_keys;
    int is_leaf;
} BPlusInternal;

typedef union {
    BPlusLeaf leaf;
    BPlusInternal internal;
} BPlusNodeData;

typedef struct {
    BPlusNodeData data;
    int is_leaf;
} BPlusNode;

// 简化：用统一结构
typedef struct BPNode {
    int keys[ORDER + 2];
    char* values[ORDER + 2];       // 叶子: 数据; 内部: NULL
    struct BPNode* children[ORDER + 3];
    struct BPNode* next;           // 叶子链表
    struct BPNode* parent;
    int num_keys;
    int is_leaf;
} BPNode;

BPNode* bp_create(int is_leaf) {
    BPNode* n = calloc(1, sizeof(BPNode));
    n->is_leaf = is_leaf;
    return n;
}

// 查找
char* bp_search(BPNode* root, int key) {
    if (!root) return NULL;
    BPNode* node = root;
    while (!node->is_leaf) {
        int i = 0;
        while (i < node->num_keys && key >= node->keys[i]) i++;
        node = node->children[i];
    }
    // 叶子节点内线性查找
    for (int i = 0; i < node->num_keys; i++) {
        if (node->keys[i] == key) {
            printf("  [B+Tree] 找到 key=%d val=%s\n", key, node->values[i]);
            return node->values[i];
        }
    }
    printf("  [B+Tree] key=%d 不存在\n", key);
    return NULL;
}

// 范围查询
void bp_range_scan(BPNode* root, int lo, int hi) {
    if (!root) return;
    // 找到lo所在的叶子
    BPNode* node = root;
    while (!node->is_leaf) {
        int i = 0;
        while (i < node->num_keys && lo >= node->keys[i]) i++;
        node = node->children[i];
    }
    printf("  [B+Tree] 范围查询 [%d, %d]:\n", lo, hi);
    int count = 0;
    while (node) {
        for (int i = 0; i < node->num_keys; i++) {
            if (node->keys[i] > hi) { printf("  共 %d 条\n", count); return; }
            if (node->keys[i] >= lo) {
                printf("    key=%d → %s\n", node->keys[i], node->values[i]);
                count++;
            }
        }
        node = node->next;
    }
    printf("  共 %d 条\n", count);
}

// 顺序扫描
void bp_full_scan(BPNode* root) {
    if (!root) return;
    BPNode* node = root;
    while (!node->is_leaf) node = node->children[0];
    printf("  [B+Tree] 顺序扫描:\n");
    int count = 0;
    while (node) {
        for (int i = 0; i < node->num_keys; i++) {
            printf("    %d:%s", node->keys[i], node->values[i]);
            if (++count % 5 == 0) printf("\n");
        }
        node = node->next;
    }
    printf("\n  共 %d 条\n", count);
}

// 插入 - 找叶子
BPNode* find_leaf(BPNode* root, int key) {
    BPNode* node = root;
    while (!node->is_leaf) {
        int i = 0;
        while (i < node->num_keys && key >= node->keys[i]) i++;
        node = node->children[i];
    }
    return node;
}

// 在叶子中插入键值对
void leaf_insert(BPNode* leaf, int key, char* val) {
    int i = leaf->num_keys - 1;
    while (i >= 0 && leaf->keys[i] > key) {
        leaf->keys[i + 1] = leaf->keys[i];
        leaf->values[i + 1] = leaf->values[i];
        i--;
    }
    leaf->keys[i + 1] = key;
    leaf->values[i + 1] = val;
    leaf->num_keys++;
}

// 分裂叶子
BPNode* split_leaf(BPNode* leaf) {
    int mid = leaf->num_keys / 2;
    BPNode* new_leaf = bp_create(1);
    new_leaf->num_keys = leaf->num_keys - mid;
    for (int i = 0; i < new_leaf->num_keys; i++) {
        new_leaf->keys[i] = leaf->keys[mid + i];
        new_leaf->values[i] = leaf->values[mid + i];
    }
    leaf->num_keys = mid;
    // 更新链表
    new_leaf->next = leaf->next;
    new_leaf->prev = leaf;
    if (leaf->next) leaf->next->prev = new_leaf;
    leaf->next = new_leaf;
    printf("  [B+Tree] 叶子分裂，新叶首键=%d\n", new_leaf->keys[0]);
    return new_leaf;
}

// 向父节点插入分裂键
void insert_to_parent(BPNode* root, BPNode* left, BPNode* right, int key);

BPNode* bp_insert(BPNode* root, int key, char* val) {
    if (!root) {
        root = bp_create(1);
        root->keys[0] = key;
        root->values[0] = val;
        root->num_keys = 1;
        return root;
    }
    BPNode* leaf = find_leaf(root, key);
    leaf_insert(leaf, key, val);
    if (leaf->num_keys <= ORDER) return root;

    // 分裂叶子
    BPNode* new_leaf = split_leaf(leaf);
    int split_key = new_leaf->keys[0];

    if (!leaf->parent) {
        // 需要新根
        BPNode* new_root = bp_create(0);
        new_root->keys[0] = split_key;
        new_root->children[0] = leaf;
        new_root->children[1] = new_leaf;
        new_root->num_keys = 1;
        leaf->parent = new_root;
        new_leaf->parent = new_root;
        printf("  [B+Tree] 新根创建，split_key=%d\n", split_key);
        return new_root;
    }

    // 向父节点插入
    insert_to_parent(root, leaf, new_leaf, split_key);
    return root;
}

void insert_to_parent(BPNode* root, BPNode* left, BPNode* right, int key) {
    BPNode* parent = left->parent;
    int idx = 0;
    while (idx < parent->num_keys && parent->keys[idx] <= key) idx++;

    // 插入键和子节点
    for (int i = parent->num_keys; i > idx; i--)
        parent->keys[i] = parent->keys[i - 1];
    for (int i = parent->num_keys + 1; i > idx + 1; i--)
        parent->children[i] = parent->children[i - 1];
    parent->keys[idx] = key;
    parent->children[idx + 1] = right;
    parent->num_keys++;
    right->parent = parent;

    if (parent->num_keys <= ORDER) return;

    // 内部节点分裂
    int mid = parent->num_keys / 2;
    int up_key = parent->keys[mid];
    BPNode* new_internal = bp_create(0);
    new_internal->num_keys = parent->num_keys - mid - 1;
    for (int i = 0; i < new_internal->num_keys; i++) {
        new_internal->keys[i] = parent->keys[mid + 1 + i];
        new_internal->children[i] = parent->children[mid + 1 + i];
        ((BPNode*)parent->children[mid + 1 + i])->parent = new_internal;
    }
    new_internal->children[new_internal->num_keys] = parent->children[parent->num_keys];
    ((BPNode*)parent->children[parent->num_keys])->parent = new_internal;
    parent->num_keys = mid;

    printf("  [B+Tree] 内部节点分裂，上提键=%d\n", up_key);

    if (!parent->parent) {
        BPNode* new_root = bp_create(0);
        new_root->keys[0] = up_key;
        new_root->children[0] = parent;
        new_root->children[1] = new_internal;
        new_root->num_keys = 1;
        parent->parent = new_root;
        new_internal->parent = new_root;
        // 注意：这里需要更新root，通过返回值
        printf("  [B+Tree] 根分裂\n");
    } else {
        insert_to_parent(root, parent, new_internal, up_key);
    }
}

// 打印
void bp_print(BPNode* node, int depth) {
    if (!node) return;
    printf("%*s[", depth*2, "");
    for (int i = 0; i < node->num_keys; i++) {
        if (node->is_leaf) printf("%d:%s", node->keys[i], node->values[i]);
        else printf("%d", node->keys[i]);
        if (i < node->num_keys - 1) printf("|");
    }
    printf("]%s\n", node->is_leaf ? "(L)" : "");
    if (!node->is_leaf) {
        for (int i = 0; i <= node->num_keys; i++)
            bp_print(node->children[i], depth + 1);
    }
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   B+树实现 (阶数=%d)                 ║\n", ORDER);
    printf("╚══════════════════════════════════════╝\n\n");

    BPNode* root = NULL;
    struct { int k; const char* v; } data[] = {
        {10,"Alice"}, {20,"Bob"}, {5,"Charlie"}, {6,"Diana"},
        {12,"Eve"}, {30,"Frank"}, {7,"Grace"}, {17,"Henry"},
        {3,"Ivy"}, {16,"Jack"}, {22,"Kate"}, {35,"Leo"},
        {40,"Mia"}, {45,"Noah"}, {8,"Olivia"}
    };
    int n = sizeof(data) / sizeof(data[0]);

    printf("--- 插入 ---\n");
    for (int i = 0; i < n; i++) {
        printf("Insert %d:%s\n", data[i].k, data[i].v);
        root = bp_insert(root, data[i].k, (char*)data[i].v);
    }

    printf("\n--- B+树结构 ---\n");
    bp_print(root, 0);

    printf("\n--- 查找 ---\n");
    bp_search(root, 12);
    bp_search(root, 35);
    bp_search(root, 99);

    printf("\n--- 范围查询 [10, 30] ---\n");
    bp_range_scan(root, 10, 30);

    printf("\n--- 顺序扫描 ---\n");
    bp_full_scan(root);

    printf("\n✅ B+树实现运行完成\n");
    return 0;
}

🐍 Python实现：B+树性能对比

"""
B+树完整实现与性能对比
"""
import time, random
from collections import deque

class BPlusNode:
    def __init__(self, order, leaf=True):
        self.order = order
        self.leaf = leaf
        self.keys = []
        self.values = [] if leaf else None  # 叶子存值
        self.children = [] if not leaf else None
        self.next = None  # 叶子链表
        self.prev = None

class BPlusTree:
    def __init__(self, order=4):
        self.order = order
        self.root = BPlusNode(order, leaf=True)
        self.splits = 0
        self.levels = 1

    def search(self, key):
        node = self.root
        while not node.leaf:
            i = 0
            while i < len(node.keys) and key >= node.keys[i]: i += 1
            node = node.children[i]
        for i, k in enumerate(node.keys):
            if k == key: return node.values[i]
        return None

    def range_query(self, lo, hi):
        node = self.root
        while not node.leaf:
            i = 0
            while i < len(node.keys) and lo >= node.keys[i]: i += 1
            node = node.children[i]
        results = []
        while node:
            for i, k in enumerate(node.keys):
                if k > hi: return results
                if lo <= k: results.append((k, node.values[i]))
            node = node.next
        return results

    def insert(self, key, value):
        root = self.root
        if len(root.keys) >= self.order:
            new_root = BPlusNode(self.order, leaf=False)
            new_root.children.append(root)
            self._split_child(new_root, 0)
            self.root = new_root
            self.levels += 1
        self._insert_non_full(self.root, key, value)

    def _split_child(self, parent, idx):
        child = parent.children[idx]
        mid = len(child.keys) // 2
        new_node = BPlusNode(self.order, leaf=child.leaf)

        if child.leaf:
            new_node.keys = child.keys[mid:]
            new_node.values = child.values[mid:]
            child.keys = child.keys[:mid]
            child.values = child.values[:mid]
            up_key = new_node.keys[0]
            # 更新链表
            new_node.next = child.next
            new_node.prev = child
            if child.next: child.next.prev = new_node
            child.next = new_node
        else:
            up_key = child.keys[mid]
            new_node.keys = child.keys[mid+1:]
            new_node.children = child.children[mid+1:]
            child.keys = child.keys[:mid]
            child.children = child.children[:mid+1]

        parent.keys.insert(idx, up_key)
        parent.children.insert(idx + 1, new_node)
        self.splits += 1

    def _insert_non_full(self, node, key, value):
        if node.leaf:
            i = 0
            while i < len(node.keys) and node.keys[i] < key: i += 1
            if i < len(node.keys) and node.keys[i] == key:
                node.values[i] = value  # update
                return
            node.keys.insert(i, key)
            node.values.insert(i, value)
        else:
            i = 0
            while i < len(node.keys) and key >= node.keys[i]: i += 1
            if len(node.children[i].keys) >= self.order:
                self._split_child(node, i)
                if key >= node.keys[i]: i += 1
            self._insert_non_full(node.children[i], key, value)

    def full_scan(self):
        node = self.root
        while not node.leaf: node = node.children[0]
        results = []
        while node:
            results.extend(zip(node.keys, node.values))
            node = node.next
        return results

# ========== B树(对比用) ==========
class BTreeNode:
    def __init__(self, order, leaf=True):
        self.order = order
        self.leaf = leaf
        self.keys = []
        self.values = []
        self.children = []

class BTree:
    def __init__(self, order=4):
        self.order = order
        self.root = BTreeNode(order, leaf=True)
        self.splits = 0

    def search(self, key):
        return self._search(self.root, key)

    def _search(self, node, key):
        i = 0
        while i < len(node.keys) and key > node.keys[i]: i += 1
        if i < len(node.keys) and node.keys[i] == key: return node.values[i]
        if node.leaf: return None
        return self._search(node.children[i], key)

    def range_query(self, lo, hi):
        results = []
        self._range(self.root, lo, hi, results)
        return results

    def _range(self, node, lo, hi, results):
        for i, k in enumerate(node.keys):
            if not node.leaf:
                self._range(node.children[i], lo, hi, results)
            if lo <= k <= hi:
                results.append((k, node.values[i]))
        if not node.leaf:
            self._range(node.children[len(node.keys)], lo, hi, results)

    def insert(self, key, value):
        if len(self.root.keys) >= self.order - 1:
            new_root = BTreeNode(self.order, leaf=False)
            new_root.children.append(self.root)
            self._split(new_root, 0)
            self.root = new_root
            self.splits += 1
        self._insert_nf(self.root, key, value)

    def _split(self, parent, idx):
        child = parent.children[idx]
        mid = len(child.keys) // 2
        new_node = BTreeNode(self.order, leaf=child.leaf)
        mid_key = child.keys[mid]
        new_node.keys = child.keys[mid+1:]
        new_node.values = child.values[mid+1:]
        if not child.leaf:
            new_node.children = child.children[mid+1:]
            child.children = child.children[:mid+1]
        child.keys = child.keys[:mid]
        child.values = child.values[:mid]
        parent.keys.insert(idx, mid_key)
        parent.values.insert(idx, child.values[mid] if mid < len(child.values) else None)
        parent.children.insert(idx+1, new_node)
        self.splits += 1

    def _insert_nf(self, node, key, value):
        i = len(node.keys) - 1
        if node.leaf:
            while i >= 0 and node.keys[i] > key: i -= 1
            node.keys.insert(i+1, key)
            node.values.insert(i+1, value)
        else:
            while i >= 0 and node.keys[i] > key: i -= 1
            i += 1
            if len(node.children[i].keys) >= self.order - 1:
                self._split(node, i)
                if key > node.keys[i]: i += 1
            self._insert_nf(node.children[i], key, value)

# ========== 对比 ==========
N = 5000
random.seed(42)
keys = list(range(N))
random.shuffle(keys)

bpt = BPlusTree(order=50)
bt = BTree(order=50)

# 插入
t0 = time.perf_counter()
for k in keys: bpt.insert(k, f"val_{k}")
bp_insert_t = time.perf_counter() - t0

t0 = time.perf_counter()
for k in keys: bt.insert(k, f"val_{k}")
bt_insert_t = time.perf_counter() - t0

# 点查
t0 = time.perf_counter()
for k in range(N): bpt.search(k % N)
bp_search_t = time.perf_counter() - t0

t0 = time.perf_counter()
for k in range(N): bt.search(k % N)
bt_search_t = time.perf_counter() - t0

# 范围查询
t0 = time.perf_counter()
for _ in range(100): bpt.range_query(random.randint(0,N//2), random.randint(N//2,N))
bp_range_t = time.perf_counter() - t0

t0 = time.perf_counter()
for _ in range(100): bt.range_query(random.randint(0,N//2), random.randint(N//2,N))
bt_range_t = time.perf_counter() - t0

print(f"{'操作':>10} | {'B+树ms':>8} | {'B树ms':>8} | {'B+树优势':>10}")
print("-" * 50)
print(f"{'插入':>10} | {bp_insert_t*1000:>7.1f} | {bt_insert_t*1000:>7.1f} | {'-':>10}")
print(f"{'点查':>10} | {bp_search_t*1000:>7.1f} | {bt_search_t*1000:>7.1f} | {'-':>10}")
print(f"{'范围查询':>10} | {bp_range_t*1000:>7.1f} | {bt_range_t*1000:>7.1f} | {bt_range_t/bp_range_t:.1f}x")
print(f"\nB+树层数: {bpt.levels}, 分裂: {bpt.splits}")
print("✅ B+树 vs B树对比完成")

🔑 关键概念总结

📝 练习

  1. 实现B+树的删除操作，包含借用和合并逻辑
  2. 添加批量加载(bulk load)功能，从有序数据高效构建B+树
  3. 实现B+树的磁盘持久化，每个节点存储为一个4KB页
  4. 对比B+树在不同阶数(10/50/100/200)下的范围查询性能
🌲

🏆 成就解锁：索引工程师

掌握B+树实现，你已理解数据库索引的工业标准数据结构！

✅ B+树插入/查找 · ✅ 叶子链表 · ✅ 范围查询优化