d-ary heap in Python
Generalization of the binary heap with d children per node, trading insert speed for extract cost.
- File
d_ary_heap.py, 100 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/d-ary-heap/. The authors are listed in prov/Y99BAR.jsonl.
from collections.abc import Iterable
from typing import Any, Generic, Protocol, TypeVar
class Ordered(Protocol):
def __lt__(self, other: Any, /) -> bool: ...
T = TypeVar("T", bound=Ordered)
class DaryHeap(Generic[T]):
"""A d-ary min-heap in a 0-based list.
The parent of i is (i - 1) // d and its children are d*i + 1 .. d*i + d.
insert and decrease_key take O(log_d n) time; extract_min takes
O(d log_d n) time; build takes O(n) time. Keys need only "<".
"""
def __init__(self, d: int = 4) -> None:
if d < 2:
raise ValueError("d must be at least 2")
self.d = d
self.a: list[T] = []
def __len__(self) -> int:
return len(self.a)
def _parent(self, i: int) -> int:
return (i - 1) // self.d # @line L1
def _child(self, i: int, j: int) -> int:
return self.d * i + j # @line L2: j = 1 .. d
def _sift_up(self, i: int) -> None:
a = self.a
while i > 0 and a[i] < a[self._parent(i)]: # @line L3
p = self._parent(i)
a[i], a[p] = a[p], a[i] # @line L4
i = p # @line L5
def _sift_down(self, i: int) -> None:
a, n = self.a, len(self.a)
while True: # @line L6
m = i # @line L7
for j in range(1, self.d + 1): # @line L8
c = self._child(i, j) # @line L9
if c >= n:
break
if a[c] < a[m]: # @line L10
m = c
if m == i: # @line L11
return
a[i], a[m] = a[m], a[i] # @line L12
i = m
def insert(self, key: T) -> None:
"""Add a key. O(log_d n) time."""
self.a.append(key) # @line L13
self._sift_up(len(self.a) - 1) # @line L14
def find_min(self) -> T:
if not self.a:
raise IndexError("heap underflow")
return self.a[0]
def extract_min(self) -> T:
"""Remove and return a smallest key. O(d log_d n) time."""
if not self.a: # @line L15
raise IndexError("heap underflow")
top = self.a[0] # @line L16
last = self.a.pop()
if self.a:
self.a[0] = last
self._sift_down(0) # @line L17
return top # @line L18
def decrease_key(self, i: int, key: T) -> None:
"""Lower the key at index i to key. O(log_d n) time."""
if not 0 <= i < len(self.a):
raise IndexError("index out of range")
if self.a[i] < key: # @line L19
raise ValueError("new key is larger than the current key")
self.a[i] = key # @line L20
self._sift_up(i)
def build(self, items: Iterable[T]) -> None:
"""Replace the contents by the given keys, in O(n) time."""
self.a = list(items) # @line L21
n = len(self.a)
for i in range(self._parent(n - 1) if n > 1 else -1, -1, -1): # @line L22
self._sift_down(i)
def check_invariant(self) -> None:
a = self.a
for i in range(1, len(a)):
assert not a[i] < a[self._parent(i)]
test_d_ary_heap.py 76 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/d-ary-heap/. The authors are listed in prov/Y99BAR.jsonl.
import heapq
import random
import pytest
from d_ary_heap import DaryHeap
def test_example_d3() -> None:
h: DaryHeap[int] = DaryHeap(3)
h.build([5, 3, 8, 1, 9, 2, 7, 6, 4])
assert h.a == [1, 2, 4, 5, 9, 3, 7, 6, 8]
h.insert(0)
assert h.a == [0, 2, 1, 5, 9, 3, 7, 6, 8, 4]
assert h.extract_min() == 0
assert h.a == [1, 2, 4, 5, 9, 3, 7, 6, 8]
h.decrease_key(7, 0)
assert h.a == [0, 2, 1, 5, 9, 3, 7, 4, 8]
h.check_invariant()
def test_edges() -> None:
with pytest.raises(ValueError):
DaryHeap(1)
h: DaryHeap[int] = DaryHeap(2)
with pytest.raises(IndexError):
h.extract_min()
with pytest.raises(IndexError):
h.find_min()
h.build([])
assert len(h) == 0
h.build([7])
assert h.find_min() == 7
with pytest.raises(ValueError):
h.decrease_key(0, 8)
with pytest.raises(IndexError):
h.decrease_key(1, 0)
assert h.extract_min() == 7 and len(h) == 0
def test_random_against_heapq() -> None:
rng = random.Random(1975)
for case in range(700):
d = 2 + case % 6
h: DaryHeap[int] = DaryHeap(d)
model: list[int] = []
if rng.random() < 0.5:
keys = [rng.randint(-50, 50) for _ in range(rng.randint(0, 40))]
h.build(keys)
model = list(keys)
heapq.heapify(model)
h.check_invariant()
for _ in range(rng.randint(0, 60)):
op = rng.random()
if op < 0.45:
k = rng.randint(-50, 50)
h.insert(k)
heapq.heappush(model, k)
elif op < 0.8:
if model:
assert h.extract_min() == heapq.heappop(model)
elif h.a:
i = rng.randrange(len(h.a))
old = h.a[i]
k = old - rng.randint(0, 30)
model.remove(old)
heapq.heapify(model)
heapq.heappush(model, k)
h.decrease_key(i, k)
h.check_invariant()
assert sorted(h.a) == sorted(model)
if model:
assert h.find_min() == model[0]
What the next level needs
- the tests did not run in this build