diff --git a/datafusion/sqllogictest/test_files/mark_join_matrix.slt b/datafusion/sqllogictest/test_files/mark_join_matrix.slt
new file mode 100644
index 0000000000000..c039a3a20f142
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/mark_join_matrix.slt
@@ -0,0 +1,206 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+
+#   http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Mark-join correctness across a config matrix. A mark join adds a boolean `mark`
+# per left row and comes from EXISTS/IN/NOT EXISTS/NOT IN inside a disjunction
+# (`WHERE <pred> OR EXISTS(..)`; see mark_join in decorrelate_predicate_subquery.rs).
+# For an equijoin mark, prefer_hash_join toggles the operator: false ->
+# SortMergeJoinExec LeftMark, true -> HashJoinExec RightMark (inputs swapped), so
+# sweeping {true,false} x batch_size {1,2,100,8192} cross-checks both directions
+# and must agree row-for-row. Range mark joins run on NestedLoopJoin regardless
+# (Part 2, batch_size only). SMJ needs target_partitions>1 and repartition_joins,
+# set below (not swept). Matrix rules: no EXPLAIN, no in-file SET of a swept knob,
+# rowsort every multi-row query.
+
+# configMatrix: datafusion.optimizer.prefer_hash_join=true,false
+# configMatrix: datafusion.execution.batch_size=1,2,100,8192
+
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+set datafusion.optimizer.repartition_joins = true;
+
+# ------------------------------------------------------------------
+# Fixtures: duplicate and NULL keys on the subquery side, a NULL key and an
+# unmatched key on the outer side.
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE mk_l(k INT, v INT) AS VALUES
+  (1, 10), (2, 20), (3, 30), (4, 40), (NULL, 50);
+
+statement ok
+CREATE TABLE mk_r(k INT) AS VALUES (1), (2), (2), (NULL);
+
+# mk_r without the NULL, for NOT IN cases that would otherwise be swallowed by
+# three-valued logic.
+statement ok
+CREATE TABLE mk_r_nn(k INT) AS VALUES (1), (2), (2);
+
+statement ok
+CREATE TABLE mk_empty(k INT);
+
+# ==================================================================
+# Part 1: Equijoin mark joins (LeftMark on SMJ vs RightMark on HashJoin)
+# ==================================================================
+
+# EXISTS in a disjunction with a sometimes-true predicate. mark(EXISTS k in
+# {1,2}) is true for k=1,2; the predicate v>35 is true for k=4 and k=NULL.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.v > 35 OR EXISTS (SELECT 1 FROM mk_r r WHERE l.k = r.k);
+----
+1 10
+2 20
+4 40
+NULL 50
+
+# NOT EXISTS in a disjunction: mark is negated, true for k not in {1,2}.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.v > 35 OR NOT EXISTS (SELECT 1 FROM mk_r r WHERE l.k = r.k);
+----
+3 30
+4 40
+NULL 50
+
+# Predicate never true (no negative k), so the result isolates the mark: the
+# EXISTS rows k in {1,2}. Exercises the mark column with the OR contributing
+# nothing.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.k < 0 OR EXISTS (SELECT 1 FROM mk_r r WHERE l.k = r.k);
+----
+1 10
+2 20
+
+# Same, negated: isolates NOT EXISTS. The NULL-keyed left row never matches, so
+# it is kept.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.k < 0 OR NOT EXISTS (SELECT 1 FROM mk_r r WHERE l.k = r.k);
+----
+3 30
+4 40
+NULL 50
+
+# IN in a disjunction: same matches as EXISTS here; the NULL in the subquery adds
+# no true values.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.v > 35 OR l.k IN (SELECT r.k FROM mk_r r);
+----
+1 10
+2 20
+4 40
+NULL 50
+
+# NOT IN over a NULL-free subquery: mark(NOT IN) is true for k not in {1,2}.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.v > 35 OR l.k NOT IN (SELECT r.k FROM mk_r_nn r);
+----
+3 30
+4 40
+NULL 50
+
+# NOT IN over a subquery with NULL. The mark join negates `l.k = r.k` and is NOT
+# null-aware, so the subquery NULL is just a non-match: k=3 and the NULL-keyed row
+# are kept (a top-level null-aware NOT IN would return nothing). SMJ and HashJoin
+# must agree.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.v > 35 OR l.k NOT IN (SELECT r.k FROM mk_r r);
+----
+3 30
+4 40
+NULL 50
+
+# Empty subquery: EXISTS is always false, so the result is just the predicate.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.v > 35 OR EXISTS (SELECT 1 FROM mk_empty r WHERE l.k = r.k);
+----
+4 40
+NULL 50
+
+# Empty subquery, negated: NOT EXISTS is always true, so every row survives.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.k < 0 OR NOT EXISTS (SELECT 1 FROM mk_empty r WHERE l.k = r.k);
+----
+1 10
+2 20
+3 30
+4 40
+NULL 50
+
+# Multiple equi keys in the mark correlation.
+statement ok
+CREATE TABLE mk2_l(k1 INT, k2 INT, v INT) AS VALUES
+  (1, 1, 10), (1, 2, 20), (2, 2, 30), (3, 3, 40);
+
+statement ok
+CREATE TABLE mk2_r(k1 INT, k2 INT) AS VALUES (1, 1), (2, 2), (1, 9);
+
+query III rowsort
+SELECT l.k1, l.k2, l.v FROM mk2_l l
+WHERE l.v > 35 OR EXISTS (SELECT 1 FROM mk2_r r WHERE l.k1 = r.k1 AND l.k2 = r.k2);
+----
+1 1 10
+2 2 30
+3 3 40
+
+# Cross-table filter in the correlation (regression #21197): unmatched mark rows
+# produce null right indices that must not corrupt non-nullable left columns.
+# EXISTS holds for the k=2 rows (partner (2,99) differs in d); k=1's partner
+# equals its d, so false.
+statement ok
+CREATE TABLE mkf_l(k INT, d INT) AS VALUES (1, 10), (2, 20), (2, 25), (3, 30);
+
+statement ok
+CREATE TABLE mkf_r(k INT, d INT) AS VALUES (1, 10), (2, 20), (2, 99);
+
+query II rowsort
+SELECT l.k, l.d FROM mkf_l l
+WHERE l.d < 0 OR EXISTS (SELECT 1 FROM mkf_r r WHERE l.k = r.k AND r.d <> l.d);
+----
+2 20
+2 25
+
+# ==================================================================
+# Part 2: Range mark joins (RightMark on NestedLoopJoin)
+# ==================================================================
+# No equi key, so these run on NestedLoopJoin in both combinations; only
+# batch_size varies. mk_r non-null keys = {1,2,2}.
+
+# EXISTS l.k > r.k is true for k>1, i.e. k in {2,3,4}. Predicate never true.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.v > 100 OR EXISTS (SELECT 1 FROM mk_r r WHERE l.k > r.k);
+----
+2 20
+3 30
+4 40
+
+# NOT EXISTS of the same range: true for k=1 (no smaller r) and the NULL key.
+query II rowsort
+SELECT l.k, l.v FROM mk_l l
+WHERE l.k < 0 OR NOT EXISTS (SELECT 1 FROM mk_r r WHERE l.k > r.k);
+----
+1 10
+NULL 50
diff --git a/datafusion/sqllogictest/test_files/piecewise_merge_join_batches.slt b/datafusion/sqllogictest/test_files/piecewise_merge_join_batches.slt
new file mode 100644
index 0000000000000..db27c9fc8bb60
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/piecewise_merge_join_batches.slt
@@ -0,0 +1,103 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+
+#   http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Plan proof for piecewise_merge_join_matrix.slt. Not a matrix file: it fixes
+# batch_size=2 and enable_piecewise_merge_join=true so it can EXPLAIN what the
+# matrix relies on but cannot check itself -- the streamed side is a range
+# LazyMemoryExec with batch_size=2 (so range(3,8), 5 rows, arrives as 3 batches)
+# feeding a PiecewiseMergeJoin, not a fallback. A VALUES source would not show
+# batch_size in the plan.
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = true;
+
+statement ok
+set datafusion.execution.batch_size = 2;
+
+statement ok
+CREATE TABLE pb_l(id INT, v INT) AS
+  SELECT CAST(value AS INT) AS id, CAST(value AS INT) AS v FROM range(1, 11);
+
+# Existence (LeftSemi): range streamed side (batch_size=2) -> LeftSemi PWMJ.
+query TT
+EXPLAIN SELECT count(*) FROM pb_l l WHERE EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v > r.value);
+----
+logical_plan
+01)Projection: count(Int64(1)) AS count(*)
+02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
+03)----Projection:
+04)------LeftSemi Join:  Filter: CAST(l.v AS Int64) > __correlated_sq_1.value
+05)--------SubqueryAlias: l
+06)----------TableScan: pb_l projection=[v]
+07)--------SubqueryAlias: __correlated_sq_1
+08)----------SubqueryAlias: r
+09)------------TableScan: range() projection=[value]
+physical_plan
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
+02)--AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
+03)----CoalescePartitionsExec
+04)------AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
+05)--------ProjectionExec: expr=[]
+06)----------PiecewiseMergeJoin: operator=Gt, join_type=LeftSemi, on=(CAST(v AS Int64) > value)
+07)------------SortPreservingMergeExec: [CAST(v@0 AS Int64) ASC]
+08)--------------SortExec: expr=[CAST(v@0 AS Int64) ASC], preserve_partitioning=[true]
+09)----------------DataSourceExec: partitions=4, partition_sizes=[2, 1, 1, 1]
+10)------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+11)--------------LazyMemoryExec: partitions=1, batch_generators=[range: start=3, end=8, batch_size=2]
+
+# Same result the matrix asserts (v > min(3) -> {4..10} = 7).
+query I
+SELECT count(*) FROM pb_l l WHERE EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v > r.value);
+----
+7
+
+# Classic Inner: same range streamed side -> Inner PWMJ.
+query TT
+EXPLAIN SELECT count(*) FROM pb_l l JOIN range(3, 8) r ON l.v < r.value;
+----
+logical_plan
+01)Projection: count(Int64(1)) AS count(*)
+02)--Aggregate: groupBy=[[]], aggr=[[count(Int64(1))]]
+03)----Projection:
+04)------Inner Join:  Filter: CAST(l.v AS Int64) < r.value
+05)--------SubqueryAlias: l
+06)----------TableScan: pb_l projection=[v]
+07)--------SubqueryAlias: r
+08)----------TableScan: range() projection=[value]
+physical_plan
+01)ProjectionExec: expr=[count(Int64(1))@0 as count(*)]
+02)--AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))]
+03)----CoalescePartitionsExec
+04)------AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))]
+05)--------ProjectionExec: expr=[]
+06)----------PiecewiseMergeJoin: operator=Lt, join_type=Inner, on=(CAST(v AS Int64) < value)
+07)------------SortPreservingMergeExec: [CAST(v@0 AS Int64) DESC]
+08)--------------SortExec: expr=[CAST(v@0 AS Int64) DESC], preserve_partitioning=[true]
+09)----------------DataSourceExec: partitions=4, partition_sizes=[2, 1, 1, 1]
+10)------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+11)--------------LazyMemoryExec: partitions=1, batch_generators=[range: start=3, end=8, batch_size=2]
+
+query I
+SELECT count(*) FROM pb_l l JOIN range(3, 8) r ON l.v < r.value;
+----
+20
+
+statement ok
+RESET datafusion.execution.batch_size;
+
+statement ok
+RESET datafusion.optimizer.enable_piecewise_merge_join;
diff --git a/datafusion/sqllogictest/test_files/piecewise_merge_join_matrix.slt b/datafusion/sqllogictest/test_files/piecewise_merge_join_matrix.slt
new file mode 100644
index 0000000000000..13aeeddfc2afa
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/piecewise_merge_join_matrix.slt
@@ -0,0 +1,661 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+
+#   http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# PiecewiseMergeJoin correctness across a config matrix. A range join with no
+# equi key runs on PiecewiseMergeJoin when enable_piecewise_merge_join=true and
+# on NestedLoopJoin when false, so sweeping {true,false} x batch_size
+# {1,2,100,8192} requires the two to agree row-for-row. PWMJ covers
+# Inner/Left/Right/Full (Part 1) and LeftSemi/LeftAnti (Part 2); explicit RIGHT
+# SEMI/ANTI JOIN and Mark (OR EXISTS) stay on NLJ. Matrix rules: no EXPLAIN, no
+# in-file SET of a swept knob, rowsort every multi-row query.
+
+# configMatrix: datafusion.optimizer.enable_piecewise_merge_join=true,false
+# configMatrix: datafusion.execution.batch_size=1,2,100,8192
+
+# ==================================================================
+# Part 1: Classic range joins (Inner / Left / Right / Full)
+# ==================================================================
+# cj_l.lv = {1,3,5,NULL}, cj_r.rv = {3,5,NULL}. A NULL key matches nothing, so it
+# appears only as an unmatched row on an outer side.
+statement ok
+CREATE TABLE cj_l(lid INT, lv INT);
+
+statement ok
+INSERT INTO cj_l VALUES (1, 1), (2, 3), (3, 5), (4, NULL);
+
+statement ok
+CREATE TABLE cj_r(rid INT, rv INT);
+
+statement ok
+INSERT INTO cj_r VALUES (1, 3), (2, 5), (3, NULL);
+
+# INNER, all four operators. `<=`/`>=` differ from `<`/`>` through the equal
+# values (lv=3=rv, lv=5=rv).
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l JOIN cj_r r ON l.lv < r.rv;
+----
+1 1 1 3
+1 1 2 5
+2 3 2 5
+
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l JOIN cj_r r ON l.lv <= r.rv;
+----
+1 1 1 3
+1 1 2 5
+2 3 1 3
+2 3 2 5
+3 5 2 5
+
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l JOIN cj_r r ON l.lv > r.rv;
+----
+3 5 1 3
+
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l JOIN cj_r r ON l.lv >= r.rv;
+----
+2 3 1 3
+3 5 1 3
+3 5 2 5
+
+# LEFT: the INNER `<` rows plus every unmatched left row (lv=5 and lv=NULL) with
+# a NULL right side.
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l LEFT JOIN cj_r r ON l.lv < r.rv;
+----
+1 1 1 3
+1 1 2 5
+2 3 2 5
+3 5 NULL NULL
+4 NULL NULL NULL
+
+# RIGHT: INNER rows plus unmatched right rows (NULL left). rid=3 has a NULL key
+# and must still be emitted (regression: streamed-side NULLs were dropped).
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l RIGHT JOIN cj_r r ON l.lv < r.rv;
+----
+1 1 1 3
+1 1 2 5
+2 3 2 5
+NULL NULL 3 NULL
+
+# RIGHT with a different operator flips the buffered-side sort direction; the
+# NULL-keyed right row rid=3 is still emitted unmatched.
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l RIGHT JOIN cj_r r ON l.lv >= r.rv;
+----
+2 3 1 3
+3 5 1 3
+3 5 2 5
+NULL NULL 3 NULL
+
+# FULL: the INNER rows plus unmatched rows from both sides, including a NULL key
+# on each side.
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l FULL JOIN cj_r r ON l.lv < r.rv;
+----
+1 1 1 3
+1 1 2 5
+2 3 2 5
+3 5 NULL NULL
+4 NULL NULL NULL
+NULL NULL 3 NULL
+
+# Empty right side: LEFT keeps every left row with NULLs, INNER is empty.
+statement ok
+CREATE TABLE cj_empty(rid INT, rv INT);
+
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l JOIN cj_empty r ON l.lv < r.rv;
+----
+
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_l l LEFT JOIN cj_empty r ON l.lv < r.rv;
+----
+1 1 NULL NULL
+2 3 NULL NULL
+3 5 NULL NULL
+4 NULL NULL NULL
+
+# Empty left side: RIGHT keeps every right row with NULLs, including rid=3's NULL
+# key.
+query IIII rowsort
+SELECT l.rid, l.rv, r.rid, r.rv FROM cj_empty l RIGHT JOIN cj_r r ON l.rv < r.rv;
+----
+NULL NULL 1 3
+NULL NULL 2 5
+NULL NULL 3 NULL
+
+# Duplicate keys on both sides: each qualifying left row must pair with the whole
+# matching right suffix. lv=1 (twice) is below rv=2 (twice) -> 4 rows; lv=3 has
+# no larger right value.
+statement ok
+CREATE TABLE cj_dup_l(lid INT, lv INT);
+
+statement ok
+INSERT INTO cj_dup_l VALUES (1, 1), (2, 1), (3, 3);
+
+statement ok
+CREATE TABLE cj_dup_r(rid INT, rv INT);
+
+statement ok
+INSERT INTO cj_dup_r VALUES (1, 2), (2, 2);
+
+query IIII rowsort
+SELECT l.lid, l.lv, r.rid, r.rv FROM cj_dup_l l JOIN cj_dup_r r ON l.lv < r.rv;
+----
+1 1 1 2
+1 1 2 2
+2 1 1 2
+2 1 2 2
+
+# ==================================================================
+# Part 2: Existence joins (LeftSemi / LeftAnti; Right/Mark fall back to NLJ)
+# ==================================================================
+# EXISTS -> LeftSemi, NOT EXISTS -> LeftAnti (the only existence joins PWMJ
+# implements). Explicit RIGHT SEMI/ANTI JOIN and Mark (OR EXISTS) stay on NLJ;
+# covered at the end.
+
+# ------------------------------------------------------------------
+# Fixtures
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE ej_l(id INT, v INT);
+
+statement ok
+INSERT INTO ej_l VALUES (1, 5), (2, 4), (3, 2), (4, 1);
+
+statement ok
+CREATE TABLE ej_r(v INT);
+
+statement ok
+INSERT INTO ej_r VALUES (2), (3), (4);
+
+# ------------------------------------------------------------------
+# Basic: every operator, both LeftSemi (EXISTS) and LeftAnti (NOT EXISTS)
+# ------------------------------------------------------------------
+# ej_l.v = {5,4,2,1}, ej_r.v = {2,3,4}.
+
+# `<` : 2<3 and 1<2 match; 5 and 4 exceed every right value.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v < r.v);
+----
+3
+4
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE NOT EXISTS (SELECT 1 FROM ej_r r WHERE l.v < r.v);
+----
+1
+2
+
+# `<=` : additionally admits v=4 through the equal right value 4.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v <= r.v);
+----
+2
+3
+4
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE NOT EXISTS (SELECT 1 FROM ej_r r WHERE l.v <= r.v);
+----
+1
+
+# `>` : 5 and 4 exceed some right value; 2 and 1 do not.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+1
+2
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE NOT EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+3
+4
+
+# `>=` : additionally admits v=2 through the equal right value 2.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v >= r.v);
+----
+1
+2
+3
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE NOT EXISTS (SELECT 1 FROM ej_r r WHERE l.v >= r.v);
+----
+4
+
+# ------------------------------------------------------------------
+# Correlation written inner-column-first, and with an expression
+# ------------------------------------------------------------------
+# `r.v < l.v` is the same join as `l.v > r.v`; the planner flips the operator so
+# the marked (left) side stays buffered. Same rows as `l.v > r.v` above.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE r.v < l.v);
+----
+1
+2
+
+# An expression on the streamed side: `l.v < r.v + 1` is `l.v <= r.v` over ints.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v < r.v + 1);
+----
+2
+3
+4
+
+# ------------------------------------------------------------------
+# NULL semantics on the buffered (left) key
+# ------------------------------------------------------------------
+# A NULL key satisfies no comparison, so it is excluded from EXISTS and kept by
+# NOT EXISTS. ej_ln.v = {NULL,3,NULL,1}, right non-null key = {2}.
+statement ok
+CREATE TABLE ej_ln(id INT, v INT);
+
+statement ok
+INSERT INTO ej_ln VALUES (1, NULL), (2, 3), (3, NULL), (4, 1);
+
+statement ok
+CREATE TABLE ej_rn(v INT);
+
+statement ok
+INSERT INTO ej_rn VALUES (2), (NULL);
+
+# `<` : only v=1 is below 2. Both NULL-keyed left rows stay in NOT EXISTS.
+query I rowsort
+SELECT l.id FROM ej_ln l WHERE EXISTS (SELECT 1 FROM ej_rn r WHERE l.v < r.v);
+----
+4
+
+query I rowsort
+SELECT l.id FROM ej_ln l WHERE NOT EXISTS (SELECT 1 FROM ej_rn r WHERE l.v < r.v);
+----
+1
+2
+3
+
+# `>` : only v=3 exceeds 2.
+query I rowsort
+SELECT l.id FROM ej_ln l WHERE EXISTS (SELECT 1 FROM ej_rn r WHERE l.v > r.v);
+----
+2
+
+query I rowsort
+SELECT l.id FROM ej_ln l WHERE NOT EXISTS (SELECT 1 FROM ej_rn r WHERE l.v > r.v);
+----
+1
+3
+4
+
+# ------------------------------------------------------------------
+# Empty inputs
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE ej_empty(v INT);
+
+# Empty streamed (right) side: no key can match, so EXISTS is empty and NOT
+# EXISTS keeps every left row.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_empty r WHERE l.v > r.v);
+----
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE NOT EXISTS (SELECT 1 FROM ej_empty r WHERE l.v > r.v);
+----
+1
+2
+3
+4
+
+statement ok
+CREATE TABLE ej_lempty(id INT, v INT);
+
+# Empty buffered (left) side: nothing to emit either way.
+query I rowsort
+SELECT l.id FROM ej_lempty l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+
+query I rowsort
+SELECT l.id FROM ej_lempty l WHERE NOT EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+
+# ------------------------------------------------------------------
+# All-NULL sides (distinct early-exit triggers in the existence path)
+# ------------------------------------------------------------------
+# An all-NULL buffered side saturates the watermark before the first poll, so no
+# streamed batch is read; EXISTS is still empty.
+statement ok
+CREATE TABLE ej_l_allnull(id INT, v INT);
+
+statement ok
+INSERT INTO ej_l_allnull VALUES (1, NULL), (2, NULL);
+
+query I rowsort
+SELECT l.id FROM ej_l_allnull l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+
+query I rowsort
+SELECT l.id FROM ej_l_allnull l WHERE NOT EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+1
+2
+
+# An all-NULL streamed side never lowers the watermark, so it behaves like an
+# empty streamed side.
+statement ok
+CREATE TABLE ej_r_allnull(v INT);
+
+statement ok
+INSERT INTO ej_r_allnull VALUES (NULL), (NULL);
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r_allnull r WHERE l.v > r.v);
+----
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE NOT EXISTS (SELECT 1 FROM ej_r_allnull r WHERE l.v > r.v);
+----
+1
+2
+3
+4
+
+# ------------------------------------------------------------------
+# Duplicate buffered keys straddling the match boundary
+# ------------------------------------------------------------------
+# The first matching buffered row is found by binary search; it must return the
+# FIRST index of a run of equal keys or earlier duplicates vanish from EXISTS.
+# ej_dup_l.v = {5,5,3,3,1}; the deciding streamed key is 3, so `<=`/`>=` land
+# inside the run of 3s.
+statement ok
+CREATE TABLE ej_dup_l(id INT, v INT);
+
+statement ok
+INSERT INTO ej_dup_l VALUES (1, 5), (2, 5), (3, 3), (4, 3), (5, 1);
+
+statement ok
+CREATE TABLE ej_dup_r(v INT);
+
+statement ok
+INSERT INTO ej_dup_r VALUES (3);
+
+# `<` : only v=1 is below 3.
+query I rowsort
+SELECT l.id FROM ej_dup_l l WHERE EXISTS (SELECT 1 FROM ej_dup_r r WHERE l.v < r.v);
+----
+5
+
+# `<=` : both v=3 rows must appear.
+query I rowsort
+SELECT l.id FROM ej_dup_l l WHERE EXISTS (SELECT 1 FROM ej_dup_r r WHERE l.v <= r.v);
+----
+3
+4
+5
+
+# `>` : only the two v=5 rows exceed 3.
+query I rowsort
+SELECT l.id FROM ej_dup_l l WHERE EXISTS (SELECT 1 FROM ej_dup_r r WHERE l.v > r.v);
+----
+1
+2
+
+# `>=` : the boundary is inside the run of 3s from the other direction.
+query I rowsort
+SELECT l.id FROM ej_dup_l l WHERE EXISTS (SELECT 1 FROM ej_dup_r r WHERE l.v >= r.v);
+----
+1
+2
+3
+4
+
+query I rowsort
+SELECT l.id FROM ej_dup_l l WHERE NOT EXISTS (SELECT 1 FROM ej_dup_r r WHERE l.v >= r.v);
+----
+5
+
+# ------------------------------------------------------------------
+# Subquery filter -> repartitioned streamed side (multi-partition final pass)
+# ------------------------------------------------------------------
+# A predicate on the inner table is pushed to the scan and lets the streamed
+# side repartition, so the final existence pass is coordinated across streamed
+# partitions. `r.v > 0` keeps all of {2,3,4}, so the rows match `l.v > r.v`.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v AND r.v > 0);
+----
+1
+2
+
+query I rowsort
+SELECT l.id FROM ej_l l WHERE NOT EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v AND r.v > 0);
+----
+3
+4
+
+# ------------------------------------------------------------------
+# Type coverage: the existence path shares the join comparator
+# ------------------------------------------------------------------
+
+# Date32 key.
+statement ok
+CREATE TABLE ej_dl(id INT, d DATE);
+
+statement ok
+INSERT INTO ej_dl VALUES (1, DATE '2022-04-23'), (2, DATE '2022-04-28'), (3, DATE '2022-04-18');
+
+statement ok
+CREATE TABLE ej_dr(d DATE);
+
+statement ok
+INSERT INTO ej_dr VALUES (DATE '2022-04-20'), (DATE '2022-04-26');
+
+query I rowsort
+SELECT l.id FROM ej_dl l WHERE EXISTS (SELECT 1 FROM ej_dr r WHERE l.d > r.d);
+----
+1
+2
+
+query I rowsort
+SELECT l.id FROM ej_dl l WHERE NOT EXISTS (SELECT 1 FROM ej_dr r WHERE l.d > r.d);
+----
+3
+
+# Float key with negative zero: -0.0 and +0.0 compare equal in SQL, so
+# `-0.0 < 0.0` is false and `-0.0 <= 0.0` is true. The comparator normalizes the
+# sign of zero before comparing.
+statement ok
+CREATE TABLE ej_fl(id INT, v DOUBLE);
+
+statement ok
+INSERT INTO ej_fl VALUES (1, -0.0), (2, 2.5);
+
+statement ok
+CREATE TABLE ej_fr(v DOUBLE);
+
+statement ok
+INSERT INTO ej_fr VALUES (0.0);
+
+query I rowsort
+SELECT l.id FROM ej_fl l WHERE EXISTS (SELECT 1 FROM ej_fr r WHERE l.v < r.v);
+----
+
+query I rowsort
+SELECT l.id FROM ej_fl l WHERE EXISTS (SELECT 1 FROM ej_fr r WHERE l.v <= r.v);
+----
+1
+
+# String key.
+statement ok
+CREATE TABLE ej_sl(id INT, s VARCHAR);
+
+statement ok
+INSERT INTO ej_sl VALUES (1, 'apple'), (2, 'cherry'), (3, 'mango');
+
+statement ok
+CREATE TABLE ej_sr(s VARCHAR);
+
+statement ok
+INSERT INTO ej_sr VALUES ('banana'), ('lemon');
+
+query I rowsort
+SELECT l.id FROM ej_sl l WHERE EXISTS (SELECT 1 FROM ej_sr r WHERE l.s > r.s);
+----
+2
+3
+
+query I rowsort
+SELECT l.id FROM ej_sl l WHERE NOT EXISTS (SELECT 1 FROM ej_sr r WHERE l.s > r.s);
+----
+1
+
+# Dictionary-encoded key: no typed arrow min/max kernel, so the extreme key per
+# streamed batch is chosen through the generic ScalarValue path.
+statement ok
+CREATE TABLE ej_dict_l AS
+  SELECT column1 AS id, arrow_cast(column2, 'Dictionary(Int32, Utf8)') AS v
+  FROM (VALUES (1, 'a'), (2, 'c'), (3, 'e'), (4, NULL));
+
+statement ok
+CREATE TABLE ej_dict_r AS
+  SELECT arrow_cast(column1, 'Dictionary(Int32, Utf8)') AS v FROM (VALUES ('c'));
+
+query I rowsort
+SELECT l.id FROM ej_dict_l l WHERE EXISTS (SELECT 1 FROM ej_dict_r r WHERE l.v > r.v);
+----
+3
+
+query I rowsort
+SELECT l.id FROM ej_dict_l l WHERE NOT EXISTS (SELECT 1 FROM ej_dict_r r WHERE l.v > r.v);
+----
+1
+2
+4
+
+query I rowsort
+SELECT l.id FROM ej_dict_l l WHERE EXISTS (SELECT 1 FROM ej_dict_r r WHERE l.v <= r.v);
+----
+1
+2
+
+# ------------------------------------------------------------------
+# Multi-batch stress: streamed side from range() (fragments by batch_size)
+# ------------------------------------------------------------------
+# range() is a LazyMemoryExec that emits ceil(n / batch_size) batches, so
+# batch_size genuinely fragments the streamed side (unlike a VALUES table, whose
+# split is incidental). Plan proof in piecewise_merge_join_batches.slt. Buffered
+# v = 1..10; streamed range(3,8) = {3,4,5,6,7}; counts are matrix-invariant.
+statement ok
+CREATE TABLE ej_range_l(id INT, v INT) AS
+  SELECT CAST(value AS INT) AS id, CAST(value AS INT) AS v FROM range(1, 11);
+
+# Existence, `>`: v > min(3) -> {4..10} = 7.
+query I
+SELECT count(*) FROM ej_range_l l WHERE EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v > r.value);
+----
+7
+
+query I
+SELECT count(*) FROM ej_range_l l WHERE NOT EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v > r.value);
+----
+3
+
+# `<`: v < max(7) -> {1..6} = 6.
+query I
+SELECT count(*) FROM ej_range_l l WHERE EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v < r.value);
+----
+6
+
+query I
+SELECT count(*) FROM ej_range_l l WHERE NOT EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v < r.value);
+----
+4
+
+# `>=` with a pushed-down inner filter that keeps every row, so the streamed side
+# also repartitions. v >= min(3) -> {3..10} = 8.
+query I
+SELECT count(*) FROM ej_range_l l WHERE EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v >= r.value AND r.value > 0);
+----
+8
+
+query I
+SELECT count(*) FROM ej_range_l l WHERE NOT EXISTS (SELECT 1 FROM range(3, 8) r WHERE l.v >= r.value AND r.value > 0);
+----
+2
+
+# Classic range join over the same range() streamed side. INNER l.v < r.value:
+# sum over v of |{r : r > v}| = 5+5+4+3+2+1 = 20.
+query I
+SELECT count(*) FROM ej_range_l l JOIN range(3, 8) r ON l.v < r.value;
+----
+20
+
+# LEFT adds the unmatched left rows (v >= 7, which is v in {7,8,9,10}): 20 + 4 = 24.
+query I
+SELECT count(*) FROM ej_range_l l LEFT JOIN range(3, 8) r ON l.v < r.value;
+----
+24
+
+# Existence result feeding an aggregate: EXISTS l.v>r.v over the small fixture is
+# {1,2}, so the count is 2.
+query I
+SELECT count(*) FROM ej_l l WHERE EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+2
+
+# ------------------------------------------------------------------
+# Other existence joins: RightSemi / RightAnti / Mark
+# ------------------------------------------------------------------
+# PWMJ rejects these, so both matrix combinations run them on NestedLoopJoin;
+# only batch_size varies. Included so the file covers every existence join type.
+statement ok
+CREATE TABLE ej_rs_l(v INT);
+
+statement ok
+INSERT INTO ej_rs_l VALUES (3), (6);
+
+statement ok
+CREATE TABLE ej_rs_r(id INT, v INT);
+
+statement ok
+INSERT INTO ej_rs_r VALUES (1, 1), (2, 4), (3, 7);
+
+# RIGHT SEMI: keep right rows that have some larger left value. left={3,6}:
+# right 1 (l=3,6), right 4 (l=6) qualify; right 7 has none.
+query I rowsort
+SELECT r.id FROM ej_rs_l l RIGHT SEMI JOIN ej_rs_r r ON l.v > r.v;
+----
+1
+2
+
+# RIGHT ANTI: the complement -> only right 7.
+query I rowsort
+SELECT r.id FROM ej_rs_l l RIGHT ANTI JOIN ej_rs_r r ON l.v > r.v;
+----
+3
+
+# Mark join: EXISTS used inside a disjunction produces a LeftMark. `l.v > r.v`
+# over the fixture yields {1,2}; `l.id = 4` adds id 4.
+query I rowsort
+SELECT l.id FROM ej_l l WHERE l.id = 4 OR EXISTS (SELECT 1 FROM ej_r r WHERE l.v > r.v);
+----
+1
+2
+4
diff --git a/datafusion/sqllogictest/test_files/sort_merge_join_batches.slt b/datafusion/sqllogictest/test_files/sort_merge_join_batches.slt
new file mode 100644
index 0000000000000..eb26d0a715104
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/sort_merge_join_batches.slt
@@ -0,0 +1,65 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+
+#   http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Plan proof for sort_merge_join_matrix.slt's cross-batch coverage. SMJ sorts
+# both inputs and SortExec re-batches to batch_size, so equal-key runs span
+# batches even from a VALUES source. Not a matrix file: it fixes batch_size=1 to
+# assert (via EXPLAIN ANALYZE) that the join then receives one row per batch
+# (input_batches=16 = input_rows), i.e. the runs of five 1s and three 2s are
+# split across batches. Fixtures match sort_merge_join_matrix.slt.
+
+statement ok
+set datafusion.optimizer.prefer_hash_join = false;
+
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+set datafusion.optimizer.repartition_joins = true;
+
+statement ok
+set datafusion.execution.batch_size = 1;
+
+statement ok
+CREATE TABLE smj_dup_l(k INT) AS VALUES
+  (1), (1), (1), (1), (1), (2), (2), (2), (9);
+
+statement ok
+CREATE TABLE smj_dup_r(k INT) AS VALUES
+  (1), (1), (1), (1), (2), (2), (3);
+
+query TT
+EXPLAIN ANALYZE SELECT count(*) FROM smj_dup_l l JOIN smj_dup_r r ON l.k = r.k;
+----
+Plan with Metrics
+01)<slt:ignore>
+02)<slt:ignore>
+03)<slt:ignore>
+04)<slt:ignore>
+05)<slt:ignore>SortMergeJoinExec:<slt:ignore>input_batches=16, input_rows=16<slt:ignore>
+<slt:ignore>
+
+query I
+SELECT count(*) FROM smj_dup_l l JOIN smj_dup_r r ON l.k = r.k;
+----
+26
+
+statement ok
+RESET datafusion.execution.batch_size;
+
+statement ok
+RESET datafusion.optimizer.prefer_hash_join;
diff --git a/datafusion/sqllogictest/test_files/sort_merge_join_matrix.slt b/datafusion/sqllogictest/test_files/sort_merge_join_matrix.slt
new file mode 100644
index 0000000000000..7f1857782cb38
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/sort_merge_join_matrix.slt
@@ -0,0 +1,315 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+
+#   http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Sort-merge-join correctness across a config matrix. Every equijoin runs on
+# SortMergeJoin (prefer_hash_join=false) and HashJoin (true), so sweeping
+# {true,false} x batch_size {1,2,100,8192} requires them to agree row-for-row.
+# batch_size=1 forces equal-key runs to span single-row batches (historically
+# bug-prone). SMJ needs target_partitions>1 and repartition_joins, set below (not
+# swept). Null-aware NOT IN always uses HashJoin, so it is out of scope (see
+# null_aware_anti_join.slt). Matrix rules: no EXPLAIN, no in-file SET of a swept
+# knob, rowsort every multi-row query.
+
+# configMatrix: datafusion.optimizer.prefer_hash_join=true,false
+# configMatrix: datafusion.execution.batch_size=1,2,100,8192
+
+statement ok
+set datafusion.execution.target_partitions = 4;
+
+statement ok
+set datafusion.optimizer.repartition_joins = true;
+
+# ------------------------------------------------------------------
+# Fixtures: duplicate keys, an unmatched key on each side, and a NULL key on
+# each side (a NULL key never matches under default equality).
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE smj_l(k INT, v VARCHAR) AS VALUES
+  (1, 'a'), (1, 'b'), (2, 'c'), (3, 'd'), (NULL, 'e');
+
+statement ok
+CREATE TABLE smj_r(k INT, w VARCHAR) AS VALUES
+  (1, 'x'), (1, 'y'), (2, 'z'), (4, 'q'), (NULL, 'n');
+
+# ------------------------------------------------------------------
+# Basic: Inner / Left / Right / Full on a single equi key
+# ------------------------------------------------------------------
+# INNER: k=1 gives 2x2, k=2 gives 1x1; k=3, k=4 and both NULLs are unmatched.
+query ITT rowsort
+SELECT l.k, l.v, r.w FROM smj_l l JOIN smj_r r ON l.k = r.k;
+----
+1 a x
+1 a y
+1 b x
+1 b y
+2 c z
+
+# LEFT: the INNER rows plus unmatched left rows (k=3, NULL) padded with NULL.
+query ITT rowsort
+SELECT l.k, l.v, r.w FROM smj_l l LEFT JOIN smj_r r ON l.k = r.k;
+----
+1 a x
+1 a y
+1 b x
+1 b y
+2 c z
+3 d NULL
+NULL e NULL
+
+# RIGHT: the INNER rows plus unmatched right rows (k=4, NULL) padded with NULL.
+query TIT rowsort
+SELECT l.v, r.k, r.w FROM smj_l l RIGHT JOIN smj_r r ON l.k = r.k;
+----
+NULL 4 q
+NULL NULL n
+a 1 x
+a 1 y
+b 1 x
+b 1 y
+c 2 z
+
+# FULL: INNER rows plus every unmatched row from both sides.
+query ITT rowsort
+SELECT l.k, l.v, r.w FROM smj_l l FULL JOIN smj_r r ON l.k = r.k;
+----
+1 a x
+1 a y
+1 b x
+1 b y
+2 c z
+3 d NULL
+NULL NULL n
+NULL NULL q
+NULL e NULL
+
+# ------------------------------------------------------------------
+# Existence joins over an equi key: LeftSemi / LeftAnti / RightSemi / RightAnti
+# ------------------------------------------------------------------
+# LeftSemi (EXISTS): left rows whose key matches, each kept once (k=1 twice
+# because the left side has two such rows, not because the right side does).
+query IT rowsort
+SELECT l.k, l.v FROM smj_l l WHERE EXISTS (SELECT 1 FROM smj_r r WHERE l.k = r.k);
+----
+1 a
+1 b
+2 c
+
+# LeftAnti (NOT EXISTS): the complementary left rows, including the NULL key.
+query IT rowsort
+SELECT l.k, l.v FROM smj_l l WHERE NOT EXISTS (SELECT 1 FROM smj_r r WHERE l.k = r.k);
+----
+3 d
+NULL e
+
+# RightSemi: right rows whose key matches some left row.
+query IT rowsort
+SELECT r.k, r.w FROM smj_l l RIGHT SEMI JOIN smj_r r ON l.k = r.k;
+----
+1 x
+1 y
+2 z
+
+# RightAnti: the complementary right rows, including the NULL key.
+query IT rowsort
+SELECT r.k, r.w FROM smj_l l RIGHT ANTI JOIN smj_r r ON l.k = r.k;
+----
+4 q
+NULL n
+
+# ------------------------------------------------------------------
+# Null-safe equality: `IS NOT DISTINCT FROM` makes NULL match NULL
+# ------------------------------------------------------------------
+# The INNER result gains the (NULL,e)-(NULL,n) pair that plain `=` drops.
+query ITT rowsort
+SELECT l.k, l.v, r.w FROM smj_l l JOIN smj_r r ON l.k IS NOT DISTINCT FROM r.k;
+----
+1 a x
+1 a y
+1 b x
+1 b y
+2 c z
+NULL e n
+
+# ------------------------------------------------------------------
+# Multiple equi keys
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE smj_mk_l(k1 INT, k2 INT, v INT) AS VALUES
+  (1, 1, 10), (1, 2, 20), (2, 2, 30), (NULL, 1, 40);
+
+statement ok
+CREATE TABLE smj_mk_r(k1 INT, k2 INT, w INT) AS VALUES
+  (1, 1, 100), (1, 2, 200), (2, 2, 300), (1, 3, 400);
+
+query IIII rowsort
+SELECT l.k1, l.k2, l.v, r.w
+FROM smj_mk_l l JOIN smj_mk_r r ON l.k1 = r.k1 AND l.k2 = r.k2;
+----
+1 1 10 100
+1 2 20 200
+2 2 30 300
+
+query IIII rowsort
+SELECT l.k1, l.k2, l.v, r.w
+FROM smj_mk_l l LEFT JOIN smj_mk_r r ON l.k1 = r.k1 AND l.k2 = r.k2;
+----
+1 1 10 100
+1 2 20 200
+2 2 30 300
+NULL 1 40 NULL
+
+# ------------------------------------------------------------------
+# Equi key plus a non-equi join filter in the ON clause
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE smj_a(k INT, x INT) AS VALUES (1, 10), (1, 20), (2, 30), (3, 40);
+
+statement ok
+CREATE TABLE smj_b(k INT, y INT) AS VALUES (1, 15), (1, 25), (2, 5), (4, 50);
+
+# INNER: only k=1 has matches, kept where x < y.
+query IIII rowsort
+SELECT l.k, l.x, r.k, r.y FROM smj_a l JOIN smj_b r ON l.k = r.k AND l.x < r.y;
+----
+1 10 1 15
+1 10 1 25
+1 20 1 25
+
+# LEFT: rows whose key never matches, or matches but fails the filter, keep the
+# left row with NULLs (k=2's x=30 fails 30<5; k=3 has no right key).
+query III rowsort
+SELECT l.k, l.x, r.y FROM smj_a l LEFT JOIN smj_b r ON l.k = r.k AND l.x < r.y;
+----
+1 10 15
+1 10 25
+1 20 25
+2 30 NULL
+3 40 NULL
+
+# FULL with the same filter: unmatched rows from both sides survive.
+query IIII rowsort
+SELECT l.k, l.x, r.k, r.y FROM smj_a l FULL JOIN smj_b r ON l.k = r.k AND l.x < r.y;
+----
+1 10 1 15
+1 10 1 25
+1 20 1 25
+2 30 NULL NULL
+3 40 NULL NULL
+NULL NULL 2 5
+NULL NULL 4 50
+
+# ------------------------------------------------------------------
+# Non-integer (string) key
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE smj_sk_l(k VARCHAR, v INT) AS VALUES ('a', 1), ('b', 2), ('a', 3);
+
+statement ok
+CREATE TABLE smj_sk_r(k VARCHAR, w INT) AS VALUES ('a', 10), ('c', 20);
+
+query TII rowsort
+SELECT l.k, l.v, r.w FROM smj_sk_l l JOIN smj_sk_r r ON l.k = r.k;
+----
+a 1 10
+a 3 10
+
+query TII rowsort
+SELECT l.k, l.v, r.w FROM smj_sk_l l FULL JOIN smj_sk_r r ON l.k = r.k;
+----
+NULL NULL 20
+a 1 10
+a 3 10
+b 2 NULL
+
+# ------------------------------------------------------------------
+# Empty inputs
+# ------------------------------------------------------------------
+statement ok
+CREATE TABLE smj_empty(k INT, w VARCHAR);
+
+# INNER against an empty side is empty; LEFT keeps every left row with NULLs.
+query ITT rowsort
+SELECT l.k, l.v, r.w FROM smj_l l JOIN smj_empty r ON l.k = r.k;
+----
+
+query ITT rowsort
+SELECT l.k, l.v, r.w FROM smj_l l LEFT JOIN smj_empty r ON l.k = r.k;
+----
+1 a NULL
+1 b NULL
+2 c NULL
+3 d NULL
+NULL e NULL
+
+# Empty left side with a RIGHT join keeps every right row with NULLs.
+query IIT rowsort
+SELECT l.k, r.k, r.w FROM smj_empty l RIGHT JOIN smj_r r ON l.k = r.k;
+----
+NULL 1 x
+NULL 1 y
+NULL 2 z
+NULL 4 q
+NULL NULL n
+
+# ------------------------------------------------------------------
+# Duplicate keys across batch boundaries (verified by count)
+# ------------------------------------------------------------------
+# SortExec re-batches to batch_size, so equal-key runs span batches (proof in
+# sort_merge_join_batches.slt). Left: five k=1, three k=2, one unmatched k=9;
+# right: four k=1, two k=2, one unmatched k=3. INNER = 5*4 + 3*2 = 26.
+statement ok
+CREATE TABLE smj_dup_l(k INT) AS VALUES
+  (1), (1), (1), (1), (1), (2), (2), (2), (9);
+
+statement ok
+CREATE TABLE smj_dup_r(k INT) AS VALUES
+  (1), (1), (1), (1), (2), (2), (3);
+
+query I
+SELECT count(*) FROM smj_dup_l l JOIN smj_dup_r r ON l.k = r.k;
+----
+26
+
+# LEFT = INNER + unmatched left k=9 = 27.
+query I
+SELECT count(*) FROM smj_dup_l l LEFT JOIN smj_dup_r r ON l.k = r.k;
+----
+27
+
+# RIGHT = INNER + unmatched right k=3 = 27.
+query I
+SELECT count(*) FROM smj_dup_l l RIGHT JOIN smj_dup_r r ON l.k = r.k;
+----
+27
+
+# FULL = INNER + both unmatched = 28.
+query I
+SELECT count(*) FROM smj_dup_l l FULL JOIN smj_dup_r r ON l.k = r.k;
+----
+28
+
+# LeftSemi = left rows with a match = 5 + 3 = 8.
+query I
+SELECT count(*) FROM smj_dup_l l WHERE EXISTS (SELECT 1 FROM smj_dup_r r WHERE l.k = r.k);
+----
+8
+
+# LeftAnti = unmatched left rows = 1 (k=9).
+query I
+SELECT count(*) FROM smj_dup_l l WHERE NOT EXISTS (SELECT 1 FROM smj_dup_r r WHERE l.k = r.k);
+----
+1