diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs b/datafusion/core/tests/physical_optimizer/sanity_checker.rs index 184125dcbe18..32f02d0f6a2f 100644 --- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs +++ b/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 { diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 24b70a22e37e..8eeb16cc9a85 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/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( diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs index fd2e0ae21c2e..9f0abd92a0e5 100644 --- a/datafusion/physical-plan/src/joins/nested_loop_join.rs +++ b/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(