This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24976-9f21155c48faac3f1658853ad6a55bd314a322ed in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 4084d4c57bc886d8a645ec7643748d09b21397db Author: Tiny_Murky <[email protected]> AuthorDate: Wed Sep 9 14:52:23 2026 +0000 fix(physical-plan): report Final emission for LeftSemi hash and neste… (#24976) ## Which issue does this PR close? - #24962 ## Rationale for this change `HashJoinExec` and `NestedLoopJoinExec` report `LeftSemi` joins as emitting incrementally, but they only emit matched build-side rows after the probe side is exhausted. Reporting `EmissionType::Final` reflects their actual behavior and allows `SanityCheckPlan` to reject pipelines with an unbounded probe side that cannot produce output. ## What changes are included in this PR? - Classify `LeftSemi` joins as `EmissionType::Final` in both operators, regardless of the probe input's emission type. - Added unit tests verifying the emission type of LeftSemi joins in both `HashJoinExec` and `NestedLoopJoinExec`. - Add an integration test covering input swapping for an unbounded left input, rejection of an unbounded probe input, and successful planning when both inputs are bounded. ## Are these changes tested? Added unit tests verifying the emission type of LeftSemi joins in both `HashJoinExec` and `NestedLoopJoinExec`. Added an integration test covering: - Unbounded left and bounded right: planning succeeds after swapping to RightSemi. - Bounded left and unbounded right: planning is rejected. - Both inputs bounded: planning succeeds. Following test commands have been executed and passed - `cargo test -p datafusion` - `cargo test --profile=ci --test sqllogictests` - `cargo test -p datafusion` - `cargo test -p datafusion-cli` ## Are there any user-facing changes? Yes. `LeftSemi` hash and nested loop joins now report Final emission. Plans with a bounded build side and an unbounded probe side are rejected instead of being accepted despite being unable to produce output. Results for bounded inputs are unchanged. --- .../tests/physical_optimizer/sanity_checker.rs | 39 ++++++++++++++ .../physical-plan/src/joins/hash_join/exec.rs | 52 ++++++++++++++----- .../physical-plan/src/joins/nested_loop_join.rs | 59 ++++++++++++++-------- 3 files changed, 116 insertions(+), 34 deletions(-) diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs b/datafusion/core/tests/physical_optimizer/sanity_checker.rs index 184125dcbe..32f02d0f6a 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 01dec6570b..5f22741662 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 { @@ -4732,6 +4735,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 fd2e0ae21c..9f0abd92a0 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( --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
