Bloom Filters

import math
import mmh3  # pip install mmh3 — MurmurHash3, a fast non-cryptographic hash

class BloomFilter:
    """Space-efficient probabilistic set membership filter.
    
    False positive rate is configurable. False negatives never occur.
    Uses the double-hashing trick (Kirsch-Mitzenmacher 2004) to generate
    k hash positions from just 2 base hashes.
    """

    def __init__(self, expected_items, fp_rate=0.01):
        """
        Args:
            expected_items: How many items you plan to add (n)
            fp_rate: Acceptable false positive probability (e.g., 0.01 = 1%)
        """
        # Calculate optimal bit array size: m = -n * ln(p) / (ln2)²
        self.m = int(-expected_items * math.log(fp_rate) / (math.log(2) ** 2))
        # Calculate optimal number of hash functions: k = (m/n) * ln(2)
        self.k = max(1, int((self.m / expected_items) * math.log(2)))
        # Bit array stored as a bytearray (8 bits per byte)
        self.bits = bytearray(self.m // 8 + 1)
        self.count = 0  # Track how many items have been added
        self.fp_rate = fp_rate

    def _get_positions(self, item):
        """Generate k bit positions using the double hashing trick.
        Two base hashes (with different seeds) generate all k positions:
        position_i = (h1 + i * h2) % m
        
        This is mathematically equivalent to k independent hash functions
        but only requires 2 hash computations."""
        s = str(item)
        h1 = mmh3.hash(s, 0) % self.m       # Base hash 1 (seed=0)
        h2 = mmh3.hash(s, 42) % self.m      # Base hash 2 (seed=42)
        return [(h1 + i * h2) % self.m for i in range(self.k)]

    def _set_bit(self, pos):
        """Set bit at position pos to 1."""
        byte_idx, bit_idx = divmod(pos, 8)
        self.bits[byte_idx] |= (1 << bit_idx)

    def _get_bit(self, pos):
        """Check if bit at position pos is 1."""
        byte_idx, bit_idx = divmod(pos, 8)
        return bool(self.bits[byte_idx] & (1 << bit_idx))

    def add(self, item):
        """Add an item to the filter. O(k) ≈ O(1).
        After this, might_contain(item) is guaranteed to return True."""
        for pos in self._get_positions(item):
            self._set_bit(pos)
        self.count += 1

    def might_contain(self, item):
        """Check if item was possibly added.
        Returns True  → item is PROBABLY in the set (could be false positive)
        Returns False → item is DEFINITELY NOT in the set (never wrong)"""
        for pos in self._get_positions(item):
            if not self._get_bit(pos):
                return False  # One bit is 0 → definitively absent
        return True  # All bits are 1 → probably present

    def estimated_fp_rate(self):
        """Estimate the current false positive rate based on how full the filter is."""
        # Count set bits
        set_bits = sum(bin(byte).count('1') for byte in self.bits)
        fill_ratio = set_bits / self.m
        return fill_ratio ** self.k

    def __len__(self):
        return self.count

    def __repr__(self):
        return (f"BloomFilter(n≈{self.count}, m={self.m} bits, "
                f"k={self.k} hashes, est_fp={self.estimated_fp_rate():.4%})")


# Usage
bf = BloomFilter(expected_items=10000, fp_rate=0.01)
print(bf)  # BloomFilter(n≈0, m=95851 bits, k=7 hashes, est_fp=0.0000%)

# Add items
for url in ["https://google.com", "https://github.com", "https://python.org"]:
    bf.add(url)

print(bf.might_contain("https://google.com"))    # True (correct — was added)
print(bf.might_contain("https://example.com"))    # False (correct — wasn't added)
print(bf.might_contain("https://evil-site.com"))  # False (probably — correct)
print(f"Using {bf.m} bits ({bf.m // 8 // 1024} KB) and {bf.k} hash functions")

class CountingBloomFilter:
    """Bloom filter variant that supports deletion.
    
    Instead of single bits, each position is a 4-bit counter (0-15).
    Adding increments the counters; deleting decrements them.
    Uses 4× more space than a standard Bloom filter.
    
    Overflow (counter > 15) is possible but extremely unlikely with
    properly sized filters. If it happens, the counter stays at 15
    (saturates) to avoid wraparound bugs.
    """

    def __init__(self, expected_items, fp_rate=0.01):
        import math
        self.m = int(-expected_items * math.log(fp_rate) / (math.log(2) ** 2))
        self.k = max(1, int((self.m / expected_items) * math.log(2)))
        # 4 bits per counter → each byte holds 2 counters
        self.counters = bytearray(self.m // 2 + 1)
        self.count = 0

    def _get_positions(self, item):
        import mmh3
        s = str(item)
        h1 = mmh3.hash(s, 0) % self.m
        h2 = mmh3.hash(s, 42) % self.m
        return [(h1 + i * h2) % self.m for i in range(self.k)]

    def _get_counter(self, pos):
        byte_idx = pos // 2
        if pos % 2 == 0:
            return self.counters[byte_idx] & 0x0F        # Lower nibble
        else:
            return (self.counters[byte_idx] >> 4) & 0x0F  # Upper nibble

    def _set_counter(self, pos, val):
        val = min(val, 15)  # Saturate at 15 to prevent overflow
        byte_idx = pos // 2
        if pos % 2 == 0:
            self.counters[byte_idx] = (self.counters[byte_idx] & 0xF0) | val
        else:
            self.counters[byte_idx] = (self.counters[byte_idx] & 0x0F) | (val << 4)

    def add(self, item):
        for pos in self._get_positions(item):
            self._set_counter(pos, self._get_counter(pos) + 1)
        self.count += 1

    def might_contain(self, item):
        return all(self._get_counter(pos) > 0 for pos in self._get_positions(item))

    def remove(self, item):
        """Remove an item. Only call if you're sure the item was added!
        Removing an item that was never added will corrupt the filter."""
        if not self.might_contain(item):
            return False
        for pos in self._get_positions(item):
            current = self._get_counter(pos)
            if current > 0 and current < 15:  # Don't decrement saturated counters
                self._set_counter(pos, current - 1)
        self.count -= 1
        return True


# Usage
cbf = CountingBloomFilter(1000, fp_rate=0.01)
cbf.add("hello")
cbf.add("world")
print(cbf.might_contain("hello"))  # True
cbf.remove("hello")
print(cbf.might_contain("hello"))  # False (successfully deleted!)

class BloomFilter {
  constructor(expectedItems, fpRate = 0.01) {
    // Calculate optimal size and hash count
    this.m = Math.ceil(-expectedItems * Math.log(fpRate) / (Math.log(2) ** 2));
    this.k = Math.max(1, Math.round((this.m / expectedItems) * Math.log(2)));
    this.bits = new Uint8Array(Math.ceil(this.m / 8));
    this.count = 0;
  }

  // FNV-1a hash variant with seed
  _hash(str, seed) {
    let h = 2166136261 ^ seed;
    for (let i = 0; i < str.length; i++) {
      h ^= str.charCodeAt(i);
      h = Math.imul(h, 16777619);  // 32-bit multiply
    }
    return Math.abs(h) % this.m;
  }

  // Generate k positions using double hashing
  _positions(item) {
    const s = String(item);
    const h1 = this._hash(s, 0);
    const h2 = this._hash(s, 42);
    return Array.from({ length: this.k }, (_, i) => (h1 + i * h2) % this.m);
  }

  add(item) {
    for (const pos of this._positions(item)) {
      this.bits[pos >> 3] |= (1 << (pos & 7));  // Set bit
    }
    this.count++;
  }

  mightContain(item) {
    return this._positions(item).every(pos =>
      (this.bits[pos >> 3] & (1 << (pos & 7))) !== 0  // Check bit
    );
  }

  get fillRatio() {
    let set = 0;
    for (const byte of this.bits) set += popcount(byte);
    return set / this.m;
  }
}

// Helper: count set bits in a byte
function popcount(n) {
  let count = 0;
  while (n) { count += n & 1; n >>= 1; }
  return count;
}

from collections import deque

class WebCrawler:
    """Simplified web crawler with Bloom filter deduplication.
    
    Without Bloom filter: 1 billion URLs × 50 bytes = ~50 GB.
    With Bloom filter at 1% FP: 1 billion × 10 bits = ~1.2 GB.
    The false positive cost: ~1% of valid URLs get skipped.
    """
    def __init__(self, seed_urls, max_pages=1000000):
        self.frontier = deque(seed_urls)
        self.seen = BloomFilter(expected_items=max_pages, fp_rate=0.01)
        self.pages_crawled = 0
        for url in seed_urls:
            self.seen.add(url)

    def crawl(self, max_iterations=100):
        while self.frontier and self.pages_crawled < max_iterations:
            url = self.frontier.popleft()
            new_links = self._fetch_and_extract(url)
            self.pages_crawled += 1
            for link in new_links:
                if not self.seen.might_contain(link):
                    self.seen.add(link)
                    self.frontier.append(link)
        return {
            "pages_crawled": self.pages_crawled,
            "urls_tracked": self.seen.count,
            "bloom_size_kb": self.seen.m // 8 // 1024,
        }

    def _fetch_and_extract(self, url):
        import hashlib
        return [f"https://example.com/{hashlib.md5(f'{url}-{i}'.encode()).hexdigest()[:8]}"
                for i in range(3)]

class SSTable:
    """Simulated sorted string table with Bloom filter guard.
    
    In real databases (Cassandra, RocksDB), each SSTable file
    on disk has an in-memory Bloom filter. Before doing a disk seek,
    the DB checks the filter. If it says 'no', the disk read is skipped.
    """
    def __init__(self, data, fp_rate=0.01):
        self.data = dict(data)  # The actual key-value pairs (on disk)
        self.bloom = BloomFilter(len(data), fp_rate)
        for key in data:
            self.bloom.add(key)
        self.disk_reads = 0

    def get(self, key):
        # First: check Bloom filter (in-memory, microseconds)
        if not self.bloom.might_contain(key):
            return None  # Definitely not here — skip disk read!

        # Bloom filter said "maybe" — do the expensive disk read
        self.disk_reads += 1
        return self.data.get(key)


# Simulate querying across multiple SSTables
tables = [
    SSTable([(f"user:{i}", f"data_{i}") for i in range(j*1000, (j+1)*1000)])
    for j in range(10)  # 10 SSTables, 1000 keys each
]

# Query a key — only 1 table does a disk read instead of all 10
for table in tables:
    result = table.get("user:5500")
    if result:
        print(f"Found: {result}")
        break

total_reads = sum(t.disk_reads for t in tables)
print(f"Disk reads: {total_reads} out of {len(tables)} tables checked")
# Typically: 1-2 disk reads instead of 10

class SafeBrowsingFilter:
    """Simplified version of Chrome's Safe Browsing check.
    
    The Bloom filter runs entirely in the browser (no network).
    Only positive results trigger a server round-trip.
    With 2 million malicious URLs at 0.01% FP rate:
    - Bloom filter size: ~4.8 MB (fits in browser memory)
    - 99.99% of safe URLs are confirmed safe instantly
    - Only 0.01% of safe URLs trigger an unnecessary server check
    """
    def __init__(self, malicious_urls):
        self.bloom = BloomFilter(len(malicious_urls), fp_rate=0.0001)
        for url in malicious_urls:
            self.bloom.add(url)

    def check_url(self, url):
        if not self.bloom.might_contain(url):
            return "safe"          # Guaranteed safe — no network needed
        # Bloom says "maybe malicious" — verify with server
        return self._verify_with_server(url)

    def _verify_with_server(self, url):
        # In real Chrome, this calls Google's Safe Browsing API
        # to confirm whether the URL is actually in the blacklist.
        # The Bloom filter eliminated 99.99%+ of these checks.
        return "checking_server..."

class RecommendationFilter:
    """Per-user Bloom filter to avoid showing already-seen content.
    
    Memory per user: ~2.4 KB (for 1000 articles at 1% FP)
    vs. hash set: ~50 KB (1000 article IDs × ~50 bytes)
    
    With 10 million users:
    - Bloom filters: ~24 GB total
    - Hash sets: ~500 GB total
    """
    def __init__(self, max_items_per_user=1000):
        self.user_filters = {}
        self.max_items = max_items_per_user

    def mark_seen(self, user_id, article_id):
        if user_id not in self.user_filters:
            self.user_filters[user_id] = BloomFilter(self.max_items, fp_rate=0.01)
        self.user_filters[user_id].add(article_id)

    def filter_recommendations(self, user_id, candidates):
        """Remove already-seen articles from recommendation candidates."""
        if user_id not in self.user_filters:
            return candidates  # New user — show everything
        bf = self.user_filters[user_id]
        return [article for article in candidates
                if not bf.might_contain(article)]
        # Occasional false positive means we skip a new article.
        # Better than showing something the user already read.