Struct BloomFilter 

pub struct BloomFilter { /* private fields */ }

Implementations

impl BloomFilter

pub fn new(expected_entries: usize) -> Self

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.

Examples found in repository

examples/sample_app.rs (line 33)

fn base_crawler_dedup() {
    println!("== base: crawler URL dedup ==");
    let mut seen = BloomFilter::new(10_000);
    let frontier = [
        "https://a.example/",
        "https://b.example/",
        "https://a.example/", // dup
        "https://c.example/",
        "https://b.example/", // dup
    ];

    let (mut fetched, mut skipped) = (0usize, 0usize);
    for url in frontier {
        if seen.might_contain(url) {
            println!("  skip  {url}");
            skipped += 1;
        } else {
            seen.add(url);
            println!("  fetch {url}");
            fetched += 1;
        }
    }
    println!("  -> fetched {fetched}, skipped {skipped}");
    for url in [
        "https://a.example/",
        "https://b.example/",
        "https://c.example/",
    ] {
        assert!(seen.might_contain(url), "no false negatives");
    }
}

/// Each shard builds its own filter over the symbols it saw, then the gateway
/// ORs them into one membership set. `union` only accepts identical geometry,
/// so every shard must be constructed with the same expected count.
/// `estimated_fpp` reports occupancy against the design point, which is how you
/// find out a filter has outgrown its sizing before the false positives do.
fn shard_merge_and_occupancy() {
    println!(
        "
== merge: per-shard filters unioned at the gateway =="
    );
    let capacity = 10_000;
    let mut gateway = BloomFilter::new(capacity);
    for shard in 0..4 {
        let mut local = BloomFilter::new(capacity);
        for i in 0..500 {
            local.add(&format!("shard{shard}-sym{i}"));
        }
        gateway.union(&local).expect("shards share one geometry");
    }
    println!("  merged 4 shards x 500 symbols");
    println!(
        "  approx distinct keys: {}",
        gateway.approximate_element_count()
    );
    println!(
        "  occupancy fpp:        {:.4}%",
        gateway.estimated_fpp() * 100.0
    );

    let mismatched = BloomFilter::new(capacity * 2);
    match gateway.union(&mismatched) {
        Err(e) => println!("  refused mismatched shard: {e}"),
        Ok(()) => unreachable!("geometry check must reject this"),
    }

    gateway.clear();
    println!(
        "  after clear -> approx distinct keys: {}",
        gateway.approximate_element_count()
    );
}

pub fn add(&mut self, key: &str)

Examples found in repository

examples/sample_app.rs (line 48)

fn base_crawler_dedup() {
    println!("== base: crawler URL dedup ==");
    let mut seen = BloomFilter::new(10_000);
    let frontier = [
        "https://a.example/",
        "https://b.example/",
        "https://a.example/", // dup
        "https://c.example/",
        "https://b.example/", // dup
    ];

    let (mut fetched, mut skipped) = (0usize, 0usize);
    for url in frontier {
        if seen.might_contain(url) {
            println!("  skip  {url}");
            skipped += 1;
        } else {
            seen.add(url);
            println!("  fetch {url}");
            fetched += 1;
        }
    }
    println!("  -> fetched {fetched}, skipped {skipped}");
    for url in [
        "https://a.example/",
        "https://b.example/",
        "https://c.example/",
    ] {
        assert!(seen.might_contain(url), "no false negatives");
    }
}

/// Each shard builds its own filter over the symbols it saw, then the gateway
/// ORs them into one membership set. `union` only accepts identical geometry,
/// so every shard must be constructed with the same expected count.
/// `estimated_fpp` reports occupancy against the design point, which is how you
/// find out a filter has outgrown its sizing before the false positives do.
fn shard_merge_and_occupancy() {
    println!(
        "
== merge: per-shard filters unioned at the gateway =="
    );
    let capacity = 10_000;
    let mut gateway = BloomFilter::new(capacity);
    for shard in 0..4 {
        let mut local = BloomFilter::new(capacity);
        for i in 0..500 {
            local.add(&format!("shard{shard}-sym{i}"));
        }
        gateway.union(&local).expect("shards share one geometry");
    }
    println!("  merged 4 shards x 500 symbols");
    println!(
        "  approx distinct keys: {}",
        gateway.approximate_element_count()
    );
    println!(
        "  occupancy fpp:        {:.4}%",
        gateway.estimated_fpp() * 100.0
    );

    let mismatched = BloomFilter::new(capacity * 2);
    match gateway.union(&mismatched) {
        Err(e) => println!("  refused mismatched shard: {e}"),
        Ok(()) => unreachable!("geometry check must reject this"),
    }

    gateway.clear();
    println!(
        "  after clear -> approx distinct keys: {}",
        gateway.approximate_element_count()
    );
}

pub fn might_contain(&self, key: &str) -> bool

Examples found in repository

examples/sample_app.rs (line 44)

fn base_crawler_dedup() {
    println!("== base: crawler URL dedup ==");
    let mut seen = BloomFilter::new(10_000);
    let frontier = [
        "https://a.example/",
        "https://b.example/",
        "https://a.example/", // dup
        "https://c.example/",
        "https://b.example/", // dup
    ];

    let (mut fetched, mut skipped) = (0usize, 0usize);
    for url in frontier {
        if seen.might_contain(url) {
            println!("  skip  {url}");
            skipped += 1;
        } else {
            seen.add(url);
            println!("  fetch {url}");
            fetched += 1;
        }
    }
    println!("  -> fetched {fetched}, skipped {skipped}");
    for url in [
        "https://a.example/",
        "https://b.example/",
        "https://c.example/",
    ] {
        assert!(seen.might_contain(url), "no false negatives");
    }
}

pub fn bit_count(&self) -> u32

pub fn k(&self) -> u32

pub fn set_bits(&self) -> u64

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 approximate_element_count(&self) -> u64

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.

Examples found in repository

examples/sample_app.rs (line 85)

fn shard_merge_and_occupancy() {
    println!(
        "
== merge: per-shard filters unioned at the gateway =="
    );
    let capacity = 10_000;
    let mut gateway = BloomFilter::new(capacity);
    for shard in 0..4 {
        let mut local = BloomFilter::new(capacity);
        for i in 0..500 {
            local.add(&format!("shard{shard}-sym{i}"));
        }
        gateway.union(&local).expect("shards share one geometry");
    }
    println!("  merged 4 shards x 500 symbols");
    println!(
        "  approx distinct keys: {}",
        gateway.approximate_element_count()
    );
    println!(
        "  occupancy fpp:        {:.4}%",
        gateway.estimated_fpp() * 100.0
    );

    let mismatched = BloomFilter::new(capacity * 2);
    match gateway.union(&mismatched) {
        Err(e) => println!("  refused mismatched shard: {e}"),
        Ok(()) => unreachable!("geometry check must reject this"),
    }

    gateway.clear();
    println!(
        "  after clear -> approx distinct keys: {}",
        gateway.approximate_element_count()
    );
}

pub fn estimated_fpp(&self) -> f64

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.

Examples found in repository

examples/sample_app.rs (line 89)

fn shard_merge_and_occupancy() {
    println!(
        "
== merge: per-shard filters unioned at the gateway =="
    );
    let capacity = 10_000;
    let mut gateway = BloomFilter::new(capacity);
    for shard in 0..4 {
        let mut local = BloomFilter::new(capacity);
        for i in 0..500 {
            local.add(&format!("shard{shard}-sym{i}"));
        }
        gateway.union(&local).expect("shards share one geometry");
    }
    println!("  merged 4 shards x 500 symbols");
    println!(
        "  approx distinct keys: {}",
        gateway.approximate_element_count()
    );
    println!(
        "  occupancy fpp:        {:.4}%",
        gateway.estimated_fpp() * 100.0
    );

    let mismatched = BloomFilter::new(capacity * 2);
    match gateway.union(&mismatched) {
        Err(e) => println!("  refused mismatched shard: {e}"),
        Ok(()) => unreachable!("geometry check must reject this"),
    }

    gateway.clear();
    println!(
        "  after clear -> approx distinct keys: {}",
        gateway.approximate_element_count()
    );
}

pub fn is_compatible(&self, other: &BloomFilter) -> bool

Two filters can be unioned only if they agree on m and k - the bit positions mean nothing otherwise.

pub fn union(&mut self, other: &BloomFilter) -> Result<(), GeometryMismatch>

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.

Examples found in repository

examples/sample_app.rs (line 80)

fn shard_merge_and_occupancy() {
    println!(
        "
== merge: per-shard filters unioned at the gateway =="
    );
    let capacity = 10_000;
    let mut gateway = BloomFilter::new(capacity);
    for shard in 0..4 {
        let mut local = BloomFilter::new(capacity);
        for i in 0..500 {
            local.add(&format!("shard{shard}-sym{i}"));
        }
        gateway.union(&local).expect("shards share one geometry");
    }
    println!("  merged 4 shards x 500 symbols");
    println!(
        "  approx distinct keys: {}",
        gateway.approximate_element_count()
    );
    println!(
        "  occupancy fpp:        {:.4}%",
        gateway.estimated_fpp() * 100.0
    );

    let mismatched = BloomFilter::new(capacity * 2);
    match gateway.union(&mismatched) {
        Err(e) => println!("  refused mismatched shard: {e}"),
        Ok(()) => unreachable!("geometry check must reject this"),
    }

    gateway.clear();
    println!(
        "  after clear -> approx distinct keys: {}",
        gateway.approximate_element_count()
    );
}

pub fn clear(&mut self)

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.

Examples found in repository

examples/sample_app.rs (line 98)

fn shard_merge_and_occupancy() {
    println!(
        "
== merge: per-shard filters unioned at the gateway =="
    );
    let capacity = 10_000;
    let mut gateway = BloomFilter::new(capacity);
    for shard in 0..4 {
        let mut local = BloomFilter::new(capacity);
        for i in 0..500 {
            local.add(&format!("shard{shard}-sym{i}"));
        }
        gateway.union(&local).expect("shards share one geometry");
    }
    println!("  merged 4 shards x 500 symbols");
    println!(
        "  approx distinct keys: {}",
        gateway.approximate_element_count()
    );
    println!(
        "  occupancy fpp:        {:.4}%",
        gateway.estimated_fpp() * 100.0
    );

    let mismatched = BloomFilter::new(capacity * 2);
    match gateway.union(&mismatched) {
        Err(e) => println!("  refused mismatched shard: {e}"),
        Ok(()) => unreachable!("geometry check must reject this"),
    }

    gateway.clear();
    println!(
        "  after clear -> approx distinct keys: {}",
        gateway.approximate_element_count()
    );
}

pub fn write_to<W: Write>(&self, out: &mut W) -> Result<()>

pub fn parse(buf: &[u8]) -> Result<Self>

Parse a serialised bloom filter from buf. Errors if the buffer is shorter than the header or truncated mid-bits.