Repository navigation
Enable TopK per Partition (datafusion.optimizer.enable_window_topn) by default #24608
Description
Is your feature request related to a problem or challenge?
While writing the DataFusion 55.0.0 blog post
- Blog post for the DataFusion
55.0.0release #24216 - Add DataFusion 55.0.0 release blog post datafusion-site#203
I wrote up the per-partition TopK window optimization as one of the headline performance improvements of the release — and only then discovered that datafusion.optimizer.enable_window_topn defaults to false, so nobody actually gets it unless they opt in.
We probably shouldn't make a big deal about a feature that is off by default, and I could not find any issue tracking turning it on. So I am filing this issue so we don't forget and we can track what is required to do so in one place rather than scattered across PR review threads.
The story we want to be able to tell
This is the text I had drafted for the 55.0.0 blog post before pulling it, included so it's clear what we're aiming for and so whoever finishes this work can reuse it:
Per-Partition TopK for Window Functions
A common analytics pattern selects the top N rows per group using a window
function:SELECT * FROM ( SELECT ROW_NUMBER() OVER (PARTITION BY category ORDER BY revenue DESC) AS rn, * FROM sales ) WHERE rn <= 5;DataFusion previously sorted the entire input to evaluate the window
function, even though only a handful of rows per partition survive the filter.
DataFusion 55 recognizes this pattern and uses a newPartitionedTopKExec
operator that keeps only the top N rows per partition, dramatically reducing
sorting and memory for high-cardinality inputs. The optimization applies to
ROW_NUMBERandRANK, resolving a feature request first filed in 2023
(#6899).
Thanks to @SubhamSinghal for implementing this feature, with reviews from
@2010YOUY01 and @kosiew. Related PRs: #21479, #22885, #23096
Why it is currently off
datafusion/common/src/config.rs:
datafusion/datafusion/common/src/config.rs
Lines 1574 to 1581 in fdf7935
| /// When set to true, the optimizer will replace | |
| /// Filter(rn<=K) → Window(ROW_NUMBER) → Sort patterns with a | |
| /// PartitionedTopKExec that maintains per-partition heaps, avoiding | |
| /// a full sort of the input. | |
| /// When the window partition key has low cardinality, enabling this optimization | |
| /// can improve performance. However, for high cardinality keys, it may | |
| /// cause regressions in both memory usage and runtime. | |
| pub enable_window_topn: bool, default = false |
The default was set to false in #21479 because of severe high-cardinality regressions against the plain-sort baseline:
If it has regressions as large as
0.03xit should [be] off by default (and
we should look if we can automatically enable it via a heuristic / stats based
on partition cardinality / rows)
— @Dandandan, #21479 (comment)
#23096 (in 55.0.0) closed most of that gap by sharing the encoder and memory
reservation across partitions:
| Partitions | main |
with #23096 | vs sort baseline |
|---|---|---|---|
| 100 | 110 ms | 105 ms | ~1.0x |
| 1,000 | 117 ms | 110 ms | ~1.0x |
| 10,000 | 640 ms | 137 ms | 1.7x faster than sort (was a regression) |
| 100,000 | 4,327 ms | 320 ms | 320 ms vs 238 ms — still slower |
but explicitly did not flip the default:
enable_window_topndefault staysfalseper the #21479 discussion — 100K+
remains slower than sort on average, so this PR doesn't motivate flipping the
default. It's the prerequisite for further optimizations that would attack the
residual 100K+ cliff.
— #23096
Describe the solution you'd like
Set enable_window_topn to true by default, once we are confident it does not
regress. Known work items:
- Close the high-cardinality cliff. -- @Dandandan's original suggestion in Perf: Window topn optimisation #21479.
- Fix RANK memory blowup — fix: reduce RANK window top-K memory from O(input) to O(K + ties) per partition #24591
- Fix wide-payload OOM — [EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns #23600
- Expose metrics — PartitionedTopKExec does not expose any metrics (
metrics()returnsNone) #24470 / fix: expose PartitionedTopKExec metrics #24495 - Broaden pattern coverage — Support FilterExec with embedded projections in WindowTopN optimization #21596 / fix(physical-optimizer): decline the WindowTopN rewrite when the FilterExec carries a fetch #23599
- Benchmark coverage in CI: WindowTopN dense_rank benchmark #24050 added a
dense_rankbenchmark; we should make sure the sweep covers high partition-cardinality as well - Flip the default, run the full benchmark suite, and update
docs/source/user-guide/configs.md.
Describe alternatives you've considered
Leave it opt-in indefinitely. 👎 -- an optimization nobody turns on is close to an optimization that doesn't exist
Additional context
- enable_window_topn panics on WHERE rn < 1 (PartitionedTopK requires k > 0) #24404 (panic on
WHERE rn < 1) was fixed in fix: skip WindowTopN rewrite when fetch is 0 to avoid PartitionedTopK panic #24405, but that landed after thebranch-55cut - Optimize "per partition" top-k :
ROW_NUMBER < 5/ TopK #6899 (original 2023 request), - Perf: Window topn optimisation #21479, perf: Extend WindowTopN to support RANK #22885, perf: share encoder/reservation across PartitionedTopKExec partition … #23096, Allow Range partitioned inputs to PartitionedTopK #23355, refactor: moving WindowTopN before EnsureRequirements #24191, [EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns #23600, PartitionedTopKExec does not expose any metrics (
metrics()returnsNone) #24470, fix: reduce RANK window top-K memory from O(input) to O(K + ties) per partition #24591.
Activity
[-]Enable `datafusion.optimizer.enable_window_topn` by default[/-][+]Enable TopK per Partition (`datafusion.optimizer.enable_window_topn`) by default[/+]perf: interleave retained rows at emit instead of a take per group en…
perf: PartitionedTopK hold retained rows in one shared RecordBatchSto…
perf: PartitionedTopKRank on the shared store with decide-then-gather (…
perf: interleave retained rows at emit instead of a take per group en…
perf: PartitionedTopK hold retained rows in one shared RecordBatchSto…
perf: PartitionedTopKRank on the shared store with decide-then-gather (…
perf: Partitioned topk emit row number (apache#26133)