fix: emit unmatched NULL-key right rows in RIGHT/FULL PiecewiseMergeJoin

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;