Bloom Filter Join Mock in Python

import hashlib
import random
import string


class BloomFilter:
    def __init__(self, size: int = 200, num_hashes: int = 3):
        self.bits = [False] * size
        self.size = size
        self.num_hashes = num_hashes

    def _hashes(self, item: str):
        result = []
        for seed in range(self.num_hashes):
            h = hashlib.blake2b(item.encode(), digest_size=8, salt=str(seed).encode())
            value = int.from_bytes(h.digest(), byteorder="big") % self.size
            result.append(value)
        return result

    def add(self, item: str):
        for index in self._hashes(item):
            self.bits[index] = True

    def contains(self, item: str) -> bool:
        return all(self.bits[index] for index in self._hashes(item))


def bloom_join(left: list[str], right: list[str]) -> list[tuple[str, str]]:
    """Mock a hash join using a bloom filter to pre-filter the right table."""
    filter_ = BloomFilter()
    for value in left:
        filter_.add(value)
    
    candidates = [value for value in right if filter_.contains(value)]
    
    result = []
    for l_value in left:
        for r_value in candidates:
            if l_value == r_value:
                result.append((l_value, r_value))
    return result


if __name__ == "__main__":
    random.seed(42)
    
    table_a = [''.join(random.choices(string.ascii_lowercase, k=5)) for _ in range(20)]
    table_b = table_a[:10] + [''.join(random.choices(string.ascii_lowercase, k=5)) for _ in range(15)]
    
    matches = bloom_join(table_a, table_b)
    print(f"Bloom filter join matched {len(matches)} rows")
    for pair in matches[:5]:
        print(pair)