fix: emit unmatched NULL-key right rows in RIGHT/FULL PiecewiseMergeJoin (#24336)

## Which issue does this PR close?

- Closes #24335.

## Rationale for this change

A `RIGHT`/`FULL` `PiecewiseMergeJoin` with a range predicate drops an
unmatched right-side row whose join key is `NULL`. A `NULL` key never
matches (`NULL < x` is UNKNOWN), so in a `RIGHT`/`FULL` join the row is
unmatched and must still be emitted with NULLs on the left — but
`PiecewiseMergeJoinExec` omits it, diverging from `NestedLoopJoin`.

```sql
create table l(v int) as values (5);
create table r(v int) as values (10), (NULL);
select l.v, r.v from l right join r on l.v < r.v;   -- drops (NULL, NULL)
```

Root cause: `resolve_classic_join` starts the match scan past the
streamed side's `NULL`-keyed rows (they sort to the front under
`nulls_first`). Those rows are never revisited, so for `Right`/`Full`
they were never added to `unmatched_indices` and got dropped.

## What changes are included in this PR?

- In `resolve_classic_join`, when skipping the streamed side's leading
`NULL`-key rows, record them as unmatched for `Right`/`Full` joins so
they are emitted (with NULLs on the buffered side).

## Are these changes tested?

Yes.

- Regression test in `pwmj.slt`: a `RIGHT JOIN` over the existing
`null_join_*` tables now emits the `(NULL, NULL)` row. The test fails on
`main` (the row is dropped) and passes with this change.
- Verified more broadly with a differential fuzz against
`NestedLoopJoin` (same SQL, `enable_piecewise_merge_join` on vs off):
1200 checks over random `RIGHT JOIN` inputs with `<`/`<=`/`>`/`>=` and
high right-side NULL density, 0 mismatches.

## Are there any user-facing changes?

`RIGHT`/`FULL` range joins via `PiecewiseMergeJoin` (behind
`enable_piecewise_merge_join`, default off) now return unmatched right
rows with `NULL` keys, matching `NestedLoopJoin`. No API changes.

Author: viirya

datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs

@@ -481,6 +481,16 @@ fn resolve_classic_join(
         buffer_idx = buffered_null_idx;
         stream_idx = stream_null_idx;
         batch_process_state.processed_null_count = true;
+
+        // The scan below starts past the streamed side's NULL-keyed rows, which
+        // sit at the front (`nulls_first`). A NULL join key never matches under
+        // `NullEqualsNothing`, so for `Right`/`Full` those rows are unmatched and
+        // must still be emitted; record them here since the scan will skip them.
+        if matches!(join_type, JoinType::Right | JoinType::Full) {
+            for row_idx in 0..stream_null_idx as u32 {
+                batch_process_state.unmatched_indices.append_value(row_idx);
+            }
+        }
     }
 
     // Our buffer_idx variable allows us to start probing on the buffered side where we last matched

datafusion/sqllogictest/test_files/pwmj.slt

@@ -342,5 +342,50 @@ ORDER BY 1,2;
 1 3
 2 3
 
+# RIGHT JOIN: every right row must be emitted, including one whose join key is
+# NULL. A NULL key never matches (NullEqualsNothing), so t2.id = NULL is
+# unmatched and must appear as (NULL, NULL). Regression test: the streamed-side
+# NULL rows are skipped by the match scan and were previously dropped instead of
+# being reported as unmatched.
+query II
+SELECT t1.id AS left_id, t2.id AS right_id
+FROM null_join_t1 t1
+RIGHT JOIN null_join_t2 t2
+  ON t1.id < t2.id
+ORDER BY 1,2;
+----
+1 3
+2 3
+NULL 1
+NULL NULL
+
+# FULL JOIN: the same fix applies to the FULL path. An unmatched right row with
+# a NULL key must still be emitted as (NULL, NULL). This uses data where every
+# left row matches, so only the right-side NULL emission fixed here is
+# exercised (the separate left-side unmatched-row drop, tracked elsewhere, is
+# not).
+statement ok
+CREATE TABLE full_null_t1 (id INT);
+
+statement ok
+CREATE TABLE full_null_t2 (id INT);
+
+statement ok
+INSERT INTO full_null_t1 VALUES (10), (20);
+
+statement ok
+INSERT INTO full_null_t2 VALUES (100), (NULL);
+
+query II
+SELECT t1.id AS left_id, t2.id AS right_id
+FROM full_null_t1 t1
+FULL JOIN full_null_t2 t2
+  ON t1.id < t2.id
+ORDER BY 1,2;
+----
+10 100
+20 100
+NULL NULL
+
 statement ok
 set datafusion.optimizer.enable_piecewise_merge_join = false;