You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/09 17:43:12 UTC
(datafusion) 05/05: fix(physical-plan): report Final emission for LeftSemi hash and neste… (#24976)
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 <mu...@gmail.com>
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org