关于数据库:实现一个简单的Database9译文

• GreatSQL 社区原创内容未经受权不得随便应用，转载请分割小编并注明起源。
• GreatSQL 是 MySQL 的国产分支版本，应用上与 MySQL 统一。
• 作者：花家舍
• 文章起源：GreatSQL 社区原创

前文回顾

• 实现一个简略的 Database 系列

译注：cstack 在 github 保护了一个简略的、相似 sqlite 的数据库实现，通过这个简略的我的项目，能够很好的了解数据库是如何运行的。本文是第九篇，次要是实现 B -tree 的二叉搜寻并解决主键反复问题

Part 9 二叉搜寻与主键反复

上次留神到咱们的 B 树存储 key 时依然是非排序的。当初咱们来解决这个问题，并检测和回绝主键的反复（插入）。

当初咱们的 execute_insert() 函数在插入时，抉择的地位是在表的结尾处。作为替换，咱们须要搜寻表（树）中正确的插入地位，而后把 key 插入到这个地位上。如果 key 在这个地位上曾经存在，那么就返回（键反复）报错。

ExecuteResult execute_insert(Statement* statement, Table* table) {void* node = get_page(table->pager, table->root_page_num);
-  if ((*leaf_node_num_cells(node) >= LEAF_NODE_MAX_CELLS)) {+  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);
-  Cursor* cursor = table_end(table);
+  uint32_t key_to_insert = row_to_insert->id;
+  Cursor* cursor = table_find(table, key_to_insert);
+
+  if (cursor->cell_num < num_cells) {+    uint32_t key_at_index = *leaf_node_key(node, cursor->cell_num);
+    if (key_at_index == key_to_insert) {
+      return EXECUTE_DUPLICATE_KEY;
+    }
+  }

   leaf_node_insert(cursor, row_to_insert->id, row_to_insert);

这样的话，咱们就不再须要 table_end() 函数了。

-Cursor* table_end(Table* table) {-  Cursor* cursor = malloc(sizeof(Cursor));
-  cursor->table = table;
-  cursor->page_num = table->root_page_num;
-
-  void* root_node = get_page(table->pager, table->root_page_num);
-  uint32_t num_cells = *leaf_node_num_cells(root_node);
-  cursor->cell_num = num_cells;
-  cursor->end_of_table = true;
-
-  return cursor;
-}

咱们用一个办法替换 table_end() 函数，办法实现应用给定的 key 来搜寻 Btree。

+/*
+Return the position of the given key.
+If the key is not present, return the position
+where it should be inserted
+*/
+Cursor* table_find(Table* table, uint32_t key) {
+  uint32_t root_page_num = table->root_page_num;
+  void* root_node = get_page(table->pager, root_page_num);
+
+  if (get_node_type(root_node) == NODE_LEAF) {+    return leaf_node_find(table, root_page_num, key);
+  } else {+    printf("Need to implement searching an internal node\n");
+    exit(EXIT_FAILURE);
+  }
+}

我先把外部节点实现的分支存根，因为我还没有实现外部节点（internal node）。咱们能够在叶子节点中应用二叉搜寻法来进行搜寻。

+Cursor* leaf_node_find(Table* table, uint32_t page_num, uint32_t key) {+  void* node = get_page(table->pager, page_num);
+  uint32_t num_cells = *leaf_node_num_cells(node);
+
+  Cursor* cursor = malloc(sizeof(Cursor));
+  cursor->table = table;
+  cursor->page_num = page_num;
+
+  // Binary search
+  uint32_t min_index = 0;
+  uint32_t one_past_max_index = num_cells;
+  while (one_past_max_index != min_index) {+    uint32_t index = (min_index + one_past_max_index) / 2;
+    uint32_t key_at_index = *leaf_node_key(node, index);
+    if (key == key_at_index) {
+      cursor->cell_num = index;
+      return cursor;
+    }
+    if (key < key_at_index) {
+      one_past_max_index = index;
+    } else {
+      min_index = index + 1;
+    }
+  }
+
+  cursor->cell_num = min_index;
+  return cursor;
+}

这将返回：

• key 的地位
• 如果咱们须要插入一个新的 key，须要挪动 key 的话，返回这个 key 的地位
• 超出最初一个 key 的地位

因为咱们当初会查看节点的类型，所以须要一个函数在节点中获取和设置节点的类型值。

+NodeType get_node_type(void* node) {+  uint8_t value = *((uint8_t*)(node + NODE_TYPE_OFFSET));
+  return (NodeType)value;
+}
+
+void set_node_type(void* node, NodeType type) {
+  uint8_t value = type;
+  *((uint8_t*)(node + NODE_TYPE_OFFSET)) = value;
+}

须要先把值转为 uint8_t 类型来确保它作为一个序列化的单字节。还须要初始化节点的类型。

-void initialize_leaf_node(void* node) {*leaf_node_num_cells(node) = 0; }
+void initialize_leaf_node(void* node) {+  set_node_type(node, NODE_LEAF);
+  *leaf_node_num_cells(node) = 0;
+}

最初，咱们须要创立并解决一个新的错误码。

-enum ExecuteResult_t {EXECUTE_SUCCESS, EXECUTE_TABLE_FULL};
+enum ExecuteResult_t {
+  EXECUTE_SUCCESS,
+  EXECUTE_DUPLICATE_KEY,
+  EXECUTE_TABLE_FULL
+};
case (EXECUTE_SUCCESS):
  printf("Executed.\n");
  break;
+      case (EXECUTE_DUPLICATE_KEY):
+        printf("Error: Duplicate key.\n");
+        break;
case (EXECUTE_TABLE_FULL):
  printf("Error: Table full.\n");
  break;

有了这些批改，咱们的测试能够更改为查看排序程序：

"db > Executed.",
"db > Tree:",
"leaf (size 3)",
-      "- 0 : 3",
-      "- 1 : 1",
-      "- 2 : 2",
+      "- 0 : 1",
+      "- 1 : 2",
+      "- 2 : 3",
"db >"
])
end

并且咱们可能为反复键减少一个新的测试：

+  it 'prints an error message if there is a duplicate id' do
+    script = [
+      "insert 1 user1 person1@example.com",
+      "insert 1 user1 person1@example.com",
+      "select",
+      ".exit",
+    ]
+    result = run_script(script)
+    expect(result).to match_array([
+      "db > Executed.",
+      "db > Error: Duplicate key.",
+      "db > (1, user1, person1@example.com)",
+      "Executed.",
+      "db >",
+    ])
+  end

就是这样！接下来：实现拆分叶节点和创立外部节点。**

Enjoy GreatSQL :)

## 对于 GreatSQL

GreatSQL 是由万里数据库保护的 MySQL 分支，专一于晋升 MGR 可靠性及性能，反对 InnoDB 并行查问个性，是实用于金融级利用的 MySQL 分支版本。

相干链接：GreatSQL 社区 Gitee GitHub Bilibili

GreatSQL 社区：

社区博客有奖征稿详情：https://greatsql.cn/thread-10…

技术交换群：

微信：扫码增加 GreatSQL 社区助手 微信好友，发送验证信息 加群。