R+-tree in Python
R-tree variant that clips objects so sibling regions never overlap, at the cost of duplicated entries.
- File
r_plus_tree.py, 230 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/r-plus-tree/. The authors are listed in prov/SFMMBA.jsonl.
"""An R+-tree in the plane, in the form of a k-d-B tree with clipping.
Every node has a half-open region [x0, x1) x [y0, y1); the root region is
the whole plane. The regions of the children of a node partition the
region of the node, so sibling regions never overlap. Objects are closed
rectangles (xmin, ymin, xmax, ymax). A leaf stores every object that
meets its region, so an object that crosses a cut is stored in several
leaves. A query reports an object once, in the leaf whose region holds
the lower left corner of the intersection of the query and the object.
Capacity M is soft: a node is split only by a cut that makes both sides
smaller than the node. If no such cut exists (more than M objects meet at
one point), the node stays over full, as Sellis et al. note.
"""
from __future__ import annotations
from typing import Hashable, List, Optional, Tuple
INF = float("inf")
Rect = Tuple[float, float, float, float]
def meets(b: Rect, reg: Rect) -> bool:
"""Closed rectangle b meets half-open region reg."""
return b[0] < reg[2] and b[2] >= reg[0] and b[1] < reg[3] and b[3] >= reg[1]
def contains_point(reg: Rect, x: float, y: float) -> bool:
return reg[0] <= x < reg[2] and reg[1] <= y < reg[3]
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]
class Node:
__slots__ = ("leaf", "region", "items", "children")
def __init__(self, leaf: bool, region: Rect) -> None:
self.leaf = leaf
self.region = region
self.items: List[Tuple[Hashable, Rect]] = [] # leaves
self.children: List[Node] = [] # internal nodes
def size(self) -> int:
return len(self.items) if self.leaf else len(self.children)
def halves(reg: Rect, axis: int, c: float) -> Tuple[Rect, Rect]:
if axis == 0:
return (reg[0], reg[1], c, reg[3]), (c, reg[1], reg[2], reg[3])
return (reg[0], reg[1], reg[2], c), (reg[0], c, reg[2], reg[3])
def cut_node(u: Node, axis: int, c: float) -> Tuple[Node, Node]:
"""Split u by the line axis = c into two nodes; children that cross the
line are cut too, down to the leaves."""
lr, rr = halves(u.region, axis, c)
left, right = Node(u.leaf, lr), Node(u.leaf, rr)
if u.leaf:
left.items = [it for it in u.items if meets(it[1], lr)] # @line L12
right.items = [it for it in u.items if meets(it[1], rr)]
return left, right
for ch in u.children:
lo, hi = ch.region[axis], ch.region[axis + 2]
if hi <= c:
left.children.append(ch)
elif lo >= c:
right.children.append(ch)
else:
a, b = cut_node(ch, axis, c) # @line L13
left.children.append(a)
right.children.append(b)
return left, right
def choose_cut(u: Node, clean: bool = False) -> Optional[Tuple[int, float]]:
"""The cut that makes the larger side smallest (ties: fewer copies), among
the sides of the entries; None if no cut makes both sides smaller.
With clean=True, only cuts that cross no entry are allowed."""
best = None
n = u.size()
for axis in (0, 1): # @line L10
lo_r, hi_r = u.region[axis], u.region[axis + 2]
if u.leaf:
cands = {b[axis] for _, b in u.items} | {b[axis + 2] for _, b in u.items}
else:
cands = {ch.region[axis] for ch in u.children} | {ch.region[axis + 2] for ch in u.children}
for c in sorted(cands):
if not lo_r < c < hi_r:
continue
if u.leaf:
nl = sum(1 for _, b in u.items if b[axis] < c)
nr = sum(1 for _, b in u.items if b[axis + 2] >= c)
else:
nl = sum(1 for ch in u.children if ch.region[axis] < c)
nr = sum(1 for ch in u.children if ch.region[axis + 2] > c)
if max(nl, nr) >= n or min(nl, nr) == 0:
continue
if clean and nl + nr > n:
continue
key = (max(nl, nr), nl + nr, axis, c)
if best is None or key < best:
best = key
return None if best is None else (best[2], best[3]) # @line L11
class RPlusTree:
def __init__(self, max_entries: int = 4) -> None:
if max_entries < 2:
raise ValueError("M must be at least 2")
self.M = max_entries
self.root = Node(True, (-INF, -INF, INF, INF))
self.n = 0
def search(self, q: Rect) -> List[Hashable]:
out: List[Hashable] = []
stack = [self.root]
while stack:
u = stack.pop()
if u.leaf:
for oid, b in u.items: # @line L3
if intersects(q, b) and contains_point(
u.region, max(q[0], b[0]), max(q[1], b[1])): # @line L4
out.append(oid)
else:
for ch in u.children: # @line L1
if meets(q, ch.region): # @line L2
stack.append(ch)
return out
def insert(self, oid: Hashable, b: Rect) -> None:
if b[0] > b[2] or b[1] > b[3]:
raise ValueError("empty rectangle")
b = tuple(b)
pieces = self._insert_at(self.root, oid, b)
while len(pieces) > 1: # @line L8
root = Node(False, self.root.region)
root.children = pieces
self.root = root
pieces = self._split_all(root, clean=True)
self.root = pieces[0]
self.n += 1
def _insert_at(self, u: Node, oid: Hashable, b: Rect) -> List[Node]:
"""Add b to every leaf below u that it meets; return the nodes that replace u."""
if u.leaf:
u.items.append((oid, b)) # @line L6
else:
new: List[Node] = []
for ch in u.children:
if meets(b, ch.region): # @line L5
new.extend(self._insert_at(ch, oid, b))
else:
new.append(ch)
u.children = new
return self._split_all(u) # @line L7
def _split_all(self, u: Node, clean: bool = False) -> List[Node]:
"""Cut u while it has more than M entries and a good cut exists."""
if u.size() <= self.M:
return [u]
cut = choose_cut(u, clean) # @line L9
if cut is None:
return [u]
a, c = cut_node(u, *cut)
return self._split_all(a, clean) + self._split_all(c, clean)
def delete(self, oid: Hashable, b: Rect) -> bool:
b = tuple(b)
found = False
stack = [self.root]
while stack: # @line L14
u = stack.pop()
if u.leaf:
k = len(u.items)
u.items = [it for it in u.items if it != (oid, b)]
found |= len(u.items) != k
else:
stack.extend(ch for ch in u.children if meets(b, ch.region))
if found:
self.n -= 1
return found
# ---- helpers ----
def leaves(self) -> List[Node]:
out, stack = [], [self.root]
while stack:
u = stack.pop()
if u.leaf:
out.append(u)
else:
stack.extend(u.children)
return out
def copies(self) -> int:
return sum(len(u.items) for u in self.leaves())
def height(self) -> int:
h, u = 0, self.root
while not u.leaf:
u = u.children[0]
h += 1
return h
def check_invariant(self, objects: dict) -> None:
"""objects: id -> rect of the stored set."""
depths = set()
stack = [(self.root, 0)]
while stack:
u, d = stack.pop()
if u.leaf:
depths.add(d)
want = sorted((k, r) for k, r in objects.items() if meets(r, u.region))
assert sorted(u.items) == want, "leaf must hold the objects that meet it"
continue
assert u.children
area = 0.0
for ch in u.children:
assert u.region[0] <= ch.region[0] < ch.region[2] <= u.region[2]
assert u.region[1] <= ch.region[1] < ch.region[3] <= u.region[3]
stack.append((ch, d + 1))
for i, a in enumerate(u.children): # disjoint siblings
for b in u.children[i + 1:]:
ra, rb = a.region, b.region
assert not (ra[0] < rb[2] and rb[0] < ra[2] and ra[1] < rb[3] and rb[1] < ra[3])
assert len(depths) == 1
test_r_plus_tree.py 94 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/r-plus-tree/. The authors are listed in prov/SFMMBA.jsonl.
import random
import pytest
from r_plus_tree import RPlusTree, choose_cut, intersects
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),
"i": (2, 2, 7, 4)}
def leaves(t):
return sorted((u.region, "".join(sorted(x for x, _ in u.items))) for u in t.leaves())
def test_page_example():
inf = float("inf")
t = RPlusTree(4)
for k in "abcde":
t.insert(k, E[k])
assert leaves(t) == [((-inf, -inf, 6, inf), "abe"), ((6, -inf, inf, inf), "cd")]
for k in "fgh":
t.insert(k, E[k])
assert leaves(t) == [((-inf, -inf, 6, 5), "aeg"), ((-inf, 5, 6, inf), "bh"),
((6, -inf, inf, inf), "cdf")]
t.insert("i", E["i"])
assert leaves(t) == [((-inf, -inf, 6, 5), "aegi"), ((-inf, 5, 6, inf), "bh"),
((6, -inf, inf, inf), "cdfi")]
assert t.copies() == 10 and t.n == 9
t.check_invariant(E)
assert sorted(t.search((2.5, 2.5, 7, 7))) == list("bdefhi")
assert t.search((4.5, 4.5, 5.5, 5.5)) == []
assert sorted(t.search((6.5, 2.5, 6.6, 2.6))) == ["i"]
assert t.delete("i", E["i"])
assert t.copies() == 8
assert not t.delete("i", E["i"])
def test_empty_and_errors():
t = RPlusTree(2)
assert t.search((0, 0, 1, 1)) == []
with pytest.raises(ValueError):
RPlusTree(1)
with pytest.raises(ValueError):
t.insert("x", (1, 0, 0, 1))
def test_same_point_overflows():
t = RPlusTree(2)
objs = {k: (1, 1, 1, 1) for k in range(5)}
for k, r in objs.items():
t.insert(k, r)
assert len(t.leaves()) == 1 and t.copies() == 5 # no cut can help
assert sorted(t.search((0, 0, 2, 2))) == list(range(5))
t.check_invariant(objs)
def brute(objs, q):
return sorted(k for k, r in objs.items() if intersects(q, r))
def rand_rect(rng, size):
x, y = rng.randint(0, 30), rng.randint(0, 30)
return (x, y, x + rng.randint(0, size), y + rng.randint(0, size))
def test_random_against_brute_force():
rng = random.Random(7)
for trial in range(600):
M = rng.randint(2, 6)
t = RPlusTree(M)
objs = {}
for k in range(rng.randint(0, 40)):
r = rand_rect(rng, rng.choice([0, 3, 8]))
t.insert(k, r)
objs[k] = r
for k in rng.sample(sorted(objs), len(objs) // 3):
assert t.delete(k, objs.pop(k))
t.check_invariant(objs)
assert t.n == len(objs)
stack = [t.root]
while stack: # a node stays over full only if no cut helps
u = stack.pop()
if u.size() > M:
assert u.leaf and choose_cut(u) is None
if not u.leaf:
stack.extend(u.children)
for _ in range(5):
q = rand_rect(rng, 12)
res = t.search(q)
assert len(res) == len(set(res))
assert sorted(res) == brute(objs, q)
What the next level needs
- the tests did not run in this build