本系列文章一共13篇，本文为第10篇，请关注公众号，后续文章会陆续发布。
系列文章列表：
《手把手教你从零开始实现一个数据库系统》
《世上最简单的SQL编译器和虚拟机》
《一个在内存中仅能做追加操作的单表数据库》
《第一次测试 （含bug处理）》
《持久化存储》
《The Cursor Abstraction》
《B树介绍》
《B树叶子节点的格式》
《二进制查询和重复键》
如果只有一个节点的话，我们的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 person15@example.com" + 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 person15@example.com Need to implement searching an internal node哎呀！谁写了那个TODO消息？：P
下次，我们将通过在多级树上执行搜索来继续史诗般的B树之旅。
原文链接：https://cstack.github.io/db_tutorial/parts/part10.html
Kubernetes实战培训
Kubernetes实战培训将于2020年7月24日在深圳开课，3天时间带你系统掌握Kubernetes，学习效果不好可以继续学习。本次培训包括：云原生介绍、微服务；Docker基础、Docker工作原理、镜像、网络、存储、数据卷、安全；Kubernetes架构、核心组件、常用对象、网络、存储、认证、服务发现、调度和服务质量保证、日志、监控、告警、Helm、实践案例等，点击下方图片或者阅读原文链接查看详情。
