fix(optimizer): fix full outer join null filter (#24899)

Author: chenzl25

e2e_test/streaming/join/full_outer_join_global_agg.slt.part

@@ -0,0 +1,117 @@
+# Regression test for: https://github.com/risingwavelabs/risingwave/issues/24790
+statement ok
+drop materialized view if exists t_stat;
+
+statement ok
+drop materialized view if exists t1_v;
+
+statement ok
+drop materialized view if exists t2_v;
+
+statement ok
+drop table if exists t;
+
+statement ok
+create table t (
+    id int primary key,
+    key string,
+    amount int
+);
+
+statement ok
+insert into t values
+    (1, 'k', 100),
+    (2, 'k', 200);
+
+statement ok
+create materialized view t1_v as
+select sum(amount) as amount, max(key) as key
+from t
+where id = 1;
+
+statement ok
+create materialized view t2_v as
+select sum(amount) as amount, max(key) as key
+from t
+where id = 2;
+
+statement ok
+create materialized view t_stat as
+select
+    coalesce(a.key, b.key) as key,
+    coalesce(a.amount, 0) as amount_1,
+    coalesce(b.amount, 0) as amount_2
+from t1_v a
+full outer join t2_v b
+    on a.key = b.key;
+
+statement ok
+flush;
+
+query TII
+select key, amount_1, amount_2 from t_stat;
+----
+k 100 200
+
+statement ok
+drop materialized view t_stat;
+
+statement ok
+drop materialized view t1_v;
+
+statement ok
+drop materialized view t2_v;
+
+statement ok
+drop table t;
+
+# Regression test for full outer join on nullable global-agg key
+statement ok
+drop materialized view if exists mvj;
+
+statement ok
+drop materialized view if exists mv;
+
+statement ok
+drop table if exists t;
+
+statement ok
+create table t (a int, b int);
+
+statement ok
+create materialized view mv as
+select max(a) ma, max(b) mb from t;
+
+statement ok
+insert into t values (1, NULL);
+
+statement ok
+flush;
+
+query II
+select ma, mb from mv;
+----
+1 NULL
+
+statement ok
+create materialized view mvj as
+select mv1.ma as ma1
+from mv as mv1
+full outer join mv as mv2
+    on mv1.mb = mv2.mb;
+
+statement ok
+flush;
+
+query I rowsort
+select ma1 from mvj;
+----
+
+statement ok
+drop materialized view mvj;
+
+statement ok
+drop materialized view mv;
+
+statement ok
+drop table t;

e2e_test/streaming/join/main.slt

@@ -5,6 +5,7 @@ include ./join.slt.part
 include ./join-3way.slt.part
 include ./asof_join.slt.part
 include ./interval_join.slt.part
+include ./full_outer_join_global_agg.slt.part
 
 statement ok
 set streaming_join_encoding to memory_optimized;
@@ -13,6 +14,7 @@ include ./join.slt.part
 include ./join-3way.slt.part
 include ./asof_join.slt.part
 include ./interval_join.slt.part
+include ./full_outer_join_global_agg.slt.part
 
 statement ok
 set streaming_join_encoding to default;

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

@@ -558,7 +558,7 @@
       └─StreamProject { exprs: [(2:Int32 * $expr1) as $expr2, ($expr1 + $expr1) as $expr3, a._row_id, b._row_id, a.x, b.x] }
         └─StreamProject { exprs: [a.x, b.x, Coalesce(a.x, b.x) as $expr1, a._row_id, b._row_id] }
           └─StreamFilter { predicate: ((2:Int32 * Coalesce(a.x, b.x)) < 10:Int32) }
-            └─StreamFilter { predicate: (IsNotNull(a._row_id) OR IsNotNull(b._row_id)) }
+            └─StreamFilter { predicate: ((IsNotNull(a._row_id) OR IsNotNull(b._row_id)) OR (IsNotNull(a.x) OR IsNotNull(b.x))) }
               └─StreamHashJoin { type: FullOuter, predicate: a.x = b.x, output: [a.x, b.x, a._row_id, b._row_id] }
                 ├─StreamExchange { dist: HashShard(a.x) }
                 │ └─StreamTableScan { table: a, columns: [a.x, a._row_id], stream_scan_type: SnapshotBackfill, stream_key: [a._row_id], pk: [_row_id], dist: UpstreamHashShard(a._row_id) }

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

@@ -107,7 +107,7 @@
     └─StreamProject { exprs: [$expr1, t1._row_id, t2._row_id, t1.v, t2.v] }
       └─StreamMaterializedExprs { exprs: [PgSleep(t1.v::Float64) as $expr1] }
         └─StreamExchange { dist: HashShard(t1.v, t1._row_id, t2._row_id, t2.v) }
-          └─StreamFilter { predicate: (IsNotNull(t1._row_id) OR IsNotNull(t2._row_id)) }
+          └─StreamFilter { predicate: ((IsNotNull(t1._row_id) OR IsNotNull(t2._row_id)) OR (IsNotNull(t1.v) OR IsNotNull(t2.v))) }
             └─StreamHashJoin { type: FullOuter, predicate: t1.v = t2.v, output: [t1.v, t1._row_id, t2._row_id, t2.v] }
               ├─StreamExchange { dist: HashShard(t1.v) }
               │ └─StreamTableScan { table: t1, columns: [t1.v, t1._row_id], stream_scan_type: SnapshotBackfill, stream_key: [t1._row_id], pk: [_row_id], dist: UpstreamHashShard(t1._row_id) }
@@ -121,7 +121,7 @@
             └── StreamExchange Hash([0, 1, 2, 3]) from 1
 
     Fragment 1
-    StreamFilter { predicate: (IsNotNull(t1._row_id) OR IsNotNull(t2._row_id)) }
+    StreamFilter { predicate: ((IsNotNull(t1._row_id) OR IsNotNull(t2._row_id)) OR (IsNotNull(t1.v) OR IsNotNull(t2.v))) }
     └── StreamHashJoin { type: FullOuter, predicate: t1.v = t2.v, output: [t1.v, t1._row_id, t2._row_id, t2.v] } { tables: [ HashJoinLeft: 1, HashJoinDegreeLeft: 2, HashJoinRight: 3, HashJoinDegreeRight: 4 ] }
         ├── StreamExchange Hash([0]) from 2
         └── StreamExchange Hash([0]) from 3