Algorithm Search⌘K

Persistent trie in Python

A trie updated by path copying, used for prefix and max-xor queries over past versions.

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

"""A persistent binary trie of w-bit integers, built by path copying.

Each version is a multiset of integers in [0, 2^w). A node stores how many
keys of its version pass through it. An update makes w + 1 new nodes along
one root-to-leaf path and shares every other node. Node 0 is the empty
trie; its two children are itself, so walks through absent parts need no
special case.
"""

from typing import List


class PersistentTrie:
    """Fully persistent multisets of w-bit integers, numbered 0, 1, 2, ...

    Version 0 is the empty multiset.
    """

    def __init__(self, width: int) -> None:
        """Make the store with version 0 empty. Cost: O(1)."""
        if width < 1:
            raise ValueError("width must be at least 1")
        self._w = width
        self._child: List[List[int]] = [[0, 0]]
        self._cnt: List[int] = [0]
        self._roots: List[int] = [0]

    def _new(self, c0: int, c1: int, cnt: int) -> int:
        self._child.append([c0, c1])
        self._cnt.append(cnt)
        return len(self._cnt) - 1

    def _root(self, version: int) -> int:
        if not 0 <= version < len(self._roots):
            raise IndexError("no such version")
        return self._roots[version]

    def _check_key(self, x: int) -> None:
        if not 0 <= x < (1 << self._w):
            raise ValueError("key out of range")

    def num_versions(self) -> int:
        """Number of versions made so far. Cost: O(1)."""
        return len(self._roots)

    def num_nodes(self) -> int:
        """Number of nodes in the pool, the empty node included. Cost: O(1)."""
        return len(self._cnt)

    def size(self, version: int) -> int:
        """Number of keys (with multiplicity) in ``version``. Cost: O(1)."""
        return self._cnt[self._root(version)]

    def _add(self, t: int, x: int, d: int) -> int:
        child, cnt = self._child, self._cnt
        root = self._new(child[t][0], child[t][1], cnt[t] + d)  # @line L1
        cur = root
        for i in range(self._w - 1, -1, -1):  # @line L2
            b = (x >> i) & 1  # @line L3
            t = child[t][b]  # @line L4
            new = self._new(child[t][0], child[t][1], cnt[t] + d)  # @line L5
            child[cur][b] = new  # @line L6
            cur = new  # @line L7
        return root  # @line L8

    def insert(self, version: int, x: int) -> int:
        """New version: ``version`` with one more copy of x.

        Returns the new version number. Cost: Θ(w) time, w + 1 new nodes.
        """
        t = self._root(version)
        self._check_key(x)
        self._roots.append(self._add(t, x, 1))
        return len(self._roots) - 1

    def remove(self, version: int, x: int) -> int:
        """New version: ``version`` with one copy of x removed.

        If x is absent, the new version shares the old root. Cost: Θ(w)
        time, w + 1 new nodes or none.
        """
        t = self._root(version)
        self._check_key(x)
        if self.count(version, x) == 0:  # @line L9
            self._roots.append(t)
        else:
            self._roots.append(self._add(t, x, -1))
        return len(self._roots) - 1

    def count(self, version: int, x: int) -> int:
        """How many copies of x are in ``version``. Cost: Θ(w)."""
        t = self._root(version)
        self._check_key(x)
        for i in range(self._w - 1, -1, -1):  # @line L10
            t = self._child[t][(x >> i) & 1]
        return self._cnt[t]  # @line L11

    def max_xor(self, before: int, after: int, y: int) -> int:
        """Max of x XOR y over the keys x of ``after`` minus ``before``.

        ``after`` must be made from ``before`` by insertions only, so that
        the difference is a multiset. Cost: Θ(w).
        """
        u = self._root(before)
        v = self._root(after)
        self._check_key(y)
        child, cnt = self._child, self._cnt
        if cnt[v] - cnt[u] <= 0:  # @line L12
            raise ValueError("the difference of the two versions is empty")
        r = 0  # @line L13
        for i in range(self._w - 1, -1, -1):
            b = (y >> i) & 1  # @line L14
            if cnt[child[v][1 - b]] - cnt[child[u][1 - b]] > 0:  # @line L15
                r |= 1 << i
                u, v = child[u][1 - b], child[v][1 - b]
            else:  # @line L16
                u, v = child[u][b], child[v][b]
        return r  # @line L17

    def keys(self, version: int) -> List[int]:
        """Sorted keys of ``version``, with multiplicity. Cost: O(n w)."""
        out: List[int] = []
        stack = [(self._root(version), 0, 0)]
        while stack:
            t, depth, prefix = stack.pop()
            if self._cnt[t] == 0:
                continue
            if depth == self._w:
                out.extend([prefix] * self._cnt[t])
                continue
            stack.append((self._child[t][1], depth + 1, prefix * 2 + 1))
            stack.append((self._child[t][0], depth + 1, prefix * 2))
        return out

    def check_invariant(self) -> None:
        """Check counts on every node reachable from a root.

        Node 0 must be empty with itself as both children. An inner node
        must have the sum of its children's counts, a node at depth w is a
        leaf whose count is the multiplicity. The walk stops at depth w, so
        it ends even though node 0 is its own child. Cost: O(total nodes).
        """
        assert self._cnt[0] == 0 and self._child[0] == [0, 0]
        seen = set()
        for root in self._roots:
            stack = [(root, 0)]
            while stack:
                t, depth = stack.pop()
                if t == 0 or (t, depth) in seen:
                    continue
                seen.add((t, depth))
                assert self._cnt[t] >= 0
                if depth == self._w:
                    continue
                c0, c1 = self._child[t]
                assert 0 <= c0 < len(self._cnt) and 0 <= c1 < len(self._cnt)
                assert self._cnt[t] == self._cnt[c0] + self._cnt[c1]
                stack.append((c0, depth + 1))
                stack.append((c1, depth + 1))
test_persistent_trie.py 101 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/persistent-trie/. The authors are listed in prov/BSY6SQ.jsonl.

import random
from collections import Counter

import pytest

from persistent_trie import PersistentTrie


def test_example() -> None:
    a = [5, 1, 6, 3]
    t = PersistentTrie(3)
    vs = [0]
    for x in a:
        vs.append(t.insert(vs[-1], x))
    assert t.num_nodes() == 1 + 4 * 4
    assert t.max_xor(vs[1], vs[3], 4) == 5
    assert t.max_xor(vs[0], vs[4], 4) == 7
    assert t.max_xor(vs[2], vs[4], 0) == 6
    assert t.keys(vs[2]) == [1, 5] and t.keys(vs[4]) == [1, 3, 5, 6]
    w = t.remove(vs[4], 6)
    assert t.keys(w) == [1, 3, 5] and t.keys(vs[4]) == [1, 3, 5, 6]
    assert t.max_xor(0, w, 0) == 5
    t.check_invariant()


def test_edges() -> None:
    with pytest.raises(ValueError):
        PersistentTrie(0)
    t = PersistentTrie(1)
    assert t.size(0) == 0 and t.keys(0) == [] and t.count(0, 1) == 0
    v = t.remove(0, 1)
    assert t.keys(v) == [] and t.num_nodes() == 1
    v = t.insert(0, 1)
    v = t.insert(v, 1)
    assert t.count(v, 1) == 2 and t.size(v) == 2
    assert t.max_xor(0, v, 1) == 0 and t.max_xor(0, v, 0) == 1
    with pytest.raises(ValueError):
        t.max_xor(v, v, 0)
    with pytest.raises(ValueError):
        t.insert(0, 2)
    with pytest.raises(IndexError):
        t.count(99, 0)
    t.check_invariant()


def test_deep() -> None:
    w = 30
    t = PersistentTrie(w)
    rng = random.Random(11)
    v = 0
    xs = [rng.randrange(1 << w) for _ in range(20000)]
    for x in xs:
        v = t.insert(v, x)
    assert t.num_nodes() == 1 + 20000 * (w + 1)
    y = rng.randrange(1 << w)
    assert t.max_xor(0, v, y) == max(x ^ y for x in xs)
    assert t.size(v) == 20000


def test_random_version_tree() -> None:
    rng = random.Random(9091)
    for _ in range(500):
        w = rng.randint(1, 5)
        t = PersistentTrie(w)
        model = [Counter()]
        for _ in range(rng.randint(1, 30)):
            v = rng.randrange(len(model))
            x = rng.randrange(1 << w)
            op = rng.random()
            if op < 0.5:
                assert t.insert(v, x) == len(model)
                model.append(model[v] + Counter([x]))
            elif op < 0.8:
                assert t.remove(v, x) == len(model)
                m = Counter(model[v])
                if m[x] > 0:
                    m[x] -= 1
                model.append(+m)
            else:
                assert t.count(v, x) == model[v][x]
        for v, m in enumerate(model):
            assert t.keys(v) == sorted(m.elements())
        t.check_invariant()


def test_random_range_max_xor() -> None:
    rng = random.Random(4242)
    for _ in range(500):
        w = rng.randint(1, 6)
        a = [rng.randrange(1 << w) for _ in range(rng.randint(1, 12))]
        t = PersistentTrie(w)
        vs = [0]
        for x in a:
            vs.append(t.insert(vs[-1], x))
        l = rng.randrange(len(a))
        r = rng.randint(l + 1, len(a))
        y = rng.randrange(1 << w)
        assert t.max_xor(vs[l], vs[r], y) == max(x ^ y for x in a[l:r])

What the next level needs

  • the tests did not run in this build

Read the full page about Persistent trie