Algorithm Search⌘K

String B-tree in Python

External-memory search tree for strings that uses longest-common-prefix information to achieve optimal I/O.

File
string_b_tree.py, 296 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
string_b_tree.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/string-b-tree/. The authors are listed in prov/EZGPK7.jsonl.

TOP = "\U0010ffff"  # a symbol above every symbol of the strings, used for prefix ranges


def _ch(s: str, i: int) -> str:
    """The symbol at position i, or the empty string past the end (smaller than every symbol)."""
    return s[i] if i < len(s) else ""


class _Node:
    """A node of the String B-tree, with the blind trie of its keys in array form.

    keys are string ids in increasing order of their strings. For an internal node the
    keys are the smallest and largest string of each child, two per child. adj[j] is the
    length of the longest common prefix of keys j - 1 and j (adj[0] = 0), and lc[j], rc[j]
    are the symbols of keys j - 1 and j at position adj[j]: the branching symbols of the
    Patricia trie of the keys.
    """

    __slots__ = ("leaf", "keys", "children", "counts", "adj", "lc", "rc")

    def __init__(self, leaf: bool) -> None:
        self.leaf = leaf
        self.keys: list[int] = []
        self.children: list[_Node] = []
        self.counts: list[int] = []
        self.adj: list[int] = []
        self.lc: list[str] = []
        self.rc: list[str] = []


class StringBTree:
    """A String B-tree over a growing list of strings, in the B+-tree form.

    Leaves hold up to b string ids; internal nodes have up to b children. Each node is
    searched with a blind search on its trie and one comparison with a string, and the
    comparison resumes at the longest common prefix found at the level above, so a search
    for a pattern P reads at most |P| + h symbols of the strings, for a tree of height h.
    The symbol U+10FFFF is reserved.
    """

    def __init__(self, b: int = 8) -> None:
        if b < 4:
            raise ValueError("b must be at least 4")
        self.b = b
        self.strings: list[str] = []
        self.root = _Node(True)
        self._rebuild(self.root)
        self.chars_read = 0  # symbols of stored strings read by comparisons
        self.nodes_read = 0

    def __len__(self) -> int:
        return len(self.strings)

    # ----- the blind trie of one node -----

    def _rebuild(self, x: _Node) -> None:
        s = self.strings
        x.adj, x.lc, x.rc = [0], [""], [""]
        for j in range(1, len(x.keys)):
            u, v = s[x.keys[j - 1]], s[x.keys[j]]
            h = 0
            while h < len(u) and h < len(v) and u[h] == v[h]:
                h += 1
            x.adj.append(h)
            x.lc.append(_ch(u, h))
            x.rc.append(_ch(v, h))

    def _blind(self, x: _Node, p: str) -> int:
        """Blind search: follow the branching symbols of p without reading any stored string."""
        lo, hi = 0, len(x.keys) - 1
        while lo < hi:  # @line L1
            d = min(x.adj[lo + 1: hi + 1])
            splits = [j for j in range(lo + 1, hi + 1) if x.adj[j] == d]
            starts = [lo] + splits
            firsts = [x.lc[splits[0]]] + [x.rc[j] for j in splits]
            c = _ch(p, d) if d < len(p) else None
            g = firsts.index(c) if c in firsts else 0  # @line L2
            lo = starts[g]
            hi = starts[g + 1] - 1 if g + 1 < len(starts) else hi
        return lo

    def _lcp_keys(self, x: _Node, i: int, j: int) -> int:
        if i == j:
            return 1 << 60
        a, b = min(i, j), max(i, j)
        return min(x.adj[a + 1: b + 1])

    def _locate(self, x: _Node, p: str, lam: int) -> tuple[int, list[int]]:
        """Return the number of keys of x that come before p, and lcp(p, key) for every key.

        p is compared with one stored string, starting at position lam, which must be at most
        the longest common prefix of p with the keys of x. A key equal to p comes before p
        unless p ends with TOP.
        """
        self.nodes_read += 1
        si = self._blind(x, p)
        s = self.strings[x.keys[si]]
        ell = lam  # @line L3
        while ell < len(p) and ell < len(s) and p[ell] == s[ell]:
            ell += 1
        self.chars_read += (ell - lam) + (1 if ell < len(s) else 0)
        pc, sc = _ch(p, ell), _ch(s, ell)
        before = 0
        lcps: list[int] = []
        for k in range(len(x.keys)):  # @line L4
            c = self._lcp_keys(x, si, k)
            lcps.append(min(c, ell))
            if c > ell:
                kc = sc
            elif c < ell:
                before += k < si
                continue
            elif k > si:
                j = k
                while x.adj[j] > c:
                    j -= 1
                kc = x.rc[j]
            else:
                j = k + 1
                while x.adj[j] > c:
                    j += 1
                kc = x.lc[j]
            before += kc <= pc
        return before, lcps

    # ----- queries -----

    def _rank(self, p: str) -> int:
        """The number of stored strings that are at most p (strictly less unless p ends in TOP)."""
        x, lam, rank = self.root, 0, 0
        if not x.keys:
            return 0
        while True:
            r, lcps = self._locate(x, p, lam)
            if x.leaf:
                return rank + r  # @line L5
            i = r // 2
            rank += sum(x.counts[:i])
            if r % 2 == 0:  # p falls between child i - 1 and child i
                return rank
            lam = max(lcps[2 * i], lcps[2 * i + 1])  # @line L6
            x = x.children[i]

    def prefix_range(self, p: str) -> tuple[int, int]:
        """Ranks [lo, hi) of the stored strings that start with p, in sorted order."""
        return self._rank_lt(p), self._rank(p + TOP)  # @line L7

    def _rank_lt(self, p: str) -> int:
        # strings < p: rank of p with a symbol below every symbol appended is the same as
        # counting strings that are at most p minus the copies equal to p
        r = self._rank(p)
        return r - (1 if self.contains(p) else 0)

    def contains(self, p: str) -> bool:
        x, lam = self.root, 0
        if not x.keys:
            return False
        while True:
            r, lcps = self._locate(x, p, lam)
            if x.leaf:
                return r > 0 and lcps[r - 1] == len(p) == len(self.strings[x.keys[r - 1]])
            i = r // 2
            if r % 2 == 0:
                return r > 0 and lcps[r - 1] == len(p) == len(self.strings[x.keys[r - 1]])
            lam = max(lcps[2 * i], lcps[2 * i + 1])
            x = x.children[i]

    def sorted_strings(self) -> list[str]:
        out: list[str] = []
        stack = [self.root]
        while stack:
            x = stack.pop()
            if x.leaf:
                out.extend(self.strings[k] for k in x.keys)
            else:
                stack.extend(reversed(x.children))
        return out

    def with_prefix(self, p: str) -> list[str]:
        lo, hi = self.prefix_range(p)
        return self.sorted_strings()[lo:hi]

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

    # ----- insertion -----

    def insert(self, w: str) -> bool:
        """Insert the string w; return False if it is already stored."""
        if TOP in w:
            raise ValueError("the symbol U+10FFFF is reserved")
        if self.contains(w):
            return False
        sid = len(self.strings)
        self.strings.append(w)
        path: list[tuple[_Node, int]] = []
        x, lam = self.root, 0
        while not x.leaf:  # @line L8
            r, lcps = self._locate(x, w, lam)
            i = min(max(r - 1, 0) // 2, len(x.children) - 1)
            if r % 2 == 1:
                lam = max(lcps[2 * i], lcps[2 * i + 1])
            else:
                lam = 0
            path.append((x, i))
            x = x.children[i]
        r, _ = self._locate(x, w, 0) if x.keys else (0, [])
        x.keys.insert(r, sid)  # @line L9
        self._rebuild(x)
        child: _Node = x
        for parent, i in reversed(path):  # @line L10
            parent.counts[i] += 1
            extra = self._split(child) if self._size(child) > self.b else None
            if extra is not None:
                parent.children.insert(i + 1, extra)
                parent.counts[i] = self._count(child)
                parent.counts.insert(i + 1, self._count(extra))
            self._refresh(parent)
            child = parent
        extra = self._split(child) if self._size(child) > self.b else None
        if extra is not None:  # @line L11
            root = _Node(False)
            root.children = [child, extra]
            root.counts = [self._count(child), self._count(extra)]
            self._refresh(root)
            self.root = root
        return True

    def _size(self, x: _Node) -> int:
        return len(x.keys) if x.leaf else len(x.children)

    def _count(self, x: _Node) -> int:
        return len(x.keys) if x.leaf else sum(x.counts)

    def _split(self, x: _Node) -> _Node:
        y = _Node(x.leaf)
        if x.leaf:
            m = len(x.keys) // 2
            y.keys, x.keys = x.keys[m:], x.keys[:m]
            self._rebuild(x)
            self._rebuild(y)
        else:
            m = len(x.children) // 2
            y.children, x.children = x.children[m:], x.children[:m]
            y.counts, x.counts = x.counts[m:], x.counts[:m]
            self._refresh(x)
            self._refresh(y)
        return y

    def _first(self, x: _Node) -> int:
        return x.keys[0]

    def _last(self, x: _Node) -> int:
        return x.keys[-1]

    def _refresh(self, x: _Node) -> None:
        """Recompute the keys of an internal node from its children, and its blind trie."""
        x.keys = []
        for c in x.children:
            x.keys += [self._first(c), self._last(c)]
        self._rebuild(x)

    # ----- invariant -----

    def check_invariant(self) -> None:
        s = self.strings
        depths: set[int] = set()
        stack: list[tuple[_Node, int]] = [(self.root, 1)]
        total = 0
        while stack:
            x, d = stack.pop()
            ks = [s[k] for k in x.keys]
            assert all(ks[j - 1] <= ks[j] for j in range(1, len(ks)))
            if x.leaf:
                depths.add(d)
                assert len(x.keys) <= self.b
                assert x is self.root or len(x.keys) >= self.b // 2
                assert len(set(ks)) == len(ks)
                total += len(x.keys)
            else:
                assert 2 <= len(x.children) <= self.b
                assert x is self.root or len(x.children) >= self.b // 2
                assert x.keys == [k for c in x.children for k in (self._first(c), self._last(c))]
                assert x.counts == [self._count(c) for c in x.children]
                stack.extend((c, d + 1) for c in x.children)
            saved = (x.adj, x.lc, x.rc)
            self._rebuild(x)
            assert saved == (x.adj, x.lc, x.rc), "stale blind trie"
        assert len(depths) == 1 and total == len(s)
        assert self.sorted_strings() == sorted(s)
test_string_b_tree.py 69 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/string-b-tree/. The authors are listed in prov/EZGPK7.jsonl.

import bisect
import random

import pytest

from string_b_tree import StringBTree

WORDS = ["abaa", "abab", "abba", "abbb", "abc", "bab", "baba", "bb", "bbaa", "bbab", "c", "caa"]


def test_example() -> None:
    t = StringBTree(4)
    for w in WORDS:
        assert t.insert(w)
        t.check_invariant()
    assert t.height() == 3 and t.sorted_strings() == sorted(WORDS)
    assert not t.insert("abc")
    assert t.with_prefix("ab") == ["abaa", "abab", "abba", "abbb", "abc"]
    assert t.with_prefix("bab") == ["bab", "baba"]
    assert t.with_prefix("bba") == ["bbaa", "bbab"]
    assert t.with_prefix("d") == [] and t.prefix_range("d") == (12, 12)
    assert t.contains("bb") and not t.contains("ba")
    t.chars_read = 0
    t.prefix_range("bbab")
    assert t.chars_read <= 2 * (len("bbab") + 1 + t.height() + 1)


def test_edges() -> None:
    t = StringBTree(4)
    assert t.prefix_range("a") == (0, 0) and not t.contains("")
    assert t.insert("")
    t.check_invariant()
    assert t.contains("") and t.with_prefix("") == [""]
    assert t.insert("a") and t.insert("aa") and t.insert("aaa")
    t.check_invariant()
    assert t.with_prefix("a") == ["a", "aa", "aaa"]
    with pytest.raises(ValueError):
        t.insert("\U0010ffff")
    with pytest.raises(ValueError):
        StringBTree(3)


def test_random_against_sorted_list() -> None:
    rng = random.Random(1999)
    for case in range(500):
        b = rng.choice([4, 5, 8])
        t = StringBTree(b)
        model: list[str] = []
        for _ in range(rng.randint(0, 60)):
            w = "".join(rng.choice("ab" if case % 2 else "abc") for _ in range(rng.randint(0, 6)))
            new = w not in model
            assert t.insert(w) == new
            if new:
                bisect.insort(model, w)
        t.check_invariant()
        assert t.sorted_strings() == model
        for _ in range(8):
            p = "".join(rng.choice("abc") for _ in range(rng.randint(0, 4)))
            want = [w for w in model if w.startswith(p)]
            lo = bisect.bisect_left(model, p)
            assert t.prefix_range(p) == (lo, lo + len(want))
            assert t.contains(p) == (p in model)
            # the lcp carried down the tree bounds the symbols read by one search
            t.chars_read = 0
            t._rank(p)
            assert t.chars_read <= len(p) + t.height()

What the next level needs

  • the tests did not run in this build

Read the full page about String B-tree