Algorithm Search⌘K

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

Read the full page about Randomized heap