SubhamSinghal opened a new pull request, #24629:
URL: https://github.com/apache/datafusion/pull/24629
## Which issue does this PR close?
Relates to the WindowTopN optimization (`ROW_NUMBER`/`RANK`/`DENSE_RANK`
per-partition top-K). It adds the missing *already-sorted-input* path.
## Rationale for this change
The `WindowTopN` rule today only handles the case where the window's input
is **unsorted**: it replaces the `SortExec` with a heap-based
`PartitionedTopKExec` (one top-K heap per distinct partition key). That is the
right trade-off when a sort would otherwise be needed, but it has two costs
that make it ship default-off:
- **O(K × P) memory** (P = number of distinct partitions), which regresses
at high partition cardinality.
- **Blocking emission** — it buffers all input before emitting.
When the window's input is **already sorted** by `(partition_keys,
order_keys)` — i.e. `EnforceSorting` elided the `SortExec` because the ordering
was already satisfied — the heaps are pure waste: the data is in the exact
order we need, so we can compute the per-partition rank in a single streaming
pass with **O(1) state** and emit incrementally, dropping rows whose rank
exceeds `K` before they reach the (expensive) full window.
Today that already-sorted case doesn't fire at all: the rule requires a
`SortExec` child, so the plan falls through to the default
full-window-then-filter. This PR closes that gap.
## What changes are included in this PR?
A new sibling operator and a restructured rule with two cases:
- **`StreamingPartitionedTopKExec`**
(`sorts/streaming_partitioned_topk.rs`) — requires input sorted by
`(partition_keys, order_keys)`, computes the per-partition rank in one pass
(carrying only `prev_partition` / `prev_order` / `rank` / `count`), and emits
filtered batches incrementally. `EmissionType::Incremental`,
`maintains_input_order = true`, `required_input_ordering = (partition, order)`,
`KeyPartitioned` distribution. Supports `ROW_NUMBER`, `RANK` and `DENSE_RANK`
(matching `WindowFnKind`);.
- **`WindowTopN` rule** split into a shared matcher plus two classifiers:
- `try_streaming` — input already sorted (no `SortExec`): insert the
streaming operator between the window and its child. Applied
**unconditionally**, because it is a pure win (cull-before-window, O(1) memory,
order-preserving).
- `try_heap` — input unsorted (`SortExec` present): today's behavior,
replace the `SortExec` with `PartitionedTopKExec`. Still gated on
`enable_window_topn`.
## Are these changes tested?
Yes
## Are there any user-facing changes?
New physical operator `StreamingPartitionedTopKExec` in `EXPLAIN` output for
qualifying queries over pre-sorted input. Results are unchanged. Because the
streaming rewrite is a pure win, it applies even when `enable_window_topn` is
disabled (that flag continues to gate only the heap path).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
