fix(physical-plan): report Final emission for LeftSemi hash and neste…

datafusion/core/tests/physical_optimizer/sanity_checker.rs

@@ -186,6 +186,45 @@ async fn test_hash_left_join_swap() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_hash_left_semi_join() -> Result<()> {
+    // LeftSemi hash joins emit matched build-side rows only after probing
+    // completes, so their emission type is Final.
+
+    let test1 = BinaryTestCase {
+        // The optimizer swaps the inputs and changes LeftSemi to RightSemi.
+        // The bounded right table becomes the build side, allowing matched
+        // rows from the unbounded left table to be emitted incrementally.
+        source_types: (SourceType::Unbounded, SourceType::Bounded),
+        expect_fail: false,
+    };
+
+    let test2 = BinaryTestCase {
+        // LeftSemi waits for the unbounded probe side to finish before
+        // emitting matched build-side rows. SanityCheckPlan must reject
+        // this pipeline because it cannot produce output.
+        source_types: (SourceType::Bounded, SourceType::Unbounded),
+        expect_fail: true,
+    };
+
+    let test3 = BinaryTestCase {
+        // Both inputs are bounded, so probing can finish and final output
+        // can be emitted.
+        source_types: (SourceType::Bounded, SourceType::Bounded),
+        expect_fail: false,
+    };
+
+    let case = QueryCase {
+        sql: "SELECT l.c1 FROM left AS l LEFT SEMI JOIN right AS r ON l.c1 = r.c1"
+            .to_string(),
+        cases: vec![Arc::new(test1), Arc::new(test2), Arc::new(test3)],
+        error_operator: "operator: HashJoinExec".to_string(),
+    };
+
+    case.run().await?;
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_hash_right_join_swap() -> Result<()> {
     let test1 = BinaryTestCase {

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -1218,19 +1218,22 @@ impl HashJoinExec {
             }
         };
 
-        let emission_type = if left.boundedness().is_unbounded() {
-            EmissionType::Final
-        } else if right.pipeline_behavior() == EmissionType::Incremental {
-            // Unmatched build-side rows can only be emitted once the probe side
-            // is exhausted; everything else is emitted incrementally.
-            if emits_unmatched_left_rows(join_type) {
-                EmissionType::Both
+        let emission_type =
+            // LeftSemi does not emit rows during probing. It records matching build-side
+            // rows in a bitmap and can only emit them after the probe side is exhausted.
+            if left.boundedness().is_unbounded() || join_type == JoinType::LeftSemi {
+                EmissionType::Final
+            } else if right.pipeline_behavior() == EmissionType::Incremental {
+                // Unmatched build-side rows can only be emitted once the probe side
+                // is exhausted; everything else is emitted incrementally.
+                if emits_unmatched_left_rows(join_type) {
+                    EmissionType::Both
+                } else {
+                    EmissionType::Incremental
+                }
             } else {
-                EmissionType::Incremental
-            }
-        } else {
-            right.pipeline_behavior()
-        };
+                right.pipeline_behavior()
+            };
 
         // If contains projection, update the PlanProperties.
         if let Some(projection) = projection {
@@ -4583,6 +4586,31 @@ mod tests {
         )
     }
 
+    #[tokio::test]
+    async fn test_semi_left_join_reports_final_emission() -> Result<()> {
+        let (left_schema, right_schema, on) = build_schema_and_on()?;
+
+        let left = TestMemoryExec::try_new_exec(&[vec![]], left_schema, None)?;
+
+        let right = TestMemoryExec::try_new_exec(&[vec![]], right_schema, None)?;
+
+        let join = HashJoinExec::try_new(
+            left,
+            right,
+            on,
+            None,
+            &JoinType::LeftSemi,
+            None,
+            PartitionMode::CollectLeft,
+            NullEquality::NullEqualsNothing,
+            false,
+        )?;
+
+        assert_eq!(join.properties().emission_type, EmissionType::Final);
+
+        Ok(())
+    }
+
     #[apply(hash_join_exec_configs)]
     #[tokio::test]
     async fn join_left_semi(

datafusion/physical-plan/src/joins/nested_loop_join.rs

@@ -390,28 +390,31 @@ impl NestedLoopJoinExec {
         let mut output_partitioning =
             asymmetric_join_output_partitioning(left, right, &join_type)?;
 
-        let emission_type = if left.boundedness().is_unbounded() {
-            EmissionType::Final
-        } else if right.pipeline_behavior() == EmissionType::Incremental {
-            match join_type {
-                // If we only need to generate matched rows from the probe side,
-                // we can emit rows incrementally.
-                JoinType::Inner
-                | JoinType::LeftSemi
-                | JoinType::RightSemi
-                | JoinType::Right
-                | JoinType::RightAnti
-                | JoinType::RightMark => EmissionType::Incremental,
-                // If we need to generate unmatched rows from the *build side*,
-                // we need to emit them at the end.
-                JoinType::Left
-                | JoinType::LeftAnti
-                | JoinType::LeftMark
-                | JoinType::Full => EmissionType::Both,
-            }
-        } else {
-            right.pipeline_behavior()
-        };
+        let emission_type =
+            // LeftSemi does not emit rows during probing. It records matching build-side
+            // rows in a bitmap and can only emit them after the probe side is exhausted.
+            if left.boundedness().is_unbounded() || join_type == JoinType::LeftSemi {
+                EmissionType::Final
+            } else if right.pipeline_behavior() == EmissionType::Incremental {
+                match join_type {
+                    // If we only need to generate matched rows from the probe side,
+                    // we can emit rows incrementally.
+                    JoinType::Inner
+                    | JoinType::LeftSemi
+                    | JoinType::RightSemi
+                    | JoinType::Right
+                    | JoinType::RightAnti
+                    | JoinType::RightMark => EmissionType::Incremental,
+                    // If we need to generate unmatched rows from the *build side*,
+                    // we need to emit them at the end.
+                    JoinType::Left
+                    | JoinType::LeftAnti
+                    | JoinType::LeftMark
+                    | JoinType::Full => EmissionType::Both,
+                }
+            } else {
+                right.pipeline_behavior()
+            };
 
         if let Some(projection) = projection {
             // construct a map from the input expressions to the output expression of the Projection
@@ -4660,6 +4663,18 @@ pub(crate) mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_left_semi_join_reports_final_emission() -> Result<()> {
+        let left = build_left_table();
+        let right = build_right_table();
+        let join =
+            NestedLoopJoinExec::try_new(left, right, None, &JoinType::LeftSemi, None)?;
+
+        assert_eq!(join.properties().emission_type, EmissionType::Final);
+
+        Ok(())
+    }
+
     #[rstest]
     #[tokio::test]
     async fn join_left_semi_with_filter(