B樹的定義
一顆M階B樹T,滿足以下條件
- 每個結點至多擁有M課子樹
- 根結點至少擁有兩顆子樹
- 除了根結點以外,其余每個分支結點至少擁有M/2課子樹
- 所有的葉結點都在同一層上
- 有k棵子樹的分支結點則存在k-1個關鍵字,關鍵字按照遞增順序進行排序
- 關鍵字數量滿足 ceil( M/2 ) - 1 <= n <= M-1
B樹與B+樹的區別
在實際磁盤存儲中往往選用的都是b+樹
b+樹相較于b樹的優點
- 關鍵字不保存數據,只用來索引,所有數據都保存在葉子結點(b樹是每個關鍵字都保存數據)
- b+樹的葉子結點是帶有指針的,且葉結點本身按關鍵字從小到大順序連接(適用于范圍查詢)
- b+樹的中間結點不保存數據,所以磁盤頁能容納更多結點元素,更“矮胖”
C++ B+樹算法
構建B+樹的基本結構
B+樹是一種多路平衡搜索樹,常用于數據庫和文件系統索引。以下是一個簡單的B+樹節點結構定義:
template <typename Key, typename Value>
class BPlusNode {
public:bool is_leaf;std::vector<Key> keys;std::vector<Value> values; // Only for leaf nodesstd::vector<BPlusNode*> children; // Only for non-leaf nodesBPlusNode* next; // Pointer to next leaf node (for range queries)
};
插入操作的實現
插入操作需要處理節點分裂和鍵的重新分配。以下是插入邏輯的核心代碼片段:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::insert(const Key& key, const Value& value) {if (root == nullptr) {root = new BPlusNode<Key, Value>(true);root->keys.push_back(key);root->values.push_back(value);return;}BPlusNode<Key, Value>* leaf = find_leaf(root, key);leaf->keys.push_back(key);leaf->values.push_back(value);std::sort(leaf->keys.begin(), leaf->keys.end());if (leaf->keys.size() > order) {split_leaf(leaf);}
}
刪除操作的實現
刪除操作需要處理節點合并和鍵的重新分配。以下是刪除邏輯的核心代碼片段:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::remove(const Key& key) {BPlusNode<Key, Value>* leaf = find_leaf(root, key);auto it = std::find(leaf->keys.begin(), leaf->keys.end(), key);if (it == leaf->keys.end()) return;size_t index = it - leaf->keys.begin();leaf->keys.erase(leaf->keys.begin() + index);leaf->values.erase(leaf->values.begin() + index);if (leaf != root && leaf->keys.size() < (order + 1) / 2) {handle_underflow(leaf);}
}
范圍查詢的實現
B+樹支持高效的范圍查詢,以下是實現代碼片段:
template <typename Key, typename Value>
std::vector<Value> BPlusTree<Key, Value>::range_query(const Key& start, const Key& end) {std::vector<Value> result;BPlusNode<Key, Value>* leaf = find_leaf(root, start);while (leaf != nullptr) {for (size_t i = 0; i < leaf->keys.size(); ++i) {if (leaf->keys[i] >= start && leaf->keys[i] <= end) {result.push_back(leaf->values[i]);}if (leaf->keys[i] > end) return result;}leaf = leaf->next;}return result;
}
完整B+樹類的定義
以下是一個完整的B+樹類定義,包含構造函數和析構函數:
template <typename Key, typename Value>
class BPlusTree {
private:int order;BPlusNode<Key, Value>* root;
public:BPlusTree(int order) : order(order), root(nullptr) {}~BPlusTree() { clear(root); }void insert(const Key& key, const Value& value);void remove(const Key& key);Value search(const Key& key);std::vector<Value> range_query(const Key& start, const Key& end);
private:BPlusNode<Key, Value>* find_leaf(BPlusNode<Key, Value>* node, const Key& key);void split_leaf(BPlusNode<Key, Value>* leaf);void handle_underflow(BPlusNode<Key, Value>* node);void clear(BPlusNode<Key, Value>* node);
};
測試B+樹的插入和查詢
以下是一個簡單的測試用例,驗證B+樹的插入和查詢功能:
void test_b_plus_tree() {BPlusTree<int, std::string> tree(3);tree.insert(1, "Alice");tree.insert(2, "Bob");tree.insert(3, "Charlie");assert(tree.search(2) == "Bob");auto results = tree.range_query(1, 3);assert(results.size() == 3);
}
處理節點分裂的邏輯
當葉子節點的鍵數量超過階數時,需要進行分裂:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::split_leaf(BPlusNode<Key, Value>* leaf) {BPlusNode<Key, Value>* new_leaf = new BPlusNode<Key, Value>(true);size_t split_pos = leaf->keys.size() / 2;new_leaf->keys.assign(leaf->keys.begin() + split_pos, leaf->keys.end());new_leaf->values.assign(leaf->values.begin() + split_pos, leaf->values.end());leaf->keys.erase(leaf->keys.begin() + split_pos, leaf->keys.end());leaf->values.erase(leaf->values.begin() + split_pos, leaf->values.end());new_leaf->next = leaf->next;leaf->next = new_leaf;insert_into_parent(leaf, new_leaf->keys[0], new_leaf);
}
處理節點下溢的邏輯
當節點的鍵數量低于最小值時,需要進行合并或借用:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::handle_underflow(BPlusNode<Key, Value>* node) {if (node == root) {if (node->keys.empty() && !node->children.empty()) {root = node->children[0];delete node;}return;}// Borrow or merge with siblingsBPlusNode<Key, Value>* parent = find_parent(root, node);// Implementation depends on sibling availability and size
}
查找父節點的輔助函數
以下是一個輔助函數,用于查找給定節點的父節點:
template <typename Key, typename Value>
BPlusNode<Key, Value>* BPlusTree<Key, Value>::find_parent(BPlusNode<Key, Value>* current, BPlusNode<Key, Value>* child) {if (current == nullptr || current->is_leaf) return nullptr;for (size_t i = 0; i < current->children.size(); ++i) {if (current->children[i] == child) {return current;}auto parent = find_parent(current->children[i], child);if (parent != nullptr) return parent;}return nullptr;
}
插入到父節點的邏輯
分裂后需要將新節點的鍵插入到父節點中:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::insert_into_parent(BPlusNode<Key, Value>* left, const Key& key, BPlusNode<Key, Value>* right) {BPlusNode<Key, Value>* parent = find_parent(root, left);if (parent == nullptr) {parent = new BPlusNode<Key, Value>(false);parent->keys.push_back(key);parent->children.push_back(left);parent->children.push_back(right);root = parent;return;}auto it = std::lower_bound(parent->keys.begin(), parent->keys.end(), key);size_t index = it - parent->keys.begin();parent->keys.insert(it, key);parent->children.insert(parent->children.begin() + index + 1, right);if (parent->keys.size() > order) {split_internal(parent);}
}
內部節點分裂的邏輯
內部節點的分裂與葉子節點類似,但需要處理子節點指針:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::split_internal(BPlusNode<Key, Value>* node) {BPlusNode<Key, Value>* new_node = new BPlusNode<Key, Value>(false);size_t split_pos = node->keys.size() / 2;Key middle_key = node->keys[split_pos];new_node->keys.assign(node->keys.begin() + split_pos + 1, node->keys.end());new_node->children.assign(node->children.begin() + split_pos + 1, node->children.end());node->keys.erase(node->keys.begin() + split_pos, node->keys.end());node->children.erase(node->children.begin() + split_pos + 1, node->children.end());insert_into_parent(node, middle_key, new_node);
}
清除B+樹的邏輯
析構時需要遞歸釋放所有節點的內存:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::clear(BPlusNode<Key, Value>* node) {if (node == nullptr) return;if (!node->is_leaf) {for (auto child : node->children) {clear(child);}}delete node;
}
搜索操作的實現
根據鍵查找對應的值:
template <typename Key, typename Value>
Value BPlusTree<Key, Value>::search(const Key& key) {BPlusNode<Key, Value>* leaf = find_leaf(root, key);auto it = std::find(leaf->keys.begin(), leaf->keys.end(), key);if (it == leaf->keys.end()) throw std::runtime_error("Key not found");return leaf->values[it - leaf->keys.begin()];
}
查找葉子節點的輔助函數
以下是一個輔助函數,用于查找包含給定鍵的葉子節點:
template <typename Key, typename Value>
BPlusNode<Key, Value>* BPlusTree<Key, Value>::find_leaf(BPlusNode<Key, Value>* node, const Key& key) {if (node == nullptr) return nullptr;if (node->is_leaf) return node;auto it = std::upper_bound(node->keys.begin(), node->keys.end(), key);size_t index = it - node->keys.begin();return find_leaf(node->children[index], key);
}
測試B+樹的刪除功能
以下是一個測試用例,驗證B+樹的刪除功能:
void test_b_plus_tree_deletion() {BPlusTree<int, std::string> tree(3);tree.insert(1, "Alice");tree.insert(2, "Bob");tree.insert(3, "Charlie");tree.remove(2);try {tree.search(2);assert(false); // Should throw} catch (const std::runtime_error& e) {assert(std::string(e.what()) == "Key not found");}
}
處理葉子節點合并的邏輯
當葉子節點的鍵數量不足時,需要與兄弟節點合并:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::merge_leaves(BPlusNode<Key, Value>* left, BPlusNode<Key, Value>* right) {left->keys.insert(left->keys.end(), right->keys.begin(), right->keys.end());left->values.insert(left->values.end(), right->values.begin(), right->values.end());left->next = right->next;remove_from_parent(right);delete right;
}
從父節點中刪除鍵的邏輯
合并后需要從父節點中刪除對應的鍵:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::remove_from_parent(BPlusNode<Key, Value>* child) {BPlusNode<Key, Value>* parent = find_parent(root, child);if (parent == nullptr) return;auto it = std::find(parent->children.begin(), parent->children.end(), child);if (it == parent->children.end()) return;size_t index = it - parent->children.begin();if (index > 0) {parent->keys.erase(parent->keys.begin() + index - 1);}parent->children.erase(it);if (parent != root && parent->keys.size() < (order + 1) / 2 - 1) {handle_underflow(parent);}
}
測試B+樹的合并功能
以下是一個測試用例,驗證B+樹的合并功能:
void test_b_plus_tree_merge() {BPlusTree<int, std::string> tree(3);for (int i = 1; i <= 4; ++i) {tree.insert(i, "Value" + std::to_string(i));}tree.remove(1);tree.remove(2);assert(tree.search(3) == "Value3");assert(tree.search(4) == "Value4");
}
B+樹的持久化存儲
將B+樹保存到文件中以便后續加載:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::serialize(const std::string& filename) {std::ofstream out(filename, std::ios::binary);serialize_node(out, root);out.close();
}
序列化節點的邏輯
遞歸序列化節點及其子節點:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::serialize_node(std::ofstream& out, BPlusNode<Key, Value>* node) {if (node == nullptr) return;out.write(reinterpret_cast<char*>(&node->is_leaf), sizeof(bool));size_t size = node->keys.size();out.write(reinterpret_cast<char*>(&size), sizeof(size_t));for (const auto& key : node->keys) {out.write(reinterpret_cast<const char*>(&key), sizeof(Key));}if (node->is_leaf) {for (const auto& value : node->values) {size_t val_size = value.size();out.write(reinterpret_cast<char*>(&val_size), sizeof(size_t));out.write(value.c_str(), val_size);}} else {for (auto child : node->children) {serialize_node(out, child);}}
}
從文件加載B+樹
從文件中加載B+樹:
template <typename Key, typename Value>
void BPlusTree<Key, Value>::deserialize(const std::string& filename) {std::ifstream in(filename, std::ios::binary);if (!in) return;clear(root);root = deserialize_node(in);in.close();
}
反序列化節點的邏輯
遞歸加載節點及其子節點:
template <typename Key, typename Value>
BPlusNode<Key, Value>* BPlusTree<Key, Value>::deserialize_node(std::ifstream& in) {if (in.eof()) return nullptr;bool is_leaf;in.read(reinterpret_cast<char*>(&is_leaf), sizeof(bool));BPlusNode<Key, Value>* node = new BPlusNode<Key, Value>(is_leaf);size_t size;in.read(reinterpret_cast<char*>(&size), sizeof(size_t));node->keys.resize(size);for (size_t i = 0; i < size; ++i) {in.read(reinterpret_cast<char*>(&node->keys[i]), sizeof(Key));}if (is_leaf) {node->values.resize(size);for (size_t i = 0; i < size; ++i) {size_t val_size;in.read(reinterpret_cast<char*>(&val_size), sizeof(size_t));char* buffer = new char[val_size + 1];in.read(buffer, val_size);buffer[val_size] = '\0';node->values[i] = std::string(buffer);delete[] buffe