Algorithm Search⌘K

Swiss table in Python

Cache-friendly open-addressing table that uses SIMD group probing and control bytes.

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

from collections.abc import Callable, Hashable
from typing import Generic, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")

EMPTY = -128  # 0b10000000, as in Abseil
DELETED = -2  # 0b11111110
MASK64 = (1 << 64) - 1


def max_fill(capacity: int) -> int:
    """The most slots that may be full or deleted: 7/8 of the slots, and at least one stays EMPTY."""
    return min(capacity * 7 // 8, capacity - 1)


class SwissTable(Generic[K, V]):
    """A map in the style of the Swiss table: groups of slots, one control byte per slot.

    A control byte is EMPTY, DELETED, or the 7-bit H2 part of the hash of
    the key in the slot. The groups are aligned and probed in the triangular
    order g, g+1, g+3, g+6, ... modulo the number of groups, a power of two.
    The scan of one group stands for one SIMD compare. Each operation costs
    O(capacity) in the worst case and O(1) expected probes in practice.
    """

    def __init__(
        self,
        group_width: int = 16,
        groups: int = 1,
        hash_fn: Callable[[K], int] = hash,
    ) -> None:
        if group_width < 1:
            raise ValueError("group_width must be positive")
        if groups < 1 or groups & (groups - 1):
            raise ValueError("groups must be a power of two")
        self._w = group_width
        self._hash = hash_fn
        self._alloc(groups)

    def _alloc(self, groups: int) -> None:
        self._groups = groups
        cap = groups * self._w
        self._ctrl: list[int] = [EMPTY] * cap
        self._keys: list[K | None] = [None] * cap
        self._vals: list[V | None] = [None] * cap
        self._n = 0
        self._dead = 0

    def __len__(self) -> int:
        return self._n

    @property
    def capacity(self) -> int:
        return len(self._ctrl)

    def _split(self, k: K) -> tuple[int, int]:
        h = self._hash(k) & MASK64
        return h >> 7, h & 0x7F  # H1, H2

    def _slots(self, g: int) -> range:
        return range(g * self._w, (g + 1) * self._w)

    def find(self, k: K) -> int | None:
        """Return the slot of k, or None."""
        h1, h2 = self._split(k)  # @line L1
        g, i = h1 % self._groups, 0
        while True:  # @line L2
            for s in self._slots(g):  # @line L3: one SIMD compare in a real table
                if self._ctrl[s] == h2 and self._keys[s] == k:  # @line L4
                    return s
            if any(self._ctrl[s] == EMPTY for s in self._slots(g)):  # @line L5
                return None
            i += 1  # @line L6
            g = (g + i) % self._groups

    def get(self, k: K, default: V | None = None) -> V | None:
        s = self.find(k)
        return default if s is None else self._vals[s]

    def __contains__(self, k: K) -> bool:
        return self.find(k) is not None

    def insert(self, k: K, v: V) -> None:
        """Insert k with value v, or replace the value of k."""
        s = self.find(k)  # @line L7
        if s is not None:  # @line L8
            self._vals[s] = v
            return
        if self._n + self._dead + 1 > max_fill(self.capacity):  # @line L9
            self._rebuild()
        self._place(k, v)

    def _place(self, k: K, v: V) -> None:
        """Lines L10 to L15: put a key that is not present into the first free slot."""
        h1, h2 = self._split(k)  # @line L10
        g, i = h1 % self._groups, 0
        while all(self._ctrl[s] >= 0 for s in self._slots(g)):  # @line L11
            i += 1  # @line L12
            g = (g + i) % self._groups
        s = next(s for s in self._slots(g) if self._ctrl[s] < 0)  # @line L13
        if self._ctrl[s] == DELETED:  # @line L14
            self._dead -= 1
        self._ctrl[s], self._keys[s], self._vals[s] = h2, k, v  # @line L15
        self._n += 1

    def erase(self, k: K) -> bool:
        """Remove k. Return True if it was present."""
        s = self.find(k)  # @line L16
        if s is None:  # @line L17
            return False
        g = s // self._w
        if any(self._ctrl[t] == EMPTY for t in self._slots(g)):  # @line L18
            self._ctrl[s] = EMPTY
        else:
            self._ctrl[s] = DELETED  # @line L19
            self._dead += 1
        self._keys[s] = self._vals[s] = None
        self._n -= 1  # @line L20
        return True

    def _rebuild(self) -> None:
        """Rehash into a table with no tombstones, doubled until n + 1 fits in half of max_fill."""
        old = self.items()
        groups = self._groups
        while 2 * (self._n + 1) > max_fill(groups * self._w):  # @line L21
            groups *= 2
        self._alloc(groups)  # @line L22
        for k, v in old:
            self._place(k, v)

    def items(self) -> list[tuple[K, V]]:
        return [(self._keys[s], self._vals[s]) for s in range(self.capacity) if self._ctrl[s] >= 0]  # type: ignore[misc]

    def ctrl(self) -> list[int]:
        return list(self._ctrl)

    def check_invariant(self) -> None:
        """Check (S1) control bytes and counts, (S2) no EMPTY group before a key's group, (S3) an EMPTY slot."""
        n = dead = 0
        seen = set()
        for s in range(self.capacity):
            c = self._ctrl[s]
            if c == DELETED:
                dead += 1
                continue
            if c == EMPTY:
                continue
            assert 0 <= c < 128
            k = self._keys[s]
            assert k not in seen
            seen.add(k)
            n += 1
            h1, h2 = self._split(k)  # type: ignore[arg-type]
            assert c == h2
            g, i = h1 % self._groups, 0
            for _ in range(self._groups):
                if g == s // self._w:
                    break
                assert all(self._ctrl[t] != EMPTY for t in self._slots(g))
                i += 1
                g = (g + i) % self._groups
            else:
                raise AssertionError("slot not on the probe sequence")
        assert n == self._n and dead == self._dead
        assert n + dead <= max_fill(self.capacity) and EMPTY in self._ctrl
test_swiss_table.py 109 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/swiss-table/. The authors are listed in prov/5PH332.jsonl.

import random

from swiss_table import DELETED, EMPTY, SwissTable, max_fill


def ident(k: int) -> int:
    return k


def layout(t: SwissTable) -> list:
    c = t.ctrl()
    return ["." if x == EMPTY else "D" if x == DELETED else t._keys[s] for s, x in enumerate(c)]


def test_page_example() -> None:
    t = SwissTable(group_width=4, groups=4, hash_fn=ident)
    for k in [3, 7, 131, 11, 15, 19, 23, 27]:
        t.insert(k, -k)
        t.check_invariant()
    assert layout(t) == [3, 7, 11, 15, 131, 19, 23, 27] + ["."] * 8
    assert t.ctrl()[:8] == [3, 7, 11, 15, 3, 19, 23, 27]
    assert t.find(27) == 7 and t.get(27) == -27
    assert t.find(259) is None  # start group 2 holds an EMPTY slot
    assert t.find(1027) is None  # H2 = 3 matches 3 and 131, then group 3 stops it
    assert t.erase(7)
    assert layout(t)[:4] == [3, "D", 11, 15]
    assert t.find(27) == 7  # the tombstone keeps group 0 "full"
    assert t.erase(27)
    assert layout(t)[4:8] == [131, 19, 23, "D"]
    t.insert(35, -35)
    assert layout(t)[:4] == [3, 35, 11, 15]
    assert len(t) == 7 and t._dead == 1
    t.check_invariant()


def test_erase_to_empty_when_group_has_empty() -> None:
    t = SwissTable(group_width=4, groups=4, hash_fn=ident)
    for k in [3, 7]:
        t.insert(k, k)
    assert t.erase(3)
    assert t.ctrl()[0] == EMPTY and t._dead == 0
    assert 7 in t and 3 not in t
    t.check_invariant()


def test_edges() -> None:
    t = SwissTable(group_width=1, groups=1, hash_fn=ident)
    assert len(t) == 0 and t.find(5) is None and not t.erase(5)
    assert t.get(5, "x") == "x"
    t.insert(5, "a")
    t.insert(5, "b")
    assert len(t) == 1 and t.get(5) == "b"
    t.check_invariant()
    for bad in [(0, 1), (4, 3), (4, 0)]:
        try:
            SwissTable(group_width=bad[0], groups=bad[1])
        except ValueError:
            pass
        else:
            raise AssertionError(bad)
    assert max_fill(16) == 14 and max_fill(4) == 3 and max_fill(1) == 0


def test_all_keys_one_group_start() -> None:
    # Every key has the same H1, so they spread over all groups in probe order.
    t = SwissTable(group_width=2, groups=8, hash_fn=lambda k: k & 0x7F)
    for k in range(13):
        t.insert(k, k)
        t.check_invariant()
    for k in range(13):
        assert t.get(k) == k
    assert t.capacity == 16


def test_tombstones_force_rebuild_not_growth() -> None:
    t = SwissTable(group_width=4, groups=4, hash_fn=ident)
    for k in [128 * i for i in range(4)]:  # H1 = i, H2 = 0
        t.insert(k, k)
    cap = t.capacity
    for r in range(200):
        k = 10_000 + r
        t.insert(k, k)
        t.erase(k)
        t.check_invariant()
    assert t.capacity == cap and len(t) == 4


def test_random_against_dict() -> None:
    rng = random.Random(2017)
    for case in range(600):
        w = rng.choice([1, 2, 4, 8, 16])
        t: SwissTable = SwissTable(group_width=w, groups=1, hash_fn=(ident if case % 2 else hash))
        ref: dict = {}
        for _ in range(rng.randint(0, 80)):
            k = rng.randrange(60) * rng.choice([1, 128])
            op = rng.random()
            if op < 0.5:
                t.insert(k, case)
                ref[k] = case
            elif op < 0.8:
                assert t.erase(k) == (k in ref)
                ref.pop(k, None)
            else:
                assert t.get(k) == ref.get(k)
            t.check_invariant()
        assert sorted(t.items()) == sorted(ref.items())

What the next level needs

  • the tests did not run in this build

Read the full page about Swiss table