# /// script
# requires-python = ">=3.10"
# dependencies = ["numpy"]
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Dedup window size, and the Bloom filter trap.

(A) Memory for exact dedup at 100,000 keys per second, 32 bytes per entry,
for an hour and a week. (B) A Bloom filter sized for 1% false positives with
1,000,000 keys: how many new keys does it wrongly call duplicates? Then the
filter in front of an exact set.
"""
import argparse, json, math
import numpy as np

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--rate", type=int, default=100_000, help="Keys per second.")
p.add_argument("--entry-bytes", type=int, default=32, help="Bytes per dedup entry.")
p.add_argument("--keys", type=int, default=1_000_000, help="Keys inserted into the filter.")
p.add_argument("--fp", type=float, default=0.01, help="Target false-positive rate.")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
rng = np.random.default_rng(args.seed)
mem = [{"window": w, "seconds": s, "gb": args.rate * args.entry_bytes * s / 1e9} for w, s in (("1 hour", 3600), ("1 week", 604_800))]
for m in mem: print(f"exact dedup for {m['window']}: {m['gb']:,.1f} GB")

m_bits = math.ceil(-args.keys * math.log(args.fp) / math.log(2) ** 2); k = round(m_bits / args.keys * math.log(2))
bits = np.zeros(m_bits, dtype=bool)
def pos(keys, i): return ((keys * np.uint64(0x9E3779B97F4A7C15 + 2 * i)) >> np.uint64(17)) % np.uint64(m_bits)
seen = rng.integers(0, 2**62, args.keys, dtype=np.uint64); new = rng.integers(2**62, 2**63, args.keys, dtype=np.uint64)
for i in range(k): bits[pos(seen, i).astype(np.int64)] = True
hit = np.ones(args.keys, dtype=bool)
for i in range(k): hit &= bits[pos(new, i).astype(np.int64)]
dropped = int(hit.sum())
exact = set(seen.tolist()); both = sum(1 for x in new[hit].tolist() if x in exact)
print(f"Bloom filter alone: {dropped:,} of {args.keys:,} new keys wrongly dropped as duplicates ({dropped / args.keys:.2%})")
print(f"Bloom filter, then the exact set: {both} dropped; the filter saved {args.keys - dropped:,} exact lookups")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"memory": mem, "summary": [{"dropped": dropped, "dropped_pct": dropped / args.keys * 100, "with_exact": both,
                   "saved_lookups": args.keys - dropped}]}}, f)