Algorithm Search⌘K

B+ tree in Python

B-tree variant that stores all keys in linked leaves, giving fast range scans and high fanout.

File
b_plus_tree.py, 225 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
b_plus_tree.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/b-plus-tree/. The authors are listed in prov/M7WT2F.jsonl.

from bisect import bisect_left, bisect_right, insort
from typing import Any, Generic, Optional, Protocol, TypeVar


class Ordered(Protocol):
    def __lt__(self, other: Any, /) -> bool: ...


K = TypeVar("K", bound=Ordered)


class Node(Generic[K]):
    """A B+ tree node. A leaf holds keys and a link to the next leaf. An
    internal node holds separators key[0..m-1] and children c[0..m], where
    c[i] holds the keys k with key[i-1] <= k < key[i]."""

    def __init__(self, leaf: bool) -> None:
        self.leaf = leaf
        self.keys: list[K] = []
        self.children: list["Node[K]"] = []
        self.next: Optional["Node[K]"] = None


class BPlusTree(Generic[K]):
    """A B+ tree of order d >= 1 over distinct keys.

    Every node except the root holds d to 2d keys. All keys are in the
    leaves, which are linked from left to right; internal nodes hold only
    separators. Insertion and deletion work bottom-up along the search path.
    search, insert and delete make O(log_d n) node reads; range(a, b) makes
    O(log_d n + z / d) reads when it reports z keys. `reads` counts the node
    reads; the root is kept in memory.
    """

    def __init__(self, d: int = 2) -> None:
        if d < 1:
            raise ValueError("the order d must be at least 1")
        self.d = d
        self.root: Node[K] = Node(leaf=True)
        self.reads = 0

    def _find_leaf(self, k: K) -> Node[K]:
        x = self.root  # @line L1
        while not x.leaf:  # @line L2
            i = bisect_right(x.keys, k)  # @line L3
            x = x.children[i]  # @line L4
            self.reads += 1
        return x  # @line L5

    def search(self, k: K) -> bool:
        """Return True when k is in the tree."""
        x = self._find_leaf(k)
        j = bisect_left(x.keys, k)
        return j < len(x.keys) and not (k < x.keys[j])

    def range(self, a: K, b: K) -> list[K]:
        """Return the keys k with a <= k <= b, in increasing order."""
        x: Optional[Node[K]] = self._find_leaf(a)  # @line L6
        out: list[K] = []
        while x is not None:  # @line L7
            for k in x.keys:  # @line L8
                if k < a:
                    continue
                if b < k:
                    return out
                out.append(k)
            if x.next is None:  # @line L9
                return out
            x = x.next  # @line L10
            self.reads += 1
        return out

    def insert(self, k: K) -> bool:
        """Insert k. Return False, and change nothing, if k is present."""
        if self.search(k):
            return False
        r = self._insert_rec(self.root, k)  # @line L11
        if r is not None:  # @line L12
            s, z = r
            new_root: Node[K] = Node(leaf=False)  # @line L13
            new_root.keys = [s]
            new_root.children = [self.root, z]
            self.root = new_root
        return True

    def _insert_rec(self, x: Node[K], k: K) -> Optional[tuple[K, Node[K]]]:
        d = self.d
        if x.leaf:  # @line L14
            insort(x.keys, k)  # @line L15
            if len(x.keys) <= 2 * d:  # @line L16
                return None
            z: Node[K] = Node(leaf=True)  # @line L17
            z.keys = x.keys[d:]
            del x.keys[d:]
            z.next, x.next = x.next, z  # @line L18
            return (z.keys[0], z)  # @line L19
        i = bisect_right(x.keys, k)  # @line L20
        self.reads += 1
        r = self._insert_rec(x.children[i], k)  # @line L21
        if r is None:  # @line L22
            return None
        s, z = r  # @line L23
        x.keys.insert(i, s)
        x.children.insert(i + 1, z)
        if len(x.keys) <= 2 * d:  # @line L24
            return None
        z2: Node[K] = Node(leaf=False)  # @line L25
        z2.keys = x.keys[d + 1 :]
        z2.children = x.children[d + 1 :]
        s2 = x.keys[d]  # @line L26
        del x.keys[d:]
        del x.children[d + 1 :]
        return (s2, z2)  # @line L27

    def delete(self, k: K) -> bool:
        """Delete k. Return False, and change nothing, if k is absent."""
        if not self.search(k):
            return False
        self._delete_rec(self.root, k)  # @line L28
        if not self.root.leaf and not self.root.keys:  # @line L29
            self.root = self.root.children[0]
        return True

    def _delete_rec(self, x: Node[K], k: K) -> bool:
        if x.leaf:  # @line L30
            j = bisect_left(x.keys, k)
            if j < len(x.keys) and not (k < x.keys[j]):
                x.keys.pop(j)
            return len(x.keys) < self.d
        i = bisect_right(x.keys, k)  # @line L31
        self.reads += 1
        if not self._delete_rec(x.children[i], k):  # @line L32
            return False
        self._fix_child(x, i)  # @line L33
        return len(x.keys) < self.d  # @line L34

    def _fix_child(self, x: Node[K], i: int) -> None:
        d = self.d
        c = x.children[i]
        left = x.children[i - 1] if i > 0 else None  # @line L35
        right = x.children[i + 1] if i + 1 < len(x.children) else None
        if left is not None and len(left.keys) > d:  # @line L36
            if c.leaf:
                c.keys.insert(0, left.keys.pop())
                x.keys[i - 1] = c.keys[0]
            else:
                c.keys.insert(0, x.keys[i - 1])
                x.keys[i - 1] = left.keys.pop()
                c.children.insert(0, left.children.pop())
            return
        if right is not None and len(right.keys) > d:
            if c.leaf:
                c.keys.append(right.keys.pop(0))
                x.keys[i] = right.keys[0]
            else:
                c.keys.append(x.keys[i])
                x.keys[i] = right.keys.pop(0)
                c.children.append(right.children.pop(0))
            return
        if right is not None:  # @line L37
            j, a, b = i, c, right
        else:
            assert left is not None
            j, a, b = i - 1, left, c
        sep = x.keys.pop(j)
        x.children.pop(j + 1)
        if a.leaf:
            a.keys.extend(b.keys)
            a.next = b.next
        else:
            a.keys.append(sep)
            a.keys.extend(b.keys)
            a.children.extend(b.children)

    def keys(self) -> list[K]:
        """Return all keys in order by walking the leaf chain."""
        x = self.root
        while not x.leaf:
            x = x.children[0]
        out: list[K] = []
        y: Optional[Node[K]] = x
        while y is not None:
            out.extend(y.keys)
            y = y.next
        return out

    def height(self) -> int:
        """Return the number of edges on a root-to-leaf path."""
        h, x = 0, self.root
        while not x.leaf:
            x, h = x.children[0], h + 1
        return h

    def check_invariant(self) -> None:
        """Raise AssertionError if a B+ tree property fails."""
        d = self.d
        leaves: list[Node[K]] = []
        depths: set[int] = set()

        def walk(x: Node[K], lo: Any, hi: Any, depth: int, is_root: bool) -> None:
            m = len(x.keys)
            assert m <= 2 * d
            assert is_root or m >= d
            assert all(x.keys[j] < x.keys[j + 1] for j in range(m - 1))
            if m:
                assert lo is None or not (x.keys[0] < lo)
                assert hi is None or x.keys[-1] < hi
            if x.leaf:
                assert not x.children
                depths.add(depth)
                leaves.append(x)
                return
            assert m >= 1 and len(x.children) == m + 1
            bounds = [lo, *x.keys, hi]
            for j, c in enumerate(x.children):
                walk(c, bounds[j], bounds[j + 1], depth + 1, False)

        walk(self.root, None, None, 0, True)
        assert len(depths) == 1
        for a, b in zip(leaves, leaves[1:]):
            assert a.next is b
        assert leaves[-1].next is None
test_b_plus_tree.py 92 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/b-plus-tree/. The authors are listed in prov/M7WT2F.jsonl.

import math
import random

import pytest

from b_plus_tree import BPlusTree, Node


def shape(x: Node[int]) -> str:
    if x.leaf:
        return "[" + " ".join(map(str, x.keys)) + "]"
    return "<" + " ".join(map(str, x.keys)) + ">(" + " ".join(shape(c) for c in x.children) + ")"


def test_example() -> None:
    t: BPlusTree[int] = BPlusTree(2)
    for k in [20, 5, 40, 10, 30, 15, 25, 35, 45, 50, 12]:
        assert t.insert(k)
        t.check_invariant()
    assert shape(t.root) == "<20 30 40>([5 10 12 15] [20 25] [30 35] [40 45 50])"
    assert t.range(12, 35) == [12, 15, 20, 25, 30, 35]
    want = [
        (40, "<20 30 40>([5 10 12 15] [20 25] [30 35] [45 50])"),
        (45, "<20 30>([5 10 12 15] [20 25] [30 35 50])"),
        (50, "<20 30>([5 10 12 15] [20 25] [30 35])"),
        (5, "<20 30>([10 12 15] [20 25] [30 35])"),
        (25, "<15 30>([10 12] [15 20] [30 35])"),
        (12, "<30>([10 15 20] [30 35])"),
        (30, "<20>([10 15] [20 35])"),
    ]
    for k, s in want:
        assert t.delete(k)
        t.check_invariant()
        assert shape(t.root) == s


def test_edges() -> None:
    with pytest.raises(ValueError):
        BPlusTree(0)
    t: BPlusTree[int] = BPlusTree(1)
    assert not t.search(3)
    assert t.range(0, 10) == []
    assert not t.delete(3)
    assert t.insert(3) and not t.insert(3)
    assert t.range(3, 3) == [3]
    assert t.range(4, 2) == []
    assert t.delete(3)
    assert t.keys() == []
    t.check_invariant()


def test_height_and_range_cost() -> None:
    d = 8
    t: BPlusTree[int] = BPlusTree(d)
    n = 5000
    for k in range(n):
        t.insert(k)
    t.check_invariant()
    h = t.height()
    # a leaf has at least d keys, an internal node other than the root at least d + 1 children
    assert h <= 1 + math.log(n / (2 * d), d + 1) + 1e-9
    r0 = t.reads
    got = t.range(1000, 1999)
    assert got == list(range(1000, 2000))
    assert t.reads - r0 <= h + math.ceil(1000 / d) + 1


def test_against_set() -> None:
    rng = random.Random(27)
    for _ in range(500):
        d = rng.randrange(1, 4)
        t: BPlusTree[int] = BPlusTree(d)
        model: set[int] = set()
        for _ in range(rng.randrange(1, 60)):
            k = rng.randrange(40)
            op = rng.randrange(4)
            if op == 0:
                assert t.insert(k) == (k not in model)
                model.add(k)
            elif op == 1:
                assert t.delete(k) == (k in model)
                model.discard(k)
            elif op == 2:
                assert t.search(k) == (k in model)
            else:
                a, b = sorted((rng.randrange(40), rng.randrange(40)))
                assert t.range(a, b) == [x for x in sorted(model) if a <= x <= b]
            t.check_invariant()
            assert t.keys() == sorted(model)

What the next level needs

  • the tests did not run in this build

Read the full page about B+ tree