Spectral Bloom filter in Python
Counting Bloom filter that estimates multiplicities of items in a multiset.
- File
spectral_bloom_filter.py, 73 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/spectral-bloom-filter/. The authors are listed in prov/MMBG5S.jsonl.
"""Spectral Bloom filter (Cohen and Matias 2003): minimum selection and minimal increase."""
import hashlib
Key = str | bytes | int
class SpectralBloomFilter:
"""m counters and k hash functions that estimate the multiplicity of a key.
estimate(x) is never below the true multiplicity of x. With
minimal_increase=False (minimum selection) the filter supports delete;
with minimal_increase=True (minimal increase) the estimates are never
larger, but delete is not allowed.
"""
def __init__(self, m: int, k: int, minimal_increase: bool = False, seed: int = 0) -> None:
if m < 1 or k < 1:
raise ValueError("m and k must be positive")
self.m = m
self.k = k
self.mi = minimal_increase
self.seed = seed
self.counters = [0] * m # @line L1
def _positions(self, key: Key) -> list[int]:
"""h_1(key), ..., h_k(key), from independent 64-bit words of BLAKE2b."""
out: list[int] = []
t = 0
while len(out) < self.k:
d = hashlib.blake2b(
repr(key).encode(),
key=self.seed.to_bytes(8, "little"),
salt=t.to_bytes(16, "little"),
).digest()
out.extend(int.from_bytes(d[w : w + 8], "little") % self.m for w in range(0, 64, 8))
t += 1
return out[: self.k] # @line L2
def _set(self, key: Key) -> list[int]:
"""The distinct positions of key, in a fixed order."""
return sorted(set(self._positions(key)))
def insert(self, key: Key, w: int = 1) -> None:
"""Add w copies of key."""
if w < 0:
raise ValueError("w must be nonnegative")
ps = self._set(key)
if not self.mi:
for p in ps: # @line L3
self.counters[p] += w
return
v = min(self.counters[p] for p in ps) # @line L4
for p in ps: # @line L5: raise only the counters below v + w
self.counters[p] = max(self.counters[p], v + w)
def delete(self, key: Key, w: int = 1) -> None:
"""Remove w copies of key. Precondition: at least w copies are present."""
if self.mi:
raise ValueError("minimal increase does not support delete")
for p in self._set(key): # @line L6
self.counters[p] -= w
def estimate(self, key: Key) -> int:
"""An upper bound on the multiplicity of key, exact with high probability."""
return min(self.counters[p] for p in self._set(key)) # @line L7
def query(self, key: Key, threshold: int = 1) -> bool:
"""True if key may have at least threshold copies; False means surely fewer."""
return self.estimate(key) >= threshold
test_spectral_bloom_filter.py 100 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/spectral-bloom-filter/. The authors are listed in prov/MMBG5S.jsonl.
import random
from collections import Counter
import pytest
from spectral_bloom_filter import Key, SpectralBloomFilter
TABLE: dict[Key, list[int]] = {
"a": [0, 3, 7],
"b": [3, 5, 9],
"c": [1, 5, 7],
"d": [0, 1, 9],
"e": [3, 5, 7],
}
STREAM = "aabcace"
class FixedSBF(SpectralBloomFilter):
def __init__(self, mi: bool) -> None:
super().__init__(10, 3, minimal_increase=mi)
def _positions(self, key: Key) -> list[int]:
return TABLE[key]
def test_example_minimum_selection() -> None:
f = FixedSBF(False)
for x in STREAM:
f.insert(x)
assert f.counters == [3, 2, 0, 5, 0, 4, 0, 6, 0, 1]
assert [f.estimate(x) for x in "abcde"] == [3, 1, 2, 1, 4]
f.delete("e")
assert [f.estimate(x) for x in "abcde"] == [3, 1, 2, 1, 3]
def test_example_minimal_increase() -> None:
f = FixedSBF(True)
for x in STREAM:
f.insert(x)
assert f.counters == [3, 2, 0, 3, 0, 3, 0, 3, 0, 1]
assert [f.estimate(x) for x in "abcde"] == [3, 1, 2, 1, 3]
assert f.query("a", 3) and not f.query("b", 2)
with pytest.raises(ValueError):
f.delete("e")
def test_edges() -> None:
f = SpectralBloomFilter(1, 4)
assert f.estimate("z") == 0 and not f.query("z")
f.insert("z", 5)
assert f.counters == [5] # one distinct position, counted once
f.insert("y", 0)
assert f.estimate("y") == 5
with pytest.raises(ValueError):
SpectralBloomFilter(0, 1)
with pytest.raises(ValueError):
f.insert("z", -1)
def test_random_against_counter() -> None:
rng = random.Random(2003)
for _ in range(500):
m = rng.randint(1, 40)
k = rng.randint(1, 5)
seed = rng.randrange(1000)
ms = SpectralBloomFilter(m, k, False, seed)
mi = SpectralBloomFilter(m, k, True, seed)
ms_only = SpectralBloomFilter(m, k, False, seed)
f: Counter[int] = Counter()
g: Counter[int] = Counter()
for _ in range(rng.randint(0, 60)):
x = rng.randrange(25)
w = rng.randint(0, 3)
ms.insert(x, w)
mi.insert(x, w)
f[x] += w
# a second filter with deletes, checked against its own multiset
if rng.random() < 0.3 and g[x] > 0:
d = rng.randint(1, g[x])
ms_only.delete(x, d)
g[x] -= d
else:
ms_only.insert(x, w)
g[x] += w
assert all(a <= b for a, b in zip(mi.counters, ms.counters))
h = {x: set(ms._set(x)) for x in range(30)}
def exact(mult: Counter[int], x: int) -> int:
# minimum selection is f_x plus the least collision mass on a position of x
return mult[x] + min(
sum(c for y, c in mult.items() if y != x and p in h[y]) for p in h[x]
)
for x in range(30):
assert ms.estimate(x) == exact(f, x)
assert f[x] <= mi.estimate(x) <= ms.estimate(x)
assert ms_only.estimate(x) == exact(g, x)
What the next level needs
- the tests did not run in this build