diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index f1bc734d7292b..55c4fb9b362a6 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -73,7 +73,7 @@ use datafusion_common::{ not_impl_err, plan_err, }; use datafusion_common::{ - TableReference, assert_eq_or_internal_err, assert_or_internal_err, + NullEquality, TableReference, assert_eq_or_internal_err, assert_or_internal_err, }; use datafusion_datasource::file_groups::FileGroup; use datafusion_datasource::memory::MemorySourceConfig; @@ -1639,9 +1639,10 @@ impl DefaultPhysicalPlanner { let prefer_hash_join = session_state.config_options().optimizer.prefer_hash_join; - // Null-aware joins are pinned to CollectLeft hash joins (see - // `HashJoinExec::null_aware`): never repartition them, and - // never route them to the sort-merge path below. + // Null-aware hash joins require CollectLeft for global NULL tracking, + // so they must not be hash-repartitioned. + // Supported null-aware LeftAnti joins can instead use the + // single-partition sort-merge path below. let can_repartition_join = session_state.config().target_partitions() > 1 && session_state.config().repartition_joins() && !*null_aware; @@ -1781,16 +1782,44 @@ impl DefaultPhysicalPlanner { None, )?) } + } else if *null_aware + && !prefer_hash_join + && *join_type == JoinType::LeftAnti + && join_on.len() == 1 + && join_filter.is_none() + && *null_equality == NullEquality::NullEqualsNothing + { + // Use null-aware sort-merge join for single-key LeftAnti joins + // without a filter. Single-partition inputs and NULLS FIRST ordering + // let the stream determine whether the entire RHS is empty or + // contains NULL before emitting any rows. + Arc::new( + SortMergeJoinExec::try_new( + physical_left, + physical_right, + join_on, + join_filter, + *join_type, + vec![SortOptions::new(false, true)], // ASC NULLS FIRST + *null_equality, + )? + .with_null_aware(true)?, + ) } else if session_state.config().target_partitions() > 1 && session_state.config().repartition_joins() && !prefer_hash_join && !*null_aware - // Null-aware joins (e.g. `NOT IN` with a nullable subquery) must - // use the CollectLeft HashJoin below: SortMergeJoinExec does not - // implement null-aware anti-join semantics and would return wrong - // results when the right side contains a null join key. + // Null-aware joins (e.g. `NOT IN` with a nullable subquery) that do not + // satisfy the conditions above must use the CollectLeft HashJoin below. + // SortMergeJoinExec currently supports null-aware semantics only for + // single-key LeftAnti joins without a filter, using NullEqualsNothing. + // Routing other null-aware joins through the ordinary sort-merge path + // would not preserve NOT IN semantics, including when the subquery + // contains a NULL key. { - // Use SortMergeJoin if hash join is not preferred + // Use ordinary sort-merge join when hash join is not preferred. + // Null-aware joins not handled above fall back to CollectLeft + // HashJoin because their semantics are not supported by sort-merge join. let join_on_len = join_on.len(); Arc::new(SortMergeJoinExec::try_new( physical_left, @@ -1802,7 +1831,7 @@ impl DefaultPhysicalPlanner { *null_equality, )?) } else { - // Null-aware joins need global probe-side state, so keep + // Remaining null-aware joins need global probe-side state, so keep // them in CollectLeft mode. let partition_mode = if can_repartition_join { PartitionMode::Auto diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs index 912903f31809c..da04ec1d849dd 100644 --- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs +++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs @@ -197,17 +197,17 @@ pub fn adjust_input_keys_ordering( requirements.children[1].data = shift_right_required(&requirements.data, left_columns_len) .unwrap_or_default(); - } else if let Some(SortMergeJoinExec { - left, - right, - on, - filter, - join_type, - sort_options, - null_equality, - .. - }) = plan.downcast_ref::() - { + } else if let Some(origin_exec) = plan.downcast_ref::() { + let SortMergeJoinExec { + left, + right, + on, + filter, + join_type, + sort_options, + null_equality, + .. + } = origin_exec; let join_constructor = |new_conditions: ( Vec<(PhysicalExprRef, PhysicalExprRef)>, Vec, @@ -220,7 +220,8 @@ pub fn adjust_input_keys_ordering( *join_type, new_conditions.1, *null_equality, - ) + )? + .with_null_aware(origin_exec.null_aware()) .map(|e| Arc::new(e) as _) }; return reorder_partitioned_join_keys( @@ -482,17 +483,17 @@ pub fn reorder_join_keys_to_inputs( .build_exec(); } } - } else if let Some(SortMergeJoinExec { - left, - right, - on, - filter, - join_type, - sort_options, - null_equality, - .. - }) = plan.downcast_ref::() - { + } else if let Some(origin_exec) = plan.downcast_ref::() { + let SortMergeJoinExec { + left, + right, + on, + filter, + join_type, + sort_options, + null_equality, + .. + } = origin_exec; let (join_keys, positions) = reorder_current_join_keys( extract_join_keys(on), Some(left.output_partitioning()), @@ -519,7 +520,8 @@ pub fn reorder_join_keys_to_inputs( *join_type, new_sort_options, *null_equality, - ) + )? + .with_null_aware(origin_exec.null_aware()) .map(|smj| Arc::new(smj) as _); } } diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs index 96bb9fa291298..3388b9ced8463 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/bitwise_stream.rs @@ -123,6 +123,7 @@ use std::cmp::Ordering; use std::sync::Arc; use crate::EmptyRecordBatchStream; +use crate::joins::sort_merge_join::exec::NullAwareMode; use crate::joins::utils::{JoinFilter, JoinKeyComparator, compare_join_arrays}; use crate::metrics::{ BaselineMetrics, Count, ExecutionPlanMetricsSet, Gauge, MetricBuilder, Time, @@ -131,7 +132,7 @@ use crate::spill::in_progress_spill_file::InProgressSpillFile; use crate::spill::spill_manager::SpillManager; use crate::stream::{ObservedStream, RecordBatchStreamAdapter}; use arrow::array::{Array, ArrayRef, BooleanArray, BooleanBufferBuilder, RecordBatch}; -use arrow::compute::{BatchCoalescer, SortOptions, filter_record_batch, not}; +use arrow::compute::{BatchCoalescer, SortOptions, and, filter_record_batch, not}; use arrow::datatypes::SchemaRef; use arrow::util::bit_chunk_iterator::UnalignedBitChunk; use arrow::util::bit_util::apply_bitwise_binary_op; @@ -198,6 +199,48 @@ fn find_key_group_end(cmp: &JoinKeyComparator, from: usize, len: usize) -> usize lo } +/// Applies SQL NOT IN semantics to the unmatched outer rows. +fn null_aware_left_anti_selection( + unmatched: BooleanArray, + outer_key: &dyn Array, + subqury_state: SubqueryState, +) -> Result { + match subqury_state { + SubqueryState::Empty => Ok(unmatched), + SubqueryState::NonEmptyNoNull => match outer_key.logical_nulls() { + // All outer keys are non-NULL, so unmatched rows can be kept. + None => Ok(unmatched), + Some(nulls) => { + // With a nonempty RHS containing no NULL keys, NOT IN keeps only + // unmatched outer rows whose keys are non-NULL. Combine the + // unmatched mask with the outer key validity bitmap to exclude + // NULL keys, for which NOT IN evaluates to UNKNOWN. + let valids = BooleanArray::new(nulls.inner().clone(), None); + and(&unmatched, &valids).map_err(DataFusionError::from) + } + }, + SubqueryState::Pending | SubqueryState::HasNull => { + internal_err!("Cannot emit null-aware anti rows before RHS validation") + } + } +} + +/// State of the right-side subquery used by a null-aware `LeftAnti` join. +/// +/// Determined before emitting left rows, using a single right partition +/// sorted with NULLS FIRST. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SubqueryState { + /// The right input has not been inspected yet. + Pending, + /// The right input contains no rows. + Empty, + /// The right input contains rows but no NULL keys. + NonEmptyNoNull, + /// The right input contains at least one NULL key. + HasNull, +} + /// Sort-Merge join stream for Semi/Anti/Mark joins. /// /// Named "bitwise" because it tracks outer-row matches via a per-batch @@ -246,6 +289,8 @@ pub(crate) struct BitwiseSortMergeJoinStream { filter: Option, sort_options: Vec, null_equality: NullEquality, + null_aware: Option, + subquery_state: SubqueryState, // Decomposed from JoinType: when RightSemi/RightAnti, outer=right, // inner=left, so we swap sides when building the filter batch. outer_is_left: bool, @@ -290,6 +335,7 @@ impl BitwiseSortMergeJoinStream { schema: SchemaRef, sort_options: Vec, null_equality: NullEquality, + null_aware: Option, outer: SendableRecordBatchStream, inner: SendableRecordBatchStream, on_outer: Vec, @@ -328,6 +374,22 @@ impl BitwiseSortMergeJoinStream { let peak_mem_used = MetricBuilder::new(metrics).peak_memory_usage("peak_mem_used", partition); + if let Some(mode) = null_aware { + let validated = NullAwareMode::try_new( + join_type, + on_outer.len(), + filter.is_some(), + null_equality, + &sort_options, + )?; + + if mode != validated { + return internal_err!( + "null-aware mode does not match the join configuration" + ); + } + } + let mut state = Self { join_type, outer, @@ -345,6 +407,8 @@ impl BitwiseSortMergeJoinStream { filter, sort_options, null_equality, + null_aware, + subquery_state: SubqueryState::Pending, outer_is_left, coalescer: BatchCoalescer::new(Arc::clone(&schema), batch_size) .with_biggest_coalesce_batch_size(Some(batch_size / 2)), @@ -586,7 +650,15 @@ impl BitwiseSortMergeJoinStream { } } JoinType::LeftAnti | JoinType::RightAnti => { - let selection = not(&BooleanArray::new(matched_buf, None))?; + let unmatched = not(&BooleanArray::new(matched_buf, None))?; + let selection = match self.null_aware { + None => unmatched, + Some(NullAwareMode::LeftAnti) => null_aware_left_anti_selection( + unmatched, + self.outer_key_arrays[0].as_ref(), + self.subquery_state, + )?, + }; let filtered = filter_record_batch(&batch, &selection)?; if filtered.num_rows() > 0 { self.coalescer.push_batch(filtered)?; @@ -1103,12 +1175,56 @@ impl BitwiseSortMergeJoinStream { } } + /// Initializes the inner input's empty/NULL summary for null-aware joins. + async fn initialized_null_aware(&mut self) -> Result<()> { + if self.subquery_state != SubqueryState::Pending { + return Ok(()); + } + + match self.null_aware { + None => return Ok(()), + Some(NullAwareMode::LeftAnti) => {} + } + + if !self.next_inner_batch().await? { + self.subquery_state = SubqueryState::Empty; + return Ok(()); + } + + // Keep the first nonempty inner batch at offset 0 for the merge scan. + // The complete right input need to be sorted with NULLS FIRST. + let first_batch_has_null = self.inner_key_arrays[0] + .logical_nulls() + .is_some_and(|null_keys| null_keys.is_null(0)); + + self.subquery_state = if first_batch_has_null { + SubqueryState::HasNull + } else { + SubqueryState::NonEmptyNoNull + }; + + Ok(()) + } + /// Main loop: a classic merge-scan over the two sorted inputs, emitting /// output batches as they complete. async fn join( &mut self, emitter: &mut TryEmitter, ) -> Result<()> { + self.initialized_null_aware().await?; + match self.null_aware { + None => {} + Some(NullAwareMode::LeftAnti) + if self.subquery_state == SubqueryState::HasNull => + { + // NOT IN cannot be TRUE for any outer row when the inner + // input contains a NULL key. (following NOT IN semetic) + return Ok(()); + } + Some(NullAwareMode::LeftAnti) => {} + } + // The `has_current_*` / `has_completed_batch` fast paths keep async // state machinery out of the per-key-group hot path; the awaiting // helpers are only entered at batch boundaries. diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs index 64f1817184f8f..5782c046cb9b2 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/exec.rs @@ -64,6 +64,61 @@ use datafusion_physical_expr_common::physical_expr::{PhysicalExprRef, fmt_sql}; use datafusion_physical_expr_common::sort_expr::{LexOrdering, OrderingRequirements}; use futures::StreamExt; +/// Null-aware (`NOT IN`) semantics of a soft merge join, derived from +/// [`SortMergeJoinExec::null_aware`] and the join type. +/// +/// Only these 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(crate) enum NullAwareMode { + /// `left.key NOT IN (right.key)`: emits left rows. + LeftAnti, +} + +impl NullAwareMode { + /// Validates that `null_aware` may be set for this join and returns its mode. + pub(crate) fn try_new( + join_type: JoinType, + num_keys: usize, + has_filter: bool, + null_equality: NullEquality, + sort_options: &[SortOptions], + ) -> Result { + if num_keys != 1 || has_filter { + return plan_err!("null-aware SortMergeJoin requires one key and no filter"); + } + + // NULL keys must not match each other. + // The null-aware execution path handles UNKNOWN results separately. + // UNKNOWN happens when `WHERE x NOT IN (NULL)` + if null_equality != NullEquality::NullEqualsNothing { + return plan_err!("null-aware SortMergeJoin requires NullEqualsNothing"); + } + + let mode = match join_type { + JoinType::LeftAnti => Self::LeftAnti, + _ => { + return plan_err!( + "null-aware SortMergeJoin only supports LeftAnti, got {join_type}" + ); + } + }; + + match mode { + Self::LeftAnti => { + // With a single right-side partition and NULLS FIRST, + // the first row reveals whether the subquery contains NULL. + if sort_options.len() != 1 || !sort_options[0].nulls_first { + return plan_err!( + "null-aware {join_type} SortMergeJoin requires NULLS FIRST ordering" + ); + } + Ok(mode) + } + } + } +} + /// Join execution plan that executes equi-join predicates on multiple partitions using Sort-Merge /// join algorithm and applies an optional filter post join. Can be used to join arbitrarily large /// inputs where one or both of the inputs don't fit in the available memory. @@ -134,6 +189,19 @@ pub struct SortMergeJoinExec { pub sort_options: Vec, /// Defines the null equality for the join. pub null_equality: NullEquality, + /// Flag to indicate if this join uses null-aware equality semantics. + /// + /// Currently supported only for uncorrelated `LeftAnti` joins with one + /// join key, no join filter, and [`NullEquality::NullEqualsNothing`]. + /// + /// Both inputs must have a single partition and be sorted with NULLS FIRST. + /// This lets the join determine whether the entire right input is empty + /// or contains a NULL key before emitting any left rows. + /// + /// An empty right input preserves all left rows, including NULL keys. + /// A right input containing a NULL key produces no rows. + /// Otherwise, only unmatched left rows with non-NULL keys are emitted. + pub null_aware: bool, /// The columns of `schema` to emit, in order. `None` emits all of them. pub projection: Option>, /// Cache holding plan properties like equivalences, output partitioning etc. @@ -208,11 +276,23 @@ impl SortMergeJoinExec { right_sort_exprs, sort_options, null_equality, + null_aware: false, projection: None, cache: Arc::new(cache), }) } + /// Enables or disables null-aware (`NOT IN`) semantics. + /// + /// Enabling requires an uncorrelated `LeftAnti` join with one key, no + /// filter, [`NullEquality::NullEqualsNothing`], and NULLS FIRST ordering. + /// Both inputs must have a single partition when executed. + pub fn with_null_aware(mut self, null_aware: bool) -> Result { + self.null_aware = null_aware; + self.null_aware_mode()?; + Ok(self) + } + /// Returns this join emitting only the columns in `projection`, in that order. /// The indices address the join's own schema, before any projection. pub fn with_projection(&self, projection: Option>) -> Result { @@ -306,6 +386,11 @@ impl SortMergeJoinExec { self.null_equality } + /// Null aware + pub fn null_aware(&self) -> bool { + self.null_aware + } + /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc. fn compute_properties( left: &Arc, @@ -377,6 +462,7 @@ impl SortMergeJoinExec { self.sort_options.clone(), self.null_equality, )? + .with_null_aware(self.null_aware)? .with_projection(swap_join_projection( left.schema().fields().len(), right.schema().fields().len(), @@ -401,6 +487,23 @@ impl SortMergeJoinExec { reorder_output_after_swap(Arc::new(new_join), &left.schema(), &right.schema()) } } + + /// 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.on.len(), + self.filter.is_some(), + self.null_equality(), + self.sort_options(), + ) + }) + .transpose() + } } impl EmbeddedProjection for SortMergeJoinExec { @@ -493,6 +596,18 @@ impl ExecutionPlan for SortMergeJoinExec { } fn input_distribution_requirements(&self) -> InputDistributionRequirements { + // Null-aware NOT IN needs facts about the entire right input. + // Require a single partition on both sides to use the existing + // sort-merge join's partition pairing without cross-partition coordination. + // + // TODO: Allow multiple left partitions by sharing the right-side summary + // and providing each left partition with the right input for matching. + if self.null_aware { + return InputDistributionRequirements::new(vec![ + Distribution::SinglePartition, + Distribution::SinglePartition, + ]); + } let (left_expr, right_expr) = self .on .iter() @@ -556,6 +671,7 @@ impl ExecutionPlan for SortMergeJoinExec { self.sort_options.clone(), self.null_equality, )? + .with_null_aware(self.null_aware)? .with_projection(self.projection.clone())?, )), _ => internal_err!("SortMergeJoin wrong number of children"), @@ -596,6 +712,20 @@ impl ExecutionPlan for SortMergeJoinExec { "Invalid SortMergeJoinExec, partition count mismatch {left_partitions}!={right_partitions},\ consider using RepartitionExec" ); + + let null_aware = self.null_aware_mode()?; + + // Null-aware execution requires a single partition on both sides. + // Validate this at execution time because callers may bypass the + // optimizer or modify the public null_aware flag after planning. + if null_aware.is_some() + && (partition != 0 || left_partitions != 1 || right_partitions != 1) + { + return internal_err!( + "null-aware SortMergeJoin requires partition 0 and single-partition inputs" + ); + } + // execute children plans let left = self.left.execute(partition, Arc::clone(&context))?; let right = self.right.execute(partition, Arc::clone(&context))?; @@ -606,6 +736,7 @@ impl ExecutionPlan for SortMergeJoinExec { schema: Arc::clone(&self.schema), sort_options: self.sort_options.clone(), null_equality: self.null_equality, + null_aware, left, right, on_left, @@ -699,15 +830,18 @@ impl ExecutionPlan for SortMergeJoinExec { self.filter().as_ref(), &column_indices, )? { - Ok(Some(Arc::new(SortMergeJoinExec::try_new( - Arc::new(projected_left_child), - Arc::new(projected_right_child), - join_on, - join_filter, - self.join_type, - self.sort_options.clone(), - self.null_equality, - )?))) + Ok(Some(Arc::new( + SortMergeJoinExec::try_new( + Arc::new(projected_left_child), + Arc::new(projected_right_child), + join_on, + join_filter, + self.join_type, + self.sort_options.clone(), + self.null_equality, + )? + .with_null_aware(self.null_aware)?, + ))) } else { try_embed_projection(projection, self) } @@ -720,6 +854,14 @@ impl ExecutionPlan for SortMergeJoinExec { ) -> Result> { use datafusion_proto_models::protobuf; + // The existing protobuf node cannot represent null-aware semantics. + // Reject encoding rather than silently decoding as an ordinary anti join. + if self.null_aware { + return datafusion_common::not_impl_err!( + "Serializing null-aware SortMergeJoinExec is not supported" + ); + } + // Destructure exhaustively (no `..`) so that a newly added field is a // compile error here instead of being silently left out of the proto. let Self { @@ -730,6 +872,7 @@ impl ExecutionPlan for SortMergeJoinExec { join_type, sort_options, null_equality, + null_aware: _, projection, // derived from the children's schemas by `try_new` on decode schema: _, @@ -864,6 +1007,7 @@ impl SortMergeJoinExec { *null_equality, "SortMergeJoinExec", )?; + let filter = filter .as_ref() .map(|filter| { @@ -912,6 +1056,7 @@ pub(crate) struct SortMergeJoinInputs { /// Sort options of the join keys, one per key, that both inputs are sorted with pub(crate) sort_options: Vec, pub(crate) null_equality: NullEquality, + pub(crate) null_aware: Option, /// Left input, sorted on `on_left` with `sort_options` pub(crate) left: SendableRecordBatchStream, /// Right input, sorted on `on_right` with `sort_options` @@ -940,6 +1085,7 @@ pub(crate) fn sort_merge_join_stream( schema, sort_options, null_equality, + null_aware, left, right, on_left, @@ -983,6 +1129,7 @@ pub(crate) fn sort_merge_join_stream( schema, sort_options, null_equality, + null_aware, streamed, buffered, on_streamed, diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs index f921812ec0619..a1f7f6cc245dc 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/tests.rs @@ -5431,6 +5431,7 @@ async fn bitwise_emits_completed_batches_before_pending_outer( Arc::clone(&schema), vec![SortOptions::default()], NullEquality::NullEqualsNothing, + None, outer, inner, vec![Arc::new(Column::new("key", 1))], @@ -5530,6 +5531,7 @@ fn join_time_test_join( outer_schema, vec![SortOptions::default()], NullEquality::NullEqualsNothing, + None, outer, inner, vec![Arc::new(Column::new("b1", 1)) as PhysicalExprRef], @@ -5915,6 +5917,7 @@ async fn filter_buffer_pending_loses_inner_rows() -> Result<()> { left_schema, // output schema = outer schema for semi vec![SortOptions::default()], NullEquality::NullEqualsNothing, + None, outer, inner, on_outer, @@ -6009,6 +6012,7 @@ async fn no_filter_boundary_pending_loses_outer_rows() -> Result<()> { left_schema, vec![SortOptions::default()], NullEquality::NullEqualsNothing, + None, outer, inner, on_outer, @@ -6100,6 +6104,7 @@ async fn no_filter_boundary_pending_with_unmatched_prefix() -> Result<()> { Arc::clone(&left_schema), vec![SortOptions::default()], NullEquality::NullEqualsNothing, + None, outer, inner, on_outer.clone(), @@ -6220,6 +6225,7 @@ async fn filtered_boundary_pending_outer_rows() -> Result<()> { left_schema, vec![SortOptions::default()], NullEquality::NullEqualsNothing, + None, outer, inner, on_outer, @@ -6965,6 +6971,7 @@ async fn spill_filtered_boundary_loses_outer_rows() -> Result<()> { Arc::clone(&left_schema), vec![SortOptions::default()], NullEquality::NullEqualsNothing, + None, outer, inner, on_outer.clone(), diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 09f648df27702..8428a923615d7 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -2819,8 +2819,8 @@ DROP TABLE sq_count_orders; # NULL the predicate is never TRUE, so the query must return zero rows. This must # hold regardless of the chosen physical join operator. Previously, with # prefer_hash_join = false and multiple partitions, the planner routed the -# null-aware anti join to SortMergeJoin (which is not null-aware) and returned -# wrong results; null-aware anti joins must use the CollectLeft HashJoin. +# null-aware anti join to an ordinary SortMergeJoin and returned wrong results. +# The null-aware SortMergeJoin path must preserve the same NOT IN semantics. statement ok set datafusion.optimizer.prefer_hash_join = false; @@ -2839,9 +2839,8 @@ query I SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null) ORDER BY x; ---- -# The null-aware anti join must be planned as a CollectLeft HashJoinExec even with -# prefer_hash_join = false: SortMergeJoinExec is not null-aware and must not be used. -# This test has no usable size statistics, so JoinSelection keeps LeftAnti. +# With prefer_hash_join = false, this single-key, unfiltered null-aware LeftAnti +# join uses SortMergeJoinExec with sorted, single-partition inputs. query TT EXPLAIN SELECT x FROM nia_left WHERE x NOT IN (SELECT y FROM nia_right_with_null); ---- @@ -2851,9 +2850,11 @@ logical_plan 03)--SubqueryAlias: __correlated_sq_1 04)----TableScan: nia_right_with_null projection=[y] physical_plan -01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(x@0, y@0)], null_aware -02)--DataSourceExec: partitions=1, partition_sizes=[1] -03)--DataSourceExec: partitions=1, partition_sizes=[1] +01)SortMergeJoinExec: join_type=LeftAnti, on=[(x@0, y@0)] +02)--SortExec: expr=[x@0 ASC], preserve_partitioning=[false] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)--SortExec: expr=[y@0 ASC], preserve_partitioning=[false] +05)----DataSourceExec: partitions=1, partition_sizes=[1] # Subquery has no NULL -> NOT IN behaves like a normal anti join. query I