Algorithm Search⌘K

Burst trie in Python

Trie whose leaf containers hold small buckets of strings that split (burst) when they grow.

File
burst_trie.py, 203 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
burst_trie.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/burst-trie/. The authors are listed in prov/4A6NMB.jsonl.

"""Burst trie: a trie of access nodes whose leaves are small containers of strings.

An access node has one slot per symbol and a value for the key that ends at the
node. A slot holds either another access node or a container: a sorted list of
(suffix, value) pairs for the keys that continue through that slot, with the
symbols of the path removed. When a container holds more than `limit` pairs it
bursts: it is replaced by an access node, and its pairs are spread over new
containers by their first symbol (the limit heuristic).
"""

from bisect import bisect_left
from typing import Generic, TypeVar

V = TypeVar("V")


class _Container(Generic[V]):
    __slots__ = ("suffixes", "vals")

    def __init__(self) -> None:
        self.suffixes: list[str] = []
        self.vals: list[V] = []

    def find(self, t: str) -> int:
        k = bisect_left(self.suffixes, t)
        return k if k < len(self.suffixes) and self.suffixes[k] == t else -1


class _Access(Generic[V]):
    __slots__ = ("slots", "val")

    def __init__(self) -> None:
        self.slots: dict[str, _Access[V] | _Container[V]] = {}
        self.val: V | None = None


class BurstTrie(Generic[V]):
    """String symbol table; values must not be None."""

    def __init__(self, limit: int = 32) -> None:
        if limit < 1:
            raise ValueError("limit must be positive")
        self.limit = limit
        self.root: _Access[V] = _Access()
        self.n = 0

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

    def get(self, s: str) -> V | None:
        """Return the value of key s, or None."""
        x: _Access[V] | _Container[V] = self.root
        d = 0
        while isinstance(x, _Access):  # @line L1
            if d == len(s):  # @line L2
                return x.val
            nxt = x.slots.get(s[d])  # @line L3
            if nxt is None:
                return None
            x, d = nxt, d + 1
        k = x.find(s[d:])  # @line L4
        return x.vals[k] if k >= 0 else None  # @line L5

    def put(self, s: str, v: V) -> None:
        """Set the value of key s to v."""
        if v is None:
            raise ValueError("a value must not be None")
        x = self.root
        d = 0
        while True:  # @line L6
            if d == len(s):
                if x.val is None:
                    self.n += 1
                x.val = v
                return
            nxt = x.slots.get(s[d])
            if nxt is None:  # @line L7
                nxt = x.slots[s[d]] = _Container()
            if isinstance(nxt, _Access):
                x, d = nxt, d + 1
                continue
            t = s[d + 1 :]  # @line L8
            k = bisect_left(nxt.suffixes, t)
            if k < len(nxt.suffixes) and nxt.suffixes[k] == t:
                nxt.vals[k] = v
                return
            nxt.suffixes.insert(k, t)  # @line L9
            nxt.vals.insert(k, v)
            self.n += 1
            if len(nxt.suffixes) > self.limit:  # @line L10
                x.slots[s[d]] = self._burst(nxt)
            return

    def _burst(self, c: _Container[V]) -> _Access[V]:
        """Replace an overfull container by an access node, bursting again where needed."""
        top: _Access[V] = _Access()
        work: list[tuple[_Access[V], _Container[V]]] = [(top, c)]
        while work:
            node, cont = work.pop()
            for t, v in zip(cont.suffixes, cont.vals):  # @line L11
                if t == "":
                    node.val = v
                    continue
                child = node.slots.get(t[0])
                if child is None:
                    child = node.slots[t[0]] = _Container()
                assert isinstance(child, _Container)
                child.suffixes.append(t[1:])  # sorted input gives sorted output
                child.vals.append(v)
            for sym, child in list(node.slots.items()):  # @line L12
                if isinstance(child, _Container) and len(child.suffixes) > self.limit:
                    new: _Access[V] = _Access()
                    node.slots[sym] = new
                    work.append((new, child))
        return top

    def delete(self, s: str) -> None:
        """Remove key s if present; an emptied container is removed from its slot."""
        x = self.root
        d = 0
        while True:
            if d == len(s):
                if x.val is not None:
                    x.val = None
                    self.n -= 1
                return
            nxt = x.slots.get(s[d])
            if nxt is None:
                return
            if isinstance(nxt, _Access):
                x, d = nxt, d + 1
                continue
            k = nxt.find(s[d + 1 :])
            if k >= 0:
                del nxt.suffixes[k]
                del nxt.vals[k]
                self.n -= 1
                if not nxt.suffixes:
                    del x.slots[s[d]]
            return

    def items_with_prefix(self, p: str) -> list[tuple[str, V]]:
        """The (key, value) pairs whose key starts with p, in order."""
        x: _Access[V] | _Container[V] = self.root
        d = 0
        while isinstance(x, _Access) and d < len(p):
            nxt = x.slots.get(p[d])
            if nxt is None:
                return []
            x, d = nxt, d + 1
        out: list[tuple[str, V]] = []
        if isinstance(x, _Container):
            head, rest = p[:d], p[d:]
            k = bisect_left(x.suffixes, rest)
            while k < len(x.suffixes) and x.suffixes[k].startswith(rest):
                out.append((head + x.suffixes[k], x.vals[k]))
                k += 1
            return out
        stack: list[tuple[_Access[V] | _Container[V], str]] = [(x, p)]
        while stack:
            y, path = stack.pop()
            if isinstance(y, _Container):
                out.extend((path + t, v) for t, v in zip(y.suffixes, y.vals))
                continue
            if y.val is not None:
                out.append((path, y.val))
            for sym in sorted(y.slots, reverse=True):
                stack.append((y.slots[sym], path + sym))
        return out

    def keys(self) -> list[str]:
        return [k for k, _ in self.items_with_prefix("")]

    def shape(self) -> tuple[int, int]:
        """The number of access nodes and of containers."""
        acc, cont = 0, 0
        stack: list[_Access[V] | _Container[V]] = [self.root]
        while stack:
            y = stack.pop()
            if isinstance(y, _Container):
                cont += 1
            else:
                acc += 1
                stack.extend(y.slots.values())
        return acc, cont

    def check_invariant(self) -> None:
        count = 0
        stack: list[_Access[V] | _Container[V]] = [self.root]
        while stack:
            y = stack.pop()
            if isinstance(y, _Container):
                assert 1 <= len(y.suffixes) <= self.limit, "container size out of range"
                assert len(y.suffixes) == len(y.vals)
                assert all(a < b for a, b in zip(y.suffixes, y.suffixes[1:])), "container not sorted"
                count += len(y.suffixes)
            else:
                count += y.val is not None
                stack.extend(y.slots.values())
        assert count == self.n
test_burst_trie.py 67 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/burst-trie/. The authors are listed in prov/4A6NMB.jsonl.

import random

import pytest

from burst_trie import BurstTrie, _Access, _Container

WORDS = "she sells sea shells by the sea shore".split()


def test_example() -> None:
    t: BurstTrie[int] = BurstTrie(limit=3)
    for i, w in enumerate(WORDS):
        t.put(w, i)
        t.check_invariant()
    root = t.root
    s = root.slots["s"]
    assert isinstance(s, _Access) and isinstance(s.slots["h"], _Container)
    assert s.slots["h"].suffixes == ["e", "ells", "ore"]
    assert s.slots["e"].suffixes == ["a", "lls"]
    assert t.shape() == (2, 4)
    assert t.get("sea") == 6 and t.get("shore") == 7 and t.get("sh") is None
    assert t.keys() == ["by", "sea", "sells", "she", "shells", "shore", "the"]
    t.put("shy", 8)
    t.check_invariant()
    assert t.shape() == (3, 6)
    assert [k for k, _ in t.items_with_prefix("sh")] == ["she", "shells", "shore", "shy"]


def test_edges() -> None:
    t: BurstTrie[int] = BurstTrie(limit=1)
    assert t.get("") is None and t.keys() == []
    t.put("", 0)
    for w in ["a", "aa", "aaa", "aaaa"]:
        t.put(w, len(w))
        t.check_invariant()
    assert t.keys() == ["", "a", "aa", "aaa", "aaaa"]
    for w in ["aaaa", "a", "", "aa", "aaa", "zz"]:
        t.delete(w)
        t.check_invariant()
    assert len(t) == 0 and t.keys() == []
    with pytest.raises(ValueError):
        BurstTrie(limit=0)


def test_random_against_dict() -> None:
    rng = random.Random(8080)
    for trial in range(500):
        t: BurstTrie[int] = BurstTrie(limit=rng.randint(1, 6))
        model: dict[str, int] = {}
        for _ in range(rng.randint(0, 60)):
            k = "".join(rng.choice("abc") for _ in range(rng.randint(0, 5)))
            if rng.random() < 0.65:
                v = rng.randint(0, 99)
                t.put(k, v)
                model[k] = v
            else:
                t.delete(k)
                model.pop(k, None)
            assert t.get(k) == model.get(k)
        t.check_invariant()
        assert len(t) == len(model)
        assert t.items_with_prefix("") == sorted(model.items())
        p = "".join(rng.choice("abc") for _ in range(rng.randint(0, 3)))
        assert t.items_with_prefix(p) == sorted((k, v) for k, v in model.items() if k.startswith(p))

What the next level needs

  • the tests did not run in this build

Read the full page about Burst trie