avatar个人博客

从头实现数据库-5

Oct 31, 2023

B-树的删除

  1. 从叶节点删除
从叶节点中删除键的代码与其他nodeReplace一样的功能。
// 删除一个key从叶子节点
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))
}
  1. 递归删除
结构类似于插入
// 从tree中删除一个key
func treeDelete(tree *BTree, node BNode, key []byte) BNode {
    // 先找到key的索引
    idx := nodeLookupLE(node, key)
    // 根据node节点的类型处理
    switch node.btype() {
    case BNODE_LEAF:
        if !bytes.Equal(key, node.getKey(idx)) {
            return BNode{} // 没找到
        }
        // 从叶子节点删除一个key
        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!")
    }
}
  1. 处理内部节点
不同的是，删除需要合并节点，而不是拆分节点。一个节点可以合并到它的左边或右边的兄弟节点之一。nodeReplace函数用于更新链接。
// part of the 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{} // not found
    }
    // 删除指针
    tree.del(kptr)
    
    // 构建新的子节点
    new := BNode{data: make([]byte, BTREE_PAGE_SIZE)}
    // check for merging
    mergeDir, sibling := shouldMerge(tree, node, idx, updated)
    switch {
    case mergeDir < 0: // left
        // 创建一个空节点 合并
        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: // right
        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
}
// replace a link with the same key
func nodeReplaceKid1ptr(new BNode, old BNode, idx uint16, ptr uint64) {
	copy(new.data, old.data[:old.nbytes()])
	new.setPtr(idx, ptr) // only the pointer is changed
}

// replace a link with multiple links
func nodeReplaceKidN(
	tree *BTree, new BNode, old BNode, idx uint16,
	kids ...BNode,
) {
	inc := uint16(len(kids))
	if inc == 1 && bytes.Equal(kids[0].getKey(0), old.getKey(idx)) {
		// common case, only replace 1 pointer
		nodeReplaceKid1ptr(new, old, idx, tree.new(kids[0]))
		return
	}

	new.setHeader(BNODE_NODE, old.nkeys()+inc-1)
	nodeAppendRange(new, old, 0, 0, idx)
	for i, node := range kids {
		nodeAppendKV(new, idx+uint16(i), tree.new(node), node.getKey(0), nil)
	}
	nodeAppendRange(new, old, idx+inc, idx+1, old.nkeys()-(idx+1))
}

// 将 2 个相邻链接替换为 1 个
func nodeReplace2Kid(
	new BNode, old BNode, idx uint16,
	ptr uint64, key []byte,
) {
	new.setHeader(BNODE_NODE, old.nkeys()-1)
	nodeAppendRange(new, old, 0, 0, idx)
	nodeAppendKV(new, idx, ptr, key, nil)
	nodeAppendRange(new, old, idx+1, idx+2, old.nkeys()-(idx+2))
}
// 合并两个节点到一个节点上
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())
}
  1. 合并的条件
合并的条件是：
  • 节点小于一个页面的1/4（这是任意的)
  • 有一个同级，合并的结果不超过一个页。
// should the updated kid be merged with a sibling?
func shouldMerge(
    tree *BTree, node BNode,
    idx uint16, updated BNode,
) (int, BNode) {
    // 大于1/4 不合并
    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{}
}
删除代码就完成了。

根节点

随着树的生长和收缩，我们需要跟踪根结点。先说删除吧
这是B树删除的最后一个接口。在下列情况下，树的高度将减少一：
  • 根结点不是叶子。
  • 根结点只有一个子结点。
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 // not found
    }

    tree.del(tree.root)
    if updated.btype() == BNODE_NODE && updated.nkeys() == 1 {
        // remove a level
        tree.root = updated.getPtr(0)
    } else {
        tree.root = tree.new(updated)
    }
    return true
}
下面是要插入的最后一个界面：
// the interface
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 {
        // create the first node
        root := BNode{data: make([]byte, BTREE_PAGE_SIZE)}
        root.setHeader(BNODE_LEAF, 2)
        // a dummy key, this makes the tree cover the whole key space.
        // thus a lookup can always find a containing node.
        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 {
        // the root was split, add a new level.
        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])
    }
}
它做两件事：
  • 当旧根被分割成多个结点时，就创建了一个新的根结点。
  • 插入第一个键时，创建第一个叶节点作为根。
这里有个小把戏。在创建第一个节点时，我们在树中插入一个空密钥。根据排序顺序，空键是最低可能的键，它使查找函数nodeLookupLE始终成功，消除了找不到包含输入键的节点的情况。

测试B-树

由于我们的数据结构代码是纯数据结构代码（没有IO），页面分配代码通过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))
}
测试用例作为练习留给读者

闭幕词

这个B-tree实现是非常小的，但是对于学习的目的来说，最小值是很好的。现实世界的实现可能要复杂得多，并且包含实际的优化
我们的B树实现有一些简单的改进:
  • 对叶节点和内部节点使用不同的格式。叶节点不需要指针，内部节点不需要值。这样可以节省一些空间。
  • 键或值的长度之一是冗余的-KV对的长度可以从下一个键的偏移量中推断出来。
  • 节点的第一个键是不需要的，因为它是从父节点的链接继承来的。
  • 添加校验和以检测数据损坏。
构建KV存储的下一步是将B树持久化到磁盘，这是下一章的主题

Copyright © 2025 later

logo