05. B树：实践（第二部分）

接上一章关于B树实现的内容。

5.1 B树删除操作

步骤1：从叶节点删除

从叶节点删除键的代码与其他节点替换函数类似。

// 从叶节点删除一个键
func leafDelete(new BNode, old BNode, idx uint16)  {
new.setHeader(BNODE_LEAF,  old.nkeys()-1)
nodeAppendRange(new,  old,  0,  0,  idx)
nodeAppendRange(new,  old,  idx,  idx+1,  old.nkeys()-(idx+1))
}

步骤2：递归删除

其结构与插入操作类似。

// 从树中删除一个键
func treeDelete(tree *BTree,  node BNode,  key []byte)  BNode {
// 在哪里查找键？
idx :=  nodeLookupLE(node,  key)
// 根据节点类型执行操作
switch  node.btype()  {
case  BNODE_LEAF:
if   !bytes.Equal(key,  node.getKey(idx))  {
return  BNode{}  // 未找到
}
// 删除叶节点中的键
new :=  BNode{data:  make([]byte,  BTREE_PAGE_SIZE)}
leafDelete(new,  node,  idx)
return  new
case  BNODE_NODE:
return  nodeDelete(tree,  node,  idx,  key)
default :
panic("bad node!")
}
}

步骤3：处理内部节点

不同之处在于，我们需要合并节点而不是拆分节点。一个节点可能会被合并到其左兄弟节点或右兄弟节点中。节点替换函数用于更新链接。

// treeDelete()的一部分
func nodeDelete(tree *BTree,  node BNode,  idx uint16,  key []byte)  BNode {
// 递归进入子节点
kptr :=  node.getPtr(idx)
updated :=  treeDelete(tree,  tree.get(kptr),  key)
if  len(updated.data)  ==  0 {
return  BNode{}  // 未找到
}
tree.del(kptr)

new :=  BNode{data:  make([]byte,  BTREE_PAGE_SIZE)}
// 检查是否合并
mergeDir,  sibling :=  shouldMerge(tree,  node,  idx,  updated)
switch  {
case  mergeDir <  0:  // 向左合并
merged :=  BNode{data:  make([]byte,  BTREE_PAGE_SIZE)}
nodeMerge(merged,  sibling,  updated)
tree.del(node.getPtr(idx -  1))
nodeReplace2Kid(new,  node,  idx-1,  tree.new(merged),  merged.getKey(0))
case  mergeDir >  0:  // 向右合并
merged :=  BNode{data:  make([]byte,  BTREE_PAGE_SIZE)}
nodeMerge(merged,  updated,  sibling)
tree.del(node.getPtr(idx +  1))
nodeReplace2Kid(new,  node,  idx,  tree.new(merged),  merged.getKey(0))
case  mergeDir ==  0:
assert(updated.nkeys()  >  0)
nodeReplaceKidN(tree,  new,  node,  idx,  updated)
}
return  new
}

// 将2个节点合并为1个
func nodeMerge(new BNode,  left BNode,   right BNode)  {
new.setHeader(left.btype(),  left.nkeys()+right.nkeys())
nodeAppendRange(new,  left,  0,  0,  left.nkeys())
nodeAppendRange(new,  right,  left.nkeys(),  0,  right.nkeys())
}

步骤4：合并条件

合并的条件是：

1. 节点小于页面的四分之一（这是任意设定的）。
2. 有一个兄弟节点，并且合并结果不超过一个页面。

// 更新后的子节点是否应与其兄弟节点合并？
func shouldMerge(
tree *BTree,  node BNode,
idx uint16,  updated BNode,
)  (int,  BNode)  {
if  updated.nbytes()  >  BTREE_PAGE_SIZE/4 {
return  0,  BNode{}
}

if  idx >  0 {
sibling :=  tree.get(node.getPtr(idx -  1))
merged :=  sibling.nbytes()  +  updated.nbytes()   -  HEADER
if  merged <=  BTREE_PAGE_SIZE {
return  -1,  sibling
}
}
if  idx+1 <  node.nkeys()  {
sibling :=  tree.get(node.getPtr(idx +  1))
merged :=  sibling.nbytes()  +  updated.nbytes()   -  HEADER
if  merged <=  BTREE_PAGE_SIZE {
return  +1,  sibling
}
}
return  0,  BNode{}
}

删除操作的代码完成了。

5.2 根节点

随着树的生长和收缩，我们需要跟踪根节点。让我们从删除操作开始。

这是B树删除操作的最终接口。当满足以下条件时，树的高度将减少一层：

1. 根节点不是叶节点。
2. 根节点只有一个子节点。

func  (tree *BTree)  Delete(key []byte)  bool {
assert(len(key)   !=  0)
assert(len(key)  <=  BTREE_MAX_KEY_SIZE)
if  tree.root ==  0 {
return  false
}

updated :=  treeDelete(tree,  tree.get(tree.root),  key)
if  len(updated.data)  ==  0 {
return  false // 未找到
}

tree.del(tree.root)
if  updated.btype()  ==  BNODE_NODE &&  updated.nkeys()  ==  1 {
// 移除一层
tree.root =  updated.getPtr(0)
}  else  {
tree.root =  tree.new(updated)
}
return  true
}

以下是插入操作的最终接口：

// 接口
func  (tree *BTree)  Insert(key []byte,  val []byte)  {
assert(len(key)   !=  0)
assert(len(key)  <=  BTREE_MAX_KEY_SIZE)
assert(len(val)  <=  BTREE_MAX_VAL_SIZE)

if  tree.root ==  0 {
// 创建第一个节点
root :=  BNode{data:  make([]byte,  BTREE_PAGE_SIZE)}
root.setHeader(BNODE_LEAF,  2)
// 一个虚拟键，这使得树覆盖整个键空间。
// 这样查找操作总能找到包含输入键的节点。
nodeAppendKV(root,  0,  0,  nil,  nil)
nodeAppendKV(root,  1,  0,  key,  val)
tree.root =  tree.new(root)
return
}

node :=  tree.get(tree.root)
tree.del(tree.root)

node =  treeInsert(tree,  node,  key,  val)
nsplit,  splitted :=  nodeSplit3(node)
if  nsplit >  1 {
// 根节点被拆分，添加新的一层。
root :=  BNode{data:  make([]byte,  BTREE_PAGE_SIZE)}
root.setHeader(BNODE_NODE,  nsplit)
for  i,  knode :=  range  splitted[:nsplit]  {
ptr,  key :=  tree.new(knode),  knode.getKey(0)
nodeAppendKV(root,  uint16(i),  ptr,  key,  nil)
}
tree.root =  tree.new(root)
}  else  {
tree.root =  tree.new(splitted[0])
}
}

插入操作执行以下两件事：

1. 当旧根节点被拆分为多个节点时，创建一个新的根节点。
2. 插入第一个键时，创建第一个叶节点作为根节点。

这里有个小技巧。在创建第一个节点时，我们向树中插入一个空键。按照排序顺序，空键是最小的键，这使得查找函数nodeLookupLE总能成功，避免了找不到包含输入键的节点的情况。

5.3 测试B树

由于我们的数据结构代码是纯粹的数据结构代码（不涉及输入输出），页面分配代码通过3个回调函数进行了隔离。以下是测试B树的容器代码，它将页面保存在内存哈希表中，而不将它们持久化到磁盘。在下一章中，我们将在不修改B树代码的情况下实现持久化。

type  C struct  { 
	tree  BTree
	ref   map[string]string
	pages map[uint64]BNode
}

func  newC()  *C {
	pages :=  map[uint64]BNode{}
	return  &C{
		tree:  BTree{
			get:  func(ptr uint64)  BNode {
				node,  ok :=  pages[ptr]
				assert(ok)
				return  node
			},
			new:  func(node BNode)  uint64 {
				assert(node.nbytes()  <=  BTREE_PAGE_SIZE)
				key :=  uint64(uintptr(unsafe.Pointer(&node.data[0])))
				assert(pages[key].data ==  nil)
				pages[key]  =  node
				return  key
			},
			del:  func(ptr uint64)  {
				_,  ok :=  pages[ptr]
				assert(ok)
				delete(pages,  ptr)
			},
		},
		ref:       map[string]string{},
		pages:  pages,
	}
}

我们使用一个引用映射来记录每次B树的更新，以便稍后验证B树的正确性。

func  (c *C)  add(key string,  val string)  {
	c.tree.Insert([]byte(key),  []byte(val))
	c.ref[key]  =  val
}

func  (c *C)  del(key string)  bool {
	delete(c.ref,  key)
	return  c.tree.Delete([]byte(key))
}

测试用例留给读者作为练习。

5.4 结束语

这个B树实现非常精简，但精简对于学习目的来说是有好处的。在实际应用中，实现可能会复杂得多，并且包含实用的优化措施。

对于我们的B树实现，有一些简单的改进方向：

1. 对叶节点和内部节点使用不同的格式。叶节点不需要指针，内部节点不需要值。这样可以节省一些空间。
2. 键或值的长度之一是多余的——键值对（KV pair）的长度可以从下一个键的偏移量推断出来。
3. 节点的第一个键是不必要的，因为它可以从其父节点的链接继承。
4. 添加校验和以检测数据损坏。

构建键值存储（KV store）的下一步是将我们的B树持久化到磁盘，这是下一章的主题。