enable_window_topn returns wrong results when a projection under the filter drops a column (wrong ranking function chosen) #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

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

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

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.