Algorithm Search⌘K

R*-tree in Python

R-tree variant with improved insertion and split heuristics that minimize overlap and perimeter.

File
r_star_tree.py, 283 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
r_star_tree.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/r-star-tree/. The authors are listed in prov/DENQWV.jsonl.
"""R*-tree of Beckmann, Kriegel, Schneider and Seeger (1990), in the plane.

Rectangles are closed, (xmin, ymin, xmax, ymax). A leaf entry is
(rect, id); an internal entry is (rect, child). The rules that differ from
Guttman's R-tree: ChooseSubtree by least overlap enlargement above the
leaves, forced reinsertion of the p entries farthest from the node center
once per level and insertion, and the split by margin, then overlap.
"""

from __future__ import annotations

from typing import Hashable, List, Optional, Set, Tuple

Rect = Tuple[float, float, float, float]


def area(r: Rect) -> float:
    return (r[2] - r[0]) * (r[3] - r[1])


def margin(r: Rect) -> float:
    return (r[2] - r[0]) + (r[3] - r[1])


def union(a: Rect, b: Rect) -> Rect:
    return (min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3]))


def overlap(a: Rect, b: Rect) -> float:
    w = min(a[2], b[2]) - max(a[0], b[0])
    h = min(a[3], b[3]) - max(a[1], b[1])
    return w * h if w > 0 and h > 0 else 0.0


def intersects(a: Rect, b: Rect) -> bool:
    return a[0] <= b[2] and b[0] <= a[2] and a[1] <= b[3] and b[1] <= a[3]


def mbr(rects: List[Rect]) -> Rect:
    r = rects[0]
    for s in rects[1:]:
        r = union(r, s)
    return r


class Node:
    __slots__ = ("leaf", "entries", "parent")

    def __init__(self, leaf: bool) -> None:
        self.leaf = leaf
        self.entries: list = []
        self.parent: Optional[Node] = None

    def rect(self) -> Rect:
        return mbr([e[0] for e in self.entries])


def _sorted_by(entries: list, side: int) -> list:
    """Sort by one side of the rectangles (0, 1: lower x, y; 2, 3: upper),
    with the opposite side on the same axis as the second key."""
    other = side + 2 if side < 2 else side - 2
    return sorted(entries, key=lambda e: (e[0][side], e[0][other]))


def rstar_split(entries: list, m: int) -> Tuple[list, list]:
    """Split by the axis of least margin sum, then the distribution of least
    overlap (ties: least area sum). Entries are (rect, payload)."""
    n = len(entries)
    best_axis, best_s = 0, None
    for axis in (0, 1):  # @line L10
        s = 0.0
        for side in (axis, axis + 2):
            srt = _sorted_by(entries, side)
            for k in range(m, n - m + 1):
                s += margin(mbr([e[0] for e in srt[:k]])) + margin(mbr([e[0] for e in srt[k:]]))
        if best_s is None or s < best_s:  # @line L11
            best_axis, best_s = axis, s
    best = None
    for side in (best_axis, best_axis + 2):  # @line L12
        srt = _sorted_by(entries, side)
        for k in range(m, n - m + 1):
            r1 = mbr([e[0] for e in srt[:k]])
            r2 = mbr([e[0] for e in srt[k:]])
            key = (overlap(r1, r2), area(r1) + area(r2))
            if best is None or key < best[0]:
                best = (key, srt[:k], srt[k:])
    assert best is not None
    return best[1], best[2]  # @line L13


class RStarTree:
    def __init__(self, max_entries: int = 4, min_entries: Optional[int] = None,
                 reinsert: Optional[int] = None) -> None:
        M = max_entries
        m = max(1, (2 * M) // 5) if min_entries is None else min_entries
        p = max(1, (3 * M) // 10) if reinsert is None else reinsert
        if M < 2 or not 1 <= m <= M // 2 or not 0 <= p <= M - m:
            raise ValueError("need M >= 2, 1 <= m <= M // 2 and 0 <= p <= M - m")
        self.M, self.m, self.p = M, m, p
        self.root = Node(True)
        self.size = 0
        self._done: Set[int] = set()

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

    def search(self, q: Rect) -> List[Hashable]:
        out: List[Hashable] = []
        stack = [self.root]
        while stack:
            u = stack.pop()
            for r, x in u.entries:
                if intersects(r, q):  # @line L17
                    if u.leaf:
                        out.append(x)
                    else:
                        stack.append(x)
        return out

    # ---- insertion ----
    def _choose(self, rect: Rect, level: int) -> Node:
        u, h = self.root, self.height()
        while h > level:
            es = u.entries
            best = None
            for k, (r, _) in enumerate(es):
                big = union(r, rect)
                d_area = area(big) - area(r)
                if h == 1:  # children are leaves: least overlap enlargement
                    d_ov = sum(overlap(big, es[j][0]) - overlap(r, es[j][0])
                               for j in range(len(es)) if j != k)
                    key = (d_ov, d_area, area(r))
                else:
                    key = (d_area, area(r), 0.0)
                if best is None or key < best[0]:
                    best = (key, k)
            assert best is not None
            u = es[best[1]][1]  # @line L2
            h -= 1
        return u

    def _fix_up(self, u: Node) -> None:
        while u.parent is not None:
            p = u.parent
            for k, (_, c) in enumerate(p.entries):
                if c is u:
                    p.entries[k] = (u.rect(), u)
                    break
            u = p

    def _insert_at(self, entry, level: int) -> None:
        u = self._choose(entry[0], level)
        if not u.leaf:
            entry[1].parent = u
        u.entries.append(entry)  # @line L3
        h = level
        while u is not None:  # @line L4
            if len(u.entries) > self.M:
                if u.parent is not None and h not in self._done and self.p > 0:  # @line L5
                    self._done.add(h)
                    c = u.rect()
                    cx, cy = (c[0] + c[2]) / 2, (c[1] + c[3]) / 2

                    def dist(e):
                        r = e[0]
                        return ((r[0] + r[2]) / 2 - cx) ** 2 + ((r[1] + r[3]) / 2 - cy) ** 2
                    order = sorted(range(len(u.entries)), key=lambda k: -dist(u.entries[k]))
                    out = set(order[: self.p])
                    removed = [u.entries[k] for k in order[: self.p]]
                    u.entries = [e for k, e in enumerate(u.entries) if k not in out]
                    self._fix_up(u)
                    for e in reversed(removed):  # close reinsert: nearest first
                        self._insert_at(e, h)
                    return
                g1, g2 = rstar_split(u.entries, self.m)  # @line L6
                v = Node(u.leaf)
                u.entries, v.entries = g1, g2
                if not u.leaf:
                    for _, ch in v.entries:
                        ch.parent = v
                    for _, ch in u.entries:
                        ch.parent = u
                if u.parent is None:  # @line L7
                    root = Node(False)
                    root.entries = [(u.rect(), u), (v.rect(), v)]
                    u.parent = v.parent = root
                    self.root = root
                    return
                v.parent = u.parent
                u.parent.entries.append((v.rect(), v))  # @line L8
            p = u.parent
            if p is not None:  # @line L9
                for k, (_, ch) in enumerate(p.entries):
                    if ch is u:
                        p.entries[k] = (u.rect(), u)
                        break
            u, h = p, h + 1

    def insert(self, oid: Hashable, rect: Rect) -> None:
        if rect[0] > rect[2] or rect[1] > rect[3]:
            raise ValueError("empty rectangle")
        self._done = set()  # @line L1
        self._insert_at((tuple(rect), oid), 0)
        self.size += 1

    # ---- deletion ----
    def delete(self, oid: Hashable, rect: Rect) -> bool:
        rect = tuple(rect)
        leaf = None
        stack = [self.root]
        while stack and leaf is None:  # @line L14
            u = stack.pop()
            for r, x in u.entries:
                if u.leaf:
                    if x == oid and r == rect:
                        leaf = u
                        break
                elif r[0] <= rect[0] and r[1] <= rect[1] and rect[2] <= r[2] and rect[3] <= r[3]:
                    stack.append(x)
        if leaf is None:
            return False
        leaf.entries = [e for e in leaf.entries if not (e[1] == oid and e[0] == rect)]
        self.size -= 1
        orphans = []
        u, h = leaf, 0
        while u.parent is not None:  # @line L15
            p = u.parent
            if len(u.entries) < self.m:
                p.entries = [e for e in p.entries if e[1] is not u]
                orphans.append((u, h))
            else:
                for k, (_, c) in enumerate(p.entries):
                    if c is u:
                        p.entries[k] = (u.rect(), u)
            u, h = p, h + 1
        for node, nh in orphans:
            for e in node.entries:
                self._done = set()
                self._insert_at(e, nh)
        while not self.root.leaf and len(self.root.entries) == 1:  # @line L16
            self.root = self.root.entries[0][1]
            self.root.parent = None
        return True

    # ---- helpers ----
    def items(self) -> List[Tuple[Hashable, Rect]]:
        out = []
        stack = [self.root]
        while stack:
            u = stack.pop()
            for r, x in u.entries:
                if u.leaf:
                    out.append((x, r))
                else:
                    stack.append(x)
        return out

    def check_invariant(self) -> None:
        depths = set()
        count = 0
        stack = [(self.root, 0)]
        while stack:
            u, d = stack.pop()
            assert len(u.entries) <= self.M
            if u is not self.root:
                assert len(u.entries) >= self.m
            elif not u.leaf:
                assert len(u.entries) >= 2
            if u.leaf:
                depths.add(d)
                count += len(u.entries)
            else:
                for r, c in u.entries:
                    assert c.parent is u and c.entries and r == c.rect()
                    stack.append((c, d + 1))
        assert len(depths) <= 1
        assert count == self.size
test_r_star_tree.py 81 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/r-star-tree/. The authors are listed in prov/DENQWV.jsonl.
import random

import pytest

from r_star_tree import RStarTree, intersects, rstar_split


E = {"a": (0, 0, 2, 2), "b": (1, 5, 3, 7), "c": (6, 0, 8, 1), "d": (7, 6, 9, 9),
     "e": (2, 1, 3, 3), "f": (6, 7, 7, 8), "g": (1, 3, 2, 4), "h": (3, 6, 4, 8)}


def leaves(t):
    nodes = [t.root] if t.root.leaf else [c for _, c in t.root.entries]
    return sorted((u.rect(), "".join(sorted(x for _, x in u.entries))) for u in nodes)


def test_page_example():
    t = RStarTree(4, 2, 1)
    for k in "abcde":
        t.insert(k, E[k])
    assert leaves(t) == [((0, 0, 3, 7), "abe"), ((6, 0, 9, 9), "cd")]
    for k in "fg":
        t.insert(k, E[k])
    assert leaves(t) == [((0, 0, 3, 7), "abeg"), ((6, 0, 9, 9), "cdf")]
    t.insert("h", E["h"])  # reinsert h once, then split on y
    assert leaves(t) == [((0, 0, 3, 4), "aeg"), ((1, 5, 4, 8), "bh"), ((6, 0, 9, 9), "cdf")]
    t.check_invariant()
    assert sorted(t.search((2.5, 2.5, 7, 7))) == list("bdefh")
    assert t.search((4.5, 1.5, 5.5, 4.5)) == []
    assert t.delete("c", E["c"])
    assert leaves(t)[2] == ((6, 6, 9, 9), "df")
    assert t.delete("d", E["d"])
    assert leaves(t) == [((0, 0, 3, 4), "aeg"), ((1, 5, 7, 8), "bfh")]
    assert not t.delete("d", E["d"])


def test_split_rule():
    es = [(E[k], k) for k in "abegh"]
    g1, g2 = rstar_split(es, 2)
    assert {x for _, x in g1} == set("aeg") and {x for _, x in g2} == set("bh")


def test_edge_cases():
    t = RStarTree(4, 2, 1)
    assert t.search((0, 0, 1, 1)) == [] and not t.delete(1, (0, 0, 1, 1))
    for k in range(30):
        t.insert(k, (5, 5, 5, 5))  # all equal points
    t.check_invariant()
    assert len(t.search((5, 5, 5, 5))) == 30
    for k in range(30):
        assert t.delete(k, (5, 5, 5, 5))
    t.check_invariant()
    assert t.root.leaf and not t.root.entries
    with pytest.raises(ValueError):
        RStarTree(4, 3)
    with pytest.raises(ValueError):
        t.insert(0, (1, 1, 0, 0))


def test_random_against_brute_force():
    rng = random.Random(1990)
    for trial in range(500):
        M = rng.randint(3, 8)
        t = RStarTree(M)
        live = {}
        for k in range(rng.randint(0, 60)):
            x, y = rng.uniform(0, 100), rng.uniform(0, 100)
            r = (x, y, x + rng.uniform(0, 10), y + rng.uniform(0, 10))
            t.insert(k, r)
            live[k] = r
            if rng.random() < 0.25 and live:
                d = rng.choice(list(live))
                assert t.delete(d, live.pop(d))
        t.check_invariant()
        assert sorted(t.items()) == sorted(live.items())
        for _ in range(4):
            x, y = rng.uniform(-5, 100), rng.uniform(-5, 100)
            q = (x, y, x + rng.uniform(0, 30), y + rng.uniform(0, 30))
            assert sorted(t.search(q)) == sorted(k for k, r in live.items() if intersects(q, r))

What the next level needs

  • the tests did not run in this build

Read the full page about R*-tree