diff --git a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs index 19f9e49973bf8..6e5193f535a5a 100644 --- a/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs +++ b/datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs @@ -45,11 +45,11 @@ use arrow::compute::{ self, BatchCoalescer, SortOptions, concat_batches, filter_record_batch, interleave, take_arrays, }; -use arrow::datatypes::SchemaRef; +use arrow::datatypes::{Schema, SchemaRef}; use datafusion_common::cast::as_uint64_array; use datafusion_common::instant::Instant; use datafusion_common::{ - DataFusionError, JoinType, NullEquality, Result, exec_err, internal_err, + DataFusionError, JoinSide, JoinType, NullEquality, Result, exec_err, internal_err, }; use datafusion_execution::memory_pool::MemoryReservation; use datafusion_execution::runtime_env::RuntimeEnv; @@ -295,6 +295,10 @@ pub(super) struct MaterializingSortMergeJoinStream { /// Cached `needs_deferred_filtering(filter, join_type)` — both inputs /// are fixed at construction time. pub deferred_filtering: bool, + /// The join filter's schema with buffered-side fields marked nullable, + /// since NULL-padded rows reach filter evaluation. `None` unless + /// `deferred_filtering`. + pub filter_schema: Option, /// Target output batch size pub batch_size: usize, @@ -551,6 +555,30 @@ impl MaterializingSortMergeJoinStream { semi/anti/mark joins use BitwiseSortMergeJoinStream" ); let join_time = join_metrics.join_time(); + let buffered_side = if join_type == JoinType::Right { + JoinSide::Left + } else { + JoinSide::Right + }; + let deferred_filtering = needs_deferred_filtering(filter.as_ref(), join_type); + let filter_schema = if deferred_filtering && let Some(f) = &filter { + let mut fields = Vec::with_capacity(f.schema().fields().len()); + for (column_index, field) in + f.column_indices().iter().zip(f.schema().fields().iter()) + { + if column_index.side == buffered_side { + fields.push(Arc::new(field.as_ref().clone().with_nullable(true))); + } else { + fields.push(Arc::clone(field)); + } + } + Some(Arc::new(Schema::new_with_metadata( + fields, + f.schema().metadata().clone(), + ))) + } else { + None + }; let mut this = Self { sort_options, null_equality, @@ -566,7 +594,8 @@ impl MaterializingSortMergeJoinStream { buffered_exhausted: false, on_streamed, on_buffered, - deferred_filtering: needs_deferred_filtering(filter.as_ref(), join_type), + deferred_filtering, + filter_schema, filter, joined_record_batches: JoinedRecordBatches { joined_batches: new_output_coalescer(Arc::clone(&schema), batch_size), @@ -1601,8 +1630,9 @@ impl MaterializingSortMergeJoinStream { if !filter_columns.is_empty() { if let Some(f) = &self.filter { + let schema = self.filter_schema.as_ref().unwrap_or_else(|| f.schema()); let filter_batch = - RecordBatch::try_new(Arc::clone(f.schema()), filter_columns)?; + RecordBatch::try_new(Arc::clone(schema), filter_columns)?; let filter_result = f .expression() .evaluate(&filter_batch)? diff --git a/datafusion/sqllogictest/test_files/sort_merge_join.slt b/datafusion/sqllogictest/test_files/sort_merge_join.slt index 5ae8117290e40..5c563c8e817c9 100644 --- a/datafusion/sqllogictest/test_files/sort_merge_join.slt +++ b/datafusion/sqllogictest/test_files/sort_merge_join.slt @@ -971,6 +971,64 @@ WHERE t1_sorted.data < 0 ---- 100 +# Regression test for #26001: outer SMJ with a join filter must not fail +# when the NULL-padded side has NOT NULL columns. target_partitions = 2 +# is needed to reproduce; with 1 partition the bug does not trigger. + +statement ok +set datafusion.execution.target_partitions = 2; + +statement ok +CREATE TABLE a (k INT NOT NULL, v INT NOT NULL) AS VALUES (1, 10), (2, 20), (3, 30); + +statement ok +CREATE TABLE b (k INT NOT NULL, w INT NOT NULL) AS VALUES (1, 100), (3, 300); + +statement ok +CREATE TABLE c (k INT NOT NULL, w INT NOT NULL) AS VALUES (1, 100), (2, 200), (3, 300); + +query IIII +SELECT * FROM a LEFT JOIN b ON a.k = b.k AND a.v < b.w ORDER BY a.k +---- +1 10 1 100 +2 20 NULL NULL +3 30 3 300 + +query IIII +SELECT * FROM a FULL JOIN b ON a.k = b.k AND a.v < b.w ORDER BY a.k +---- +1 10 1 100 +2 20 NULL NULL +3 30 3 300 + +query IIII +SELECT * FROM b RIGHT JOIN c ON b.k = c.k AND b.w <= c.w ORDER BY c.k +---- +1 100 1 100 +NULL NULL 2 200 +3 300 3 300 + +# NULL-padded rows reach the filter; their zero-valued slots must not +# cause divide-by-zero errors +query IIII +SELECT * FROM a LEFT JOIN b ON a.k = b.k AND a.v / b.w < 1 ORDER BY a.k +---- +1 10 1 100 +2 20 NULL NULL +3 30 3 300 + +statement ok +DROP TABLE a; + +statement ok +DROP TABLE b; + +statement ok +DROP TABLE c; + +statement ok +set datafusion.execution.target_partitions = 4; + # Config reset statement ok RESET datafusion.optimizer.prefer_hash_join;