Python中B树实现未通过测试，疑似_split_node函数存在错误

我正在开发一个B树项目，已经完成了B树类的编写，但无法通过测试用例。测试用例通过递归断言检查B树的结构属性，但始终不通过，我怀疑问题出在_split_node函数中，希望得到排查指导。

B树实现代码 #

class BTree:
    def __init__(self, M: int, L: int):
        self.M = M
        self.L = L
        self.root_addr: Address = DISK.new()
        DISK.write(self.root_addr, BTreeNode(self.root_addr, None, None, True))

    def _find_leaf(self, key: KT) -> Address:
        current_addr = self.root_addr
        while True:
            node = DISK.read(current_addr)
            if node.is_leaf:
                return current_addr

            idx = node.find_idx(key)
            if idx < len(node.keys) and node.keys[idx] == key:
                idx += 1
            current_addr = node.children_addrs[idx]

    def insert(self, key: KT, value: VT) -> None:
        leaf_addr = self._find_leaf(key)
        leaf = DISK.read(leaf_addr)

        leaf.insert_data(key, value)

        if len(leaf.keys) > self.M - 1:
            self._split_node(leaf_addr)

        DISK.write(leaf_addr, leaf)

    def _split_node(self, node_addr: Address) -> None:
        node = DISK.read(node_addr)
        middle_index = len(node.keys) // 2
        middle_key = node.keys[middle_index]

        left_node = BTreeNode(DISK.new(), node.parent_addr, None, node.is_leaf)
        right_node = BTreeNode(DISK.new(), node.parent_addr, None, node.is_leaf)

        left_node.keys = node.keys[:middle_index]
        right_node.keys = node.keys[middle_index + 1:]

        if node.is_leaf:
            left_node.data = node.data[:middle_index]
            right_node.data = node.data[middle_index + 1:]
        else:
            left_node.children_addrs = node.children_addrs[:middle_index + 1]
            right_node.children_addrs = node.children_addrs[middle_index + 1:]

        DISK.write(left_node.my_addr, left_node)
        DISK.write(right_node.my_addr, right_node)

        if node.parent_addr is not None:
            parent = DISK.read(node.parent_addr)
            insert_index = parent.find_idx(middle_key)

            parent.keys.insert(insert_index, middle_key)
            parent.children_addrs[insert_index] = left_node.my_addr
            parent.children_addrs.insert(insert_index + 1, right_node.my_addr)

            DISK.write(node.parent_addr, parent)
        else:
            new_root = BTreeNode(DISK.new(), None, None, False)
            new_root.keys = [middle_key]
            new_root.children_addrs = [left_node.my_addr, right_node.my_addr]
            self.root_addr = new_root.my_addr

            DISK.write(new_root.my_addr, new_root)

    def find(self, key: KT) -> Optional[VT]:
        current_addr = self.root_addr

        while True:
            node = DISK.read(current_addr)

            idx = node.find_idx(key)

            if node.is_leaf:
                if idx < len(node.keys) and node.keys[idx] == key:
                    return node.data[idx]
                return None
            else:
                if idx < len(node.keys) and node.keys[idx] == key:
                    idx += 1
                current_addr = node.children_addrs[idx]

测试代码 #

def btree_properties_recurse(root_node_addr, node, M, L):

    assert sorted(node.keys) == node.keys # Keys should remain sorted so that a binary search is possible

    if node.is_leaf:
        # Leaf node general properties
        assert len(node.children_addrs) == 0
        assert len(node.keys) == len(node.data)
        assert len(node.data) <= L
    else:
        # Non-leaf node general properties
        assert len(node.data) == 0
        assert len(node.keys) == len(node.children_addrs) - 1
        assert len(node.children_addrs) <= M

    if node.my_addr == root_node_addr:
        # Root node properties
        assert node.parent_addr is None
        assert node.index_in_parent is None
        if not node.is_leaf:
            assert len(node.children_addrs) >= 2
    else:
        # Non-root node properties
        assert node.parent_addr is not None
        assert node.index_in_parent is not None
        if node.is_leaf:
            assert len(node.data) >= (L+1)//2
        else:
            assert len(node.children_addrs) >= (M+1)//2
        
    # Run the assertions on all children
    for child_addr in node.children_addrs:
        btree_properties_recurse(root_node_addr, DISK.read(child_addr), M, L)

def test_btree_properties_even() -> None:
    M = 6
    L = 6
    btree = BTree(M, L)
    for i in range(100):
        btree.insert(i, str(i))
    for i in range(0, -100, -1):
        btree.insert(i, str(i))

    root_addr = btree.root_addr
    btree_properties_recurse(root_addr, DISK.read(root_addr), M, L)

问题排查与修复建议 #

你的_split_node函数存在几个关键问题，直接导致测试断言失败：

1. 叶子节点拆分触发条件错误 #

在insert方法中，你用self.M - 1作为叶子节点的键数上限，但实际上叶子节点的最大键数应该是self.L（测试代码里也明确断言叶子节点len(data) <= L）。当前逻辑会导致叶子节点还没达到容量上限就被拆分，或者超过上限才拆分，违反B树的叶子节点容量规则。

修复：将insert中的判断条件改为：

if len(leaf.keys) > self.L:
    self._split_node(leaf_addr)

2. 非叶子节点拆分时，子节点的父地址未更新 #

当拆分非叶子节点时，原节点的子节点现在属于left_node或right_node，但这些子节点的parent_addr仍然指向原节点，没有更新为新的父节点地址。这会触发测试中“非根节点parent_addr不为None”的断言失败，同时破坏树的父子关系。

修复：在拆分非叶子节点后，遍历左右节点的子节点，更新它们的父地址：

else:
    left_node.children_addrs = node.children_addrs[:middle_index + 1]
    right_node.children_addrs = node.children_addrs[middle_index + 1:]
    # 更新左节点子节点的父地址
    for child_addr in left_node.children_addrs:
        child = DISK.read(child_addr)
        child.parent_addr = left_node.my_addr
        DISK.write(child_addr, child)
    # 更新右节点子节点的父地址
    for child_addr in right_node.children_addrs:
        child = DISK.read(child_addr)
        child.parent_addr = right_node.my_addr
        DISK.write(child_addr, child)

3. 未设置新节点的index_in_parent属性 #

测试代码明确断言非根节点的index_in_parent不为None，但你的_split_node函数中创建的left_node和right_node的index_in_parent始终为None，直接触发断言失败。

修复：在父节点中插入新节点后，设置左右节点的index_in_parent并写入磁盘：

if node.parent_addr is not None:
    parent = DISK.read(node.parent_addr)
    insert_index = parent.find_idx(middle_key)

    parent.keys.insert(insert_index, middle_key)
    parent.children_addrs[insert_index] = left_node.my_addr
    parent.children_addrs.insert(insert_index + 1, right_node.my_addr)

    # 设置左右节点的index_in_parent
    left_node.index_in_parent = insert_index
    DISK.write(left_node.my_addr, left_node)
    right_node.index_in_parent = insert_index + 1
    DISK.write(right_node.my_addr, right_node)

    DISK.write(node.parent_addr, parent)
else:
    # 根节点拆分时，新根的子节点index_in_parent需要设置
    new_root = BTreeNode(DISK.new(), None, None, False)
    new_root.keys = [middle_key]
    new_root.children_addrs = [left_node.my_addr, right_node.my_addr]
    self.root_addr = new_root.my_addr

    left_node.index_in_parent = 0
    DISK.write(left_node.my_addr, left_node)
    right_node.index_in_parent = 1
    DISK.write(right_node.my_addr, right_node)

    DISK.write(new_root.my_addr, new_root)

4. 拆分后原叶子节点的无效写回 #

在拆分叶子节点后，你仍然将原叶子节点写回磁盘（insert方法最后一行的DISK.write(leaf_addr, leaf)），但这个原节点已经被拆分后的左右节点替代，留在磁盘中会导致无效数据。实际上，拆分后原节点已经不再属于树结构，不需要再写回。

修复：将insert方法中的写回操作移到拆分判断之前：

def insert(self, key: KT, value: VT) -> None:
    leaf_addr = self._find_leaf(key)
    leaf = DISK.read(leaf_addr)

    leaf.insert_data(key, value)
    # 先写回插入后的节点
    DISK.write(leaf_addr, leaf)

    if len(leaf.keys) > self.L:
        self._split_node(leaf_addr)

完成以上修复后，重新运行测试，应该能通过所有断言检查。

内容的提问来源于stack exchange，提问作者Daniel Branysh