[sesteves]

Which issue does this PR close?

Closes #20967.

Rationale

Finding the most frequently occurring values in a column is a very common analytical pattern — top error codes, most popular products, most active users, frequent URL paths, etc. The exact approach (GROUP BY value ORDER BY COUNT(*) DESC LIMIT k) requires materializing all distinct groups, which is memory-intensive and slow on high-cardinality columns.

approx_top_k solves this with a streaming approximation algorithm that operates in bounded memory, making it practical for large-scale analytics.

This function is already available in ClickHouse (topK) and via extensions in PostgreSQL and Druid, and is a natural addition to DataFusion's existing family of approximate aggregate functions (approx_distinct, approx_median, approx_percentile_cont).

What changes are included in this PR?

Core implementation (datafusion/functions-aggregate/src/approx_top_k.rs)

• SpaceSavingSummary — Implements the Filtered Space-Saving algorithm (Metwally et al., 2005):

  • Fixed-size counter summary with eviction of minimum-count items when full
  • Alpha map for improved accuracy — tracks recently evicted items and filters low-frequency noise before it enters the main summary (same approach used by ClickHouse)
  • CAPACITY_MULTIPLIER = 3 (matching ClickHouse's default) — the summary tracks k × 3 counters internally
  • Merge support for parallel/distributed execution
  • Binary serialization/deserialization for intermediate state transfer

• ApproxTopK — AggregateUDFImpl with #[user_doc] documentation, supporting any hashable scalar type

• ApproxTopKAccumulator — Accumulator with update_batch, merge_batch, evaluate (returns List(Struct({value: T, count: UInt64})) ordered by count descending), and state (serialized summary + data type)

• 9 unit tests covering basic operations, eviction, alpha map, merge, serialization, accumulator update/evaluate, merge_batch, and a distributed merge simulation

Registration (datafusion/functions-aggregate/src/lib.rs)

• Added module, expr_fn re-export, and registration in all_default_aggregate_functions()

SQL logic tests (datafusion/sqllogictest/test_files/approx_top_k.slt)

• 8 tests: basic string/int, GROUP BY, k=1, NULL handling, error cases (invalid k), WHERE clause filtering

Documentation

• docs/source/user-guide/sql/aggregate_functions.md — TOC entry + full documentation section
• docs/source/user-guide/expressions.md — quick-reference table entry

Integration tests

• Proto roundtrip (datafusion/proto/tests/cases/roundtrip_logical_plan.rs) — approx_top_k(col("a"), lit(3))
• DataFrame (datafusion/core/tests/dataframe/dataframe_functions.rs) — test_fn_approx_top_k with insta::assert_snapshot!

Are these changes tested?

Yes:

• 9 unit tests in approx_top_k.rs (algorithm correctness, serialization, accumulator lifecycle, distributed merge)
• 8 SQL logic tests in approx_top_k.slt (end-to-end SQL behavior including edge cases and errors)
• 1 DataFrame integration test with snapshot assertion
• 1 proto roundtrip test

Are there any user-facing changes?

Yes — new approx_top_k aggregate function available in SQL and the DataFrame API:

-- Top 3 most frequent values
SELECT approx_top_k(url_path, 3) FROM access_logs;

-- Top 3 per group
SELECT region, approx_top_k(product_id, 3) FROM sales GROUP BY region;

// DataFrame API
use datafusion::functions_aggregate::approx_top_k::approx_top_k;
df.aggregate(vec![], vec![approx_top_k(col("url_path"), lit(5))])?;

[jonathanc-n]

cc @Dandandan @martin-g Should be good for another look

[sesteves]

Thanks for the review @Dandandan @martin-g , I believe I've addressed all of your feedback.

[martin-g]

Unfortunately, after the squash+force push I will have to review the whole PR again ...
Please don't squash your commits in a PR!
It just makes incremental PR reviews hard/impossible.
The merge will squash them anyway, with separate commit message for all commits.

[sesteves]

Unfortunately, after the squash+force push I will have to review the whole PR again ... Please don't squash your commits in a PR! It just makes incremental PR reviews hard/impossible. The merge with squash them anyway, with separate commit message for all commits.

Sorry about that, the squash slipped through accidentally!

[martin-g]

No problem!
It would be good a second person to review the PR anyway.

[sesteves]

Hi @martin-g and @Dandandan ! Friendly ping on this when you have a moment. I know you mentioned wanting a second person to review - let me know if there is someone specific I should tag, or if there is anything else I can do from my side to help get this across the finish line.

[Dandandan]

@sesteves

starts to look good!

Can you also remove the testing submodule commit / update? And fix remaining CI issues?

[sesteves]

@sesteves

starts to look good!

Can you also remove the testing submodule commit / update? And fix remaining CI issues?

Done, please lmk if there are still any remaining issues @Dandandan

[sesteves]

@Dandandan @martin-g when should I expect the next round of reviews on this? thanks!

[github-actions]

Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.

[sesteves]

@Dandandan @martin-g when should I expect the next round of reviews on this? thanks!

ping

[timsaucer]

I took a pass at it myself and I also directed an agent to review it. I've marked which comments came from the agent. It also had these suggestions:

Should fix

1. 144 signature variants. L691–696 builds 18 value types × 8 k types as TypeSignature::Exact. approx_percentile_cont handles the same shape with TypeSignature::Coercible + TypeSignatureClass::Integer, which collapses k to a single coercion and accepts any integer width for free.
2. Missing the types this function is most for. No Utf8View/BinaryView — the default for Parquet string reads under force_view_types — and no Dictionary. They'll be coerced to Utf8/Binary, so it works, but you eat a full cast per batch on the exact workload (top URL paths, top error codes) the PR motivates. Also absent: Decimal128/256, Boolean, Time32/64. Separately, update_batch's match (L1037–1078) has to be kept in sync with the signature list by hand; a RowConverter or ScalarValue-keyed encoding would make the type support fall out for free instead of being 150 lines of per-type to_le_bytes plus a mirrored 150 lines in build_value_array.
3. Literal extraction is bespoke. L750–766 downcasts to expressions::Literal directly. crate::utils::get_scalar_value evaluates against an empty batch, so it also handles cast-wrapped or otherwise foldable literals and keeps the error text consistent with sibling functions.
4. Corrupt state silently becomes NULL. build_value_array uses <[u8; N]>::try_from(..).ok() and String::from_utf8(..).ok() throughout (L830–980), so a wrong-length or non-UTF-8 item yields NULL rather than an error. The _ => fallback (L974–980) reinterprets arbitrary types as UTF-8 strings; since the signature already restricts types, that should be internal_err!.
5. Hash choice. hash_item (L157–165) is byte-at-a-time FNV-1a. It's slow on long strings, and the alpha index uses hash as usize & mask (L246) — FNV's low bits are its weakest, and hashbrown::HashTable also derives its bucket index from the low bits, so you get correlated clustering in both structures. foldhash is already a dependency of this crate.
6. item_heap_bytes uses raw +=/-= (L259, L290, L400, L446) after the rest of the file moved to saturating_*; a bookkeeping slip is a debug panic. Also #[derive(Clone)] re-derives Vec capacities, so a plain .clone() leaves item_heap_bytes overstated — the merge fast path works around this at L353–355 but nothing else does.

Nits

• return_type's arg_types.is_empty() → Utf8 branch (L714–724) is dead; the signature guarantees arity 2.
• SpaceSavingSummary::new(0) leaves alpha_map empty, and the at-capacity branch then does self.alpha_map.len() - 1 → underflow panic. Unreachable through accumulator() (k ≥ 1) but reachable from tests; worth a debug_assert or an explicit guard.
• LargeUtf8 → Utf8 narrowing of the return type is commented in code but not surfaced in #[user_doc]; it's user-visible.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 81.06%. Comparing base (3e3a92d) to head (ea08c6a).
⚠️ Report is 10 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #20968      +/-   ##
==========================================
+ Coverage   80.91%   81.06%   +0.14%     
==========================================
  Files        1103     1107       +4     
  Lines      377219   382932    +5713     
  Branches   377219   382932    +5713     
==========================================
+ Hits       305244   310433    +5189     
- Misses      53775    54205     +430     
- Partials    18200    18294      +94     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[sesteves]

Thank you for the thorough review @timsaucer . I have addressed the feedback and aligned the implementation closer to the original paper rather than clickhouse variant. We have already been running this in our production environment since we need to provide top-k with constant memory utilization regardless of dataset size.

[github-actions]

Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.

[sesteves]

Hi @timsaucer, thanks again for reviewing this PR. I’ve addressed the feedback and updated the implementation. Could you take another look when you have a chance?