Algorithm Search⌘K

Sample sort in Python

Pick splitters from a random sample, bucket the keys by them, then sort each bucket independently.

File
sample_sort.py, 75 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
sample_sort.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/sample-sort/. The authors are listed in prov/X40DBH.jsonl.

"""Sample sort with p buckets, simulated one processor at a time.

Keys are paired with their input index and ordered by key, then index.
So all items are distinct, the bucket bound holds with duplicate keys,
and the sort is stable.
"""

import bisect
import math
import random
from typing import Protocol, TypeVar


class Ordered(Protocol):
    def __lt__(self, other: "Ordered", /) -> bool: ...


T = TypeVar("T", bound=Ordered)


class Item:
    """A key with its input index, ordered by key and then by index."""

    __slots__ = ("key", "index")

    def __init__(self, key: Ordered, index: int) -> None:
        self.key, self.index = key, index

    def __lt__(self, other: "Item") -> bool:
        if self.key < other.key:
            return True
        if other.key < self.key:
            return False
        return self.index < other.index


def choose_splitters(items: list[Item], p: int, s: int, rng: random.Random) -> list[Item]:
    """Draw p*s items with replacement, sort them, and keep every s-th one."""
    sample = [items[rng.randrange(len(items))] for _ in range(p * s)]  # @line L1
    sample.sort()  # @line L2
    return [sample[j * s] for j in range(1, p)]  # @line L3


def sample_sort(a: list[T], p: int, s: int | None = None, rng: random.Random | None = None) -> tuple[list[T], list[int]]:
    """Return a stably sorted copy of a, and the sizes of the p buckets."""
    if p < 1:
        raise ValueError("need p >= 1")
    n = len(a)
    if n == 0:
        return [], [0] * p
    if s is None:
        s = max(1, math.ceil(12 * math.log(max(n, 2))))
    if rng is None:
        rng = random.Random(0)
    items = [Item(x, i) for i, x in enumerate(a)]
    splitters = choose_splitters(items, p, s, rng)
    bucket_of = [bisect.bisect_left(splitters, it) for it in items]  # @line L4
    sizes = [0] * p
    for b in bucket_of:  # @line L5: counts, a prefix sum and a scatter
        sizes[b] += 1
    starts, total = [0] * p, 0
    for j in range(p):
        starts[j], total = total, total + sizes[j]
    placed = list(items)
    nxt = list(starts)
    for it, b in zip(items, bucket_of):
        placed[nxt[b]] = it
        nxt[b] += 1
    for j in range(p):  # @line L6: each processor sorts its own bucket
        lo, hi = starts[j], starts[j] + sizes[j]
        placed[lo:hi] = sorted(placed[lo:hi])
    return [it.key for it in placed], sizes  # @line L7
test_sample_sort.py 61 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/sample-sort/. The authors are listed in prov/X40DBH.jsonl.

import math
import random

import pytest

from sample_sort import sample_sort


def test_example() -> None:
    a = [41, 7, 93, 25, 60, 12, 88, 3, 54, 31, 76, 19]
    out, sizes = sample_sort(a, 3, 2, random.Random(1))
    assert out == sorted(a)
    assert sum(sizes) == len(a) and len(sizes) == 3


def test_edges() -> None:
    assert sample_sort([], 4) == ([], [0, 0, 0, 0])
    assert sample_sort([5], 4)[0] == [5]
    assert sample_sort([2, 1], 1)[0] == [1, 2]
    with pytest.raises(ValueError):
        sample_sort([1], 0)


def test_bucket_bound_with_high_probability() -> None:
    rng = random.Random(8)
    for n, p in ((2000, 8), (5000, 16), (20000, 32)):
        a = rng.sample(range(10 * n), n)
        out, sizes = sample_sort(a, p, None, rng)
        assert out == sorted(a)
        assert max(sizes) <= math.ceil(2 * n / p)


def test_all_equal_keys_are_spread() -> None:
    out, sizes = sample_sort([7] * 4000, 8, None, random.Random(3))
    assert out == [7] * 4000
    assert max(sizes) <= 1000


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(2)
    a = [Key(rng.randint(0, 4), i) for i in range(300)]
    out, _ = sample_sort(a, 5, 4, rng)
    assert [(x.k, x.tag) for x in out] == sorted((x.k, x.tag) for x in a)


def test_random_against_sorted() -> None:
    rng = random.Random(10)
    for _ in range(200):
        a = [rng.randint(-50, 50) for _ in range(rng.randint(1, 90))]
        out, _ = sample_sort(a, rng.randint(1, 8), rng.randint(1, 5), rng)
        assert out == sorted(a)

What the next level needs

  • the tests did not run in this build

Read the full page about Sample sort