SubhamSinghal opened a new issue, #26416:
URL: https://github.com/apache/datafusion/issues/26416
### Describe the bug
With `datafusion.optimizer.enable_window_topn = true`, the `WindowTopN` rule
can build `PartitionedTopKExec` from the **wrong window expression** and return
wrong results: missing tied rows, or rows that violate the query's own `WHERE`
clause.
In `WindowTopN::try_transform`
(`datafusion/physical-optimizer/src/window_topn.rs`), `col_idx` is read from
the `FilterExec` predicate, so it indexes the filter's input, i.e. the output
of the `ProjectionExec` below it. `find_window_below` walks through that
projection to the `BoundedWindowAggExec`, but `col_idx` is never translated
through it. The rule compares it directly with the window's **input** width
and computes `window_expr_idx = col_idx input_field_count`. When the projection
drops a column the window used, every later index shifts down by one, so with
two ranking functions the filter's column resolves to the other one. The rule
then removes the `FilterExec`, relying on `PartitionedTopKExec`'s `fetch` to
replace it, so nothing corrects the result.
### To Reproduce
```sql
CREATE TABLE t (pk INT, v INT, id INT) AS VALUES
(1, 0, 1), (1, 0, 2), (1, 1, 3), (2, 0, 4), (2, 0, 5), (2, 1, 6);
SET datafusion.optimizer.enable_window_topn = true;
```
**A: filter on `rank()`, rewritten as `row_number()`, so tied rows are lost**
```sql
SELECT pk, id, rn, rk FROM (
SELECT pk, v, id,
row_number() OVER (PARTITION BY pk ORDER BY v) AS rn,
rank() OVER (PARTITION BY pk ORDER BY v) AS rk
FROM t)
WHERE rk <= 1 ORDER BY pk, id;
```
`enable_window_topn = false` (correct):
```
+----+----+----+----+
| pk | id | rn | rk |
+----+----+----+----+
| 1 | 1 | 1 | 1 |
| 1 | 2 | 2 | 1 |
| 2 | 4 | 1 | 1 |
| 2 | 5 | 2 | 1 |
+----+----+----+----+
```
`enable_window_topn = true`:
```
+----+----+----+----+
| pk | id | rn | rk |
+----+----+----+----+
| 1 | 1 | 1 | 1 |
| 2 | 4 | 1 | 1 |
+----+----+----+----+
```
Plan: `PartitionedTopKExec: fn=row_number, fetch=1`, with no `FilterExec`.
**B: filter on `row_number()`, rewritten as `rank()`, so rows with `rn = 2`
are returned**
```sql
SELECT pk, id, rk, rn FROM (
SELECT pk, v, id,
rank() OVER (PARTITION BY pk ORDER BY v) AS rk,
row_number() OVER (PARTITION BY pk ORDER BY v) AS rn
FROM t)
WHERE rn <= 1 ORDER BY pk, id;
```
`enable_window_topn = false` (correct):
```
+----+----+----+----+
| pk | id | rk | rn |
+----+----+----+----+
| 1 | 1 | 1 | 1 |
| 2 | 4 | 1 | 1 |
+----+----+----+----+
```
`enable_window_topn = true`:
```
+----+----+----+----+
| pk | id | rk | rn |
+----+----+----+----+
| 1 | 1 | 1 | 1 |
| 1 | 2 | 1 | 2 |
| 2 | 4 | 1 | 1 |
| 2 | 5 | 1 | 2 |
+----+----+----+----+
```
Plan: `PartitionedTopKExec: fn=rank, fetch=1`, with no `FilterExec`.
**Why**, for query A:
```
Filter: rank() <= 1 -- col_idx = 3 (rank
in the projection's output)
Projection: t.pk, t.id, row_number(), rank() -- drops v
WindowAggr: [row_number(), rank()] -- output [pk, v, id,
row_number, rank]
TableScan: t projection=[pk, v, id] -- window input width
= 3
```
`rank()` is window output 4 but projection output 3. The rule computes
`window_expr_idx = 3 - 3 = 0` and picks
`row_number()`.
### Expected behavior
Results identical to `enable_window_topn = false`: the rule should use the
window expression the filter actually
references, or decline.
--
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]
