Algorithm Search⌘K

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
d_ary_heap.pyRawGitHub
# 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

Read the full page about d-ary heap