[branch-54] fix: allow NULL-padded rows in sort-merge join filter evaluation (#26027) - #27356

…che#26027)

- Closes apache#26001.

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.

**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.

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.

No. Bug fix only; no public API change.

(cherry picked from commit 8248a57)

Backport notes (branch-54): the sort-merge join on this branch predates the
generator refactor, so the cherry-pick conflicted in `materializing_stream.rs`.
Resolved by hand: there is no cached `deferred_filtering` field here, so the
filter schema is built from `needs_deferred_filtering(&filter, join_type)`;
the filter evaluation change and the sqllogictest cases are as upstream.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>