HAT-trie in Python
Cache-conscious trie whose leaves are hash-table-backed array buckets.
- File
hat_trie.py, 229 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/hat-trie/. The authors are listed in prov/HZ5P9K.jsonl.
from typing import Generic, TypeVar
V = TypeVar("V")
def _hash(s: str) -> int:
"""Shift-add-xor string hash, deterministic across runs."""
h = 220373
for c in s:
h ^= ((h << 5) + (h >> 2) + ord(c)) & 0xFFFFFFFF
return h
class Container(Generic[V]):
"""An array hash table: each slot is one array of (suffix, value) entries.
A pure container hangs from one pointer and stores the suffixes without the
symbol of that pointer. A hybrid container hangs from the pointers lo .. hi - 1
of its parent and keeps the first symbol of each suffix.
"""
def __init__(self, slots: int, pure: bool, lo: int, hi: int) -> None:
self.slots: list[list[tuple[str, V]]] = [[] for _ in range(slots)]
self.pure = pure
self.lo, self.hi = lo, hi
self.size = 0
def find(self, t: str) -> V | None:
for k, v in self.slots[_hash(t) % len(self.slots)]: # @line L5
if k == t:
return v
return None
def put(self, t: str, v: V) -> bool:
"""Insert or update t; return True if t is new."""
slot = self.slots[_hash(t) % len(self.slots)]
for i, (k, _) in enumerate(slot):
if k == t:
slot[i] = (t, v)
return False
slot.append((t, v)) # @line L7
self.size += 1
return True
def items(self) -> list[tuple[str, V]]:
return [e for slot in self.slots for e in slot]
class Node(Generic[V]):
__slots__ = ("child", "value")
def __init__(self, r: int) -> None:
self.child: list[Node[V] | Container[V]] = []
self.value: V | None = None # the value of the key that ends at this node
class HatTrie(Generic[V]):
"""A HAT-trie over the symbols with codes 0 .. r - 1.
Trie nodes have an array of r pointers; leaves are array hash containers of at
most `limit` entries. Values must not be None.
"""
def __init__(self, r: int = 128, limit: int = 1024, slots: int = 64) -> None:
if limit < 1 or slots < 1:
raise ValueError("the limit and the number of slots must be positive")
self.r, self.limit, self.nslots = r, limit, slots
self.root: Node[V] = self._node()
self.n = 0
self.bursts = 0
def _node(self) -> Node[V]:
x: Node[V] = Node(self.r)
h: Container[V] = Container(self.nslots, False, 0, self.r)
x.child = [h] * self.r
return x
def _code(self, c: str) -> int:
k = ord(c)
if k >= self.r:
raise ValueError(f"symbol {c!r} is outside the alphabet")
return k
def _descend(self, s: str) -> tuple[Node[V], int]:
"""The deepest trie node on the path of s and the number of symbols it consumed."""
x, d = self.root, 0 # @line L1
while d < len(s): # @line L2
y = x.child[self._code(s[d])] # @line L3
if not isinstance(y, Node):
break
x, d = y, d + 1
return x, d
def get(self, s: str) -> V | None:
x, d = self._descend(s)
if d == len(s):
return x.value
y = x.child[self._code(s[d])]
assert isinstance(y, Container)
return y.find(s[d + 1:] if y.pure else s[d:]) # @line L4
def put(self, s: str, v: V) -> None:
if v is None:
raise ValueError("a value must not be None")
x, d = self._descend(s) # @line L6
if d == len(s):
if x.value is None:
self.n += 1
x.value = v
return
y = x.child[self._code(s[d])]
assert isinstance(y, Container)
if y.put(s[d + 1:] if y.pure else s[d:], v):
self.n += 1
if y.size > self.limit: # @line L8
self._burst(x, y)
def _burst(self, x: Node[V], y: Container[V]) -> None:
work = [(x, y)]
while work: # @line L12
x, y = work.pop()
if y.size <= self.limit:
continue
self.bursts += 1
if y.pure: # @line L9
z = self._node()
x.child[y.lo] = z
h = z.child[0]
assert isinstance(h, Container)
for t, v in y.items():
if t == "":
z.value = v
else:
h.put(t, v)
work.append((z, h))
continue
counts = [0] * (y.hi - y.lo)
for t, _ in y.items():
counts[ord(t[0]) - y.lo] += 1
if y.hi - y.lo == 1: # @line L11
p: Container[V] = Container(self.nslots, True, y.lo, y.hi)
for t, v in y.items():
p.put(t[1:], v)
x.child[y.lo] = p
work.append((x, p))
continue
used = [y.lo + i for i, k in enumerate(counts) if k]
if len(used) == 1: # @line L10
c = used[0] # one symbol only: cut out its range
cuts = [y.lo, c, c + 1, y.hi]
else:
best, split, left = None, used[1], 0
for s in range(used[0] + 1, used[-1] + 1):
left += counts[s - 1 - y.lo]
gap = abs(2 * left - y.size)
if best is None or gap < best:
best, split = gap, s
cuts = [y.lo, split, y.hi]
for lo, hi in zip(cuts, cuts[1:]):
if lo == hi:
continue
part: Container[V] = Container(self.nslots, False, lo, hi)
for t, v in y.items():
if lo <= ord(t[0]) < hi:
part.put(t, v)
for c in range(lo, hi):
x.child[c] = part
work.append((x, part))
def size(self) -> int:
return self.n
def keys(self) -> list[str]:
"""All keys in sorted order; each container is sorted when it is reached."""
out: list[str] = []
stack: list[tuple[Node[V] | None, str, Container[V] | None]] = [(self.root, "", None)]
while stack:
x, prefix, c = stack.pop()
if c is not None:
out.extend(sorted(prefix + t for t, _ in c.items()))
continue
assert x is not None
if x.value is not None:
out.append(prefix)
entries: list[tuple[Node[V] | None, str, Container[V] | None]] = []
k = 0
while k < self.r:
y = x.child[k]
if isinstance(y, Node):
entries.append((y, prefix + chr(k), None))
k += 1
elif y.pure:
entries.append((None, prefix + chr(k), y))
k += 1
else:
entries.append((None, prefix, y))
k = y.hi
stack.extend(reversed(entries))
return out
def check_invariant(self) -> None:
count = 0
stack = [self.root]
while stack:
x = stack.pop()
assert len(x.child) == self.r
count += x.value is not None
k = 0
while k < self.r:
y = x.child[k]
if isinstance(y, Node):
stack.append(y)
k += 1
continue
assert y.lo == k and y.size <= self.limit
assert all(x.child[c] is y for c in range(y.lo, y.hi))
assert y.size == len(y.items())
if y.pure:
assert y.hi == k + 1
else:
assert all(t != "" and y.lo <= ord(t[0]) < y.hi for t, _ in y.items())
for t, _ in y.items():
assert y.slots[_hash(t) % len(y.slots)].count((t, y.find(t))) == 1
count += y.size
k = y.hi
assert count == self.n
test_hat_trie.py 69 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/hat-trie/. The authors are listed in prov/HZ5P9K.jsonl.
import random
import pytest
from hat_trie import Container, HatTrie, Node
WORDS = ["cab", "cat", "car", "cart", "dog", "do", "cats"]
def test_example() -> None:
t: HatTrie[int] = HatTrie(r=128, limit=3, slots=4)
for w in WORDS[:3]:
t.put(w, len(w))
assert t.bursts == 0 and isinstance(t.root.child[ord("c")], Container)
t.put("cart", 4)
t.check_invariant()
assert t.bursts == 7
c = t.root.child[ord("c")]
assert isinstance(c, Node)
ca = c.child[ord("a")]
assert isinstance(ca, Node)
for w in WORDS[4:]:
t.put(w, len(w))
t.check_invariant()
assert t.bursts == 8
leaf = ca.child[ord("t")]
assert isinstance(leaf, Container) and not leaf.pure
assert sorted(k for k, _ in leaf.items()) == ["t", "ts"]
assert t.keys() == sorted(WORDS)
assert t.get("cart") == 4 and t.get("ca") is None and t.get("cats") == 4
assert t.get("dot") is None and t.size() == 7
def test_edges() -> None:
t: HatTrie[int] = HatTrie(r=128, limit=2, slots=2)
t.put("", 0)
for k in range(1, 40):
t.put("a" * k, k)
t.check_invariant()
assert t.get("") == 0 and t.get("a" * 39) == 39 and t.get("a" * 40) is None
assert t.keys() == ["a" * k for k in range(40)]
t.put("a", 100)
assert t.get("a") == 100 and t.size() == 40
with pytest.raises(ValueError):
t.put("é", 1)
with pytest.raises(ValueError):
t.put("x", None) # type: ignore[arg-type]
def test_random_against_dict() -> None:
rng = random.Random(2007)
for _ in range(500):
t: HatTrie[int] = HatTrie(r=128, limit=rng.randint(1, 6), slots=rng.choice([1, 2, 4, 8]))
model: dict[str, int] = {}
alpha = rng.choice(["ab", "abc", "abcdefgh"])
for _ in range(rng.randint(0, 60)):
k = "".join(rng.choice(alpha) for _ in range(rng.randint(0, 6)))
v = rng.randint(0, 99)
t.put(k, v)
model[k] = v
assert t.get(k) == v
t.check_invariant()
assert t.size() == len(model) and t.keys() == sorted(model)
for _ in range(5):
k = "".join(rng.choice(alpha) for _ in range(rng.randint(0, 6)))
assert t.get(k) == model.get(k)
What the next level needs
- the tests did not run in this build