fix: Fix the incorrect null joined rows for outer join with join filter

datafusion/core/tests/sql/joins.rs

@@ -235,3 +235,47 @@ async fn join_change_in_planner_without_sort_not_allowed() -> Result<()> {
     }
     Ok(())
 }
+
+#[tokio::test]
+async fn test_smj_right_filtered() -> Result<()> {
+    let ctx: SessionContext = SessionContext::new();
+
+    let sql = "set datafusion.optimizer.prefer_hash_join = false;";
+    let _ = ctx.sql(sql).await?.collect().await?;
+
+    let sql = "set datafusion.execution.batch_size = 100";
+    let _ = ctx.sql(sql).await?.collect().await?;
+
+    let sql = "
+            select * from (
+            with t as (
+                select id, id % 5 id1 from (select unnest(range(0,10)) id)
+            ), t1 as (
+                select id % 10 id, id + 2 id1 from (select unnest(range(0,10)) id)
+            )
+            select * from t right join t1 on t.id1 = t1.id and t.id > t1.id1
+            ) order by 1, 2, 3, 4
+        ";
+
+    let actual = ctx.sql(sql).await?.collect().await?;
+
+    let expected: Vec<&str> = vec![
+        "+----+-----+----+-----+",
+        "| id | id1 | id | id1 |",
+        "+----+-----+----+-----+",
+        "| 5  | 0   | 0  | 2   |",
+        "| 6  | 1   | 1  | 3   |",
+        "| 7  | 2   | 2  | 4   |",
+        "| 8  | 3   | 3  | 5   |",
+        "| 9  | 4   | 4  | 6   |",
+        "|    |     | 5  | 7   |",
+        "|    |     | 6  | 8   |",
+        "|    |     | 7  | 9   |",
+        "|    |     | 8  | 10  |",
+        "|    |     | 9  | 11  |",
+        "+----+-----+----+-----+",
+    ];
+    datafusion_common::assert_batches_eq!(expected, &actual);
+
+    Ok(())
+}