diff --git a/datafusion/common/src/join_type.rs b/datafusion/common/src/join_type.rs index 50a4d88f3f2c3..e6a90db2dc3eb 100644 --- a/datafusion/common/src/join_type.rs +++ b/datafusion/common/src/join_type.rs @@ -113,20 +113,6 @@ impl JoinType { | JoinType::RightMark ) } - - /// Returns true when an empty build side necessarily produces an empty - /// result for this join type. - pub fn empty_build_side_produces_empty_result(self) -> bool { - matches!( - self, - JoinType::Inner - | JoinType::Left - | JoinType::LeftSemi - | JoinType::LeftAnti - | JoinType::LeftMark - | JoinType::RightSemi - ) - } } impl Display for JoinType { diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index 781f7ce879b98..25b320f985507 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -2215,110 +2215,6 @@ mod tests { ) } - fn empty_build_with_probe_error_inputs() - -> (Arc, Arc, JoinOn) { - let left_batch = - build_table_i32(("a1", &vec![]), ("b1", &vec![]), ("c1", &vec![])); - let left_schema = left_batch.schema(); - let left: Arc = TestMemoryExec::try_new_exec( - &[vec![left_batch]], - Arc::clone(&left_schema), - None, - ) - .unwrap(); - - let err = exec_err!("bad data error"); - let right_batch = - build_table_i32(("a2", &vec![]), ("b1", &vec![]), ("c2", &vec![])); - let right_schema = right_batch.schema(); - let on = vec![( - Arc::new(Column::new_with_schema("b1", &left_schema).unwrap()) as _, - Arc::new(Column::new_with_schema("b1", &right_schema).unwrap()) as _, - )]; - let right: Arc = Arc::new( - MockExec::new(vec![Ok(right_batch), err], right_schema).with_use_task(false), - ); - - (left, right, on) - } - - async fn assert_empty_build_probe_behavior( - join_types: &[JoinType], - expect_probe_error: bool, - with_filter: bool, - ) { - let (left, right, on) = empty_build_with_probe_error_inputs(); - let filter = prepare_join_filter(); - - for join_type in join_types { - let join = if with_filter { - join_with_filter( - Arc::clone(&left), - Arc::clone(&right), - on.clone(), - filter.clone(), - join_type, - NullEquality::NullEqualsNothing, - ) - .unwrap() - } else { - join( - Arc::clone(&left), - Arc::clone(&right), - on.clone(), - join_type, - NullEquality::NullEqualsNothing, - ) - .unwrap() - }; - - let result = common::collect( - join.execute(0, Arc::new(TaskContext::default())).unwrap(), - ) - .await; - - if expect_probe_error { - let result_string = result.unwrap_err().to_string(); - assert!( - result_string.contains("bad data error"), - "actual: {result_string}" - ); - } else { - let batches = result.unwrap(); - assert!( - batches.is_empty(), - "expected no output batches for {join_type}, got {batches:?}" - ); - } - } - } - - fn hash_join_with_dynamic_filter( - left: Arc, - right: Arc, - on: JoinOn, - join_type: JoinType, - ) -> Result<(HashJoinExec, Arc)> { - let dynamic_filter = HashJoinExec::create_dynamic_filter(&on); - let mut join = HashJoinExec::try_new( - left, - right, - on, - None, - &join_type, - None, - PartitionMode::CollectLeft, - NullEquality::NullEqualsNothing, - false, - )?; - join.dynamic_filter = Some(HashJoinExecDynamicFilter { - filter: Arc::clone(&dynamic_filter), - build_accumulator: OnceLock::new(), - }); - - Ok((join, dynamic_filter)) - } - async fn join_collect( left: Arc, right: Arc, @@ -5027,70 +4923,6 @@ mod tests { } } - #[tokio::test] - async fn join_does_not_consume_probe_when_empty_build_fixes_output() { - assert_empty_build_probe_behavior( - &[ - JoinType::Inner, - JoinType::Left, - JoinType::LeftSemi, - JoinType::LeftAnti, - JoinType::LeftMark, - JoinType::RightSemi, - ], - false, - false, - ) - .await; - } - - #[tokio::test] - async fn join_does_not_consume_probe_when_empty_build_fixes_output_with_filter() { - assert_empty_build_probe_behavior( - &[ - JoinType::Inner, - JoinType::Left, - JoinType::LeftSemi, - JoinType::LeftAnti, - JoinType::LeftMark, - JoinType::RightSemi, - ], - false, - true, - ) - .await; - } - - #[tokio::test] - async fn join_still_consumes_probe_when_empty_build_needs_probe_rows() { - assert_empty_build_probe_behavior( - &[ - JoinType::Right, - JoinType::Full, - JoinType::RightAnti, - JoinType::RightMark, - ], - true, - false, - ) - .await; - } - - #[tokio::test] - async fn join_still_consumes_probe_when_empty_build_needs_probe_rows_with_filter() { - assert_empty_build_probe_behavior( - &[ - JoinType::Right, - JoinType::Full, - JoinType::RightAnti, - JoinType::RightMark, - ], - true, - true, - ) - .await; - } - #[tokio::test] async fn join_split_batch() { let left = build_table( @@ -5534,8 +5366,26 @@ mod tests { Arc::new(Column::new_with_schema("b1", &right.schema())?) as _, )]; - let (join, dynamic_filter) = - hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?; + // Create a dynamic filter manually + let dynamic_filter = HashJoinExec::create_dynamic_filter(&on); + let dynamic_filter_clone = Arc::clone(&dynamic_filter); + + // Create HashJoinExec with the dynamic filter + let mut join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + false, + )?; + join.dynamic_filter = Some(HashJoinExecDynamicFilter { + filter: dynamic_filter, + build_accumulator: OnceLock::new(), + }); // Execute the join let stream = join.execute(0, task_ctx)?; @@ -5543,7 +5393,7 @@ mod tests { // After the join completes, the dynamic filter should be marked as complete // wait_complete() should return immediately - dynamic_filter.wait_complete().await; + dynamic_filter_clone.wait_complete().await; Ok(()) } @@ -5565,8 +5415,26 @@ mod tests { Arc::new(Column::new_with_schema("b1", &right.schema())?) as _, )]; - let (join, dynamic_filter) = - hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?; + // Create a dynamic filter manually + let dynamic_filter = HashJoinExec::create_dynamic_filter(&on); + let dynamic_filter_clone = Arc::clone(&dynamic_filter); + + // Create HashJoinExec with the dynamic filter + let mut join = HashJoinExec::try_new( + left, + right, + on, + None, + &JoinType::Inner, + None, + PartitionMode::CollectLeft, + NullEquality::NullEqualsNothing, + false, + )?; + join.dynamic_filter = Some(HashJoinExecDynamicFilter { + filter: dynamic_filter, + build_accumulator: OnceLock::new(), + }); // Execute the join let stream = join.execute(0, task_ctx)?; @@ -5574,28 +5442,7 @@ mod tests { // Even with empty build side, the dynamic filter should be marked as complete // wait_complete() should return immediately - dynamic_filter.wait_complete().await; - - Ok(()) - } - - #[tokio::test] - async fn test_hash_join_skips_probe_on_empty_build_after_partition_bounds_report() - -> Result<()> { - let task_ctx = Arc::new(TaskContext::default()); - let (left, right, on) = empty_build_with_probe_error_inputs(); - - // Keep an extra consumer reference so execute() enables dynamic filter pushdown - // and enters the WaitPartitionBoundsReport path before deciding whether to poll - // the probe side. - let (join, dynamic_filter) = - hash_join_with_dynamic_filter(left, right, on, JoinType::Inner)?; - - let stream = join.execute(0, task_ctx)?; - let batches = common::collect(stream).await?; - assert!(batches.is_empty()); - - dynamic_filter.wait_complete().await; + dynamic_filter_clone.wait_complete().await; Ok(()) } diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index d64b35a4dd24d..b31982ea3b7b4 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -406,21 +406,6 @@ impl HashJoinStream { } } - /// Returns the next state after the build side has been fully collected - /// and any required build-side coordination has completed. - fn state_after_build_ready( - join_type: JoinType, - left_data: &JoinLeftData, - ) -> HashJoinStreamState { - if left_data.map().is_empty() - && join_type.empty_build_side_produces_empty_result() - { - HashJoinStreamState::Completed - } else { - HashJoinStreamState::FetchProbeBatch - } - } - /// Separate implementation function that unpins the [`HashJoinStream`] so /// that partial borrows work correctly fn poll_next_impl( @@ -484,9 +469,7 @@ impl HashJoinStream { if let Some(ref mut fut) = self.build_waiter { ready!(fut.get_shared(cx))?; } - let build_side = self.build_side.try_as_ready()?; - self.state = - Self::state_after_build_ready(self.join_type, build_side.left_data.as_ref()); + self.state = HashJoinStreamState::FetchProbeBatch; Poll::Ready(Ok(StatefulStreamResult::Continue)) } @@ -557,8 +540,7 @@ impl HashJoinStream { })); self.state = HashJoinStreamState::WaitPartitionBoundsReport; } else { - self.state = - Self::state_after_build_ready(self.join_type, left_data.as_ref()); + self.state = HashJoinStreamState::FetchProbeBatch; } self.build_side = BuildSide::Ready(BuildSideReadyState { left_data }); @@ -661,14 +643,10 @@ impl HashJoinStream { } } - // If the build side is empty, this stream only reaches ProcessProbeBatch for - // join types whose output still depends on probe rows. + // if the left side is empty, we can skip the (potentially expensive) join operation let is_empty = build_side.left_data.map().is_empty(); - if is_empty { - // Invariant: state_after_build_ready should have already completed - // join types whose result is fixed to empty when the build side is empty. - debug_assert!(!self.join_type.empty_build_side_produces_empty_result()); + if is_empty && self.filter.is_none() { let result = build_batch_empty_build_side( &self.schema, build_side.left_data.batch(), diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs index 56ab8366cfd85..cf4bf2cd163fd 100644 --- a/datafusion/physical-plan/src/joins/utils.rs +++ b/datafusion/physical-plan/src/joins/utils.rs @@ -1060,35 +1060,47 @@ pub(crate) fn build_batch_empty_build_side( column_indices: &[ColumnIndex], join_type: JoinType, ) -> Result { - if join_type.empty_build_side_produces_empty_result() { - // These join types only return data if the left side is not empty. - return Ok(RecordBatch::new_empty(Arc::new(schema.clone()))); - } + match join_type { + // these join types only return data if the left side is not empty, so we return an + // empty RecordBatch + JoinType::Inner + | JoinType::Left + | JoinType::LeftSemi + | JoinType::RightSemi + | JoinType::LeftAnti + | JoinType::LeftMark => Ok(RecordBatch::new_empty(Arc::new(schema.clone()))), - // The remaining joins return right-side rows and nulls for the left side. - let num_rows = probe_batch.num_rows(); - if schema.fields().is_empty() { - return new_empty_schema_batch(schema, num_rows); - } + // the remaining joins will return data for the right columns and null for the left ones + JoinType::Right | JoinType::Full | JoinType::RightAnti | JoinType::RightMark => { + let num_rows = probe_batch.num_rows(); + if schema.fields().is_empty() { + return new_empty_schema_batch(schema, num_rows); + } + let mut columns: Vec> = + Vec::with_capacity(schema.fields().len()); + + for column_index in column_indices { + let array = match column_index.side { + // left -> null array + JoinSide::Left => new_null_array( + build_batch.column(column_index.index).data_type(), + num_rows, + ), + // right -> respective right array + JoinSide::Right => Arc::clone(probe_batch.column(column_index.index)), + // right mark -> unset boolean array as there are no matches on the left side + JoinSide::None => Arc::new(BooleanArray::new( + BooleanBuffer::new_unset(num_rows), + None, + )), + }; - let columns = column_indices - .iter() - .map(|column_index| match column_index.side { - // left -> null array - JoinSide::Left => new_null_array( - build_batch.column(column_index.index).data_type(), - num_rows, - ), - // right -> respective right array - JoinSide::Right => Arc::clone(probe_batch.column(column_index.index)), - // right mark -> unset boolean array as there are no matches on the left side - JoinSide::None => { - Arc::new(BooleanArray::new(BooleanBuffer::new_unset(num_rows), None)) + columns.push(array); } - }) - .collect(); - Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?) + Ok(RecordBatch::try_new(Arc::new(schema.clone()), columns)?) + } + } } /// The input is the matched indices for left and right and