Algorithm Search⌘K

k-d-B tree in Python

Disk-oriented hybrid of the k-d tree and the B-tree in which all leaves are at the same depth.

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

from typing import Optional

INF = float("inf")
Point = tuple[int, int]
Vec = tuple[float, float]


class Page:
    """A page of the tree with the region [lo, hi), half-open on each axis.

    A point page (leaf) has children None and holds points. A region page has
    a list of child pages whose regions partition its own region.
    """

    __slots__ = ("lo", "hi", "points", "children")

    def __init__(self, lo: Vec, hi: Vec, children: Optional[list["Page"]] = None) -> None:
        self.lo, self.hi = lo, hi
        self.points: list[Point] = []
        self.children = children

    def holds(self, p: Point) -> bool:
        return all(self.lo[a] <= p[a] < self.hi[a] for a in range(2))

    def meets(self, lo: Point, hi: Point) -> bool:
        """Whether the region meets the closed box [lo, hi]."""
        return all(self.lo[a] <= hi[a] and lo[a] < self.hi[a] for a in range(2))


def _with(v: Vec, a: int, x: float) -> Vec:
    return (x, v[1]) if a == 0 else (v[0], x)


def split_page(u: Page, a: int, v: float) -> tuple[Page, Page]:
    """Cut the page u by the line x_a = v, which crosses its region, into two pages.

    The children that cross the line are cut too, down to the point pages:
    these are the forced splits of the k-d-B tree.
    """
    left = Page(u.lo, _with(u.hi, a, v))
    right = Page(_with(u.lo, a, v), u.hi)
    if u.children is None:
        left.points = [p for p in u.points if p[a] < v]
        right.points = [p for p in u.points if p[a] >= v]
        return left, right
    left.children, right.children = [], []
    for c in u.children:
        if c.hi[a] <= v:
            left.children.append(c)
        elif c.lo[a] >= v:
            right.children.append(c)
        else:
            cl, cr = split_page(c, a, v)
            left.children.append(cl)
            right.children.append(cr)
    return left, right


class KdBTree:
    """A two-dimensional k-d-B tree with B points per point page and M entries per region page.

    Invariant: the regions of the children of a region page partition its
    region, each point lies in the region of its point page, every point page
    is at the same depth, point pages hold at most B points, and region pages
    hold between 1 and M children. Pages may be underfull, even empty.
    """

    def __init__(self, leaf_capacity: int = 2, fanout: int = 3) -> None:
        if leaf_capacity < 1 or fanout < 2:
            raise ValueError("need leaf_capacity >= 1 and fanout >= 2")
        self.B, self.M = leaf_capacity, fanout
        self.root = Page((-INF, -INF), (INF, INF))
        self.size = 0

    def __len__(self) -> int:
        return self.size

    def _path(self, p: Point) -> list[Page]:
        path = [self.root]
        while path[-1].children is not None:
            path.append(next(c for c in path[-1].children if c.holds(p)))  # type: ignore[union-attr]
        return path

    def __contains__(self, p: Point) -> bool:
        return (int(p[0]), int(p[1])) in self._path(p)[-1].points

    def _leaf_plane(self, u: Page) -> tuple[int, float]:
        """A line through the median of the distinct coordinates on the axis of larger spread."""
        best = None
        for a in range(2):
            d = sorted({p[a] for p in u.points})
            if len(d) >= 2:
                key = (-(d[-1] - d[0]), a)
                if best is None or key < best[0]:
                    best = (key, a, d[len(d) // 2])
        assert best is not None, "a full page of distinct points differs on some axis"
        return best[1], best[2]

    def _region_plane(self, u: Page) -> tuple[int, float]:
        """The line at a child boundary that balances the halves, then cuts the fewest children."""
        kids = u.children or []
        best = None
        for a in range(2):
            for v in sorted({c.lo[a] for c in kids if c.lo[a] > u.lo[a]}):
                left = sum(1 for c in kids if c.lo[a] < v)
                right = sum(1 for c in kids if c.hi[a] > v)
                if left <= self.M and right <= self.M:
                    key = (max(left, right), left + right, a, v)
                    if best is None or key < best:
                        best = key
        assert best is not None, "a guillotine partition always has a cut that splits no child"
        return best[2], best[3]

    def insert(self, p: Point) -> bool:
        """Add p. Returns False if p was stored."""
        p = (int(p[0]), int(p[1]))
        path = self._path(p)  # @line L1
        leaf = path[-1]
        if p in leaf.points:
            return False
        leaf.points.append(p)  # @line L2
        self.size += 1
        i = len(path) - 1
        while i >= 0:  # @line L3
            u = path[i]
            full = len(u.points) > self.B if u.children is None else len(u.children) > self.M
            if not full:  # @line L4
                break
            a, v = self._leaf_plane(u) if u.children is None else self._region_plane(u)  # @line L5
            left, right = split_page(u, a, v)  # @line L6
            if i == 0:  # @line L7
                self.root = Page((-INF, -INF), (INF, INF), [left, right])
                break
            parent = path[i - 1]
            k = parent.children.index(u)  # type: ignore[union-attr]
            parent.children[k : k + 1] = [left, right]  # type: ignore[index]  # @line L8
            i -= 1  # @line L9
        return True

    def delete(self, p: Point) -> bool:
        """Remove p from its point page. Pages are not merged."""
        p = (int(p[0]), int(p[1]))
        leaf = self._path(p)[-1]  # @line L14
        if p not in leaf.points:
            return False
        leaf.points.remove(p)  # @line L15
        self.size -= 1
        return True

    def range_search(self, lo: Point, hi: Point) -> list[Point]:
        """The points in the closed box [lo, hi]."""
        out: list[Point] = []
        stack = [self.root]
        while stack:  # @line L10
            u = stack.pop()
            if not u.meets(lo, hi):  # @line L11
                continue
            if u.children is None:  # @line L12
                out.extend(q for q in u.points if lo[0] <= q[0] <= hi[0] and lo[1] <= q[1] <= hi[1])
            else:
                stack.extend(u.children)  # @line L13
        return out

    def height(self) -> int:
        h, u = 0, self.root
        while u.children is not None:
            u, h = u.children[0], h + 1
        return h

    def leaves(self) -> list[Page]:
        out, stack = [], [self.root]
        while stack:
            u = stack.pop()
            if u.children is None:
                out.append(u)
            else:
                stack.extend(reversed(u.children))
        return out

    def points(self) -> list[Point]:
        return [q for leaf in self.leaves() for q in leaf.points]

    def check_invariant(self) -> None:
        assert self.root.lo == (-INF, -INF) and self.root.hi == (INF, INF)
        depths: set[int] = set()
        total = 0
        stack: list[tuple[Page, int]] = [(self.root, 0)]
        while stack:
            u, d = stack.pop()
            assert all(u.lo[a] < u.hi[a] for a in range(2)), "empty region"
            if u.children is None:
                depths.add(d)
                assert len(u.points) <= self.B, "point page overflow"
                assert all(u.holds(q) for q in u.points), "point outside its page"
                total += len(u.points)
                continue
            assert 1 <= len(u.children) <= self.M, "region page has a wrong size"
            area_ok = True
            for c in u.children:
                assert all(u.lo[a] <= c.lo[a] and c.hi[a] <= u.hi[a] for a in range(2)), "child outside"
                stack.append((c, d + 1))
            kids = u.children
            for i in range(len(kids)):
                for j in range(i + 1, len(kids)):
                    x, y = kids[i], kids[j]
                    overlap = all(max(x.lo[a], y.lo[a]) < min(x.hi[a], y.hi[a]) for a in range(2))
                    assert not overlap, "child regions overlap"
            # cover: the lower left corner of each cell of the grid of child boundaries lies in one child
            xs = sorted({c.lo[0] for c in kids} | {c.hi[0] for c in kids if c.hi[0] < u.hi[0]} | {u.lo[0]})
            ys = sorted({c.lo[1] for c in kids} | {c.hi[1] for c in kids if c.hi[1] < u.hi[1]} | {u.lo[1]})
            for x in xs:
                for y in ys:
                    probe = (x if x > -INF else -(10**18), y if y > -INF else -(10**18))
                    n = sum(1 for c in kids if all(c.lo[a] <= probe[a] < c.hi[a] for a in range(2)))
                    area_ok = area_ok and n == 1
            assert area_ok, "child regions do not cover the page"
        assert len(depths) <= 1, "point pages at different depths"
        assert total == self.size == len(set(self.points())), "size is wrong"
test_kd_b_tree.py 93 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/kd-b-tree/. The authors are listed in prov/TMRMPW.jsonl.

import random

import pytest

from kd_b_tree import INF, KdBTree, Page


def shape(u: Page) -> object:
    box = (u.lo, u.hi)
    if u.children is None:
        return (box, sorted(u.points))
    return (box, [shape(c) for c in u.children])


def test_page_example() -> None:
    t = KdBTree(1, 4)
    for p in [(3, 1), (5, 1), (5, 8), (7, 7)]:
        assert t.insert(p)
        t.check_invariant()
    assert t.height() == 1 and len(t.root.children) == 4
    assert t.insert((7, 8))
    t.check_invariant()
    assert not t.insert((7, 8))
    assert t.height() == 2
    assert shape(t.root) == (
        ((-INF, -INF), (INF, INF)),
        [
            (
                ((-INF, -INF), (INF, 8)),
                [
                    (((-INF, -INF), (5, 8)), [(3, 1)]),
                    (((5, -INF), (INF, 7)), [(5, 1)]),
                    (((5, 7), (INF, 8)), [(7, 7)]),
                ],
            ),
            (
                ((-INF, 8), (INF, INF)),
                [
                    (((-INF, 8), (5, INF)), []),
                    (((5, 8), (7, INF)), [(5, 8)]),
                    (((7, 8), (INF, INF)), [(7, 8)]),
                ],
            ),
        ],
    )
    assert sorted(t.range_search((4, 0), (6, 9))) == [(5, 1), (5, 8)]
    assert t.delete((3, 1)) and not t.delete((3, 1))
    t.check_invariant()
    assert len(t) == 4 and t.height() == 2


def test_edges() -> None:
    t = KdBTree(1, 2)
    assert t.range_search((0, 0), (9, 9)) == [] and (1, 1) not in t
    for i in range(64):
        assert t.insert((i, i))
        t.check_invariant()
    assert sorted(t.range_search((10, 10), (12, 12))) == [(10, 10), (11, 11), (12, 12)]
    assert t.height() >= 6
    u = KdBTree(2, 3)
    for i in range(30):
        u.insert((5, i))
    u.check_invariant()
    with pytest.raises(ValueError):
        KdBTree(0, 3)
    with pytest.raises(ValueError):
        KdBTree(2, 1)


def test_random_against_oracle() -> None:
    rng = random.Random(20261104)
    for case in range(600):
        t = KdBTree(rng.randint(1, 4), rng.randint(2, 6))
        live: set = set()
        span = rng.choice([5, 20, 1000])
        for _ in range(rng.randint(1, 60)):
            p = (rng.randrange(span), rng.randrange(span))
            if live and rng.random() < 0.2:
                d = rng.choice(sorted(live))
                assert d in t and t.delete(d)
                live.discard(d)
            assert t.insert(p) == (p not in live)
            live.add(p)
        t.check_invariant()
        assert sorted(t.points()) == sorted(live)
        for _ in range(3):
            x0, x1 = sorted(rng.randrange(-1, span + 1) for _ in range(2))
            y0, y1 = sorted(rng.randrange(-1, span + 1) for _ in range(2))
            want = sorted(q for q in live if x0 <= q[0] <= x1 and y0 <= q[1] <= y1)
            assert sorted(t.range_search((x0, y0), (x1, y1))) == want

What the next level needs

  • the tests did not run in this build

Read the full page about k-d-B tree