Algorithm Search⌘K

External samplesort in Python

Pick splitters from a sample, partition records into buckets in one pass, and sort each bucket recursively.

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

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)


def blocks(length: int, b: int) -> int:
    """The number of blocks of b items that hold a file of the given length."""
    return -(-length // b)


def choose_splitters(sample: list[T], a: int, s: int) -> list[T]:
    """The distinct items of ranks a, 2a, ..., (s - 1)a of the sorted sample, in order."""
    sample = sorted(sample)
    out: list[T] = []
    for r in range(a, min(s * a, len(sample) + 1), a):
        x = sample[r - 1]
        if not out or out[-1] < x:
            out.append(x)
    return out


def distribute(f: list[T], splitters: list[T]) -> list[list[T]]:
    """The buckets U_0, E_1, U_1, ..., E_k, U_k, each in the order of f.

    U_j holds the items strictly between s_j and s_(j+1), and E_j the items equal to s_j.
    """
    k = len(splitters)
    out: list[list[T]] = [[] for _ in range(2 * k + 1)]
    for x in f:
        j = bisect.bisect_left(splitters, x)
        if j < k and not x < splitters[j]:
            out[2 * j + 1].append(x)  # x equals s_(j+1)
        else:
            out[2 * j].append(x)
    return out


def external_samplesort(f: list[T], m: int = 16, b: int = 2, seed: int = 0, a: int = 0) -> int:
    """Sort f in place as an external samplesort with a memory of m items and blocks of b items.

    Returns the number of block I/Os. Precondition: b >= 1 and m >= 4 * b.
    A read or write of a file of length L costs ceil(L / b) I/Os, and the
    sample costs one read per distinct block that holds a sampled position.
    a is the oversampling factor; 0 means ceil(8 ln S). The sort is stable.
    With S = m / (2b) buckets per level the expected cost is O(n log_m n).
    """
    if b < 1 or m < 4 * b:
        raise ValueError("need b >= 1 and m >= 4 * b")
    rng = random.Random(seed)

    def sort(g: list[T]) -> tuple[list[T], int]:
        if len(g) <= m:  # @line L1
            return sorted(g), 2 * blocks(len(g), b)
        s = max(2, min(m // (2 * b), -(-4 * len(g) // m)))  # @line L2
        over = a if a > 0 else math.ceil(8 * math.log(s))
        pos = [rng.randrange(len(g)) for _ in range(over * s)]  # @line L3
        ios = len({p // b for p in pos})
        splitters = choose_splitters([g[p] for p in pos], over, s)  # @line L4
        buckets = distribute(g, splitters)  # @line L5
        ios += blocks(len(g), b) + sum(blocks(len(u), b) for u in buckets)
        out: list[T] = []
        for i, u in enumerate(buckets):  # @line L6
            if i % 2 == 0:
                u, c = sort(u)  # @line L7
                ios += c
            out.extend(u)  # @line L8
        return out, ios

    result, total = sort(list(f))
    f[:] = result
    return total
test_external_samplesort.py 78 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/external-samplesort/. The authors are listed in prov/PAB348.jsonl.

import math
import random

import pytest

from external_samplesort import choose_splitters, distribute, external_samplesort

F = [14, 3, 9, 3, 17, 6, 12, 3, 20, 1, 8, 15, 3, 11, 5, 18, 7, 3, 16, 10]


def test_example_buckets() -> None:
    sample = [F[p] for p in (0, 1, 6, 10, 12, 15)]
    splitters = choose_splitters(sample, 2, 3)
    assert splitters == [3, 12]
    assert distribute(F, splitters) == [
        [1], [3, 3, 3, 3, 3], [9, 6, 8, 11, 5, 7, 10], [12], [14, 17, 20, 15, 18, 16],
    ]


def test_example_sort() -> None:
    f = F[:]
    external_samplesort(f, 12, 2, a=2)
    assert f == sorted(F)


def test_edges() -> None:
    for g in ([], [1], [7] * 200, list(range(300)), list(range(300, 0, -1))):
        f = g[:]
        external_samplesort(f, 16, 2)
        assert f == sorted(g)
    with pytest.raises(ValueError):
        external_samplesort([1], 7, 2)


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(5)
    for _ in range(60):
        a = [Key(rng.randint(0, 5), i) for i in range(rng.randint(0, 200))]
        want = sorted((x.k, x.tag) for x in a)
        external_samplesort(a, 16, 2, seed=rng.randint(0, 10**6))
        assert [(x.k, x.tag) for x in a] == want


def test_bucket_bound() -> None:
    rng = random.Random(8)
    s, n = 16, 4000
    a = math.ceil(8 * math.log(s))
    limit = 2 * -(-2 * n // s)
    good = 0
    for _ in range(200):
        g = [rng.random() for _ in range(n)]
        sample = [g[rng.randrange(n)] for _ in range(a * s)]
        buckets = distribute(g, choose_splitters(sample, a, s))
        good += all(len(u) < limit for u in buckets[0::2])
    assert good >= 190


def test_random_against_sorted() -> None:
    rng = random.Random(1970)
    for _ in range(600):
        n = rng.randint(0, 150)
        b = rng.randint(1, 3)
        m = b * rng.randint(4, 8)
        f = [rng.randint(-30, 30) for _ in range(n)]
        want = sorted(f)
        ios = external_samplesort(f, m, b, seed=rng.randint(0, 10**6))
        assert f == want
        assert ios >= 2 * -(-n // b)

What the next level needs

  • the tests did not run in this build

Read the full page about External samplesort