Blog post about TopK per partition (PartitionedTopKExec / enable_window_topn) #27233

Is your feature request related to a problem or challenge?

There is a lot of great work from @SubhamSinghal, @jayzhan211 and others on the "TopK per partition" optimization for window queries like

SELECT * FROM (
    SELECT ROW_NUMBER() OVER (PARTITION BY category ORDER BY revenue DESC) AS rn, *
    FROM sales
) WHERE rn <= 5;

The original request is from 2023:

• Optimize "per partition" top-k : ROW_NUMBER < 5 / TopK #6899

The optimization landed in DataFusion 55 behind datafusion.optimizer.enable_window_topn, and the EPIC to turn it on by default is

• Enable TopK per Partition (datafusion.optimizer.enable_window_topn) by default #24608
• feat: enable datafusion.optimizer.enable_window_topn by default #27197

As always, telling the community about it will be good

1. To advertise the feature for others who have the same use case,
2. To document in an easier to understand way how the feature works
3. The authors to raise their visibility.

I had drafted a section about it for the DataFusion 55 release post (see below) but a release post isn't great as it has limited space to explain how the operator works

• Blog post for the DataFusion 55.0.0 release #24216
• Add DataFusion 55.0.0 release blog post datafusion-site#203

Describe the solution you'd like

Thus I suggest a complete self contained blog post on https://datafusion.apache.org/blog/ (make a PR to https://github.com/apache/datafusion-site) about TopK per partition.

I suggest the standard structure of "informative post with low key DataFusion pitch"

For example, the pattern of

• Dynamic Filters (TopK and hash join, 25x faster): https://datafusion.apache.org/blog/2025/09/10/dynamic-filters/
• Fast high cardinality grouping: https://datafusion.apache.org/blog/2023/08/05/datafusion_fast_grouping/

Here are some ideas for an outline:

1. An understandable representative usecase: "top N per group" (e.g latest N events per device) so readers can tell if it applies to them. Mention this is db-benchmark / h2o query 8 shape.
2. Background / technical intuition: the naive plan sorts the entire input to compute the window function and then throws almost all of it away. The TopK per partition operator keeps a bounded heap per partition instead. Explain how ROW_NUMBER, RANK and DENSE_RANK differ in what has to be retained.
3. How to use it in DataFusion: What version is it introduced in (57 I hope, show EXPLAIN before/after with PartitionedTopKExec), and how to turn it off)
4. A bit about how it is implemented: the WindowTopN physical optimizer rule, PartitionedTopKExec, and then some of the highlights of making it work for high cardinality:
   • the 100K+ partition cliff, Perf: Window topn optimisation #21479 (comment)
   • Memory optimizations, etc
   • Emitting the rank column directly from the operator so the window node can be removed (perf: Partitioned topk emit row number #26133)
5. Call for help 🎣: Any remaining work that needs to happen

I think the best time to publish is around 57 release (when #27197 is merged and released, so the post can say the optimization is on by default.

Describe alternatives you've considered

Here is the original text I had drafted for the 55 release post (from #24608), which could be a starting point:

SELECT * FROM (
    SELECT ROW_NUMBER() OVER (PARTITION BY category ORDER BY revenue DESC) AS rn, *
    FROM sales
) WHERE rn <= 5;

Additional context

Tracking issues:

• Optimize "per partition" top-k : ROW_NUMBER < 5 / TopK #6899 (original 2023 request)
• Enable TopK per Partition (datafusion.optimizer.enable_window_topn) by default #24608 (enable by default EPIC)
• Further Optimize "per partition" top-k : ROW_NUMBER < 5 / TopK #24060 (follow on optimizations)
• [EPIC] Improve window function performance (especially for large windows) #23197 (window function performance EPIC)

Core operator and optimizer rule:

• Perf: Window topn optimisation #21479 (ROW_NUMBER, @SubhamSinghal)
• perf: Extend WindowTopN to support RANK #22885 (RANK, @SubhamSinghal)
• perf: Extend WindowTopN to dense_rank #23869 (DENSE_RANK, @SubhamSinghal)
• refactor: moving WindowTopN before EnsureRequirements #24191 (run WindowTopN before EnforceRequirements, @saadtajwar)
• Allow Range partitioned inputs to PartitionedTopK #23355 (range partitioned inputs, @stuhood)
• perf: Partitioned topk emit row number #26133 (emit the rank column from the operator, @SubhamSinghal)
• feat: enable datafusion.optimizer.enable_window_topn by default #27197 (enable by default, @SubhamSinghal)

Closing the high cardinality gap:

• perf: share encoder/reservation across PartitionedTopKExec partition … #23096 (share encoder and reservation across partitions)
• fix: reduce RANK window top-K memory from O(input) to O(K + ties) per partition #24591 (RANK memory from O(input) to O(K + ties), @jayzhan211)
• perf: interleave retained rows at emit instead of a take per group entry #25468 (interleave retained rows at emit)
• perf: PartitionedTopK hold retained rows in one shared RecordBatchStore #25730 (shared RecordBatchStore for ROW_NUMBER)
• perf: PartitionedTopKRank on the shared store with decide-then-gather #25968 (RANK on the shared store with decide-then-gather)

Correctness fixes and metrics:

• fix: skip WindowTopN rewrite when fetch is 0 to avoid PartitionedTopK panic #24405 (@viirya)
• fix(physical-optimizer): decline the WindowTopN rewrite when the FilterExec carries a fetch #23599 (@zhuqi-lucas)
• fix: expose PartitionedTopKExec metrics #24495 (metrics, @shinzoxD)
• fix: WindowTopN resolves the filter column through intermediate proje… #26417 (filter column through projections)
• fix: PartitionedTopKRank counts tie rows twice in output_rows #25535 (@jayzhan211)

Benchmarks:

• WindowTopN dense_rank benchmark #24050 (dense_rank benchmark)
• bench(h2o): window Top-N sweep over a declared-sorted input #24732 (h2o window_sorted sweep)

Open follow on work:

• PartitionedTopKExec keeps whole Utf8View input buffers alive for the rows it retains #26381 (Utf8View buffers kept alive)
• PartitionedTopKExec with DENSE_RANK retains every input batch a retained row came from #26876 / fix: port PartitionedTopKDenseRank onto the shared store #27136 (DENSE_RANK on the shared store)
• Support FilterExec with embedded projections in WindowTopN optimization #21596 (filters with embedded projections)
• Support equality with one in opt-in WindowTopN rule #26160 (rn = 1)
• [EPIC] Memory-efficient dedup on wide payload: fix ROW_NUMBER()=1 and DISTINCT ON patterns #23600 (ROW_NUMBER() = 1 / DISTINCT ON dedup)
• feat(window-topn): streaming per-partition top-K for pre-sorted input #24629 (streaming path for pre-sorted input, closed but a good idea)

Related prior posts for style / context:

• Optimizing for Almost Sorted Data: Sort Pushdown
• Dynamic Filters
• Previous blog post tickets: Blog post about TopK filter pushdown #15513, Blog post about Range Partitioning #24667