Robin Hood hashing in Python
Linear probing that swaps keys so probe lengths are evened out.
- File
robin_hood_hashing.py, 139 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/robin-hood-hashing/. The authors are listed in prov/CR9J17.jsonl.
from collections.abc import Callable, Hashable
from typing import Generic, TypeVar
K = TypeVar("K", bound=Hashable)
V = TypeVar("V")
class RobinHoodTable(Generic[K, V]):
"""A map with Robin Hood linear probing and backward shift deletion.
Slot j is empty when dist[j] is -1; else it holds keys[j], vals[j] and
dist[j] = (j - h(keys[j])) mod m. The table grows (doubles) before the
number of keys passes max_load * m, and it always keeps two empty slots
free for an insertion. Each operation costs O(1) expected time for a
random hash and a fixed max_load below 1, and O(m) in the worst case.
"""
def __init__(
self,
capacity: int = 8,
max_load: float = 0.875,
hash_fn: Callable[[K], int] = hash,
) -> None:
if capacity < 2:
raise ValueError("capacity must be at least 2")
if not 0.0 < max_load < 1.0:
raise ValueError("max_load must be in (0, 1)")
self._hash = hash_fn
self._max_load = max_load
self._alloc(capacity)
def _alloc(self, m: int) -> None:
self._keys: list[K | None] = [None] * m
self._vals: list[V | None] = [None] * m
self._dist: list[int] = [-1] * m
self._n = 0
def __len__(self) -> int:
return self._n
@property
def capacity(self) -> int:
return len(self._keys)
def _home(self, k: K) -> int:
return self._hash(k) % len(self._keys)
def search(self, k: K) -> int | None:
"""Return the slot of k, or None."""
m = len(self._keys)
j, d = self._home(k), 0 # @line L1
while True: # @line L2
if self._dist[j] < d: # @line L3: an empty slot has dist -1 < d
return None
if self._keys[j] == k: # @line L4
return j
j, d = (j + 1) % m, d + 1 # @line L5
def get(self, k: K, default: V | None = None) -> V | None:
j = self.search(k)
return default if j is None else self._vals[j]
def __contains__(self, k: K) -> bool:
return self.search(k) is not None
def put(self, k: K, v: V) -> None:
"""Insert k with value v, or replace the value of k."""
j = self.search(k)
if j is not None:
self._vals[j] = v
return
m = len(self._keys)
if self._n + 1 > self._max_load * m or self._n + 2 > m - 1:
self._grow()
self._insert(k, v)
def _insert(self, k: K, v: V) -> None:
"""RH-INSERT: k is not in the table, and two slots are empty."""
m = len(self._keys)
x, val, j, d = k, v, self._home(k), 0 # @line L6
while True: # @line L7
if self._dist[j] < 0: # @line L8
self._keys[j], self._vals[j], self._dist[j] = x, val, d
self._n += 1
return
if self._dist[j] < d: # @line L9: the poorer key takes the slot
x, self._keys[j] = self._keys[j], x # type: ignore[assignment]
val, self._vals[j] = self._vals[j], val # type: ignore[assignment]
d, self._dist[j] = self._dist[j], d
j, d = (j + 1) % m, d + 1 # @line L10
def delete(self, k: K) -> bool:
"""Remove k with a backward shift. Return True if k was present."""
m = len(self._keys)
j = self.search(k) # @line L11
if j is None: # @line L12
return False
i = (j + 1) % m # @line L13
while self._dist[i] > 0: # @line L14: empty is -1, home slot is 0
self._keys[j], self._vals[j] = self._keys[i], self._vals[i] # @line L15
self._dist[j] = self._dist[i] - 1
j, i = i, (i + 1) % m # @line L16
self._keys[j], self._vals[j], self._dist[j] = None, None, -1 # @line L17
self._n -= 1
return True
def _grow(self) -> None:
old = [(self._keys[j], self._vals[j]) for j in range(len(self._keys)) if self._dist[j] >= 0]
self._alloc(2 * len(self._keys))
for k, v in old:
self._insert(k, v) # type: ignore[arg-type]
def items(self) -> list[tuple[K, V]]:
return [(self._keys[j], self._vals[j]) for j in range(len(self._keys)) if self._dist[j] >= 0] # type: ignore[misc]
def distances(self) -> list[int]:
"""The distance of each key, in slot order."""
return [d for d in self._dist if d >= 0]
def check_invariant(self) -> None:
"""Check the stored distances, invariant (R), one empty slot, and the count."""
m = len(self._keys)
n = 0
seen = set()
for j in range(m):
d = self._dist[j]
if d < 0:
continue
n += 1
k = self._keys[j]
assert k not in seen
seen.add(k)
assert d == (j - self._home(k)) % m # type: ignore[arg-type]
if d > 0:
assert self._dist[j - 1] >= d - 1 # also fails if slot j - 1 is empty
assert n == self._n and n < m
test_robin_hood_hashing.py 113 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/robin-hood-hashing/. The authors are listed in prov/CR9J17.jsonl.
import random
import pytest
from robin_hood_hashing import RobinHoodTable
def ident(k: int) -> int:
return k
def linear_probing_distances(keys: list[int], m: int) -> list[int]:
"""Plain first-come linear probing, as an independent oracle for the sum of distances."""
slot: list[int | None] = [None] * m
for k in keys:
j = k % m
while slot[j] is not None:
j = (j + 1) % m
slot[j] = k
return [(j - k % m) % m for j, k in enumerate(slot) if k is not None]
def test_example() -> None:
t: RobinHoodTable[int, int] = RobinHoodTable(capacity=8, max_load=0.875, hash_fn=ident)
for k in (8, 1, 16, 2, 24, 9):
t.put(k, 10 * k)
t.check_invariant()
assert t._keys == [8, 16, 24, 1, 9, 2, None, None]
assert t._dist == [0, 1, 2, 2, 3, 3, -1, -1]
assert t.search(9) == 4
assert t.search(17) is None and t.search(32) is None
assert t.delete(16)
assert t._keys == [8, 24, 1, 9, 2, None, None, None]
assert t._dist == [0, 1, 1, 2, 2, -1, -1, -1]
assert t.get(9) == 90 and t.get(2) == 20
t.check_invariant()
def test_same_total_less_spread_than_linear_probing() -> None:
keys = [8, 1, 16, 2, 24, 9]
plain = linear_probing_distances(keys, 8)
t: RobinHoodTable[int, int] = RobinHoodTable(capacity=8, max_load=0.875, hash_fn=ident)
for k in keys:
t.put(k, k)
rh = t.distances()
assert sum(plain) == sum(rh) == 11
assert sum(d * d for d in plain) == 37 and sum(d * d for d in rh) == 27
assert max(plain) == 4 and max(rh) == 3
def test_edges() -> None:
t: RobinHoodTable[int, str] = RobinHoodTable(capacity=2)
assert len(t) == 0 and t.get(1) is None and not t.delete(1)
t.put(1, "a")
t.put(1, "b")
assert len(t) == 1 and t.get(1) == "b"
assert t.delete(1) and len(t) == 0 and 1 not in t
t.check_invariant()
with pytest.raises(ValueError):
RobinHoodTable(capacity=1)
def test_wrap_around_and_one_home() -> None:
t: RobinHoodTable[int, int] = RobinHoodTable(capacity=8, max_load=0.875, hash_fn=lambda k: 6)
for k in range(5):
t.put(k, k)
t.check_invariant()
assert t.distances() == [2, 3, 4, 0, 1] # slots 0, 1, 2 wrap around after 6, 7
t.delete(0)
t.check_invariant()
assert sorted(k for k, _ in t.items()) == [1, 2, 3, 4]
def test_random_against_dict() -> None:
rng = random.Random(1985)
for trial in range(500):
m = rng.choice([2, 4, 8, 16])
hash_fn = (lambda k: k % 5) if trial % 4 == 0 else hash
t: RobinHoodTable[int, int] = RobinHoodTable(capacity=m, max_load=rng.choice([0.5, 0.75, 0.9]), hash_fn=hash_fn)
model: dict[int, int] = {}
for _ in range(rng.randint(0, 80)):
k = rng.randint(-30, 30)
op = rng.random()
if op < 0.55:
v = rng.randint(0, 99)
t.put(k, v)
model[k] = v
elif op < 0.8:
assert t.delete(k) == (k in model)
model.pop(k, None)
else:
assert t.get(k) == model.get(k)
t.check_invariant()
assert len(t) == len(model)
assert sorted(t.items()) == sorted(model.items())
def test_random_sum_and_squares_against_linear_probing() -> None:
rng = random.Random(2004)
for _ in range(500):
m = rng.choice([4, 8, 16, 32])
keys = rng.sample(range(200), rng.randint(0, m - 2))
t: RobinHoodTable[int, int] = RobinHoodTable(capacity=m, max_load=0.99, hash_fn=ident)
for k in keys:
t.put(k, k)
assert t.capacity == m
plain = linear_probing_distances(keys, m)
rh = t.distances()
assert sum(rh) == sum(plain)
assert sum(d * d for d in rh) <= sum(d * d for d in plain)
What the next level needs
- the tests did not run in this build