Counting Bloom filter in Python
Bloom filter with small counters in place of bits, allowing deletions.
- File
counting_bloom_filter.py, 60 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/counting-bloom-filter/. The authors are listed in prov/EHFR24.jsonl.
"""Counting Bloom filter (Fan, Cao, Almeida and Broder 2000)."""
import hashlib
Key = str | bytes | int
class CountingBloomFilter:
"""m counters of c bits each and k hash functions.
insert, delete and query take O(k) time. A counter that reaches the
largest value 2^c - 1 sticks there, so it never causes a false negative.
delete(x) is allowed only for a key x that is in the multiset of keys
inserted and not yet deleted; deleting another key can cause false
negatives, and the code cannot detect it.
"""
def __init__(self, m: int, k: int, c: int = 4, seed: int = 0) -> None:
if m < 1 or k < 1 or c < 1:
raise ValueError("m, k and c must be positive")
self.m = m
self.k = k
self.seed = seed
self.cmax = (1 << c) - 1
self.counters = [0] * m # @line L1
def _positions(self, key: Key) -> list[int]:
"""h_1(key), ..., h_k(key) by double hashing on BLAKE2b."""
d = hashlib.blake2b(
repr(key).encode(), digest_size=16, key=self.seed.to_bytes(8, "little")
).digest()
h1 = int.from_bytes(d[:8], "little")
h2 = int.from_bytes(d[8:], "little") | 1
return [(h1 + i * h2) % self.m for i in range(self.k)] # @line L2
def insert(self, key: Key) -> None:
"""Add one copy of key."""
for p in self._positions(key): # @line L3
if self.counters[p] < self.cmax: # @line L4: a full counter sticks
self.counters[p] += 1
def delete(self, key: Key) -> None:
"""Remove one copy of key. Precondition: a copy of key is present."""
for p in self._positions(key): # @line L5
if self.counters[p] < self.cmax: # @line L6: a stuck counter stays
self.counters[p] -= 1
def query(self, key: Key) -> bool:
"""False if key is surely absent; True if it may be present."""
for p in self._positions(key): # @line L7
if self.counters[p] == 0: # @line L8
return False
return True # @line L9
def saturated(self) -> int:
"""The number of counters stuck at the largest value."""
return sum(1 for v in self.counters if v == self.cmax)
test_counting_bloom_filter.py 92 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/counting-bloom-filter/. The authors are listed in prov/EHFR24.jsonl.
import random
from collections import Counter
import pytest
from counting_bloom_filter import CountingBloomFilter, Key
class FixedCBF(CountingBloomFilter):
def __init__(self, table: dict[Key, list[int]], c: int = 4) -> None:
super().__init__(10, 3, c)
self.table = table
def _positions(self, key: Key) -> list[int]:
return self.table[key]
TABLE: dict[Key, list[int]] = {"a": [0, 3, 7], "b": [3, 5, 9], "c": [1, 5, 7], "d": [0, 1, 9]}
def test_example() -> None:
f = FixedCBF(TABLE)
f.insert("a")
assert f.counters == [1, 0, 0, 1, 0, 0, 0, 1, 0, 0]
f.insert("b")
assert f.counters == [1, 0, 0, 2, 0, 1, 0, 1, 0, 1]
f.insert("c")
assert f.counters == [1, 1, 0, 2, 0, 2, 0, 2, 0, 1]
assert f.query("d") # a false positive: 0, 1 and 9 are all nonzero
f.delete("b")
assert f.counters == [1, 1, 0, 1, 0, 1, 0, 2, 0, 0]
assert not f.query("b") and not f.query("d")
assert f.query("a") and f.query("c")
def test_saturation() -> None:
f = FixedCBF(TABLE, c=2) # counters stick at 3
for _ in range(4):
f.insert("a")
assert f.counters[0] == 3 and f.saturated() == 3
for _ in range(4):
f.delete("a")
assert f.counters[0] == 3 # stuck, so no false negative is possible
assert f.query("a")
def test_edges() -> None:
f = CountingBloomFilter(1, 1)
assert not f.query(5)
f.insert(5)
f.insert(5)
f.delete(5)
assert f.query(5)
f.delete(5)
assert not f.query(5) and f.counters == [0]
with pytest.raises(ValueError):
CountingBloomFilter(4, 2, c=0)
def test_random_against_multiset() -> None:
rng = random.Random(2000)
for _ in range(500):
m = rng.randint(1, 60)
k = rng.randint(1, 5)
c = rng.randint(1, 4)
f = CountingBloomFilter(m, k, c, seed=rng.randrange(100))
live: Counter[int] = Counter()
ever_saturated = False
for _ in range(rng.randint(0, 60)):
x = rng.randrange(30)
r = rng.random()
if r < 0.45:
f.insert(x)
live[x] += 1
elif r < 0.8 and live[x] > 0:
f.delete(x)
live[x] -= 1
else:
assert f.query(x) or live[x] == 0
ever_saturated = ever_saturated or f.saturated() > 0
assert all(f.query(y) for y in live if live[y] > 0)
assert all(0 <= v <= f.cmax for v in f.counters)
if not ever_saturated:
# with no saturation, each counter is the exact count of positions
want = [0] * m
for y, mult in live.items():
for p in f._positions(y):
want[p] += mult
assert f.counters == want
What the next level needs
- the tests did not run in this build