Repository navigation
Conversation
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
approx_top_kaggregate function for approximate heavy hitter / top-k frequency estimation #20967.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 everydistinct value, which is slow and memory hungry on high-cardinality columns.
approx_top_k(expression, k)answers the same question in bounded memory pergroup, similar to ClickHouse's
topK, and complements DataFusion's existingapproximate aggregates (
approx_distinct,approx_median,approx_percentile_cont).What changes are included in this PR?
approx_top_k(expression, k)aggregate indatafusion-functions-aggregatereturning
List<Struct{value, count}>, ranked by estimated count.k * 3counters; thealpha filter rejects low-frequency values before they displace tracked ones.
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.
countis always an upper bound of the truefrequency.
GroupsAccumulator(includingconvert_to_statefor skipped partialaggregation). Per-group memory is allocated lazily, so singleton groups stay
small, and per-row states are merged without building a summary.
alpha map); corrupt states produce errors.
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 BYinside 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.benches/approx_top_k.rs.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: theconvert_to_state(skip partialaggregation) path.
same-bucket replacement, per-bucket merge bounds, folding exact states, state
validation).
Accumulator/GroupsAccumulatorstate compatibility test(
functions-aggregate/tests/state_compat.rs) coversapprox_top_k.Are there any user-facing changes?
Yes, a new
approx_top_kaggregate function:Changes since the last review
Rebased onto
main; the new work is one commit on top of the earlier ones.kis read withget_scalar_value, andexpr_fn::approx_top_ktakes(expression, k).encodings are rejected during planning.
skipping partial aggregation gives the same result as aggregating directly
(the
state_compattest now passes).filter bucket instead of the largest alpha in the summary.
header and applied without building a summary. This halves the merge time in
the
merge convertedbenchmark (42.7 ms to 21.6 ms for k=10), close toaggregating the rows directly (18.1 ms).
ORDER BYinside the call is ignored, and empty window frames and unmatchedcorrelated subqueries return
[]instead of NULL.