Replacement selection in Python
Form runs using a heap of fixed size so that runs are on average about twice the memory size.
- File
replacement_selection.py, 69 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/replacement-selection/. The authors are listed in prov/H4VBFM.jsonl.
import heapq
from typing import Iterable, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: "Ordered", /) -> bool: ...
T = TypeVar("T", bound=Ordered)
class _Entry:
"""An item x in the heap, tagged with its run number r and its input position s."""
__slots__ = ("r", "x", "s")
def __init__(self, r: int, x: T, s: int) -> None:
self.r, self.x, self.s = r, x, s
def __lt__(self, other: "_Entry") -> bool:
if self.r != other.r:
return self.r < other.r
if self.x < other.x:
return True
if other.x < self.x:
return False
return self.s < other.s
def replacement_selection(f: Iterable[T], p: int) -> list[list[T]]:
"""Cut the input stream f into sorted runs with a heap of p items.
Precondition: p >= 1. Each run is sorted; equal items keep their input
order inside a run, and an item never goes to an earlier run than an equal
item that was read before it. Each run except the last has at least p
items. On random input the runs have about 2p items on average. The time is
O(n log p) for n items and the extra space is O(p).
"""
if p < 1:
raise ValueError("need p >= 1")
it = iter(f)
seq = 0
heap: list[_Entry] = [] # @line L1
for x in it: # @line L2
heap.append(_Entry(0, x, seq))
seq += 1
if len(heap) == p:
break
heapq.heapify(heap)
runs: list[list[T]] = [] # @line L3
while heap: # @line L4
e = heap[0] # @line L5
if e.r == len(runs): # @line L6
runs.append([])
runs[e.r].append(e.x) # @line L7
y = next(it, _END) # @line L8
if y is not _END:
r = e.r if not y < e.x else e.r + 1 # @line L9
heapq.heapreplace(heap, _Entry(r, y, seq)) # @line L10
seq += 1
else:
heapq.heappop(heap) # @line L11
return runs # @line L12
_END = object()
test_replacement_selection.py 75 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/replacement-selection/. The authors are listed in prov/H4VBFM.jsonl.
import random
import pytest
from replacement_selection import replacement_selection
def check(f: list[int], p: int, runs: list[list[int]]) -> None:
assert sorted(x for r in runs for x in r) == sorted(f)
for r in runs:
assert r and all(r[i] <= r[i + 1] for i in range(len(r) - 1))
for r in runs[:-1]:
assert len(r) >= p
def test_example() -> None:
f = [5, 1, 9, 3, 7, 2, 8, 6, 4, 10, 0, 11]
assert replacement_selection(f, 3) == [[1, 3, 5, 7, 8, 9], [2, 4, 6, 10, 11], [0]]
def test_edges() -> None:
assert replacement_selection([], 3) == []
assert replacement_selection([4], 3) == [[4]]
assert replacement_selection([2, 1], 5) == [[1, 2]]
assert replacement_selection(list(range(10)), 2) == [list(range(10))]
assert replacement_selection(list(range(9, -1, -1)), 3) == [[7, 8, 9], [4, 5, 6], [1, 2, 3], [0]]
assert replacement_selection([3, 2, 1], 1) == [[3], [2], [1]]
assert replacement_selection([7] * 6, 2) == [[7] * 6]
with pytest.raises(ValueError):
replacement_selection([1], 0)
def test_stable() -> None:
class Key:
def __init__(self, k: int, tag: int) -> None:
self.k, self.tag = k, tag
def __lt__(self, other: "Key") -> bool:
return self.k < other.k
rng = random.Random(5)
for _ in range(100):
a = [Key(rng.randint(0, 4), i) for i in range(rng.randint(0, 40))]
runs = replacement_selection(a, rng.randint(1, 5))
for r in runs:
for u, v in zip(r, r[1:]):
assert u.k < v.k or (u.k == v.k and u.tag < v.tag)
first: dict[int, int] = {}
for i, r in enumerate(runs):
for x in r:
first.setdefault(x.tag, i)
for x in a:
for y in a:
if x.k == y.k and x.tag < y.tag:
assert first[x.tag] <= first[y.tag]
def test_random_against_sorted() -> None:
rng = random.Random(9001)
for _ in range(700):
n = rng.randint(0, 80)
p = rng.randint(1, 8)
f = [rng.randint(-25, 25) for _ in range(n)]
check(f, p, replacement_selection(f, p))
def test_mean_length_near_two_p() -> None:
rng = random.Random(77)
p, n = 50, 50000
runs = replacement_selection([rng.random() for _ in range(n)], p)
mean = n / len(runs)
assert 1.9 * p < mean < 2.1 * p
What the next level needs
- the tests did not run in this build