diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index cc396da278fa4..24b70a22e37e5 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -41,8 +41,8 @@ use crate::joins::hash_join::stream::{ }; use crate::joins::join_hash_map::{JoinHashMapU32, JoinHashMapU64}; use crate::joins::utils::{ - OnceAsync, OnceFut, asymmetric_join_output_partitioning, reorder_output_after_swap, - swap_join_projection, update_hash, + OnceAsync, OnceFut, asymmetric_join_output_partitioning, emits_unmatched_left_rows, + is_existence_join, reorder_output_after_swap, swap_join_projection, update_hash, }; use crate::joins::{JoinOn, JoinOnRef, PartitionMode, SharedBitmapBuilder}; use crate::metrics::{Count, MetricBuilder, MetricCategory}; @@ -212,6 +212,63 @@ pub(super) struct NullValueScopeMap { pub(super) build_indices: UInt64Array, } +/// Null-aware (`NOT IN`) semantics of a hash join, derived from +/// [`HashJoinExec::null_aware`] and the join type. +/// +/// Only these three combinations are legal (see [`Self::try_new`]), so the +/// stream matches on this instead of re-checking `null_aware && join_type == ..`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum NullAwareMode { + /// Uncorrelated `build.key NOT IN (probe.key)`: emits build rows, and + /// none of them once any probe key is NULL. + LeftAnti, + /// Uncorrelated `probe.key NOT IN (build.key)`: emits probe rows, and + /// none of them once any build key is NULL. + RightAnti, + /// `NOT IN` as a nullable mark column on the build rows. `correlated` + /// means `on[1..]` are correlation scope keys (see + /// [`HashJoinExec::null_aware`]). + LeftMark { correlated: bool }, +} + +impl NullAwareMode { + /// Validates that `null_aware` may be set for this join and returns its mode. + pub(super) fn try_new( + join_type: JoinType, + partition_mode: PartitionMode, + num_keys: usize, + has_filter: bool, + ) -> Result { + let mode = match (join_type, partition_mode) { + (JoinType::LeftAnti, _) => Self::LeftAnti, + // `PartitionMode::CollectLeft` is safe because `RightAnti` is probe-driven + (JoinType::RightAnti, PartitionMode::CollectLeft) => Self::RightAnti, + (JoinType::LeftMark, _) => Self::LeftMark { + correlated: num_keys > 1, + }, + _ => { + return plan_err!( + "null_aware can only be true for LeftAnti joins and RightAnti joins with `CollectLeft` `PartitionMode`, or LeftMark joins, got {join_type} with {partition_mode}" + ); + } + }; + match mode { + Self::LeftAnti | Self::RightAnti if num_keys != 1 => plan_err!( + "null_aware {join_type} joins only support single column join key, got {num_keys} columns" + ), + Self::LeftMark { .. } if partition_mode == PartitionMode::Partitioned => { + plan_err!( + "null_aware joins require PartitionMode::CollectLeft, got PartitionMode::Partitioned" + ) + } + Self::RightAnti if has_filter => { + plan_err!("null_aware RightAnti join does not support a join filter") + } + _ => Ok(mode), + } + } +} + /// HashTable and input data for the left (build side) of a join pub(super) struct JoinLeftData { /// The hash table with indices into `batch` @@ -474,45 +531,7 @@ impl HashJoinExecBuilder { } = self; // Validate null_aware flag - if exec.null_aware { - let join_type = exec.join_type(); - let partition_mode = exec.partition_mode(); - if !matches!( - (join_type, partition_mode), - (JoinType::LeftAnti, _) - | (JoinType::RightAnti, PartitionMode::CollectLeft) // `PartitionMode::CollectLeft` is safe because `RightAnti` is probe-driven - | (JoinType::LeftMark, _) - ) { - return plan_err!( - "null_aware can only be true for LeftAnti joins and RightAnti joins with `CollectLeft` `PartitionMode`, or LeftMark joins, got {join_type} with {partition_mode}" - ); - } - let on = exec.on(); - if *join_type == JoinType::LeftAnti && on.len() != 1 { - return plan_err!( - "null_aware LeftAnti joins only support single column join key, got {} columns", - on.len() - ); - } - if *join_type == JoinType::RightAnti && on.len() != 1 { - return plan_err!( - "null_aware RightAnti joins only support single column join key, got {} columns", - on.len() - ); - } - if *join_type == JoinType::LeftMark - && matches!(partition_mode, PartitionMode::Partitioned) - { - return plan_err!( - "null_aware joins require PartitionMode::CollectLeft, got PartitionMode::Partitioned" - ); - } - if *join_type == JoinType::RightAnti && exec.filter.is_some() { - return plan_err!( - "null_aware RightAnti join does not support a join filter" - ); - } - } + exec.null_aware_mode()?; if preserve_properties { return Ok(exec); @@ -1121,19 +1140,28 @@ impl HashJoinExec { Ok(self) } + /// The null-aware semantics of this join, if [`Self::null_aware`] is set. + /// + /// Errors if `null_aware` is set on a join that does not support it. + pub(super) fn null_aware_mode(&self) -> Result> { + self.null_aware + .then(|| { + NullAwareMode::try_new( + self.join_type, + self.mode, + self.on.len(), + self.filter.is_some(), + ) + }) + .transpose() + } + /// Calculate order preservation flags for this hash join. fn maintains_input_order(join_type: JoinType) -> Vec { - vec![ - false, - matches!( - join_type, - JoinType::Inner - | JoinType::Right - | JoinType::RightAnti - | JoinType::RightSemi - | JoinType::RightMark - ), - ] + // The output follows the probe (right) order only when every output + // row is produced while scanning the probe side. Joins that also emit + // rows from the build-side visited bitmap at the end break that order. + vec![false, !need_produce_result_in_final(join_type)] } /// Get probe side information for the hash join. @@ -1193,21 +1221,12 @@ impl HashJoinExec { 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, + // 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 { right.pipeline_behavior() @@ -1285,17 +1304,10 @@ impl HashJoinExec { )) .with_partition_mode(partition_mode) .build()?; - // In case of anti / semi joins or if there is embedded projection in HashJoinExec, output column order is preserved, no need to add projection again - if matches!( - self.join_type(), - JoinType::LeftSemi - | JoinType::RightSemi - | JoinType::LeftAnti - | JoinType::RightAnti - | JoinType::LeftMark - | JoinType::RightMark - ) || self.projection.is_some() - { + // Semi / anti / mark joins output columns from one side only, so + // swapping does not reorder them; an embedded projection already fixes + // the order too. Otherwise a projection restores the original order. + if is_existence_join(self.join_type) || self.projection.is_some() { Ok(Arc::new(new_join)) } else { reorder_output_after_swap(Arc::new(new_join), &left.schema(), &right.schema()) @@ -1589,10 +1601,7 @@ impl ExecutionPlan for HashJoinExec { .flatten() .flatten(); - // The extra scope maps + null bitmap are only built for correlated - // null-aware LeftMark joins (`on[1..]` are correlation scope keys). - let with_null_aware_mark_state = - self.null_aware && self.join_type == JoinType::LeftMark && on_left.len() > 1; + let null_aware = self.null_aware_mode()?; let left_fut = match self.mode { PartitionMode::CollectLeft => self.left_fut.try_once(|| { @@ -1612,9 +1621,8 @@ impl ExecutionPlan for HashJoinExec { enable_dynamic_filter_pushdown, Arc::clone(context.session_config().options()), self.null_equality, - self.null_aware && self.join_type == JoinType::RightAnti, + null_aware, array_map_created_count, - with_null_aware_mark_state, )) })?, PartitionMode::Partitioned => { @@ -1634,11 +1642,8 @@ impl ExecutionPlan for HashJoinExec { enable_dynamic_filter_pushdown, Arc::clone(context.session_config().options()), self.null_equality, - false, + null_aware, array_map_created_count, - // Partitioned mode is rejected for null-aware joins (see - // `try_new`), so the extra null-aware state is never built. - false, )) } PartitionMode::Auto => { @@ -1688,7 +1693,7 @@ impl ExecutionPlan for HashJoinExec { self.right.output_ordering().is_some(), build_accumulator, self.mode, - self.null_aware, + null_aware, self.fetch, ))) } @@ -2700,7 +2705,7 @@ fn new_join_hashmap( /// # Returns /// `JoinLeftData` containing the hash map, consolidated batch, join key values, /// visited indices bitmap, and computed bounds (if requested). -#[expect(clippy::too_many_arguments, clippy::fn_params_excessive_bools)] +#[expect(clippy::too_many_arguments)] async fn collect_left_input( random_state: RandomState, left_stream: SendableRecordBatchStream, @@ -2712,12 +2717,18 @@ async fn collect_left_input( should_compute_dynamic_filters: bool, config: Arc, null_equality: NullEquality, - compute_build_side_has_null: bool, + null_aware: Option, array_map_created_count: Count, - with_null_aware_mark_state: bool, ) -> Result { let schema = left_stream.schema(); + // The extra scope maps + null bitmap are only built for correlated + // null-aware LeftMark joins (`on_left[1..]` are correlation scope keys). + let with_null_aware_mark_state = matches!( + null_aware, + Some(NullAwareMode::LeftMark { correlated: true }) + ); + let is_phj_candidate = is_perfect_hash_join_candidate(&on_left, &schema)?; let initial = BuildSideState::try_new( @@ -2954,7 +2965,7 @@ async fn collect_left_input( bounds = None; } - let build_has_null = compute_build_side_has_null + let build_has_null = null_aware == Some(NullAwareMode::RightAnti) && !left_values.is_empty() && left_values[0].logical_null_count() > 0; diff --git a/datafusion/physical-plan/src/joins/hash_join/stream.rs b/datafusion/physical-plan/src/joins/hash_join/stream.rs index cf2db5b037eb0..ff08828e94262 100644 --- a/datafusion/physical-plan/src/joins/hash_join/stream.rs +++ b/datafusion/physical-plan/src/joins/hash_join/stream.rs @@ -28,7 +28,7 @@ use crate::coalesce::{LimitedBatchCoalescer, PushBatchStatus}; use crate::joins::Map; use crate::joins::MapOffset; use crate::joins::PartitionMode; -use crate::joins::hash_join::exec::JoinLeftData; +use crate::joins::hash_join::exec::{JoinLeftData, NullAwareMode}; use crate::joins::hash_join::shared_bounds::{ PartitionBounds, PartitionBuildData, SharedBuildAccumulator, }; @@ -345,8 +345,8 @@ pub(super) struct HashJoinStream { /// Output buffer for coalescing small batches into larger ones with optional fetch limit. /// Uses `LimitedBatchCoalescer` to efficiently combine batches and absorb limit with 'fetch' output_buffer: LimitedBatchCoalescer, - /// Whether this is a null-aware anti or mark join - null_aware: bool, + /// Null-aware (`NOT IN`) semantics of this join, if any + null_aware: Option, } impl RecordBatchStream for HashJoinStream { @@ -496,7 +496,7 @@ impl HashJoinStream { right_side_ordered: bool, build_accumulator: Option>, mode: PartitionMode, - null_aware: bool, + null_aware: Option, fetch: Option, ) -> Self { // Create output buffer with coalescing and optional fetch limit. @@ -767,103 +767,37 @@ impl HashJoinStream { let timer = self.join_metrics.join_time.timer(); - // Null-aware anti join semantics: - - // For LeftAnti: output LEFT (build) rows where LEFT.key NOT IN RIGHT.key - // 1. If RIGHT (probe) contains NULL in any batch, no LEFT rows should be output - // 2. LEFT rows with NULL keys should not be output (handled in final stage) - - // For RightAnti: output RIGHT (probe) rows where RIGHT.key NOT IN LEFT.key - // 1. If LEFT (build) contains NULL, no RIGHT rows should be output - // 2. RIGHT rows with NULL keys should not be output - // 3. If LEFT (build) is empty, all RIGHT rows should be output - if self.null_aware { - match self.join_type { - JoinType::RightAnti => { - if build_side.left_data.build_side_has_null { - timer.done(); - self.state = HashJoinStreamState::FetchProbeBatch; - return Ok(StatefulStreamResult::Continue); - } - } - JoinType::LeftAnti => { - // Mark that we've seen a probe batch with actual rows (probe side is non-empty) - // Only set this if batch has rows - empty batches don't count - // Use shared atomic state so all partitions can see this global information - if state.batch.num_rows() > 0 { - build_side - .left_data - .probe_side_non_empty - .store(true, Ordering::Relaxed); - } - - // Check if probe side (RIGHT) contains NULL - // Since null_aware validation ensures single column join, we only check the first column - let probe_key_column = &state.values[0]; - let probe_has_null = if self.filter.is_some() { - probe_key_column.null_count() > 0 - } else { - probe_key_column.logical_null_count() > 0 - }; - if probe_has_null { - // Found NULL in probe side - set shared flag to prevent any output - build_side - .left_data - .probe_side_has_null - .store(true, Ordering::Relaxed); - } - - // If probe side has NULL (detected in this or any other partition), return empty result - if build_side - .left_data - .probe_side_has_null - .load(Ordering::Relaxed) - { - timer.done(); - self.state = HashJoinStreamState::FetchProbeBatch; - return Ok(StatefulStreamResult::Continue); - } - } - JoinType::LeftMark => { - if state.batch.num_rows() > 0 { - build_side - .left_data - .probe_side_non_empty - .store(true, Ordering::Relaxed); - } - - let probe_key_column = &state.values[0]; - if probe_key_column.null_count() > 0 { - build_side - .left_data - .probe_side_has_null - .store(true, Ordering::Relaxed); - } - } - _ => {} + if let Some(mode) = self.null_aware { + if null_aware_skip_probe_batch( + mode, + state, + &build_side.left_data, + self.filter.is_some(), + ) { + timer.done(); + self.state = HashJoinStreamState::FetchProbeBatch; + return Ok(StatefulStreamResult::Continue); } - } - // For correlated null-aware LeftMark (`on[1..]` scope keys, hence - // values len > 1), record this batch's UNKNOWN candidates once, before - // the first chunked lookup (offset == (0, None)). - // - // Must precede the empty-build-map return below: an all-NULL-key build - // side has an empty full-key map but still needs UNKNOWN marks. - if self.null_aware - && self.join_type == JoinType::LeftMark - && state.values.len() > 1 - && state.offset == (0, None) - { - mark_null_candidates_for_probe_batch( - build_side, - state, - &self.random_state, - self.batch_size, - &mut self.null_mark_hashes_buffer, - &mut self.null_mark_probe_indices_buffer, - &mut self.null_mark_build_indices_buffer, - )?; + // For correlated null-aware LeftMark, record this batch's UNKNOWN + // candidates once, before the first chunked lookup + // (offset == (0, None)). + // + // Must precede the empty-build-map return below: an all-NULL-key + // build side has an empty full-key map but still needs UNKNOWN marks. + if matches!(mode, NullAwareMode::LeftMark { correlated: true }) + && state.offset == (0, None) + { + mark_null_candidates_for_probe_batch( + build_side, + state, + &self.random_state, + self.batch_size, + &mut self.null_mark_hashes_buffer, + &mut self.null_mark_probe_indices_buffer, + &mut self.null_mark_build_indices_buffer, + )?; + } } let is_empty = !build_side.left_data.has_matchable_build_rows(); @@ -989,24 +923,8 @@ impl HashJoinStream { self.right_side_ordered, )?; - // If null-aware RightAnti join, we don't want to emit NULL probe keys - if self.join_type == JoinType::RightAnti && self.null_aware { - let probe_key = &state.values[0]; - // if the valid_keys mask is available, use that, else use the probe key - let mask = state - .valid_keys - .clone() - .or_else(|| probe_key.logical_nulls()); - // we only need this copy if there are NULLs - if let Some(mask) = mask.filter(|m| m.null_count() > 0) { - let filtered_right_indices = right_indices - .values() - .iter() - .copied() - .filter(|idx| !mask.is_null(*idx as usize)) - .collect::>(); - right_indices = UInt32Array::from(filtered_right_indices); - } + if self.null_aware == Some(NullAwareMode::RightAnti) { + right_indices = drop_null_probe_keys(state, right_indices); } // Build output batch and push to coalescer @@ -1071,8 +989,7 @@ impl HashJoinStream { // For null-aware anti join, if probe side had NULL, no rows should be output // Check shared atomic state to get global knowledge across all partitions - if self.null_aware - && self.join_type == JoinType::LeftAnti + if self.null_aware == Some(NullAwareMode::LeftAnti) && build_side .left_data .probe_side_has_null @@ -1090,80 +1007,39 @@ impl HashJoinStream { } // use the global left bitmap to produce the left indices and right indices - let (mut left_side, mut right_side) = get_final_indices_from_shared_bitmap( + let (left_side, right_side) = get_final_indices_from_shared_bitmap( build_side.left_data.visited_indices_bitmap(), self.join_type, true, ); - // For null-aware anti join, filter out LEFT rows with NULL in join keys - // BUT only if the probe side (RIGHT) was non-empty. If probe side is empty, - // NULL NOT IN (empty) = TRUE, so NULL rows should be returned. - // Use shared atomic state to get global knowledge across all partitions - if self.null_aware - && self.join_type == JoinType::LeftAnti - && build_side - .left_data - .probe_side_non_empty - .load(Ordering::Relaxed) - { - // Since null_aware validation ensures single column join, we only check the first column - let build_key_column = &build_side.left_data.values()[0]; - - // Filter out indices where the key is NULL - let filtered_indices: Vec = left_side - .iter() - .filter_map(|idx| { - let idx = idx.expect( - "LeftAnti final indices should always contain build-side rows", - ); - let idx_usize = idx as usize; - if build_key_column.is_null(idx_usize) { - None // Skip rows with NULL keys - } else { - Some(idx) - } - }) - .collect(); - - left_side = UInt64Array::from(filtered_indices); - right_side = UInt32Array::new_null(left_side.len()); - } + // Null-aware joins post-process the build rows under SQL three-valued + // logic; see the helpers for the rules. + let (left_side, right_side, mark_column) = match self.null_aware { + Some(NullAwareMode::LeftAnti) => { + let (left_side, right_side) = null_aware_left_anti_final_indices( + &build_side.left_data, + left_side, + right_side, + ); + (left_side, right_side, None) + } + Some(NullAwareMode::LeftMark { .. }) => { + let mark_column = null_aware_left_mark_column( + &build_side.left_data, + &left_side, + &right_side, + ); + (left_side, right_side, Some(mark_column)) + } + _ => (left_side, right_side, None), + }; self.join_metrics.input_batches.add(1); self.join_metrics.input_rows.add(left_side.len()); // Push final unmatched indices to output buffer if !left_side.is_empty() { - let mark_column = if self.null_aware && self.join_type == JoinType::LeftMark { - let probe_side_has_null = build_side - .left_data - .probe_side_has_null - .load(Ordering::Relaxed); - let probe_side_non_empty = build_side - .left_data - .probe_side_non_empty - .load(Ordering::Relaxed); - let build_key_column = &build_side.left_data.values()[0]; - let null_indices_bitmap = - if build_side.left_data.null_aware_mark_scope_map().is_some() { - Some(build_side.left_data.null_indices_bitmap().lock()) - } else { - None - }; - - Some(build_null_aware_left_mark_column( - &left_side, - &right_side, - build_key_column.as_ref(), - null_indices_bitmap.as_deref(), - probe_side_has_null, - probe_side_non_empty, - )) - } else { - None - }; - let empty_right_batch = RecordBatch::new_empty(self.right.schema()); let batch = build_batch_from_indices( &self.schema, @@ -1191,6 +1067,121 @@ impl HashJoinStream { } } +/// Applies the pre-lookup bookkeeping of a null-aware join to a probe batch +/// and returns `true` if the batch cannot contribute any output row. +/// +/// Under SQL three-valued logic a NULL on either side of `NOT IN` makes the +/// predicate UNKNOWN for every row of the other side: +/// - `RightAnti` emits probe rows, so a NULL build key silences every batch. +/// - `LeftAnti` emits build rows in the final stage, so a NULL probe key seen +/// by any partition silences everything; the shared flags carry this to the +/// final stage. +/// - `LeftMark` only records the shared flags; the mark column is built in +/// the final stage from them. +fn null_aware_skip_probe_batch( + mode: NullAwareMode, + state: &ProcessProbeBatchState, + left_data: &JoinLeftData, + has_filter: bool, +) -> bool { + match mode { + NullAwareMode::RightAnti => left_data.build_side_has_null, + NullAwareMode::LeftAnti | NullAwareMode::LeftMark { .. } => { + // Only batches with rows count: `NULL NOT IN (empty)` is TRUE. + if state.batch.num_rows() > 0 { + left_data + .probe_side_non_empty + .store(true, Ordering::Relaxed); + } + // `on[0]` is the `NOT IN` value key for both modes. + let probe_key_column = &state.values[0]; + let probe_has_null = match mode { + NullAwareMode::LeftAnti if !has_filter => { + probe_key_column.logical_null_count() > 0 + } + _ => probe_key_column.null_count() > 0, + }; + if probe_has_null { + left_data.probe_side_has_null.store(true, Ordering::Relaxed); + } + mode == NullAwareMode::LeftAnti + && left_data.probe_side_has_null.load(Ordering::Relaxed) + } + } +} + +/// Drops probe rows whose `NOT IN` key is NULL from the output of a null-aware +/// `RightAnti` join: `NULL NOT IN (..)` is never TRUE. +fn drop_null_probe_keys( + state: &ProcessProbeBatchState, + right_indices: UInt32Array, +) -> UInt32Array { + let probe_key = &state.values[0]; + // if the valid_keys mask is available, use that, else use the probe key + let mask = state + .valid_keys + .clone() + .or_else(|| probe_key.logical_nulls()); + // we only need this copy if there are NULLs + match mask.filter(|m| m.null_count() > 0) { + Some(mask) => UInt32Array::from_iter_values( + right_indices + .values() + .iter() + .copied() + .filter(|idx| !mask.is_null(*idx as usize)), + ), + None => right_indices, + } +} + +/// Final-stage rule of a null-aware `LeftAnti` join: a NULL build key means +/// `NULL NOT IN (probe)`, which is UNKNOWN (row dropped) unless the probe side +/// was empty, where it is TRUE (row kept). +fn null_aware_left_anti_final_indices( + left_data: &JoinLeftData, + left_side: UInt64Array, + right_side: UInt32Array, +) -> (UInt64Array, UInt32Array) { + if !left_data.probe_side_non_empty.load(Ordering::Relaxed) { + return (left_side, right_side); + } + // null_aware validation ensures a single join key + let build_key_column = &left_data.values()[0]; + let left_side = UInt64Array::from_iter_values(left_side.iter().filter_map(|idx| { + let idx = + idx.expect("LeftAnti final indices should always contain build-side rows"); + (!build_key_column.is_null(idx as usize)).then_some(idx) + })); + let right_side = UInt32Array::new_null(left_side.len()); + (left_side, right_side) +} + +/// Builds the nullable mark column of a null-aware `LeftMark` join from the +/// final indices and the shared NULL-tracking state. +fn null_aware_left_mark_column( + left_data: &JoinLeftData, + left_side: &UInt64Array, + right_side: &UInt32Array, +) -> ArrayRef { + let probe_side_has_null = left_data.probe_side_has_null.load(Ordering::Relaxed); + let probe_side_non_empty = left_data.probe_side_non_empty.load(Ordering::Relaxed); + let build_key_column = &left_data.values()[0]; + // Correlated joins precomputed the UNKNOWN decision per build row. + let null_indices_bitmap = left_data + .null_aware_mark_scope_map() + .is_some() + .then(|| left_data.null_indices_bitmap().lock()); + build_null_aware_left_mark_column( + left_side, + right_side, + build_key_column.as_ref(), + null_indices_bitmap.as_deref(), + probe_side_has_null, + probe_side_non_empty, + ) +} + /// Records which build rows of a correlated null-aware `LeftMark` join are /// UNKNOWN candidates for this probe batch. /// diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs index 798f22b1d5770..04ff9e30bf5f8 100644 --- a/datafusion/physical-plan/src/joins/utils.rs +++ b/datafusion/physical-plan/src/joins/utils.rs @@ -1192,6 +1192,31 @@ pub(crate) fn need_produce_result_in_final(join_type: JoinType) -> bool { ) } +/// Whether the join emits left rows that found no match on the right side +/// (`LeftMark` emits them with a `false` mark). Those rows can only be produced +/// once the right side has been fully consumed. +pub(crate) fn emits_unmatched_left_rows(join_type: JoinType) -> bool { + matches!( + join_type, + JoinType::Left | JoinType::LeftAnti | JoinType::LeftMark | JoinType::Full + ) +} + +/// Whether the join only tests for the existence of a match (semi, anti and +/// mark joins). Such joins output the columns of a single input, plus the +/// mark column for mark joins. +pub(crate) fn is_existence_join(join_type: JoinType) -> bool { + matches!( + join_type, + JoinType::LeftSemi + | JoinType::RightSemi + | JoinType::LeftAnti + | JoinType::RightAnti + | JoinType::LeftMark + | JoinType::RightMark + ) +} + pub(crate) fn get_final_indices_from_shared_bitmap( shared_bitmap: &SharedBitmapBuilder, join_type: JoinType,