Fusion tree in Python
Integer search tree achieving O(log n / log w) time per operation on w-bit words using word-parallel tricks.
- File
fusion_tree.py, 187 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/fusion-tree/. The authors are listed in prov/SD1PX6.jsonl.
"""A static fusion tree for predecessor search on w-bit integers.
Each node holds at most B keys. A node keeps the distinguishing bit positions
of its keys, the exact sketches of the keys, and one packed word with all the
sketches, so that the rank of a sketch among the node's sketches is found with
one subtraction, one AND and one population count. This version computes the
exact sketch with a loop over the positions; the structure of Fredman and
Willard computes an approximate sketch with one multiplication instead.
"""
from bisect import bisect_right
from typing import List, Optional
def msb(x: int) -> int:
"""Position of the highest set bit of x > 0."""
return x.bit_length() - 1
class FusionNode:
__slots__ = ("keys", "children", "pos", "r", "width", "packed", "mult", "high")
def __init__(self, keys: List[int]) -> None:
self.keys = keys
self.children: List[Optional["FusionNode"]] = [None] * (len(keys) + 1)
d = {msb(a ^ b) for a, b in zip(keys, keys[1:])}
self.pos = sorted(d, reverse=True) # distinguishing positions, high to low
self.r = len(self.pos)
self.width = self.r + 1 # one sketch plus a sentinel bit per field
self.mult = 0
self.high = 0
self.packed = 0
for j, x in enumerate(keys):
shift = j * self.width
self.mult |= 1 << shift
self.high |= 1 << (shift + self.r)
self.packed |= ((1 << self.r) | self.sketch(x)) << shift
def sketch(self, z: int) -> int:
s = 0
for b in self.pos:
s = (s << 1) | ((z >> b) & 1)
return s
def packed_rank(self, s: int) -> int:
"""The number of keys whose sketch is less than s, for 0 <= s <= 2^r."""
q = s * self.mult # @line L9
t = (self.packed - q) & self.high # @line L10
return len(self.keys) - bin(t).count("1") # @line L11
def node_pred(self, q: int) -> int:
"""Index of the largest key <= q in this node, or -1."""
keys = self.keys
s = self.sketch(q) # @line L1
i = self.packed_rank(s) # @line L2
y = None # @line L3
for j in (i - 1, i):
if 0 <= j < len(keys) and (y is None or (keys[j] ^ q) < (y ^ q)):
y = keys[j]
if y == q: # @line L4
return i # keys[i - 1] has a smaller sketch, so q is keys[i]
p = msb(q ^ y) # @line L5
if (q >> p) & 1: # @line L6
e = ((q >> (p + 1)) << (p + 1)) | ((1 << p) - 1)
rank = self.packed_rank(self.sketch(e) + 1)
else: # @line L7
e = ((q >> (p + 1)) << (p + 1)) | (1 << p)
rank = self.packed_rank(self.sketch(e))
return rank - 1 # @line L8
class FusionTree:
def __init__(self, keys, w: int = 64, b: int = 8) -> None:
if b < 1 or b * b > w:
raise ValueError("need 1 <= b and b * b <= w so that a node fits in a word")
self.w, self.b = w, b
a = sorted(set(keys))
for x in a:
if not 0 <= x < (1 << w):
raise ValueError("keys must be w-bit unsigned integers")
self.n = len(a)
self.root = self._build(a)
def _build(self, a: List[int]) -> Optional[FusionNode]:
if not a:
return None
root_holder: List[Optional[FusionNode]] = [None]
stack = [(a, None, 0)]
while stack:
part, parent, slot = stack.pop()
m = len(part)
if m <= self.b:
node = FusionNode(part)
rest: List[List[int]] = []
else:
k = self.b
idx = [(j + 1) * (m + 1) // (k + 1) - 1 for j in range(k)]
node = FusionNode([part[t] for t in idx])
bounds = [-1] + idx + [m]
rest = [part[bounds[j] + 1:bounds[j + 1]] for j in range(k + 1)]
if parent is None:
root_holder[0] = node
else:
parent.children[slot] = node
for j, sub in enumerate(rest):
if sub:
stack.append((sub, node, j))
return root_holder[0]
def __len__(self) -> int:
return self.n
def predecessor(self, q: int) -> Optional[int]:
"""The largest key <= q, or None."""
v = self.root # @line L12
best = None
while v is not None: # @line L13
j = v.node_pred(q) # @line L14
if j >= 0 and v.keys[j] == q: # @line L15
return q
if j >= 0: # @line L16
best = v.keys[j]
v = v.children[j + 1] # @line L17
return best # @line L18
def successor(self, q: int) -> Optional[int]:
"""The smallest key >= q, or None, with the same node ranks."""
v = self.root
best = None
while v is not None:
j = v.node_pred(q)
if j >= 0 and v.keys[j] == q:
return q
if j + 1 < len(v.keys):
best = v.keys[j + 1]
v = v.children[j + 1]
return best
def contains(self, q: int) -> bool:
return self.predecessor(q) == q
def keys(self) -> List[int]:
"""All keys in increasing order, by an in-order walk with an explicit stack."""
out: List[int] = []
stack: List[tuple] = [(False, self.root)]
while stack:
is_key, item = stack.pop()
if is_key:
out.append(item)
elif item is not None:
v = item
for j in range(len(v.keys), 0, -1):
stack.append((False, v.children[j]))
stack.append((True, v.keys[j - 1]))
stack.append((False, v.children[0]))
return out
def height(self) -> int:
h, v = 0, self.root
while v is not None:
h, v = h + 1, v.children[0]
return h
def check_invariant(self) -> None:
stack = [(self.root, -1, 1 << self.w)]
count = 0
while stack:
v, lo, hi = stack.pop()
if v is None:
continue
k = v.keys
assert 1 <= len(k) <= self.b
assert all(lo < x < hi for x in k)
assert all(a < b for a, b in zip(k, k[1:]))
sk = [v.sketch(x) for x in k]
assert all(a < b for a, b in zip(sk, sk[1:])), "sketches keep the order"
assert len(k) * v.width <= self.w, "the packed node fits in a word"
for s in range((1 << v.r) + 1):
assert v.packed_rank(s) == sum(1 for t in sk if t < s)
assert len(v.children) == len(k) + 1
count += len(k)
bounds = [lo] + k + [hi]
for j, c in enumerate(v.children):
stack.append((c, bounds[j], bounds[j + 1]))
assert count == self.n
test_fusion_tree.py 88 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/fusion-tree/. The authors are listed in prov/SD1PX6.jsonl.
import bisect
import random
import pytest
from fusion_tree import FusionNode, FusionTree
def oracle_pred(a, q):
i = bisect.bisect_right(a, q)
return a[i - 1] if i else None
def oracle_succ(a, q):
i = bisect.bisect_left(a, q)
return a[i] if i < len(a) else None
def test_node_example() -> None:
v = FusionNode([2, 9, 12, 26])
assert v.pos == [4, 3, 2]
assert [v.sketch(x) for x in v.keys] == [0, 2, 3, 6]
assert v.packed_rank(3) == 2
assert v.sketch(13) == 3 and v.packed_rank(v.sketch(13)) == 2
assert v.keys[v.node_pred(13)] == 12 # the sketch rank alone would give 9
assert v.keys[v.node_pred(21)] == 12
assert v.node_pred(1) == -1 and v.keys[v.node_pred(31)] == 26
def test_small_and_empty() -> None:
t = FusionTree([], w=16, b=4)
t.check_invariant()
assert t.predecessor(5) is None and t.successor(5) is None and t.keys() == []
t = FusionTree([7], w=16, b=4)
assert t.predecessor(6) is None and t.predecessor(7) == 7 and t.successor(8) is None
with pytest.raises(ValueError):
FusionTree([1], w=16, b=5)
with pytest.raises(ValueError):
FusionTree([1 << 16], w=16, b=4)
def test_exhaustive_small_universe() -> None:
rng = random.Random(11)
for _ in range(200):
a = sorted(rng.sample(range(64), rng.randrange(1, 30)))
t = FusionTree(a, w=16, b=4)
t.check_invariant()
assert t.keys() == a
for q in range(64):
assert t.predecessor(q) == oracle_pred(a, q)
assert t.successor(q) == oracle_succ(a, q)
def test_random_against_bisect() -> None:
rng = random.Random(20251104)
for case in range(600):
w = rng.choice([16, 32, 64])
b = rng.choice([2, 3, 4]) if w == 16 else rng.choice([4, 5, 8]) if w == 64 else rng.choice([3, 4, 5])
n = rng.randrange(1, 300)
if rng.random() < 0.3:
base = rng.randrange(1 << (w - 8))
a = sorted({(base << 8) | rng.randrange(256) for _ in range(n)})
else:
a = sorted({rng.randrange(1 << w) for _ in range(n)})
sa = set(a)
t = FusionTree(a, w=w, b=b)
if case % 20 == 0:
t.check_invariant()
assert t.keys() == a
for _ in range(30):
if rng.random() < 0.3:
q = rng.choice(a) + rng.choice([-1, 0, 1])
q = min(max(q, 0), (1 << w) - 1)
else:
q = rng.randrange(1 << w)
assert t.predecessor(q) == oracle_pred(a, q)
assert t.successor(q) == oracle_succ(a, q)
assert t.contains(q) == (q in sa)
def test_height() -> None:
a = list(range(0, 3 * 10 ** 5, 3))
t = FusionTree(a, w=64, b=8)
assert t.height() <= 7
assert t.predecessor(299999) == 299997 and t.predecessor(300000 * 7) == 299997
What the next level needs
- the tests did not run in this build