Support join-filter pushdown for semi/anti join

datafusion/core/tests/sql/joins.rs

@@ -2112,6 +2112,34 @@ async fn left_semi_join() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn left_semi_join_pushdown() -> Result<()> {
+    let ctx = create_left_semi_anti_join_context_with_null_ids("t1_id", "t2_id", false)
+        .unwrap();
+
+    // assert logical plan
+    let sql = "SELECT t1.t1_id, t1.t1_name FROM t1 left semi join t2 on (t1.t1_id = t2.t2_id and t2.t2_int > 1)";
+    let msg = format!("Creating logical plan for '{sql}'");
+    let dataframe = ctx.sql(&("explain ".to_owned() + sql)).await.expect(&msg);
+    let plan = dataframe.into_optimized_plan()?;
+    let expected = vec![
+        "Explain [plan_type:Utf8, plan:Utf8]",
+        "  Projection: t1.t1_id, t1.t1_name [t1_id:UInt32;N, t1_name:Utf8;N]",
+        "    LeftSemi Join: t1.t1_id = t2.t2_id [t1_id:UInt32;N, t1_name:Utf8;N]",
+        "      TableScan: t1 projection=[t1_id, t1_name] [t1_id:UInt32;N, t1_name:Utf8;N]",
+        "      Filter: t2.t2_int > UInt32(1) [t2_id:UInt32;N, t2_int:UInt32;N]",
+        "        TableScan: t2 projection=[t2_id, t2_int] [t2_id:UInt32;N, t2_int:UInt32;N]", 
+    ];
+    let formatted = plan.display_indent_schema().to_string();
+    let actual: Vec<&str> = formatted.trim().lines().collect();
+    assert_eq!(
+        expected, actual,
+        "\n\nexpected:\n\n{expected:#?}\nactual:\n\n{actual:#?}\n\n"
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn left_anti_join() -> Result<()> {
     let test_repartition_joins = vec![true, false];

datafusion/core/tests/sqllogictests/test_files/join.slt

@@ -42,3 +42,42 @@ SELECT s.*, g.grade FROM students s join grades g on s.mark between g.min and g.
 Amina 89 5
 Salma 77 4
 Christen 50 3
+
+# two tables for join
+statement ok
+CREATE TABLE t1(t1_id INT, t1_name TEXT, t1_int INT) AS VALUES
+(11, 'a', 1),
+(22, 'b', 2),
+(33, 'c', 3),
+(44, 'd', 4);
+
+statement ok
+CREATE TABLE t2(t2_id INT, t2_name TEXT, t2_int INT) AS VALUES
+(11, 'z', 3),
+(22, 'y', 1),
+(44, 'x', 3),
+(55, 'w', 3);
+
+# left semi with wrong where clause
+query error DataFusion error: Schema error: No field named 't2'.'t2_id'. Valid fields are 't1'.'t1_id', 't1'.'t1_name', 't1'.'t1_int'.
+SELECT t1.t1_id,
+       t1.t1_name,
+       t1.t1_int
+FROM   t1 LEFT SEMI
+JOIN   t2
+ON     (
+              t1.t1_id = t2.t2_id)
+WHERE  t2.t2_id > 1
+
+# left semi join with on-filter
+query ITI rowsort
+SELECT t1.t1_id,
+       t1.t1_name,
+       t1.t1_int
+FROM   t1 LEFT SEMI
+JOIN   t2
+ON     (
+              t1.t1_id = t2.t2_id and t2.t2_int > 1)
+----
+11 a 1
+44 d 4