fix(over window): fix over window predicate pushdown

e2e_test/over_window/generated/batch/main.slt.part

@@ -8,3 +8,4 @@ include ./rank_func/mod.slt.part
 include ./expr_in_win_func/mod.slt.part
 include ./agg_in_win_func/mod.slt.part
 include ./opt_agg_then_join/mod.slt.part
+include ./with_filter/mod.slt.part

e2e_test/over_window/generated/batch/with_filter/mod.slt.part

@@ -0,0 +1,27 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+# Test window functions together with filters.
+# https://github.com/risingwavelabs/risingwave/issues/13653
+
+statement ok
+create table t (id int, cat varchar, rule varchar, at timestamptz);
+
+statement ok
+insert into t values
+  (1, 'foo', 'A', '2023-11-23T12:00:42Z')
+, (2, 'foo', 'B', '2023-11-23T12:01:15Z');
+
+query TT
+select rule, lag(rule) over (partition by cat order by at) from t where rule = 'B';
+----
+B  NULL
+
+query TT
+select * from (select rule, lag(rule) over (partition by cat order by at) as prev_rule from t) where rule = 'B';
+----
+B  A
+
+query TT
+select * from (select rule, at, row_number() over (partition by cat order by at) as rank from t) where at = '2023-11-23T12:01:15Z'::timestamptz;
+----
+B  2023-11-23 12:01:15+00:00  2

e2e_test/over_window/generated/streaming/main.slt.part

@@ -8,3 +8,4 @@ include ./rank_func/mod.slt.part
 include ./expr_in_win_func/mod.slt.part
 include ./agg_in_win_func/mod.slt.part
 include ./opt_agg_then_join/mod.slt.part
+include ./with_filter/mod.slt.part

e2e_test/over_window/generated/streaming/with_filter/mod.slt.part

@@ -0,0 +1,27 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+# Test window functions together with filters.
+# https://github.com/risingwavelabs/risingwave/issues/13653
+
+statement ok
+create table t (id int, cat varchar, rule varchar, at timestamptz);
+
+statement ok
+insert into t values
+  (1, 'foo', 'A', '2023-11-23T12:00:42Z')
+, (2, 'foo', 'B', '2023-11-23T12:01:15Z');
+
+query TT
+select rule, lag(rule) over (partition by cat order by at) from t where rule = 'B';
+----
+B  NULL
+
+query TT
+select * from (select rule, lag(rule) over (partition by cat order by at) as prev_rule from t) where rule = 'B';
+----
+B  A
+
+query TT
+select * from (select rule, at, row_number() over (partition by cat order by at) as rank from t) where at = '2023-11-23T12:01:15Z'::timestamptz;
+----
+B  2023-11-23 12:01:15+00:00  2

e2e_test/over_window/templates/main.slt.part

@@ -6,3 +6,4 @@ include ./rank_func/mod.slt.part
 include ./expr_in_win_func/mod.slt.part
 include ./agg_in_win_func/mod.slt.part
 include ./opt_agg_then_join/mod.slt.part
+include ./with_filter/mod.slt.part

e2e_test/over_window/templates/with_filter/mod.slt.part

@@ -0,0 +1,25 @@
+# Test window functions together with filters.
+# https://github.com/risingwavelabs/risingwave/issues/13653
+
+statement ok
+create table t (id int, cat varchar, rule varchar, at timestamptz);
+
+statement ok
+insert into t values
+  (1, 'foo', 'A', '2023-11-23T12:00:42Z')
+, (2, 'foo', 'B', '2023-11-23T12:01:15Z');
+
+query TT
+select rule, lag(rule) over (partition by cat order by at) from t where rule = 'B';
+----
+B  NULL
+
+query TT
+select * from (select rule, lag(rule) over (partition by cat order by at) as prev_rule from t) where rule = 'B';
+----
+B  A
+
+query TT
+select * from (select rule, at, row_number() over (partition by cat order by at) as rank from t) where at = '2023-11-23T12:01:15Z'::timestamptz;
+----
+B  2023-11-23 12:01:15+00:00  2

src/frontend/planner_test/tests/testdata/input/over_window_function.yaml

@@ -492,3 +492,19 @@
     - stream_plan
     - optimized_logical_plan_for_batch
     - batch_plan
+
+# With filter
+- sql: |
+    create table t (id int, cat varchar, rule varchar, at timestamptz);
+    select * from (select cat, rule, at, lag(rule) over (partition by cat order by at) as prev_rule from t) as with_prev
+    where rule = 'B' and cat is not null and at = '2023-11-23T12:00:42Z'::timestamptz;
+  expected_outputs:
+    - optimized_logical_plan_for_stream
+    - optimized_logical_plan_for_batch
+- sql: |
+    create table t (id int, cat varchar, rule varchar, at timestamptz);
+    select cat, rule, at, lag(rule) over (partition by cat order by at) as prev_rule from t
+    where rule = 'B' and cat is not null and at = '2023-11-23T12:00:42Z'::timestamptz;
+  expected_outputs:
+    - optimized_logical_plan_for_stream
+    - optimized_logical_plan_for_batch

src/frontend/planner_test/tests/testdata/output/column_pruning.yaml

@@ -151,8 +151,10 @@
   optimized_logical_plan_for_batch: |-
     LogicalProject { exprs: [row_number] }
     └─LogicalOverWindow { window_functions: [row_number() OVER(PARTITION BY t.x ORDER BY t.y ASC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] }
-      └─LogicalTopN { order: [t.y ASC], limit: 3, offset: 0, group_key: [t.x] }
-        └─LogicalScan { table: t, output_columns: [t.x, t.y], required_columns: [t.x, t.y, t.z], predicate: (t.z > 0:Int32) AND (t.y > 0:Int32) AND (t.x > 0:Int32) }
+      └─LogicalProject { exprs: [t.x, t.y] }
+        └─LogicalFilter { predicate: (t.z > 0:Int32) AND (t.y > 0:Int32) }
+          └─LogicalTopN { order: [t.y ASC], limit: 3, offset: 0, group_key: [t.x] }
+            └─LogicalScan { table: t, columns: [t.x, t.y, t.z], predicate: (t.x > 0:Int32) }
 - name: mixed
   sql: |
     create table t (v1 bigint, v2 double precision, v3 int);