Randomized heap in Python
A mergeable heap that merges by randomly choosing which child to descend into, giving expected logarithmic insert, merge and extract-min.
- File
randomized_heap.py, 120 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/randomized-heap/. The authors are listed in prov/E1PH4P.jsonl.
import random
from typing import Any, Generic, Iterator, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: Any, /) -> bool: ...
class Coin(Protocol):
def random(self) -> float: ...
K = TypeVar("K", bound=Ordered)
class Node(Generic[K]):
"""A node: a key and two subtrees. There is no balance field."""
__slots__ = ("key", "left", "right")
def __init__(self, key: K) -> None:
self.key = key
self.left: Node[K] | None = None
self.right: Node[K] | None = None
def meld(a: "Node[K] | None", b: "Node[K] | None", coin: Coin) -> "Node[K] | None":
"""Randomized meld of two heap-ordered trees; return the new root.
This is the recursive MELD of the page as a loop: at each node of the
path, a fair coin chooses the child that the rest is melded into.
"""
if a is None: # @line L1
return b
if b is None:
return a
if b.key < a.key: # @line L2
a, b = b, a
root = a
while True:
go_left = coin.random() < 0.5 # @line L3
sub = a.left if go_left else a.right
if sub is not None and b.key < sub.key:
sub, b = b, sub # the smaller root continues the path
if go_left: # @line L4
a.left = sub if sub is not None else b
else:
a.right = sub if sub is not None else b
if sub is None: # MELD(NIL, b) = b
break
a = sub
return root # @line L5
class RandomizedHeap(Generic[K]):
"""A min-heap with randomized meld (Gambin and Malinowski, 1998)."""
def __init__(self, coin: Coin | None = None) -> None:
self.coin: Coin = coin if coin is not None else random.Random()
self.root: Node[K] | None = None
self.n = 0
def __len__(self) -> int:
return self.n
def is_empty(self) -> bool:
"""Return True if the heap holds no item."""
return self.root is None
def insert(self, key: K) -> None:
"""Add key to the heap in O(log n) expected time."""
self.root = meld(self.root, Node(key), self.coin) # @line L6
self.n += 1
def find_min(self) -> K:
"""Return a smallest key; the heap must not be empty."""
if self.root is None:
raise IndexError("find_min on an empty heap")
return self.root.key
def delete_min(self) -> K:
"""Remove and return a smallest key in O(log n) expected time."""
r = self.root
if r is None:
raise IndexError("delete_min on an empty heap")
self.root = meld(r.left, r.right, self.coin) # @line L7
self.n -= 1
return r.key # @line L8
def meld(self, other: "RandomizedHeap[K]") -> None:
"""Move all items of other into this heap; other becomes empty."""
self.root = meld(self.root, other.root, self.coin)
self.n += other.n
other.root, other.n = None, 0
def keys(self) -> Iterator[K]:
"""Yield all keys in preorder."""
stack = [self.root] if self.root is not None else []
while stack:
v = stack.pop()
yield v.key
for c in (v.right, v.left):
if c is not None:
stack.append(c)
def check_invariant(self) -> None:
"""Check heap order and the count."""
count = 0
stack = [self.root] if self.root is not None else []
while stack:
v = stack.pop()
count += 1
for c in (v.left, v.right):
if c is not None:
assert not (c.key < v.key), "heap order"
stack.append(c)
assert count == self.n, "count"
test_randomized_heap.py 115 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/randomized-heap/. The authors are listed in prov/E1PH4P.jsonl.
import heapq
import math
import random
import pytest
from randomized_heap import Node, RandomizedHeap, meld
class Script:
"""A coin that follows a script: 'L' gives 0.0 (left), 'R' gives 0.9 (right)."""
def __init__(self, flips: str) -> None:
self.flips = list(flips)
self.used = 0
def random(self) -> float:
self.used += 1
return 0.0 if self.flips.pop(0) == "L" else 0.9
class Counting:
"""A seeded coin that counts its flips."""
def __init__(self, seed: int) -> None:
self.rng = random.Random(seed)
self.flips = 0
def random(self) -> float:
self.flips += 1
return self.rng.random()
def tree(key: int, left: "Node[int] | None" = None, right: "Node[int] | None" = None) -> Node[int]:
v = Node(key)
v.left, v.right = left, right
return v
def sh(v: "Node[int] | None") -> str:
if v is None:
return "-"
if v.left is None and v.right is None:
return str(v.key)
return f"{v.key}({sh(v.left)}, {sh(v.right)})"
def test_example_meld() -> None:
x = tree(2, tree(6), tree(4, tree(9)))
y = tree(3, tree(5))
coin = Script("RLR")
r = meld(x, y, coin)
assert sh(r) == "2(6, 3(4(9, 5), -))"
assert coin.used == 3
def test_example_other_coins() -> None:
x = tree(2, tree(6), tree(4, tree(9)))
y = tree(3, tree(5))
r = meld(x, y, Script("LR"))
assert sh(r) == "2(3(5, 6), 4(9, -))"
def test_empty() -> None:
h: RandomizedHeap[int] = RandomizedHeap(random.Random(1))
assert meld(None, None, random.Random(1)) is None
with pytest.raises(IndexError):
h.find_min()
with pytest.raises(IndexError):
h.delete_min()
h.insert(4)
assert h.delete_min() == 4 and h.is_empty()
def test_expected_path_is_logarithmic() -> None:
coin = Counting(5)
h: RandomizedHeap[int] = RandomizedHeap(coin)
n = 4096
for k in range(n):
h.insert(k)
assert coin.flips / n <= math.log2(n + 1) + 2
coin.flips = 0
out = [h.delete_min() for _ in range(n)]
assert out == list(range(n))
assert coin.flips / n <= 2 * math.log2(n + 1) + 2
def test_random_against_heapq() -> None:
rng = random.Random(1998)
for _ in range(600):
heaps = [RandomizedHeap[int](random.Random(rng.random())) for _ in range(3)]
refs: list[list[int]] = [[] for _ in range(3)]
for _ in range(rng.randrange(1, 60)):
i = rng.randrange(3)
op = rng.random()
if op < 0.55:
k = rng.randrange(-20, 20)
heaps[i].insert(k)
heapq.heappush(refs[i], k)
elif op < 0.85:
if refs[i]:
assert heaps[i].find_min() == refs[i][0]
assert heaps[i].delete_min() == heapq.heappop(refs[i])
else:
j = rng.randrange(3)
if i != j:
heaps[i].meld(heaps[j])
refs[i] = refs[i] + refs[j]
heapq.heapify(refs[i])
refs[j] = []
heaps[i].check_invariant()
assert sorted(heaps[i].keys()) == sorted(refs[i])
What the next level needs
- the tests did not run in this build