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

…che#26027)

## Which issue does this PR close?

- Closes apache#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.

…luation (#26027) (#27354)

## Which issue does this PR close?

- Part of #26868 (the 55.3.0 patch release tracking issue).
- Backport of #26027 to `branch-55`. #26027 fixed #26001.

## Rationale for this change

With a 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. This is a failed query, which
makes it eligible for backport as a correctness fix.

When a streamed row has no match but the buffered side still holds a
batch, the NULL-padded row is materialized in the same chunk as matched
rows, so its NULL buffered-side columns reach filter evaluation, whose
intermediate schema declares them non-nullable. #26027 fixed it on
`main` (labelled `v56.0.0`); it is on `branch-56` but not on
`branch-55`, so it is in neither 55.1.0 nor 55.2.0. It was requested for
55.3.0 on #26868
(#26868 (comment)).

## What changes are included in this PR?

- A `git cherry-pick -x` of 8248a57
(#26027, by @CuteChuanChuan).
-
`datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs`:
`MaterializingSortMergeJoinStream::try_new` builds the filter schema
once with the buffered-side fields marked nullable, and filter
evaluation uses it. It is only built when the join needs deferred
filtering (outer join with a filter); inner joins keep the original
schema. The filter result of a NULL-padded row does not affect the
output, so this does not change results.
- `datafusion/sqllogictest/test_files/sort_merge_join.slt`: the `LEFT`,
`FULL` and `RIGHT` cases from the issue, and a `LEFT` case whose filter
divides by a buffered-side column.
- One conflict, in `materializing_stream.rs`: on this branch
`needs_deferred_filtering` takes `&Option<JoinFilter>` (on `main` it
takes `Option<&JoinFilter>`), so the new local passes `&filter`. The
rest is as upstream.

## Are these changes tested?

Yes. On this branch (Rust 1.95.0) I ran:

- `cargo test -p datafusion-sqllogictest --test sqllogictests -- join`:
all 16 matching files pass, including `sort_merge_join.slt` with the new
cases.
- With the change in `materializing_stream.rs` reverted and the new slt
cases kept, the four new queries fail with the Arrow error above
(`sort_merge_join.slt` lines 990, 997, 1004 and 1013).
- `cargo test -p datafusion-physical-plan --lib`: 1700 passed.
- `cargo fmt --all -- --check` and `cargo clippy -p
datafusion-physical-plan --all-targets --all-features -- -D warnings`
pass.

## Are there any user-facing changes?

No. Bug fix only; no public API change.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Yu-Chuan Hung <[email protected]>
Co-authored-by: Claude Sonnet 5.5 <[email protected]>

…luation (#26027) (#27356)

## Which issue does this PR close?

- Part of #23805 (the 54.2.0 patch release tracking issue).
- Backport of #26027 to `branch-54`. #26027 fixed #26001.

## Rationale for this change

With a 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. This is a failed query, which
makes it eligible for backport as a correctness fix. The four new
sqllogictest queries fail on `branch-54` exactly as on `main`.

#26027 fixed it on `main` (labelled `v56.0.0`); it is on `branch-56` but
not on `branch-54`. It was requested for `branch-54` on #26868
(#26868 (comment)),
"to prevent regressions on upgrade to 54".

## What changes are included in this PR?

- A `git cherry-pick -x` of 8248a57
(#26027, by @CuteChuanChuan), with the conflicts in
`materializing_stream.rs` resolved by hand. The sort-merge join on this
branch predates the generator refactor (#23761, #23976), so the
surrounding code differs.
-
`datafusion/physical-plan/src/joins/sort_merge_join/materializing_stream.rs`:
`MaterializingSortMergeJoinStream::try_new` builds the filter schema
once with the buffered-side fields marked nullable, and filter
evaluation uses it. It is only built when the join needs deferred
filtering. This branch has no cached `deferred_filtering` field and
calls `needs_deferred_filtering(&self.filter, self.join_type)` where it
is needed, so the filter schema is built from that call; there is no
other change in the constructor. The filter result of a NULL-padded row
does not affect the output, so this does not change results.
- `datafusion/sqllogictest/test_files/sort_merge_join.slt`: the cases
from #26027, unchanged (they applied cleanly).

## Are these changes tested?

Yes. On this branch (Rust 1.95.0) I ran:

- `cargo test -p datafusion-sqllogictest --test sqllogictests -- join`:
all 14 matching files pass, including `sort_merge_join.slt` with the new
cases.
- With the change in `materializing_stream.rs` reverted and the new slt
cases kept, the four new queries fail with the Arrow error above
(`sort_merge_join.slt` lines 990, 997, 1004 and 1013).
- `cargo test -p datafusion-physical-plan --lib`: 1439 passed.
- `cargo fmt --all -- --check` and `cargo clippy -p
datafusion-physical-plan --all-targets --all-features -- -D warnings`
pass.

## Are there any user-facing changes?

No. Bug fix only; no public API change.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Yu-Chuan Hung <[email protected]>
Co-authored-by: Claude Sonnet 5.5 <[email protected]>