Skip to content

feat: add approx_top_k aggregate function - #20968

Open
sesteves wants to merge 6 commits into
apache:mainfrom
sesteves:feat/approx-top-k
Open

sesteves wants to merge 6 commits into
apache:mainfrom
sesteves:feat/approx-top-k

Conversation

@sesteves

@sesteves sesteves commented Mar 16, 2026 •

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Rationale for this change

Finding the most frequent values in a column (top error codes, most popular
products, frequent URL paths, ...) is a common analytical pattern. The exact
approach (GROUP BY value ORDER BY COUNT(*) DESC LIMIT k) materializes every
distinct value, which is slow and memory hungry on high-cardinality columns.

approx_top_k(expression, k) answers the same question in bounded memory per
group, similar to ClickHouse's topK, and complements DataFusion's existing
approximate aggregates (approx_distinct, approx_median,
approx_percentile_cont).

What changes are included in this PR?

  • New approx_top_k(expression, k) aggregate in datafusion-functions-aggregate
    returning List<Struct{value, count}>, ranked by estimated count.
    • Filtered Space-Saving (Homem & Carvalho, 2010) with k * 3 counters; the
      alpha filter rejects low-frequency values before they displace tracked ones.
    • Full partial states are merged with Parallel Space-Saving, using each
      partition's alpha for the value's filter bucket as the bound for values that
      partition did not track. States that never filled up hold exact counts and
      are folded in with the weighted Filtered Space-Saving update, so the per-row
      states produced when partial aggregation is skipped give the same result as
      aggregating the rows directly. count is always an upper bound of the true
      frequency.
    • Native GroupsAccumulator (including convert_to_state for skipped partial
      aggregation). Per-group memory is allocated lazily, so singleton groups stay
      small, and per-row states are merged without building a summary.
    • Compact, versioned, validated intermediate state (sparse or bitmap-encoded
      alpha map); corrupt states produce errors.
    • Supported input types: strings (incl. views), binary (incl. views and
      fixed-size binary such as UUIDs), booleans, integers, 32/64-bit floats,
      decimals, dates, times and timestamps. Dictionary inputs are decoded to
      their value type; other encodings are rejected during planning.
    • Order insensitive (an ORDER BY inside the call is ignored). Empty input,
      empty window frames and unmatched correlated subqueries return [].
  • expr_fn::approx_top_k(expression, k) for the DataFrame API.
  • Criterion benchmark benches/approx_top_k.rs.
  • User documentation.

What is the testing strategy for this PR?

  • approx_top_k.slt: basic usage, GROUP BY, FILTER, NULL / empty input
    (including empty window frames and correlated subqueries), argument
    validation, all supported types and return types, multi-partition merges, and
    high-cardinality inputs that force filtering/eviction.
  • aggregate_skip_partial.slt: the convert_to_state (skip partial
    aggregation) path.
  • Unit tests for algorithm internals not reachable from SQL (filter admission,
    same-bucket replacement, per-bucket merge bounds, folding exact states, state
    validation).
  • The Accumulator/GroupsAccumulator state compatibility test
    (functions-aggregate/tests/state_compat.rs) covers approx_top_k.
  • DataFrame and proto round-trip tests.

Are there any user-facing changes?

Yes, a new approx_top_k aggregate function:

SELECT approx_top_k(url_path, 3) FROM access_logs;
SELECT region, approx_top_k(product_id, 3) FROM sales GROUP BY region;
use datafusion::functions_aggregate::expr_fn::approx_top_k;
df.aggregate(vec![], vec![approx_top_k(col("url_path"), lit(5))])?;

Changes since the last review

Rebased onto main; the new work is one commit on top of the earlier ones.

  • k is read with get_scalar_value, and expr_fn::approx_top_k takes
    (expression, k).
  • New input types: booleans, decimals, times and fixed-size binary. Unsupported
    encodings are rejected during planning.
  • Corrupt intermediate states return errors instead of panicking.
  • States that never filled up are folded in with the weighted update, so
    skipping partial aggregation gives the same result as aggregating directly
    (the state_compat test now passes).
  • Merging two full summaries bounds a missing value by the alpha of its own
    filter bucket instead of the largest alpha in the summary.
  • Per-row states from skipped partial aggregation are recognized by their
    header and applied without building a summary. This halves the merge time in
    the merge converted benchmark (42.7 ms to 21.6 ms for k=10), close to
    aggregating the rows directly (18.1 ms).
  • ORDER BY inside the call is ignored, and empty window frames and unmatched
    correlated subqueries return [] instead of NULL.
  • SPDX license headers and documentation fixes.

Loading
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core Core DataFusion crate documentation Improvements or additions to documentation functions Changes to functions implementation proto Related to proto crate sqllogictest SQL Logic Tests (.slt)

Projects

None yet

Development

Successfully merging this pull request may close these issues.

feat: Add approx_top_k aggregate function for approximate heavy hitter / top-k frequency estimation

7 participants