"""Original educational Bloom lab; Python 3.11+, no network/dependencies. Deterministic SHA-256 probes demonstrate membership semantics, not a production hash implementation, Redis compatibility, or a performance benchmark. """ import hashlib import json import math import platform class Bloom: def __init__(self, m, k): self.m, self.k = m, k self.bits = bytearray((m + 7) // 8) def positions(self, value): return [int.from_bytes(hashlib.sha256( (str(i) + ":" + value).encode("utf-8")).digest()[:8], "big") % self.m for i in range(self.k)] def add(self, value): for j in self.positions(value): self.bits[j // 8] |= 1 << (j % 8) def maybe(self, value): return all(self.bits[j // 8] & (1 << (j % 8)) for j in self.positions(value)) def run(): results = {} # Chosen positions, not a hash demonstration: A and B share bit 4. positions = {"A": [1, 4, 7], "B": [2, 4, 6], "C": [1, 2, 6], "D": [0, 2, 6]} bits = [0] * 8 for key in ("A", "B"): for j in positions[key]: bits[j] = 1 maybe = lambda key: all(bits[j] for j in positions[key]) assert maybe("A") and maybe("B") and maybe("C") and not maybe("D") results["toy"] = {"inserted": ["A", "B"], "bits": bits.copy(), "positions": positions, "absent_C_positive": maybe("C"), "absent_D_negative": not maybe("D")} for j in positions["A"]: bits[j] = 0 assert not maybe("B") results["unsafe_delete"] = {"B_still_in_set": True, "filter_says_B": maybe("B")} m, k = math.ceil(-1000 * math.log(.01) / math.log(2) ** 2), 7 bloom = Bloom(m, k) members = {f"item:{i}" for i in range(1000)} for key in members: bloom.add(key) assert all(bloom.maybe(key) for key in members) before = bytes(bloom.bits) bloom.add("item:0") assert before == bytes(bloom.bits) fp = sum(bloom.maybe(f"absent:{i}") for i in range(10000)) assert 0 < fp < 10000 # Not a statistical acceptance test of the target rate. results["finite_sample"] = {"m_bits": m, "k": k, "inserted": 1000, "inserted_false_negatives": 0, "duplicate_bits_unchanged": True, "absent_queries": 10000, "false_positives": fp, "observed_rate": fp / 10000} # Deliberately violate the sizing assumption, keeping the same bit array. for i in range(1000, 2000): bloom.add(f"item:{i}") fp2 = sum(bloom.maybe(f"absent:{i}") for i in range(10000)) assert fp2 > fp results["overfilled"] = {"inserted": 2000, "false_positives": fp2, "absent_queries": 10000, "observed_rate": fp2 / 10000, "approximate_rate": (1 - math.exp(-k * 2000 / m)) ** k} # DB contains a just-written row which has not been added to the filter. stale = Bloom(128, 3) stale.add("old") db = {"old", "new"} assert not stale.maybe("new") and "new" in db def lookup(key, coverage_ready): if coverage_ready and not stale.maybe(key): return False, "skipped" return key in db, "exact-store" assert lookup("new", True) == (False, "skipped") assert lookup("new", False) == (True, "exact-store") results["coverage_gap"] = {"naive_result": lookup("new", True), "coverage_not_ready_result": lookup("new", False)} results["sizing"] = [] for p in (.01, .001): bits_per_key = -math.log(p) / math.log(2) ** 2 results["sizing"].append({"n": 1000000, "target_p": p, "bits_per_key": bits_per_key, "ideal_MiB": bits_per_key * 1000000 / 8 / 2**20, "k_ceil": math.ceil(-math.log(p) / math.log(2))}) results["workload_expectations"] = [] for q in (.9, .05, .99): n, p = 10000, .01 results["workload_expectations"].append({"queries": n, "absent_fraction": q, "false_positive_rate": p, "expected_store_lookups": n*((1-q)+q*p), "expected_false_positives": n*q*p, "expected_avoided_lookups": n*q*(1-p), "false_fraction_among_positives": q*p/((1-q)+q*p)}) results["scope"] = "Finite synthetic membership checks and hypothetical arithmetic; no Redis/RocksDB/DB/network/performance measurement." return results if __name__ == "__main__": print(json.dumps({"python": platform.python_version(), "results": run()}, ensure_ascii=False, indent=2))