Algorithm Search⌘K

Persistent vector in Python

A wide, shallow persistent trie giving near-constant indexing and update by copying a short path.

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

"""A persistent vector: a wide trie of fixed-size leaves plus a tail buffer.

This is the layout of the Clojure vector. Each node has up to M = 2^B
children (B = 5 gives M = 32). Nodes are tuples, so they cannot change.
Index i is found by its digits in base M. An update copies one node per
level, and push and pop at the end mostly copy only the tail.
Recursion depth is the height of the trie, which is about log n / B.
"""

from typing import Generic, Iterator, Optional, Tuple, TypeVar

T = TypeVar("T")

Node = Tuple[object, ...]


class PersistentVector(Generic[T]):
    """An immutable sequence with index, set, push and pop at the end."""

    __slots__ = ("_b", "_m", "_size", "_shift", "_root", "_tail")

    def __init__(self, bits: int = 5) -> None:
        """The empty vector with branching factor 2^bits. Cost: O(1)."""
        if bits < 1:
            raise ValueError("bits must be at least 1")
        self._b = bits
        self._m = 1 << bits
        self._size = 0
        self._shift = bits
        self._root: Node = ()
        self._tail: Node = ()

    def _make(self, size: int, shift: int, root: Node, tail: Node) -> "PersistentVector[T]":
        v: PersistentVector[T] = PersistentVector.__new__(PersistentVector)
        v._b, v._m = self._b, self._m
        v._size, v._shift, v._root, v._tail = size, shift, root, tail
        return v

    @classmethod
    def from_iterable(cls, items: "Iterator[T] | list[T]", bits: int = 5) -> "PersistentVector[T]":
        """Build by repeated push. Cost: O(n) amortized over the pushes."""
        v: PersistentVector[T] = cls(bits)
        for x in items:
            v = v.push(x)
        return v

    def _tailoff(self) -> int:
        if self._size < self._m:
            return 0
        return ((self._size - 1) >> self._b) << self._b

    def __len__(self) -> int:
        """Number of items. Cost: O(1)."""
        return self._size

    def _leaf_for(self, i: int) -> Node:
        if i >= self._tailoff():
            return self._tail
        node = self._root
        level = self._shift
        while level > 0:  # @line L2
            node = node[(i >> level) & (self._m - 1)]  # type: ignore[assignment]
            level -= self._b
        return node

    def get(self, i: int) -> T:
        """Item at index i. Cost: Θ(1 + log n / B)."""
        if not 0 <= i < self._size:
            raise IndexError("index out of range")
        if i >= self._tailoff():  # @line L1
            return self._tail[i - self._tailoff()]  # type: ignore[return-value]
        return self._leaf_for(i)[i & (self._m - 1)]  # type: ignore[return-value]  # @line L3

    def __getitem__(self, i: int) -> T:
        return self.get(i)

    def _set_path(self, node: Node, level: int, i: int, x: T) -> Node:
        c = list(node)  # @line L6
        if level == 0:  # @line L7
            c[i & (self._m - 1)] = x
        else:
            j = (i >> level) & (self._m - 1)
            c[j] = self._set_path(node[j], level - self._b, i, x)  # type: ignore[arg-type]
        return tuple(c)  # @line L8

    def set(self, i: int, x: T) -> "PersistentVector[T]":
        """A vector equal to this one with item i replaced by x.

        Cost: Θ(1 + log n / B) time; one copied node of at most M slots per
        level.
        """
        if not 0 <= i < self._size:
            raise IndexError("index out of range")
        off = self._tailoff()
        if i >= off:  # @line L4
            tail = list(self._tail)
            tail[i - off] = x
            return self._make(self._size, self._shift, self._root, tuple(tail))
        root = self._set_path(self._root, self._shift, i, x)  # @line L5
        return self._make(self._size, self._shift, root, self._tail)

    def _new_path(self, level: int, leaf: Node) -> Node:
        node = leaf
        while level > 0:
            node = (node,)
            level -= self._b
        return node

    def _push_tail(self, node: Node, level: int, i: int, leaf: Node) -> Node:
        j = (i >> level) & (self._m - 1)  # @line L13
        c = list(node)
        if level == self._b:  # @line L14
            child: Node = leaf
        elif j < len(node):  # @line L15
            child = self._push_tail(node[j], level - self._b, i, leaf)  # type: ignore[arg-type]
        else:  # @line L16
            child = self._new_path(level - self._b, leaf)
        if j < len(c):
            c[j] = child
        else:
            c.append(child)
        return tuple(c)

    def push(self, x: T) -> "PersistentVector[T]":
        """A vector with x added at the end.

        Cost: Θ(1) when the tail has room (one copy of the tail), else
        Θ(1 + log n / B); O(1) amortized over a run of pushes.
        """
        if self._size - self._tailoff() < self._m:  # @line L9
            return self._make(self._size + 1, self._shift, self._root, self._tail + (x,))
        shift = self._shift
        if (self._size >> self._b) > (1 << self._shift):  # @line L10
            root: Node = (self._root, self._new_path(self._shift, self._tail))
            shift += self._b
        else:  # @line L11
            root = self._push_tail(self._root, self._shift, self._size - 1, self._tail)
        return self._make(self._size + 1, shift, root, (x,))  # @line L12

    def _pop_tail(self, node: Node, level: int, i: int) -> Optional[Node]:
        j = (i >> level) & (self._m - 1)
        if level > self._b:  # @line L23
            c = self._pop_tail(node[j], level - self._b, i)  # type: ignore[arg-type]
            if c is None and j == 0:
                return None
            rest = list(node[:j])
            if c is not None:
                rest.append(c)
            return tuple(rest)
        if j == 0:  # @line L24
            return None
        return node[:j]  # @line L25

    def pop(self) -> "PersistentVector[T]":
        """A vector without its last item. Cost: Θ(1 + log n / B)."""
        if self._size == 0:
            raise IndexError("pop from empty vector")
        if self._size == 1:  # @line L17
            return self._make(0, self._b, (), ())
        if self._size - self._tailoff() > 1:  # @line L18
            return self._make(self._size - 1, self._shift, self._root, self._tail[:-1])
        new_tail = self._leaf_for(self._size - 2)  # @line L19
        root = self._pop_tail(self._root, self._shift, self._size - 2)  # @line L20
        if root is None:
            root = ()
        shift = self._shift
        if shift > self._b and len(root) == 1:  # @line L21
            root = root[0]  # type: ignore[assignment]
            shift -= self._b
        return self._make(self._size - 1, shift, root, new_tail)  # @line L22

    def __iter__(self) -> Iterator[T]:
        off = self._tailoff()
        i = 0
        while i < off:
            leaf = self._leaf_for(i)
            for x in leaf:
                yield x  # type: ignore[misc]
            i += self._m
        for x in self._tail:
            yield x  # type: ignore[misc]

    def to_list(self) -> "list[T]":
        """All items in order. Cost: Θ(n)."""
        return list(iter(self))

    def height(self) -> int:
        """Levels of the trie below the root (shift / B). Cost: O(1)."""
        return self._shift // self._b

    def check_invariant(self) -> None:
        """Check the layout: full leaves, left-packed inner nodes, tail size.

        The trie holds the first tailoff items in full leaves of M items,
        packed from the left, and the tail holds the rest (1 to M items
        when the vector is not empty). The root has room for the trie.
        Cost: Θ(n / M) nodes visited.
        """
        off = self._tailoff()
        assert len(self._tail) == self._size - off
        assert self._size == 0 or 1 <= len(self._tail) <= self._m
        assert off % self._m == 0
        leaves = off // self._m
        assert self._shift >= self._b and self._shift % self._b == 0
        assert leaves <= 1 << self._shift
        if self._shift > self._b:
            assert leaves > 1 << (self._shift - self._b)
        count = 0
        stack = [(self._root, self._shift)]
        while stack:
            node, level = stack.pop()
            assert len(node) <= self._m
            if level == 0:
                assert len(node) == self._m
                count += 1
                continue
            if node is not self._root:
                assert len(node) >= 1
            for child in node:
                stack.append((child, level - self._b))  # type: ignore[arg-type]
        assert count == leaves
test_persistent_vector.py 109 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/persistent-vector/. The authors are listed in prov/E5Q5G6.jsonl.

import random

import pytest

from persistent_vector import PersistentVector


def test_example() -> None:
    # Branching factor M = 4 (bits = 2), as on the page.
    v10: PersistentVector[int] = PersistentVector.from_iterable(range(10), bits=2)
    assert v10._root == ((0, 1, 2, 3), (4, 5, 6, 7)) and v10._tail == (8, 9)
    assert v10.height() == 1 and v10.get(6) == 6
    w = v10.set(6, 60)
    assert w._root[0] is v10._root[0] and w._tail is v10._tail
    assert w.to_list() == [0, 1, 2, 3, 4, 5, 60, 7, 8, 9]
    assert v10.to_list() == list(range(10))
    v = v10
    for x in (10, 11, 12, 13, 14, 15, 16):
        v = v.push(x)
    # 17 items: 4 full leaves in the trie, tail (16,)
    assert len(v._root) == 4 and v._tail == (16,) and v.height() == 1
    v = v.push(17)
    assert v._tail == (16, 17)
    v = v.push(18).push(19).push(20)
    # 21 items: the fifth leaf overflows a root of 4 leaves, so the trie grows.
    assert v.height() == 2 and v._tail == (20,)
    assert v.to_list() == list(range(10)) + list(range(10, 21))
    p = v.pop()
    assert p.height() == 1 and p._tail == (16, 17, 18, 19)
    q = p.pop().pop().pop().pop()
    assert len(q) == 16 and q.height() == 1 and q._tail == (12, 13, 14, 15)
    for x in (v10, w, v, p, q):
        x.check_invariant()


def test_edges() -> None:
    with pytest.raises(ValueError):
        PersistentVector(0)
    e: PersistentVector[str] = PersistentVector()
    assert len(e) == 0 and e.to_list() == []
    with pytest.raises(IndexError):
        e.pop()
    with pytest.raises(IndexError):
        e.get(0)
    one = e.push("a")
    assert one.get(0) == "a" and one.pop().to_list() == [] and e.to_list() == []
    with pytest.raises(IndexError):
        one.set(1, "b")
    with pytest.raises(IndexError):
        one.get(-1)
    b1: PersistentVector[int] = PersistentVector.from_iterable(range(9), bits=1)
    assert b1.to_list() == list(range(9))
    while len(b1):
        b1.check_invariant()
        b1 = b1.pop()
    e.check_invariant()


def test_deep() -> None:
    n = 200000
    v: PersistentVector[int] = PersistentVector.from_iterable(range(n))
    assert len(v) == n and v.get(n - 1) == n - 1 and v.get(12345) == 12345
    assert v.height() == 3  # 6249 full leaves: more than 32^2, at most 32^3
    u = v
    for i in range(0, n, 1013):
        u = u.set(i, -i)
    assert all(u.get(i) == (-i if i % 1013 == 0 else i) for i in range(0, n, 7))
    assert v.get(1013) == 1013
    while len(u) > 1000:
        u = u.pop()
    assert len(u) == 1000 and u.height() == 1
    u.check_invariant()
    narrow: PersistentVector[int] = PersistentVector.from_iterable(range(5000), bits=1)
    assert narrow.to_list() == list(range(5000))
    narrow.check_invariant()


def test_random_version_tree() -> None:
    rng = random.Random(1979)
    for _ in range(500):
        bits = rng.randint(1, 3)
        versions = [PersistentVector(bits)]
        model: "list[list[int]]" = [[]]
        for _ in range(rng.randint(1, 60)):
            v = rng.randrange(len(versions))
            op = rng.random()
            if op < 0.5:
                x = rng.randint(0, 99)
                versions.append(versions[v].push(x))
                model.append(model[v] + [x])
            elif op < 0.7 and model[v]:
                versions.append(versions[v].pop())
                model.append(model[v][:-1])
            elif op < 0.9 and model[v]:
                i = rng.randrange(len(model[v]))
                x = rng.randint(0, 99)
                versions.append(versions[v].set(i, x))
                m = list(model[v])
                m[i] = x
                model.append(m)
            elif model[v]:
                i = rng.randrange(len(model[v]))
                assert versions[v].get(i) == model[v][i]
        for t, m in zip(versions, model):
            assert t.to_list() == m and len(t) == len(m)
            t.check_invariant()

What the next level needs

  • the tests did not run in this build

Read the full page about Persistent vector