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
# 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