fix: allow NULL-padded rows in sort-merge join filter evaluation (#26027)

## Which issue does this PR close?

- Closes #26001.

## Rationale for this change

With sort-merge join (`datafusion.optimizer.prefer_hash_join = false`),
a `LEFT`, `RIGHT` or `FULL` join with an extra join filter fails with
`Arrow error: Invalid argument error: Column '...' is declared as
non-nullable but contains null values` when the NULL-padded side has
`NOT NULL` columns referenced by the filter. Hash join returns the
correct result for the same queries.

## What changes are included in this PR?

**Root cause.** As @jayzhan211 pointed out in the issue, when a streamed
row has no match but the buffered side still holds a batch,
`null_join_streamed_row` appends it with `Some(scanning_batch_idx)`, so
it is materialized in the same chunk as matched rows. Its NULL-padded
buffered-side columns then reach join filter evaluation, but the
filter's intermediate schema declares them non-nullable, so
`RecordBatch::try_new` fails.

**Fix.** Following the fix suggested in the issue,
`MaterializingSortMergeJoinStream::try_new` builds the filter schema
once with buffered-side fields marked nullable (`JoinSide::Right` for
`Left`/`Full`, `JoinSide::Left` for `Right`), and filter evaluation uses
it. It is only built when `deferred_filtering` is set (outer join with a
filter); inner joins keep the original schema.

**Why this is safe.** The filter result for a NULL-padded row does not
affect the output. Such a row is the only entry for its streamed row, so
`get_corrected_filter_mask` keeps it as-is if the filter returns true
and null-joins it if false; either way the same NULL-padded row is
emitted. The `Full` join filter-status tracking already skips NULL
buffered indices, so this path already expects such rows.

**Alternative considered.** Keep NULL-padded rows out of the filter by
appending them with `buffered_batch_idx = None`. Done naively, this
reorders output: `freeze_streamed` pushes `None` chunks immediately but
pushes matched chunks together at the end, so sort order would be
silently broken. Flushing matched chunks before each `None` chunk would
preserve order, but splits the single batched filter evaluation in
`freeze_streamed_matched`, which matters most when keys are near-unique.
Making the schema nullable avoids both.

## What is the testing strategy for this PR?

Added sqllogictest cases to `sort_merge_join.slt` with `NOT NULL`
columns and `target_partitions = 2` (needed to reproduce; the queries
pass with 1 partition):

- `LEFT`, `FULL` and `RIGHT` joins from the issue, with expected results
matching hash join.
- A `LEFT` join whose filter divides by a buffered-side column (`a.v /
b.w < 1`), to check that the zero-valued slots under NULL-padded rows do
not cause a divide-by-zero error.

All four queries fail on `main` with the Arrow error above and pass with
this change. The full extended test suite also passes locally.

## Are there any user-facing changes?

No. Bug fix only; no public API change.

Author: CuteChuanChuan

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;