test: add end-to-end SQL coverage for the embedded-projection rewrite

Adds a PROJ group to window_topn.slt covering the shape the rule now
handles: the outer query drops `rn`, so projection pushdown folds the
ProjectionExec into the FilterExec.

- PROJ1 asserts results are correct through the rewrite.
- PROJ2 shows the plan is rewritten to PartitionedTopKExec.
- PROJ3 shows the same query with the rule off, where the FilterExec
  still carries `projection=[...]` -- the shape that used to make the
  rule bail.
- PROJ4 covers an outer LIMIT, confirming the limit survives as its own
  GlobalLimitExec rather than being folded into a discarded filter.

Author: zhuqi-lucas

datafusion/sqllogictest/test_files/window_topn.slt

@@ -1592,3 +1592,115 @@ set datafusion.execution.batch_size = 8192;
 
 statement ok
 set datafusion.optimizer.enable_window_topn = false;
+
+###
+### Embedded-projection group: the outer query drops `rn`, so projection
+### pushdown folds the ProjectionExec into the FilterExec. Before this was
+### supported the rule bailed on `filter.projection().is_some()` and the
+### rewrite never fired for this (very common) shape.
+###
+
+statement ok
+CREATE TABLE window_topn_t (id INT, pk INT, val INT) AS VALUES
+  (1, 1, 10),
+  (2, 1, 20),
+  (3, 1, 30),
+  (4, 1, 40),
+  (5, 2, 5),
+  (6, 2, 15),
+  (7, 2, 25),
+  (8, 3, 100),
+  (9, 3, 50),
+  (10, 3, 75);
+
+statement ok
+set datafusion.optimizer.enable_window_topn = true;
+
+# Test PROJ1: results are correct when the filter carries a projection.
+query III rowsort
+SELECT id, pk, val FROM (
+  SELECT *, ROW_NUMBER() OVER (PARTITION BY pk ORDER BY val) as rn FROM window_topn_t
+) WHERE rn <= 3;
+----
+1 1 10
+10 3 75
+2 1 20
+3 1 30
+5 2 5
+6 2 15
+7 2 25
+8 3 100
+9 3 50
+
+statement ok
+SET datafusion.explain.physical_plan_only = true;
+
+# Test PROJ2: the plan is rewritten to PartitionedTopKExec, and the filter's
+# projection is preserved as the top-level ProjectionExec.
+query TT
+EXPLAIN SELECT id, pk, val FROM (
+  SELECT *, ROW_NUMBER() OVER (PARTITION BY pk ORDER BY val) as rn FROM window_topn_t
+) WHERE rn <= 3;
+----
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, pk@1 as pk, val@2 as val]
+02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [window_topn_t.pk] ORDER BY [window_topn_t.val ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [window_topn_t.pk] ORDER BY [window_topn_t.val ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+03)----RepartitionExec: partitioning=Hash([pk@1], 4), input_partitions=1, maintains_sort_order=true
+04)------PartitionedTopKExec: fn=row_number, fetch=3, partition=[pk@1], order=[val@2 ASC NULLS LAST]
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Test PROJ3: with the rule off, the same query shows the shape the rule now
+# handles -- a FilterExec carrying `projection=[...]` over the window.
+statement ok
+set datafusion.optimizer.enable_window_topn = false;
+
+query TT
+EXPLAIN SELECT id, pk, val FROM (
+  SELECT *, ROW_NUMBER() OVER (PARTITION BY pk ORDER BY val) as rn FROM window_topn_t
+) WHERE rn <= 3;
+----
+physical_plan
+01)FilterExec: row_number() PARTITION BY [window_topn_t.pk] ORDER BY [window_topn_t.val ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@3 <= 3, projection=[id@0, pk@1, val@2]
+02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [window_topn_t.pk] ORDER BY [window_topn_t.val ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [window_topn_t.pk] ORDER BY [window_topn_t.val ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+03)----SortExec: expr=[pk@1 ASC NULLS LAST, val@2 ASC NULLS LAST], preserve_partitioning=[false]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+set datafusion.optimizer.enable_window_topn = true;
+
+# Test PROJ4: adding an outer LIMIT must not lose the limit. `WindowTopN` runs
+# before `LimitPushdown`, so the rewrite happens while the filter still has no
+# fetch, and the limit lands as its own `GlobalLimitExec` -- it is never folded
+# into a filter this rule then discards. `target_partitions = 1` is what makes
+# `LimitPushdown` able to reach the filter at all.
+statement ok
+SET datafusion.execution.target_partitions = 1;
+
+query TT
+EXPLAIN SELECT id, pk, val FROM (
+  SELECT *, ROW_NUMBER() OVER (PARTITION BY pk ORDER BY val) as rn FROM window_topn_t
+) WHERE rn <= 3 LIMIT 2;
+----
+physical_plan
+01)ProjectionExec: expr=[id@0 as id, pk@1 as pk, val@2 as val]
+02)--GlobalLimitExec: skip=0, fetch=2
+03)----BoundedWindowAggExec: wdw=[row_number() PARTITION BY [window_topn_t.pk] ORDER BY [window_topn_t.val ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { "row_number() PARTITION BY [window_topn_t.pk] ORDER BY [window_topn_t.val ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": UInt64 }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
+04)------PartitionedTopKExec: fn=row_number, fetch=3, partition=[pk@1], order=[val@2 ASC NULLS LAST]
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
+SET datafusion.explain.physical_plan_only = false;
+
+query III
+SELECT id, pk, val FROM (
+  SELECT *, ROW_NUMBER() OVER (PARTITION BY pk ORDER BY val) as rn FROM window_topn_t
+) WHERE rn <= 3 LIMIT 2;
+----
+1 1 10
+2 1 20
+
+statement ok
+SET datafusion.execution.target_partitions = 4;
+
+statement ok
+set datafusion.optimizer.enable_window_topn = false;