perf: Extend WindowTopN to support RANK - #22885
Conversation
|
@kumarUjjawal @2010YOUY01 will you be able to review this PR. |
|
might be related #23021 |
a2585c8 to
173b53c
Compare
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
|
@kosiew can you help in reviewing this PR. |
kosiew
left a comment
There was a problem hiding this comment.
@SubhamSinghal
Looks good overall.
I left a couple of suggestions to keep the docs accurate and to call out a possible memory-accounting follow-up.
There was a problem hiding this comment.
Small docs nit: the Limitations section still says this operator is only activated for ROW_NUMBER. Could you update this sentence to mention RANK and the non-empty ORDER BY requirement?
|
|
||
| // Commit this batch's ties as a single entry. | ||
| if !equal_indices.is_empty() { | ||
| state.ties.push(TieEntry { |
There was a problem hiding this comment.
This looks correct for preserving tied rows. One possible follow-up: TieEntry keeps a clone of the full input RecordBatch and charges input_batch_bytes per partition tie entry, which could over-reserve memory when a single batch has boundary ties across many partitions.
It may be worth looking at compact tie batches or shared batch accounting later.
There was a problem hiding this comment.
I will file a followup issue for this.
|
@SubhamSinghal Leaving this open for a few days for more reviews as this is a big PR. |
|
@kosiew can we merge this PR? |
|
Thank you @SubhamSinghal and @kosiew |
## Which issue does this PR close? - Related to apache#6899 (DENSE_RANK to follow in a separate PR will close it). ## Rationale for this change PR apache#21479 introduced `WindowTopN` for `ROW_NUMBER` only; `RANK` and `DENSE_RANK` were explicitly out of scope. This PR extends the rule to `RANK`, replacing the full sort under `Filter(rk≤K) → Window(RANK) → Sort` with a per-partition heap-of-K plus a boundary-tie buffer. ## What changes are included in this PR? - **`datafusion/physical-plan/src/topk/mod.rs`** — new `pub(crate) struct PartitionedTopKRank` (sibling of `PartitionedTopK` from apache#23096) with per-partition `RankPartitionState { TopKHeap, Vec<TieEntry> }`. - **`datafusion/physical-plan/src/sorts/partitioned_topk.rs`** — `WindowFnKind` enum (`RowNumber` / `Rank`). `do_partitioned_topk` dispatches on `fn_kind` to `PartitionedTopK::try_new` or `PartitionedTopKRank::try_new`; - **`datafusion/physical-optimizer/src/window_topn.rs`** — `is_row_number` → `supported_window_fn(expr) -> Option<WindowFnKind>`; empty-`order_by` guard for RANK; `WindowFnKind` plumbed through `PartitionedTopKExec::try_new`. - **`datafusion/sqllogictest/test_files/window_topn.slt`** — RANK SLT cases: basic, strict (`<`), flipped (`>=` / `>`), boundary ties, ties spanning ob values, empty-`ORDER BY` (rule must NOT fire), mixed window functions, ASC/DESC × NULLS FIRST/LAST, QUALIFY. - **`datafusion/core/tests/physical_optimizer/window_topn.rs`** — 6 new RANK rule unit tests covering predicate matching, partition-by/order-by guards, dense_rank skip. - **`benchmarks/queries/h2o/window.sql`** — six new RANK queries (Q14–Q17, Q22, Q23) covering partition counts from ~100 to ~100K, low and heavy tie densities. h2o `window` benchmark, 10M-row `large` table, RANK top-2, 3-iteration average. Toggle via `DATAFUSION_OPTIMIZER_ENABLE_WINDOW_TOPN`. | Variant | Partitions | OFF (rule disabled) | ON (rule enabled) | Δ | |---|---:|---:|---:|:---:| | RANK low ties (`id3 % 100`) | ~100 | 305 ms | **107 ms** | **2.84× faster** ✓ | | RANK low ties (`id3 % 1000`) | ~1K | 263 ms | **120 ms** | **2.19× faster** ✓ | | RANK heavy ties (`id3 % 1000`, `v2 % 10` OB) | ~1K | 282 ms | **125 ms** | **2.25× faster** ✓ | | RANK low ties (`id2`) | ~10K | 363 ms | **140 ms** | **2.59× faster** ✓ | | RANK heavy ties (`id2`, `v2 % 10` OB) | ~10K | 291 ms | **143 ms** | **2.04× faster** ✓ | | RANK low ties (`id3 % 100K`) | ~100K | 241 ms | 422 ms | 1.75× slower | ## Are these changes tested? Yes: - `cargo test -p datafusion-physical-plan --lib` — 1455 passed - `cargo test -p datafusion-physical-optimizer --lib` — 27 passed - `cargo test -p datafusion --test core_integration physical_optimizer::window_topn::` — 13 passed (7 ROW_NUMBER + 6 RANK) - `cargo test --test sqllogictests -- window_topn` — passed ## Are there any user-facing changes? The existing `optimizer.enable_window_topn` config flag (default `false`) now also covers `RANK` queries. No public API additions
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes apache#123` indicates that this PR will close issue apache#123. --> - Part of apache#21706 ## Rationale for this change Updated sql queries to match existing benchmark queries, mostly apache#22885 and apache#24050 ## What changes are included in this PR? sql benchmark query updates. ## Are these changes tested? Yes ## Are there any user-facing changes? No
Which issue does this PR close?
ROW_NUMBER < 5/ TopK #6899 (DENSE_RANK to follow in a separate PR will close it).Rationale for this change
PR #21479 introduced
WindowTopNforROW_NUMBERonly;RANKandDENSE_RANKwere explicitly out of scope. This PR extends the rule toRANK, replacing the full sort underFilter(rk≤K) → Window(RANK) → Sortwith a per-partition heap-of-K plus a boundary-tie buffer.What changes are included in this PR?
datafusion/physical-plan/src/topk/mod.rs— newpub(crate) struct PartitionedTopKRank(sibling ofPartitionedTopKfrom perf: share encoder/reservation across PartitionedTopKExec partition … #23096) with per-partitionRankPartitionState { TopKHeap, Vec<TieEntry> }.datafusion/physical-plan/src/sorts/partitioned_topk.rs—WindowFnKindenum (RowNumber/Rank).do_partitioned_topkdispatches onfn_kindtoPartitionedTopK::try_neworPartitionedTopKRank::try_new;datafusion/physical-optimizer/src/window_topn.rs—is_row_number→supported_window_fn(expr) -> Option<WindowFnKind>; empty-order_byguard for RANK;WindowFnKindplumbed throughPartitionedTopKExec::try_new.datafusion/sqllogictest/test_files/window_topn.slt— RANK SLT cases: basic, strict (<), flipped (>=/>),boundary ties, ties spanning ob values, empty-
ORDER BY(rule must NOT fire), mixed window functions, ASC/DESC × NULLS FIRST/LAST, QUALIFY.datafusion/core/tests/physical_optimizer/window_topn.rs— 6 new RANK rule unit tests covering predicatematching, partition-by/order-by guards, dense_rank skip.
benchmarks/queries/h2o/window.sql— six new RANK queries (Q14–Q17, Q22, Q23) covering partition counts from ~100 to ~100K, low and heavy tie densities.h2o
windowbenchmark, 10M-rowlargetable, RANK top-2, 3-iteration average. Toggle viaDATAFUSION_OPTIMIZER_ENABLE_WINDOW_TOPN.id3 % 100)id3 % 1000)id3 % 1000,v2 % 10OB)id2)id2,v2 % 10OB)id3 % 100K)Are these changes tested?
Yes:
cargo test -p datafusion-physical-plan --lib— 1455 passedcargo test -p datafusion-physical-optimizer --lib— 27 passedcargo test -p datafusion --test core_integration physical_optimizer::window_topn::— 13 passed (7 ROW_NUMBER + 6 RANK)cargo test --test sqllogictests -- window_topn— passedAre there any user-facing changes?
The existing
optimizer.enable_window_topnconfig flag (defaultfalse) now also coversRANKqueries. No public API additions