subms-bloom-filter 0.10.0

//! Minimal bloom filter - standalone, reusable, zero-dependency.
//!
//! Standard double-hashed bloom filter: FNV-1a 64-bit produces two 32-bit
//! subhashes for the double-hashing trick. Sizing defaults to ~10 bits per
//! key and k=7, which gives ~1% false-positive rate. Suitable as a building
//! block for other cookbook samples (LSM tree SSTables, in particular).
//!
//! ```
//! use subms_bloom_filter::BloomFilter;
//!
//! let mut bf = BloomFilter::new(10_000);
//! bf.add("alice");
//! assert!(bf.might_contain("alice"));   // stored keys always report present
//! assert!(!bf.might_contain("bob"));    // absent keys usually report absent
//! ```
//!
//! The on-disk layout is fixed and language-agnostic:
//!
//! ```text
//! bit_count: u32 (big-endian)
//! k:         u32 (big-endian)
//! words:     u32 (big-endian) - number of u64 words
//! bits:      (u64 big-endian) * words
//! ```
//!
//! Full writeup, design notes and measured benchmarks:
//! <https://www.submillisecond.com/cookbook/recipes/subms-bloom-filter>

#[cfg(feature = "harness")]
pub mod recipe;

// Opt-in feature modules. Each is independent of the base filter and
// gated by its own Cargo feature; `cargo add subms-bloom-filter` alone
// keeps the base zero-dep + std-only shape.
//
// See README and the cookbook page for the per-feature p99 numbers,
// memory cost, and composition guidance.
#[cfg(any(feature = "counting", feature = "scalable", feature = "partitioned"))]
pub mod features;

#[cfg(feature = "counting")]
pub use features::counting::CountingBloomFilter;
#[cfg(feature = "partitioned")]
pub use features::partitioned::PartitionedBloomFilter;
#[cfg(feature = "scalable")]
pub use features::scalable::ScalableBloomFilter;

use std::io::{self, Write};

pub(crate) const FNV_OFFSET: u64 = 0xcbf29ce484222325;
pub(crate) const FNV_PRIME: u64 = 0x100000001b3;

pub struct BloomFilter {
    bit_count: u32,
    k: u32,
    bits: Vec<u64>,
}

impl BloomFilter {
    /// Build an empty filter sized for `expected_entries` at ~1% FPR
    /// (10 bits/key, k=7). The 64-bit floor matters when expected_entries is small.
    pub fn new(expected_entries: usize) -> Self {
        let bit_count = expected_entries.saturating_mul(10).max(64) as u32;
        let words = (bit_count as usize).div_ceil(64);
        Self {
            bit_count,
            k: 7,
            bits: vec![0u64; words],
        }
    }

    pub fn add(&mut self, key: &str) {
        let h = fnv1a64(key);
        let h1 = h as u32;
        let h2 = ((h >> 32) as u32) | 1;
        for i in 0..self.k {
            let idx = h1.wrapping_add(i.wrapping_mul(h2)) % self.bit_count;
            self.bits[(idx / 64) as usize] |= 1u64 << (idx % 64);
        }
    }

    pub fn might_contain(&self, key: &str) -> bool {
        let h = fnv1a64(key);
        let h1 = h as u32;
        let h2 = ((h >> 32) as u32) | 1;
        for i in 0..self.k {
            let idx = h1.wrapping_add(i.wrapping_mul(h2)) % self.bit_count;
            if self.bits[(idx / 64) as usize] & (1u64 << (idx % 64)) == 0 {
                return false;
            }
        }
        true
    }

    pub fn bit_count(&self) -> u32 {
        self.bit_count
    }

    pub fn k(&self) -> u32 {
        self.k
    }

    /// Population count of the bit array. Walks every word, so keep it off
    /// the hot path; it is the input to both saturation estimators below.
    pub fn set_bits(&self) -> u64 {
        self.bits.iter().map(|w| w.count_ones() as u64).sum()
    }

    /// Swamidass-Baldi estimate of how many distinct keys were added:
    /// `-(m/k) * ln(1 - X/m)` for `X` set bits. Diverges once the array
    /// saturates, so a fully set filter reports `u64::MAX` rather than a
    /// number that reads as real.
    pub fn approximate_element_count(&self) -> u64 {
        let m = self.bit_count as f64;
        let x = self.set_bits() as f64;
        if x >= m {
            return u64::MAX;
        }
        let n = -(m / self.k as f64) * (1.0 - x / m).ln();
        n.round() as u64
    }

    /// Current false-positive probability given actual occupancy:
    /// `(X/m)^k`. This is the measured rate, not the design-point ~1%,
    /// so it is what tells you the filter has outgrown its sizing.
    pub fn estimated_fpp(&self) -> f64 {
        let ratio = self.set_bits() as f64 / self.bit_count as f64;
        ratio.powi(self.k as i32)
    }

    /// Two filters can be unioned only if they agree on `m` and `k` -
    /// the bit positions mean nothing otherwise.
    pub fn is_compatible(&self, other: &BloomFilter) -> bool {
        self.bit_count == other.bit_count
            && self.k == other.k
            && self.bits.len() == other.bits.len()
    }

    /// OR another filter's bits into this one. The result is the filter you
    /// would have built by adding both key sets to one array, which is what
    /// makes a shard-per-producer build mergeable at fan-in.
    pub fn union(&mut self, other: &BloomFilter) -> Result<(), GeometryMismatch> {
        if !self.is_compatible(other) {
            return Err(GeometryMismatch {
                lhs: (self.bit_count, self.k),
                rhs: (other.bit_count, other.k),
            });
        }
        for (dst, src) in self.bits.iter_mut().zip(&other.bits) {
            *dst |= *src;
        }
        Ok(())
    }

    /// Zero the bits, keeping the allocation. A generation boundary that
    /// rebuilds membership from a source of truth reuses the array instead
    /// of dropping and re-allocating it.
    pub fn clear(&mut self) {
        self.bits.fill(0);
    }

    pub fn write_to<W: Write>(&self, out: &mut W) -> io::Result<()> {
        out.write_all(&self.bit_count.to_be_bytes())?;
        out.write_all(&self.k.to_be_bytes())?;
        out.write_all(&(self.bits.len() as u32).to_be_bytes())?;
        for w in &self.bits {
            out.write_all(&w.to_be_bytes())?;
        }
        Ok(())
    }

    /// Parse a serialised bloom filter from `buf`. Errors if the buffer is
    /// shorter than the header or truncated mid-bits.
    pub fn parse(buf: &[u8]) -> io::Result<Self> {
        if buf.len() < 12 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "bloom section too short",
            ));
        }
        let bit_count = u32::from_be_bytes(buf[0..4].try_into().unwrap());
        let k = u32::from_be_bytes(buf[4..8].try_into().unwrap());
        let words = u32::from_be_bytes(buf[8..12].try_into().unwrap()) as usize;
        if buf.len() < 12 + words * 8 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "bloom section truncated",
            ));
        }
        let mut bits = Vec::with_capacity(words);
        for i in 0..words {
            let off = 12 + i * 8;
            bits.push(u64::from_be_bytes(buf[off..off + 8].try_into().unwrap()));
        }
        Ok(Self { bit_count, k, bits })
    }
}

/// Returned by [`BloomFilter::union`] when the two filters were sized
/// differently. Bit `i` of one filter has no relationship to bit `i` of the
/// other unless `m` and `k` match, so the merge is refused rather than
/// silently producing a filter with false negatives.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GeometryMismatch {
    /// `(bit_count, k)` of the filter being merged into.
    pub lhs: (u32, u32),
    /// `(bit_count, k)` of the filter being merged from.
    pub rhs: (u32, u32),
}

impl std::fmt::Display for GeometryMismatch {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "incompatible bloom geometry: m={} k={} vs m={} k={}",
            self.lhs.0, self.lhs.1, self.rhs.0, self.rhs.1
        )
    }
}

impl std::error::Error for GeometryMismatch {}

pub(crate) fn fnv1a64(key: &str) -> u64 {
    let mut h = FNV_OFFSET;
    for &b in key.as_bytes() {
        h ^= b as u64;
        h = h.wrapping_mul(FNV_PRIME);
    }
    h
}

// Unit tests live in colocated files (org convention: `<module>_tests.rs`
// alongside the module), not the top-level `tests/` dir. Zero-dep std-only,
// so no genuine integration tests exist.
#[cfg(test)]
#[path = "lib_tests.rs"]
mod lib_tests;

#[cfg(test)]
#[path = "sample_app_tests.rs"]
mod sample_app_tests;