FilterChain Tutorial

Mix.install([
  {:ex_data_sketch, "~> 0.10"}
],
config: [
    ex_data_sketch: [
      backend: ExDataSketch.Backend.Rust,
      integrations: [opentelemetry: false]
    ]
  ])

Introduction

ExDataSketch.FilterChain composes multiple membership filters into a single query pipeline -- a cheap, larger first-pass filter to reject most non-members fast, backed by a smaller, more precise second-pass filter for the items that survive the first check. It's not a new sketch algorithm; it's a capability-aware wrapper around the filters covered in the other tutorials here (Bloom, Cuckoo, Quotient, CQF, XorFilter) plus IBLT as a non-queryable "adjunct."

member?/2 short-circuits on the first stage that says "definitely not," put/2 fans an insert out to every stage that supports writes (skipping static XorFilter stages), and delete/2 requires every stage to support deletion or it raises.

Sample data (cached locally)

cache_path = Path.join(System.tmp_dir!(), "ex_data_sketch_livebook_cache/filter_chain_sample.bin")

{inserted, novel} =
  if File.exists?(cache_path) do
    IO.puts("Loading cached sample data from #{cache_path}")
    cache_path |> File.read!() |> :erlang.binary_to_term()
  else
    IO.puts("Generating sample data (this takes a few seconds)...")
    inserted = for i <- 1..500_000, do: "user_#{i}"
    novel = for i <- 500_001..1_000_000, do: "user_#{i}"
    File.mkdir_p!(Path.dirname(cache_path))
    File.write!(cache_path, :erlang.term_to_binary({inserted, novel}))
    {inserted, novel}
  end

IO.puts("#{length(inserted)} inserted, #{length(novel)} novel users")

Building a chain

alias ExDataSketch.{Bloom, Cuckoo, FilterChain}

chain =
  FilterChain.new()
  |> FilterChain.add_stage(Bloom.new(capacity: 500_000, false_positive_rate: 0.05))
  |> FilterChain.add_stage(Cuckoo.new(capacity: 500_000))

FilterChain.stages(chain) |> Enum.map(& &1.__struct__)

Inserting: fans out to every writable stage

{:ok, chain} = FilterChain.put(chain, "hello")
FilterChain.member?(chain, "hello")

put_many/2 for a batch -- it batches through each stage's own put_many/2 (Rust-accelerated where a stage's backend supports it) instead of looping put/2 once per item, which matters a lot at this scale: for 500,000 items across two 500,000-capacity stages, item-by-item put/2 calls would take on the order of a minute, since every family's single-item put/2 always runs in the Pure backend (no per-item Rust NIF exists for any family) and reconstructs its entire state binary per call. Returns {:ok, chain} or {:error, :full, partial_chain}:

{:ok, chain} = FilterChain.put_many(chain, inserted)
IO.puts("Both stages now contain the inserted set")

update_many/2 (its Sketch-behaviour-compatible name, raising instead of returning {:error, :full, ...}) delegates to put_many/2 and is equally fast.

Querying: short-circuit AND across stages

member?/2 only reports "yes" if every stage agrees -- it stops at the first "no." A false positive requires every stage to independently false-positive on the same item, which is far less likely than any one filter false-positiving alone -- the compounding accuracy gain is the whole point of chaining:

false_positives = Enum.count(novel, &FilterChain.member?(chain, &1))
observed_fpr = false_positives / length(novel)
IO.puts("Chain false positives: #{false_positives} / #{length(novel)} (#{Float.round(observed_fpr * 100, 5)}%)")

# Compare against the first stage (Bloom, configured at a loose 5% FPR) alone:
bloom_only = Bloom.new(capacity: 500_000, false_positive_rate: 0.05) |> Bloom.put_many(inserted)
bloom_fps = Enum.count(Enum.take(novel, 100_000), &Bloom.member?(bloom_only, &1))
IO.puts("Bloom stage alone (5% target): #{Float.round(bloom_fps / 100_000 * 100, 2)}% observed")

Deletion requires every stage to support it

This chain has a Bloom stage, and Bloom has no delete/2 -- calling FilterChain.delete/2 on it raises:

try do
  FilterChain.delete(chain, "hello")
rescue
  e in ExDataSketch.Errors.UnsupportedOperationError -> IO.puts("Raised as expected: #{Exception.message(e)}")
end

A chain built entirely from delete-capable stages (Cuckoo, Quotient, CQF) supports it:

alias ExDataSketch.Quotient

deletable_chain =
  FilterChain.new()
  |> FilterChain.add_stage(Cuckoo.new(capacity: 1000))
  |> FilterChain.add_stage(Quotient.new(q: 12, r: 8))

{:ok, deletable_chain} = FilterChain.put(deletable_chain, "temp_item")
IO.puts("Before delete: #{FilterChain.member?(deletable_chain, "temp_item")}")

deletable_chain = FilterChain.delete(deletable_chain, "temp_item")
IO.puts("After delete: #{FilterChain.member?(deletable_chain, "temp_item")}")

A static terminal stage: XorFilter

XorFilter can only be the last query stage (no incremental put/2 after construction) -- useful as a precise, space-efficient final check behind a mutable first pass that absorbs new writes.

There's a real limitation worth understanding here: member?/2 requires every stage to agree, XorFilter included. put/2 correctly skips writing to the static XorFilter stage, so a genuinely new item lands in Cuckoo but not in XorFilter -- meaning FilterChain.member?/2 on the whole chain will (correctly) say "no" for it, forever, until the XorFilter stage is rebuilt to include it. Absorbing new writes into Cuckoo doesn't make them queryable through the combined chain; it just means Cuckoo, queried on its own, already has the answer XorFilter doesn't have yet:

alias ExDataSketch.XorFilter

{:ok, xor} = XorFilter.build(Enum.take(inserted, 100_000))

hybrid_chain =
  FilterChain.new()
  |> FilterChain.add_stage(Cuckoo.new(capacity: 500_000))
  |> FilterChain.add_stage(xor)

# put/2 skips the static XorFilter stage automatically -- only the
# Cuckoo stage actually receives new writes.
{:ok, hybrid_chain} = FilterChain.put(hybrid_chain, "newly_seen_item")

[cuckoo_stage, _xor_stage] = FilterChain.stages(hybrid_chain)
IO.puts("Present in Cuckoo stage alone: #{Cuckoo.member?(cuckoo_stage, "newly_seen_item")}")
IO.puts("Present via the whole chain (Cuckoo AND XorFilter): #{FilterChain.member?(hybrid_chain, "newly_seen_item")}")

Serialization

binary = FilterChain.serialize(chain)
{:ok, restored} = FilterChain.deserialize(binary)
IO.puts("Round-tripped membership check: #{FilterChain.member?(restored, hd(inserted))}")