19
一起做个简单的数据库（十）：叶子节点的拆分
source link: http://dockone.io/article/9935
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
用C语言从零开始实现SQLite clone系列：
- REPL的介绍和设置
- 世上最简单的SQL编译器和虚拟机
- 一个在内存中仅能做追加操作的单表数据库
- 第一次测试 （含bug处理）
- 持久化存储
- The Cursor Abstraction
- B树介绍
- B树叶子节点的格式
- 二进制查询和重复键
如果只有一个节点的话，我们的B树看起来不像是颗树。为了解决这个问题，我会用一些代码在把叶子节点拆成一对节点。拆分以后我们还要创建一个内部节点作为两个叶子节点的父节点。
基本上我们的目标就把下图：
变成这样：
首先，我们删除完整叶节点的错误处理：
void leaf_node_insert(Cursor* cursor, uint32_t key, Row* value) {
void* node = get_page(cursor->table->pager, cursor->page_num);
uint32_t num_cells = *leaf_node_num_cells(node);
if (num_cells >= LEAF_NODE_MAX_CELLS) {
// Node full
- printf("Need to implement splitting a leaf node.\n");
- exit(EXIT_FAILURE);
+ leaf_node_split_and_insert(cursor, key, value);
+ return;
}
ExecuteResult execute_insert(Statement* statement, Table* table) {
void* node = get_page(table->pager, table->root_page_num);
uint32_t num_cells = (*leaf_node_num_cells(node));
- if (num_cells >= LEAF_NODE_MAX_CELLS) {
- return EXECUTE_TABLE_FULL;
- }
Row* row_to_insert = &(statement->row_to_insert);
uint32_t key_to_insert = row_to_insert->id;
拆分算法
简单的部分结束了。 接下来是对我们需要做的事情的描述：SQLite 数据库系统：设计和实施。
如果叶子节点上没有空间，我们会将驻留在该节点上的现有条目和新条目（已插入）拆分成两个相等的部分：下半部分和上半部分。 （上半部分的键值必须大于下半部分的键值。）我们分配一个新的叶子节点，然后将上半部分移到新节点。
让我们获取旧节点的句柄并创建新节点：
+void leaf_node_split_and_insert(Cursor* cursor, uint32_t key, Row* value) {
+ /*
+ Create a new node and move half the cells over.
+ Insert the new value in one of the two nodes.
+ Update parent or create a new parent.
+ */
+
+ void* old_node = get_page(cursor->table->pager, cursor->page_num);
+ uint32_t new_page_num = get_unused_page_num(cursor->table->pager);
+ void* new_node = get_page(cursor->table->pager, new_page_num);
+ initialize_leaf_node(new_node);
接下来把每个单元格拷贝到新位置：
+ /*
+ All existing keys plus new key should be divided
+ evenly between old (left) and new (right) nodes.
+ Starting from the right, move each key to correct position.
+ */
+ for (int32_t i = LEAF_NODE_MAX_CELLS; i >= 0; i--) {
+ void* destination_node;
+ if (i >= LEAF_NODE_LEFT_SPLIT_COUNT) {
+ destination_node = new_node;
+ } else {
+ destination_node = old_node;
+ }
+ uint32_t index_within_node = i % LEAF_NODE_LEFT_SPLIT_COUNT;
+ void* destination = leaf_node_cell(destination_node, index_within_node);
+
+ if (i == cursor->cell_num) {
+ serialize_row(value, destination);
+ } else if (i > cursor->cell_num) {
+ memcpy(destination, leaf_node_cell(old_node, i - 1), LEAF_NODE_CELL_SIZE);
+ } else {
+ memcpy(destination, leaf_node_cell(old_node, i), LEAF_NODE_CELL_SIZE);
+ }
+ }
在每个节点头部显示单元格的数量：
+ /* Update cell count on both leaf nodes */ + *(leaf_node_num_cells(old_node)) = LEAF_NODE_LEFT_SPLIT_COUNT; + *(leaf_node_num_cells(new_node)) = LEAF_NODE_RIGHT_SPLIT_COUNT;
接下来我们更新节点的父节点。 如果原始节点是根，则它没有父节点。 在这种情况下，需要创建一个新的根节点来充当父节点。 我现在暂存另一个分支：
+ if (is_node_root(old_node)) {
+ return create_new_root(cursor->table, new_page_num);
+ } else {
+ printf("Need to implement updating parent after split\n");
+ exit(EXIT_FAILURE);
+ }
+}
分配新页面：
让我们重新定义一些新函数和常量。当我们新创建叶子节点时，我们用get_unused_page_num()来做判断：
+/*
+Until we start recycling free pages, new pages will always
+go onto the end of the database file
+*/
+uint32_t get_unused_page_num(Pager* pager) { return pager->num_pages; }
现在，我们假设在具有N页的数据库中分配了从0到N-1的页码。 因此，我们始终可以为新页面分配页码N。在我们删除操作后，某些页面可能会变空并且其页码会闲置。 为了提高效率，我们可以重新分配那些空闲页面。
叶子节点的大小：
为了让树平衡，我们在两个节点间平均分配单元。如果一个叶子节点可以承载N个单元，那么在整个过程中我们需要分配N+1个单元（N个初始单元加一个新单元）如果N+1为奇数，我们把它存在任意左侧节点。
+const uint32_t LEAF_NODE_RIGHT_SPLIT_COUNT = (LEAF_NODE_MAX_CELLS + 1) / 2; +const uint32_t LEAF_NODE_LEFT_SPLIT_COUNT = + (LEAF_NODE_MAX_CELLS + 1) - LEAF_NODE_RIGHT_SPLIT_COUNT;
创建新根：
下面关于如何创建根节点流程的解释来自《SQLite 数据库系统》：
让N作为根节点，首先分配2个节点，比如L和R。把N的键值较低的部分放入L并把键值高的部分放到R。现在N是空的。将L,K,R加入N，K是L中最大的值。页面N仍然是根，树的深度增加1，树仍旧遵从B+树的规则，保持平衡。
至此我们把简直较高的一半移到右侧的节点。我们的函数会将正确的键值放入左侧的节点并分配新的页面。
+void create_new_root(Table* table, uint32_t right_child_page_num) {
+ /*
+ Handle splitting the root.
+ Old root copied to new page, becomes left child.
+ Address of right child passed in.
+ Re-initialize root page to contain the new root node.
+ New root node points to two children.
+ */
+
+ void* root = get_page(table->pager, table->root_page_num);
+ void* right_child = get_page(table->pager, right_child_page_num);
+ uint32_t left_child_page_num = get_unused_page_num(table->pager);
+ void* left_child = get_page(table->pager, left_child_page_num);
旧的根目录被复制到左子节点，因此我们可以重复使用根目录页：
+ /* Left child has data copied from old root */ + memcpy(left_child, root, PAGE_SIZE); + set_node_root(left_child, false);
最终我们把根节点的页面初始化为带两个子节点的内部节点
+ /* Root node is a new internal node with one key and two children */ + initialize_internal_node(root); + set_node_root(root, true); + *internal_node_num_keys(root) = 1; + *internal_node_child(root, 0) = left_child_page_num; + uint32_t left_child_max_key = get_node_max_key(left_child); + *internal_node_key(root, 0) = left_child_max_key; + *internal_node_right_child(root) = right_child_page_num; +}
内部节点的格式
现在我们终于要创建了内部节点，我们要把它的结构样式定义好。它以通用的头信息开始，然后包含键的数量，紧跟着右侧子页面的页码。内部节点的子指针总是比键多，额外的子指针也存在头部信息里。
+/* + * Internal Node Header Layout + */ +const uint32_t INTERNAL_NODE_NUM_KEYS_SIZE = sizeof(uint32_t); +const uint32_t INTERNAL_NODE_NUM_KEYS_OFFSET = COMMON_NODE_HEADER_SIZE; +const uint32_t INTERNAL_NODE_RIGHT_CHILD_SIZE = sizeof(uint32_t); +const uint32_t INTERNAL_NODE_RIGHT_CHILD_OFFSET = + INTERNAL_NODE_NUM_KEYS_OFFSET + INTERNAL_NODE_NUM_KEYS_SIZE; +const uint32_t INTERNAL_NODE_HEADER_SIZE = COMMON_NODE_HEADER_SIZE + + INTERNAL_NODE_NUM_KEYS_SIZE + + INTERNAL_NODE_RIGHT_CHILD_SIZE
主体是一个单元格数组，其中每个单元格都包含一个子指针和一个键。 每个键应该是子级左侧包含的最大键。
+/* + * Internal Node Body Layout + */ +const uint32_t INTERNAL_NODE_KEY_SIZE = sizeof(uint32_t); +const uint32_t INTERNAL_NODE_CHILD_SIZE = sizeof(uint32_t); +const uint32_t INTERNAL_NODE_CELL_SIZE = + INTERNAL_NODE_CHILD_SIZE + INTERNAL_NODE_KEY_SIZE;
基于这些常量，下面是内部节点的布局：
注意我们巨大的分支。 由于每个子指针/键对都很小，因此我们可以在每个内部节点中容纳510个键和511个子指针。 这意味着我们无需遍历树的许多层就能找到给定的键！
实际上，由于标头，键以及被浪费空间的开销，我们无法为每个叶节点存储完整的4 KB数据。 但是我们可以从磁盘上仅加载4页来搜索500 GB的数据。 这就是B树是数据库有用的数据结构的原因。
以下是读取和写入内部节点的方法：
+uint32_t* internal_node_num_keys(void* node) {
+ return node + INTERNAL_NODE_NUM_KEYS_OFFSET;
+}
+
+uint32_t* internal_node_right_child(void* node) {
+ return node + INTERNAL_NODE_RIGHT_CHILD_OFFSET;
+}
+
+uint32_t* internal_node_cell(void* node, uint32_t cell_num) {
+ return node + INTERNAL_NODE_HEADER_SIZE + cell_num * INTERNAL_NODE_CELL_SIZE;
+}
+
+uint32_t* internal_node_child(void* node, uint32_t child_num) {
+ uint32_t num_keys = *internal_node_num_keys(node);
+ if (child_num > num_keys) {
+ printf("Tried to access child_num %d > num_keys %d\n", child_num, num_keys);
+ exit(EXIT_FAILURE);
+ } else if (child_num == num_keys) {
+ return internal_node_right_child(node);
+ } else {
+ return internal_node_cell(node, child_num);
+ }
+}
+
+uint32_t* internal_node_key(void* node, uint32_t key_num) {
+ return internal_node_cell(node, key_num) + INTERNAL_NODE_CHILD_SIZE;
+}
对于内部节点，最大键始终是其右键。 对于叶节点，它是最大索引处键：
+uint32_t get_node_max_key(void* node) {
+ switch (get_node_type(node)) {
+ case NODE_INTERNAL:
+ return *internal_node_key(node, *internal_node_num_keys(node) - 1);
+ case NODE_LEAF:
+ return *leaf_node_key(node, *leaf_node_num_cells(node) - 1);
+ }
+}
保持对根的追踪：
我们最终在通用头部内使用了is_root字段。回想一下物品们用它来决定如何拆分叶子节点：
if (is_node_root(old_node)) {
return create_new_root(cursor->table, new_page_num);
} else {
printf("Need to implement updating parent after split\n");
exit(EXIT_FAILURE);
}
}
+bool is_node_root(void* node) {
+ uint8_t value = *((uint8_t*)(node + IS_ROOT_OFFSET));
+ return (bool)value;
+}
+
+void set_node_root(void* node, bool is_root) {
+ uint8_t value = is_root;
+ *((uint8_t*)(node + IS_ROOT_OFFSET)) = value;
+}
初始化这两种类型的节点都应默认将is_root设置为false：
// New database file. Initialize page 0 as leaf node. void* root_node = get_page(pager, 0); initialize_leaf_node(root_node); + set_node_root(root_node, true); } return table;
打印树
为了帮助我们可视化数据库的状态，我们应该更新.btree元命令以打印多级树。
我将替换当前的print_leaf_node（）函数
-void print_leaf_node(void* node) {
- uint32_t num_cells = *leaf_node_num_cells(node);
- printf("leaf (size %d)\n", num_cells);
- for (uint32_t i = 0; i < num_cells; i++) {
- uint32_t key = *leaf_node_key(node, i);
- printf(" - %d : %d\n", i, key);
- }
-}
带有一个新的递归函数，该函数接受任何节点，然后打印该节点及其子节点。 它以缩进级别作为参数，每次递归调用时都会增加。 我还要添加一个小的辅助函数来缩进。
+void indent(uint32_t level) {
+ for (uint32_t i = 0; i < level; i++) {
+ printf(" ");
+ }
+}
+
+void print_tree(Pager* pager, uint32_t page_num, uint32_t indentation_level) {
+ void* node = get_page(pager, page_num);
+ uint32_t num_keys, child;
+
+ switch (get_node_type(node)) {
+ case (NODE_LEAF):
+ num_keys = *leaf_node_num_cells(node);
+ indent(indentation_level);
+ printf("- leaf (size %d)\n", num_keys);
+ for (uint32_t i = 0; i < num_keys; i++) {
+ indent(indentation_level + 1);
+ printf("- %d\n", *leaf_node_key(node, i));
+ }
+ break;
+ case (NODE_INTERNAL):
+ num_keys = *internal_node_num_keys(node);
+ indent(indentation_level);
+ printf("- internal (size %d)\n", num_keys);
+ for (uint32_t i = 0; i < num_keys; i++) {
+ child = *internal_node_child(node, i);
+ print_tree(pager, child, indentation_level + 1);
+
+ indent(indentation_level + 1);
+ printf("- key %d\n", *internal_node_key(node, i));
+ }
+ child = *internal_node_right_child(node);
+ print_tree(pager, child, indentation_level + 1);
+ break;
+ }
+}
同时更新对打印功能的调用，缩进级别为零。
} else if (strcmp(input_buffer->buffer, ".btree") == 0) {
printf("Tree:\n");
- print_leaf_node(get_page(table->pager, 0));
+ print_tree(table->pager, 0, 0);
return META_COMMAND_SUCCESS;
下面是对打印函数的测试：
+ it 'allows printing out the structure of a 3-leaf-node btree' do
+ script = (1..14).map do |i|
+ "insert #{i} user#{i} person#{i}@example.com"
+ end
+ script << ".btree"
+ script << "insert 15 user15 [email protected]"
+ script << ".exit"
+ result = run_script(script)
+
+ expect(result[14...(result.length)]).to match_array([
+ "db > Tree:",
+ "- internal (size 1)",
+ " - leaf (size 7)",
+ " - 1",
+ " - 2",
+ " - 3",
+ " - 4",
+ " - 5",
+ " - 6",
+ " - 7",
+ " - key 7",
+ " - leaf (size 7)",
+ " - 8",
+ " - 9",
+ " - 10",
+ " - 11",
+ " - 12",
+ " - 13",
+ " - 14",
+ "db > Need to implement searching an internal node",
+ ])
+ end
新格式有所简化，因此我们需要更新现有的.btree测试：
"db > Executed.", "db > Executed.", "db > Tree:", - "leaf (size 3)", - " - 0 : 1", - " - 1 : 2", - " - 2 : 3", + "- leaf (size 3)", + " - 1", + " - 2", + " - 3", "db > " ]) end
下面是测试结果：
Tree: - internal (size 1) - leaf (size 7) - 1 - 2 - 3 - 4 - 5 - 6 - 7 - key 7 - leaf (size 7) - 8 - 9 - 10 - 11 - 12 - 13 - 14
在最小缩进级别上，我们看到根节点（内部节点）。 之所以说是1号是因为它只有一把钥匙。 缩进一个级别，我们看到一个叶节点，一个键和另一个叶节点。 根节点（7）中的密钥是第一个叶节点中的最大密钥。 每个大于7的键都在第二个叶子节点中。
主要问题：
如果你一直跟着我做，你可能发现我们忽略了一些重要问题。比如，让我们看看当我们插入另一行会发生什么？
db > insert 15 user15 [email protected] Need to implement searching an internal node
哎呀！ 谁写了那个TODO消息？ ：P
下次，我们将通过在多级树上执行搜索来继续史诗般的B树之旅。
原文链接： Part 10 - Binary Search and Duplicate Keys （翻译：吴世曦）
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK