fix: allow NULL-padded rows in sort-merge join filter evaluation

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<SchemaRef>,
     /// 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)?

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;