Algorithm Search⌘K

Semi-persistent array in Python

A persistent array with O(1) access to the newest version, where older versions are reconstructed by diffs.

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

"""A semi-persistent array: one shared array plus a chain of diffs.

The valid versions form a path v0 -> v1 -> ... -> vk, where vk (the newest)
holds the real array and each older vi is a diff "vi equals v(i+1) except
that a[idx] = val". Reads of any valid version walk the diffs toward vk.
An update of an older version vi first rolls the array back to vi, which
makes v(i+1), ..., vk invalid, and then writes. All loops are iterative.
"""

from typing import Generic, List, Optional, Sequence, TypeVar

T = TypeVar("T")

_ARR, _DIFF, _DEAD = 0, 1, 2


class _Store(Generic[T]):
    __slots__ = ("a", "root", "current", "length")

    def __init__(self, items: Sequence[T]) -> None:
        self.a: List[T] = list(items)
        self.root: "Optional[SemiPersistentArray[T]]" = None
        self.current: "Optional[SemiPersistentArray[T]]" = None
        self.length = 1  # number of valid versions


class SemiPersistentArray(Generic[T]):
    """A handle to one version. Only versions on the path from the first
    version to the newest one are valid; using another raises ValueError.
    """

    __slots__ = ("_store", "_kind", "_idx", "_val", "_next")

    def __init__(self, items: Sequence[T]) -> None:
        """The first version, holding ``items``. Cost: Θ(n)."""
        store: _Store[T] = _Store(items)
        self._store = store
        self._kind = _ARR
        self._idx = 0
        self._val: Optional[T] = None
        self._next: "Optional[SemiPersistentArray[T]]" = None
        store.root = self
        store.current = self

    def __len__(self) -> int:
        """Number of slots. Cost: O(1)."""
        return len(self._store.a)

    def is_valid(self) -> bool:
        """Whether this version can still be used. Cost: O(1)."""
        return self._kind != _DEAD

    def _check(self, i: int) -> None:
        if self._kind == _DEAD:
            raise ValueError("this version is no longer valid")
        if not 0 <= i < len(self._store.a):
            raise IndexError("index out of range")

    def get(self, i: int) -> T:
        """Item i of this version. Cost: Θ(1 + d), d the distance to the newest."""
        self._check(i)  # @line L1
        t: SemiPersistentArray[T] = self  # @line L2
        while t._kind == _DIFF:  # @line L3
            if t._idx == i:  # @line L4
                return t._val  # type: ignore[return-value]
            assert t._next is not None
            t = t._next  # @line L5
        return self._store.a[i]  # @line L6

    def _reroot(self) -> None:
        if self._kind == _DEAD:  # @line L11
            raise ValueError("this version is no longer valid")
        path: List[SemiPersistentArray[T]] = []
        t: SemiPersistentArray[T] = self
        while t._kind == _DIFF:  # @line L12
            path.append(t)
            assert t._next is not None
            t = t._next
        a = self._store.a
        for d in reversed(path):  # @line L13
            a[d._idx] = d._val  # type: ignore[assignment]
        for d in path[1:]:  # @line L14
            d._kind = _DEAD
            d._next = None
            d._val = None
        if path:
            t._kind = _DEAD
            self._store.length -= len(path)
        self._kind = _ARR  # @line L15
        self._next = None
        self._val = None
        self._store.current = self

    def set(self, i: int, x: T) -> "SemiPersistentArray[T]":
        """A new version equal to this one with item i set to x.

        Versions made after this one become invalid. Cost: Θ(1 + d).
        """
        self._check(i)
        self._reroot()  # @line L7
        store = self._store
        w: SemiPersistentArray[T] = SemiPersistentArray.__new__(SemiPersistentArray)  # @line L8
        w._store, w._kind, w._idx, w._val, w._next = store, _ARR, 0, None, None
        self._kind, self._idx, self._val, self._next = _DIFF, i, store.a[i], w  # @line L9
        store.a[i] = x  # @line L10
        store.current = w
        store.length += 1
        return w

    def to_list(self) -> List[T]:
        """All items of this version. Cost: Θ(n + d)."""
        if self._kind == _DEAD:
            raise ValueError("this version is no longer valid")
        out = list(self._store.a)
        path: List[SemiPersistentArray[T]] = []
        t: SemiPersistentArray[T] = self
        while t._kind == _DIFF:
            path.append(t)
            assert t._next is not None
            t = t._next
        for d in reversed(path):
            out[d._idx] = d._val  # type: ignore[assignment]
        return out

    def check_invariant(self) -> None:
        """Check that the valid versions form one path of diffs ending at the
        version that owns the array, and that this version is on it (or dead).
        Cost: Θ(k) for k valid versions.
        """
        store = self._store
        t = store.root
        assert t is not None
        count = 1
        on_path = t is self
        while t._kind == _DIFF:
            assert t._next is not None and 0 <= t._idx < len(store.a)
            t = t._next
            count += 1
            on_path = on_path or t is self
        assert t._kind == _ARR and t is store.current
        assert count == store.length
        assert on_path or self._kind == _DEAD
test_semi_persistent_array.py 94 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/semi-persistent-array/. The authors are listed in prov/07TV6Q.jsonl.

import random

import pytest

from semi_persistent_array import SemiPersistentArray


def test_example() -> None:
    v0 = SemiPersistentArray([0, 0, 0, 0])
    v1 = v0.set(1, 5)
    v2 = v1.set(3, 7)
    v3 = v2.set(1, 9)
    assert v3.to_list() == [0, 9, 0, 7]
    assert v3._store.a == [0, 9, 0, 7]
    assert v0.get(1) == 0 and v1.get(1) == 5 and v2.get(1) == 5 and v2.get(3) == 7
    assert v1.to_list() == [0, 5, 0, 0]
    # Update v1: the array is rolled back to v1, and v2, v3 become invalid.
    v4 = v1.set(2, 4)
    assert v4.to_list() == [0, 5, 4, 0]
    assert v1.to_list() == [0, 5, 0, 0] and v0.to_list() == [0, 0, 0, 0]
    assert not v2.is_valid() and not v3.is_valid()
    with pytest.raises(ValueError):
        v3.get(0)
    with pytest.raises(ValueError):
        v2.set(0, 1)
    for v in (v0, v1, v4):
        v.check_invariant()


def test_edges() -> None:
    e: SemiPersistentArray[int] = SemiPersistentArray([])
    assert len(e) == 0 and e.to_list() == []
    with pytest.raises(IndexError):
        e.get(0)
    one = SemiPersistentArray(["a"])
    two = one.set(0, "b")
    same = two.set(0, "b")
    assert same.to_list() == ["b"] and one.get(0) == "a"
    again = one.set(0, "c")
    assert again.get(0) == "c" and not two.is_valid() and not same.is_valid()
    with pytest.raises(IndexError):
        again.set(1, "x")
    again.check_invariant()


def test_deep() -> None:
    n = 1000
    v0 = SemiPersistentArray(list(range(n)))
    v = v0
    versions = [v0]
    for k in range(200000):
        v = v.set(k % n, -k)
        if k % 5000 == 0:
            versions.append(v)
    assert v.get(5) == -199005
    assert v0.get(5) == 5  # walks 200000 diffs without recursion
    w = v0.set(0, 42)  # rolls back 200000 diffs
    assert w.to_list()[:3] == [42, 1, 2]
    assert all(not x.is_valid() for x in versions[1:])
    w.check_invariant()


def test_random_semi_persistent_use() -> None:
    rng = random.Random(8080)
    for _ in range(500):
        n = rng.randint(1, 6)
        start = [rng.randint(0, 9) for _ in range(n)]
        v0 = SemiPersistentArray(start)
        # path holds the valid versions, oldest first, with their models.
        path = [(v0, list(start))]
        dead = []
        for _ in range(rng.randint(1, 40)):
            j = rng.randrange(len(path))
            v, m = path[j]
            if rng.random() < 0.5:
                i, x = rng.randrange(n), rng.randint(0, 9)
                w = v.set(i, x)
                dead.extend(p for p, _ in path[j + 1 :])
                m2 = list(m)
                m2[i] = x
                path = path[: j + 1] + [(w, m2)]
            else:
                i = rng.randrange(n)
                assert v.get(i) == m[i]
        for v, m in path:
            assert v.is_valid() and v.to_list() == m
            v.check_invariant()
        for v in dead:
            assert not v.is_valid()
            with pytest.raises(ValueError):
                v.get(0)

What the next level needs

  • the tests did not run in this build

Read the full page about Semi-persistent array