CQF (Counting Quotient Filter) Tutorial

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

Introduction

ExDataSketch.CQF is ExDataSketch.Quotient's data structure extended to track multiplicity -- how many times each item was inserted, not just whether it was inserted at all. It's a multiset membership filter: member?/2 for presence, estimate_count/2 for (an over-estimate of, never under) how many times, delete/2 to decrement.

Sample data (cached locally)

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

events =
  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)...")

    # 1,000,000 rate-limit-check events over 50,000 distinct API keys,
    # skewed (some keys are far more active than others).
    events =
      for _ <- 1..1_000_000 do
        rank = trunc(:math.pow(:rand.uniform(), 2) * 49_999) + 1
        "api_key_#{rank}"
      end

    File.mkdir_p!(Path.dirname(cache_path))
    File.write!(cache_path, :erlang.term_to_binary(events))
    events
  end

true_counts = Enum.frequencies(events)
IO.puts("#{length(events)} events across #{map_size(true_counts)} distinct keys")

Basic usage

alias ExDataSketch.CQF

cqf = CQF.new(q: 18, r: 8) |> CQF.put!("x") |> CQF.put!("x") |> CQF.put!("x")
CQF.estimate_count(cqf, "x")

put_many/2 for a batch. It returns {:ok, cqf} (or {:error, :full, partial_cqf} if the table fills up partway through -- see the sizing section below):

{:ok, sketch} = CQF.new(q: 21, r: 8) |> CQF.put_many(events)

{busiest_key, true_count} = Enum.max_by(true_counts, fn {_, c} -> c end)
estimate = CQF.estimate_count(sketch, busiest_key)

IO.puts("#{busiest_key}: true count=#{true_count}, CQF estimate=#{estimate}")
IO.puts("Sketch size: #{CQF.size_bytes(sketch)} bytes")

member? vs estimate_count

member?/2 is the cheap yes/no question; estimate_count/2 is the more detailed (and slightly more expensive) "how many":

IO.puts("member?: #{CQF.member?(sketch, busiest_key)}")
IO.puts("estimate_count: #{CQF.estimate_count(sketch, busiest_key)}")
IO.puts("member? for a truly novel key: #{CQF.member?(sketch, "never_seen_key")}")

Accuracy: overestimate-only, like CMS

Same guarantee shape as ExDataSketch.CMS -- collisions can only add extra weight, never remove it:

sample_keys = true_counts |> Map.keys() |> Enum.take_random(20)

results =
  for key <- sample_keys do
    true_c = Map.fetch!(true_counts, key)
    est = CQF.estimate_count(sketch, key)
    {key, true_c, est, est - true_c}
  end

never_undercounts? = Enum.all?(results, fn {_, _, _, diff} -> diff >= 0 end)
IO.puts("Every sampled estimate >= true count: #{never_undercounts?}")

Deletion

f = CQF.new(q: 10, r: 8) |> CQF.put!("x") |> CQF.put!("x")
IO.puts("Count before delete: #{CQF.estimate_count(f, "x")}")

f = CQF.delete(f, "x")
IO.puts("Count after one delete: #{CQF.estimate_count(f, "x")}")

Sizing: q must budget for total occurrences, not distinct keys

Unlike most of this library's filters, CQF's slot budget (2^q slots) is consumed by every occurrence of every item, not just distinct keys -- each repeat of an already-seen item costs one more physical slot (there's no compact run-length counter; a count of N is literally N-1 duplicate slots plus the original). This dataset has 1,000,000 events across only 50,000 distinct keys, so q has to be sized against the 1,000,000, not the 50,000: q: 18 (262,144 slots) is under a third of what 1,000,000 occurrences need. put/2/put_many/2 return {:error, :full, partial} once the table has no room left (mirroring ExDataSketch.Cuckoo), so undersizing no longer fails silently -- but it's still expensive to get there: as the table fills, each insert searches further for a free slot before finally failing. q: 21 (2,097,152 slots) gives this dataset roughly 2x headroom over its raw occurrence count, so size generously rather than relying on the error signal alone.

What "full" looks like

Force it with a tiny q:

tiny = CQF.new(q: 4, r: 4)

result =
  Enum.reduce_while(1..10_000, {:ok, tiny}, fn i, {:ok, f} ->
    case CQF.put(f, "item_#{i}") do
      {:ok, updated} -> {:cont, {:ok, updated}}
      {:error, :full} -> {:halt, {:error, :full, i - 1}}
    end
  end)

case result do
  {:error, :full, items_inserted} ->
    IO.puts("Filter reported full after #{items_inserted} inserts (q was 4, 16 slots)")

  {:ok, _} ->
    IO.puts("Never filled -- try a smaller q")
end

put!/2 raises ExDataSketch.Errors.FilterFullError instead of returning the error tuple, for callers who'd rather crash than handle it explicitly.

Sizing: r controls per-item collision rate

for r <- [4, 8, 12] do
  {:ok, s} = CQF.new(q: 21, r: r) |> CQF.put_many(events)
  est = CQF.estimate_count(s, busiest_key)
  IO.puts("r=#{r} (#{CQF.size_bytes(s)} bytes): #{busiest_key} estimate=#{est} (true: #{true_count})")
end

Merging

half = div(length(events), 2)
{first_half, second_half} = Enum.split(events, half)

{:ok, worker_a} = CQF.new(q: 21, r: 8) |> CQF.put_many(first_half)
{:ok, worker_b} = CQF.new(q: 21, r: 8) |> CQF.put_many(second_half)

merged = CQF.merge(worker_a, worker_b)
IO.puts("Merged estimate for #{busiest_key}: #{CQF.estimate_count(merged, busiest_key)}")

Serialization

binary = CQF.serialize(sketch)
{:ok, restored} = CQF.deserialize(binary)
IO.puts("Round-tripped count: #{CQF.estimate_count(restored, busiest_key)}")