You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/08 15:12:49 UTC
(datafusion) branch gh-readonly-queue/main/pr-26043-415f25db4b768cc2090f5740823fbcf561518074 created (now 277ddcc24f)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-26043-415f25db4b768cc2090f5740823fbcf561518074
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 277ddcc24f fix: align nested piecewise merge join sorting and comparison (#26043)
This branch includes the following new commits:
new 277ddcc24f fix: align nested piecewise merge join sorting and comparison (#26043)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: align nested piecewise merge join sorting and comparison (#26043)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26043-415f25db4b768cc2090f5740823fbcf561518074
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 277ddcc24f22c266edf1372646cf3aa77b284df4
Author: mika <sp...@outlook.com>
AuthorDate: Thu Oct 8 15:12:24 2026 +0000
fix: align nested piecewise merge join sorting and comparison (#26043)
## Which issue does this PR close?
- Closes #25957.
## Rationale for this change
Piecewise merge joins can return incorrect results for nested range keys
containing inner NULLs, and fail when sorting Struct or List-of-List
keys. Sorting, extreme-key selection, and comparison must use the same
ordering.
## What changes are included in this PR?
- Sort `<`/`<=` keys descending with NULLs last, reversing both value
and inner-NULL ordering to agree with SQL comparisons.
- Bound classic and left existence scans to non-null keys, preserving
trailing NULL rows in outer, anti, and mark results.
- Select nested min/max keys with Arrow's comparator.
- Sort streamed keys with `lexsort_to_indices`, supporting Struct and
List-of-List keys.
- Execute nested range joins with `PiecewiseMergeJoinExec`.
## What is the testing strategy for this PR?
SQL logic tests compare complete result multisets against NestedLoopJoin
for List, Struct, and List-of-List keys, all four range operators,
classic joins, semi/anti joins, and an EXISTS mark expression. The
matrix covers batch sizes 1, 2, and 8192 and partition counts 1 and 4.
Plan assertions check that Struct and List-of-List classic joins use
PWMJ.
Validation passed: PWMJ SQL logic tests and fuzz tests, the extended
workspace suite (including all 527 SQL test files), formatting, Clippy
with all targets/features, and the repository lint checks.
## Are there any user-facing changes?
Nested range joins produce correct results and support Struct and
List-of-List keys with piecewise merge joins enabled. No public API
changes.
---------
Co-authored-by: mikamikasuki <21...@users.noreply.github.com>
---
datafusion/core/tests/fuzz_cases/join_fuzz.rs | 4 +-
.../src/joins/piecewise_merge_join/classic_join.rs | 58 +--
.../src/joins/piecewise_merge_join/exec.rs | 12 +-
.../joins/piecewise_merge_join/existence_join.rs | 97 ++--
.../src/joins/piecewise_merge_join/utils.rs | 28 ++
.../test_files/piecewise_merge_join_batches.slt | 4 +-
.../piecewise_merge_join_nested_nulls.slt | 558 +++++++++++++++++++++
datafusion/sqllogictest/test_files/pwmj.slt | 64 ++-
8 files changed, 748 insertions(+), 77 deletions(-)
diff --git a/datafusion/core/tests/fuzz_cases/join_fuzz.rs b/datafusion/core/tests/fuzz_cases/join_fuzz.rs
index 86d4d5de77..7df8dcbe3f 100644
--- a/datafusion/core/tests/fuzz_cases/join_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/join_fuzz.rs
@@ -1604,14 +1604,14 @@ fn pwmj_plan(
join_type: JoinType,
) -> Arc<dyn ExecutionPlan> {
// Matches `PiecewiseMergeJoinExec::required_input_ordering`: descending for `<`/`<=`,
- // ascending for `>`/`>=`, NULLs first either way. Right existence joins require no
+ // ascending for `>`/`>=`, reversing NULL placement too. Right existence joins require no
// ordering at all -- they only read the buffered side's min/max -- so they are fed the
// left side unsorted, which is the input shape they will see in a real plan.
let buffered = match join_type {
JoinType::RightSemi | JoinType::RightAnti | JoinType::RightMark => left,
_ => {
let sort_options = match op {
- Operator::Lt | Operator::LtEq => SortOptions::new(true, true),
+ Operator::Lt | Operator::LtEq => SortOptions::new(true, false),
Operator::Gt | Operator::GtEq => SortOptions::new(false, true),
other => panic!("not a range operator: {other:?}"),
};
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
index 4eebbcf079..63885aa652 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
@@ -17,12 +17,12 @@
//! Stream Implementation for PiecewiseMergeJoin's Classic Join (Left, Right, Full, Inner)
-use arrow::array::{Array, PrimitiveBuilder, new_null_array};
+use arrow::array::{PrimitiveBuilder, new_null_array};
use arrow::compute::{BatchCoalescer, take};
use arrow::datatypes::UInt32Type;
use arrow::{
array::{ArrayRef, RecordBatch, UInt32Array},
- compute::{sort_to_indices, take_record_batch},
+ compute::{SortColumn, lexsort_to_indices, take_record_batch},
};
use arrow_schema::{Schema, SchemaRef, SortOptions};
use datafusion_common::NullEquality;
@@ -37,7 +37,9 @@ use std::{sync::Arc, task::Poll};
use crate::handle_state;
use crate::joins::piecewise_merge_join::exec::{BufferedSide, BufferedSideReadyState};
-use crate::joins::piecewise_merge_join::utils::need_produce_result_in_final;
+use crate::joins::piecewise_merge_join::utils::{
+ need_produce_result_in_final, non_null_range, unmatched_buffered_batch,
+};
use crate::joins::utils::JoinKeyComparator;
use crate::joins::utils::{BuildProbeJoinMetrics, StatefulStreamResult};
use crate::stream::EmptyRecordBatchStream;
@@ -244,9 +246,11 @@ impl ClassicPWMJStream {
self.join_metrics.input_rows.add(batch.num_rows());
// Sort stream values and change the streamed record batch accordingly
- let indices = sort_to_indices(
- stream_values.as_ref(),
- Some(self.sort_option),
+ let indices = lexsort_to_indices(
+ &[SortColumn {
+ values: Arc::clone(&stream_values),
+ options: Some(self.sort_option),
+ }],
None,
)?;
let stream_batch = take_record_batch(&batch, &indices)?;
@@ -336,16 +340,11 @@ impl ClassicPWMJStream {
let buffered_data = Arc::clone(&self.buffered_side.try_as_ready()?.buffered_data);
let buffered_batch = buffered_data.batch();
- // Every match marks the suffix `[k, buffered_len)`, so the buffered rows that were
- // never matched are exactly the complementary prefix `[0, min_marked)` -- which
- // includes the null-keyed rows, since nulls sort first and the scan starts past
- // them. That makes the final pass a zero-copy slice instead of building an index
- // array and running `take` over it.
- let min_marked = buffered_data
- .min_marked
- .load(AtomicOrdering::SeqCst)
- .min(buffered_batch.num_rows());
- let new_buffered_batch = buffered_batch.slice(0, min_marked);
+ let (_, non_null_end) =
+ non_null_range(buffered_data.values().as_ref(), self.sort_option);
+ let min_marked = buffered_data.min_marked.load(AtomicOrdering::SeqCst);
+ let new_buffered_batch =
+ unmatched_buffered_batch(buffered_batch, min_marked, non_null_end)?;
let mut buffered_columns = new_buffered_batch.columns().to_vec();
let streamed_columns: Vec<ArrayRef> = self
@@ -444,7 +443,8 @@ fn resolve_classic_join(
join_type: JoinType,
batch_process_state: &mut BatchProcessState,
) -> Result<RecordBatch> {
- let buffered_len = buffered_side.buffered_data.values().len();
+ let (buffered_start, buffered_len) =
+ non_null_range(buffered_side.buffered_data.values().as_ref(), sort_options);
let stream_values = stream_batch.compare_key_values();
// Build comparator once for the batch pair
@@ -459,26 +459,28 @@ fn resolve_classic_join(
let mut stream_idx = batch_process_state.start_stream_idx;
if !batch_process_state.processed_null_count {
- let buffered_null_idx = buffered_side.buffered_data.values().null_count();
- let stream_null_idx = stream_values[0].null_count();
- buffer_idx = buffered_null_idx;
- stream_idx = stream_null_idx;
+ let (stream_start, stream_end) =
+ non_null_range(stream_values[0].as_ref(), sort_options);
+ buffer_idx = buffered_start;
+ stream_idx = stream_start;
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.
+ // NULL keys never match, but outer joins must emit them on either end.
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);
+ for row_idx in
+ (0..stream_start).chain(stream_end..stream_batch.batch.num_rows())
+ {
+ batch_process_state
+ .unmatched_indices
+ .append_value(row_idx as u32);
}
}
}
// Our buffer_idx variable allows us to start probing on the buffered side where we last matched
// in the previous stream row.
- for row_idx in stream_idx..stream_batch.batch.num_rows() {
+ let (_, stream_end) = non_null_range(stream_values[0].as_ref(), sort_options);
+ for row_idx in stream_idx..stream_end {
while buffer_idx < buffered_len {
let compare = cmp.compare(row_idx, buffer_idx);
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs
index f2395211be..fb878b7747 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs
@@ -330,7 +330,7 @@ impl PiecewiseMergeJoinExec {
// Take the operator and enforce a sort order on the streamed + buffered side based on
// the operator type.
let sort_options = match operator {
- Operator::Lt | Operator::LtEq => SortOptions::new(true, true),
+ Operator::Lt | Operator::LtEq => SortOptions::new(true, false),
Operator::Gt | Operator::GtEq => SortOptions::new(false, true),
_ => {
return internal_err!(
@@ -1034,14 +1034,14 @@ pub(super) struct BufferedSideData {
pub(super) batch: RecordBatch,
values: ArrayRef,
pub(super) remaining_partitions: AtomicUsize,
- /// The start of the matched suffix of the buffered side, or `usize::MAX` before the
- /// first match. `[min_marked, len)` *is* the matched set and `[0, min_marked)` the
- /// unmatched one -- no bitmap is allocated.
+ /// The start of the matched non-null suffix, or `usize::MAX` before the
+ /// first match. `[min_marked, non_null_end)` is the matched set; the prefix
+ /// and any trailing NULL keys are unmatched. No bitmap is allocated.
///
/// Both stream kinds only ever mark a suffix, which is what makes one index enough:
- /// - `ExistencePWMJStream` marks `[k, len)` for the first buffered row `k` matching
+ /// - `ExistencePWMJStream` marks `[k, non_null_end)` for the first row `k` matching
/// a streamed batch's extreme key.
- /// - `ClassicPWMJStream` emits `buffered[k..] x streamed_row` on each match, so the
+ /// - `ClassicPWMJStream` emits `buffered[k..non_null_end] x streamed_row`, so the
/// rows it marks are exactly that same suffix.
///
/// Shared so each partition benefits from what the others have marked; it only ever
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs
index 2ea33239c1..274577b4ed 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs
@@ -65,22 +65,22 @@
//! so a partition that starts late reads nothing at all.
//!
//! Rows whose join key is NULL never satisfy a comparison predicate. Buffered NULLs sort to
-//! the front, so the scan starts past them and null-keyed buffered rows are left unmarked —
+//! either end, outside the searched range, and null-keyed buffered rows are left unmarked —
//! correctly excluded from `LeftSemi` and included in `LeftAnti`. The extreme key is picked
//! from each streamed batch with NULLs ignored, so it is non-null unless the whole batch is.
//!
//! # Output
//!
//! Marking only ever covers a suffix, and each mark lowers the watermark to its own start,
-//! so the matched set is always exactly `[min_marked, buffered_len)`. A bitmap would be a
+//! so the matched set is always exactly `[min_marked, non_null_end)`. A bitmap would be a
//! less compact encoding of that one index, so none is allocated. `ClassicPWMJStream` marks
//! the same way, which is why the watermark lives in `BufferedSideData` rather than here.
//!
//! Once every streamed partition has been consumed, the last one to finish slices the
-//! buffered batch: `LeftSemi` takes `[min_marked, len)`, `LeftAnti` the complementary
-//! prefix `[0, min_marked)`, which is where the null-keyed rows live. `LeftMark` takes
+//! buffered batch: `LeftSemi` takes `[min_marked, non_null_end)`, `LeftAnti` the complementary
+//! prefix `[0, min_marked)` plus any trailing null-keyed rows. `LeftMark` takes
//! neither slice: every buffered row is preserved, with a `mark` column built from the same
-//! watermark (`true` from `min_marked` on, `false` before it) appended instead of any row
+//! watermark (`true` in the matched range, `false` elsewhere) appended instead of any row
//! being dropped. Only the buffered (left) columns, plus that `mark` column for `LeftMark`,
//! are produced.
//!
@@ -91,8 +91,12 @@ use std::sync::Arc;
use std::sync::atomic::Ordering as AtomicOrdering;
use std::task::{Poll, ready};
-use arrow::array::{Array, ArrayRef, BooleanArray, BooleanBufferBuilder, RecordBatch};
-use arrow::compute::BatchCoalescer;
+use arrow::array::{
+ Array, ArrayRef, BooleanArray, BooleanBufferBuilder, RecordBatch, UInt32Array,
+ new_null_array,
+};
+use arrow::compute::{BatchCoalescer, take};
+use arrow_ord::ord::make_comparator;
use arrow_schema::{SchemaRef, SortOptions};
use datafusion_common::{NullEquality, Result, internal_err};
use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream};
@@ -101,6 +105,7 @@ use datafusion_functions_aggregate_common::min_max::{max_batch, min_batch};
use datafusion_physical_expr::PhysicalExprRef;
use futures::{Stream, StreamExt};
+use super::utils::{non_null_range, unmatched_buffered_batch};
use crate::handle_state;
use crate::joins::piecewise_merge_join::exec::{BufferedSide, BufferedSideReadyState};
use crate::joins::utils::{
@@ -265,7 +270,8 @@ impl ExistencePWMJStream {
let min_marked = buffered_data.min_marked.load(AtomicOrdering::SeqCst);
let buffered_values = buffered_data.values();
- Ok(min_marked.min(buffered_values.len()) <= buffered_values.null_count())
+ let (start, end) = non_null_range(buffered_values.as_ref(), self.sort_option);
+ Ok(min_marked.min(end) <= start)
}
/// Marks this partition done with the streamed side: releases the input pipeline and,
@@ -301,20 +307,18 @@ impl ExistencePWMJStream {
{
let buffered_data = &self.buffered_side.try_as_ready()?.buffered_data;
let buffered_values = buffered_data.values();
- let buffered_len = buffered_values.len();
- // NULL keys can never match, and `sort_options` uses `nulls_first` for every
- // operator (see `try_new`), so buffered nulls sit at the front -- skip past them.
- let first_non_null_buffered = buffered_values.null_count();
+ let (first_non_null_buffered, non_null_end) =
+ non_null_range(buffered_values.as_ref(), sort_option);
- // `[min_marked, buffered_len)` was already marked, by this partition or
+ // `[min_marked, non_null_end)` was already marked, by this partition or
// another, so a match found there would write nothing. Stop the scan at the
// watermark: that bounds the comparisons this batch performs, not just the
// bits it writes.
let scan_limit = buffered_data
.min_marked
.load(AtomicOrdering::SeqCst)
- .min(buffered_len);
+ .min(non_null_end);
// The extreme key is the only one that can decide anything: it reaches the
// smallest matching `buffer_idx`, and every other row in the batch matches a
@@ -370,7 +374,7 @@ impl ExistencePWMJStream {
if buffer_idx < scan_limit {
// Everything from `buffer_idx` on matches, so lowering the
// watermark to it records the match: the marked set is exactly
- // `[min_marked, buffered_len)` and needs no bitmap.
+ // `[min_marked, non_null_end)` and needs no bitmap.
//
// INVARIANT: sound only because the buffered side and each
// streamed batch are sorted the same way for this operator
@@ -403,30 +407,29 @@ impl ExistencePWMJStream {
let buffered_batch = buffered_data.batch();
let buffered_len = buffered_batch.num_rows();
- // The marked rows are always the contiguous suffix `[min_marked, len)`: each
- // match covers `[k, previous min_marked)` and then lowers the watermark to
- // `k`, so the union is `[k, len)`. The result is therefore a slice, with no
- // index array to materialize and no `take`.
+ // The matched non-null suffix is `[min_marked, non_null_end)`.
+ // Each match lowers the watermark; trailing NULL keys stay unmarked.
+ let (_, non_null_end) =
+ non_null_range(buffered_data.values().as_ref(), self.sort_option);
let min_marked = buffered_data
.min_marked
.load(AtomicOrdering::SeqCst)
- .min(buffered_len);
+ .min(non_null_end);
let (num_rows, columns) = match self.join_type {
JoinType::LeftSemi => {
let sliced =
- buffered_batch.slice(min_marked, buffered_len - min_marked);
+ buffered_batch.slice(min_marked, non_null_end - min_marked);
(sliced.num_rows(), sliced.columns().to_vec())
}
// `LeftMark` keeps every buffered row -- nothing to slice -- and appends
- // the watermark as a `mark` column instead of using it to drop rows: `false`
- // for the unmatched prefix `[0, min_marked)`, `true` for the matched suffix
- // `[min_marked, len)` -- the same split `LeftSemi`/`LeftAnti` slice the
- // buffered batch on, just kept as one column instead of used to drop rows.
+ // the watermark as a `mark` column: true only within the matched
+ // non-null range, false for the prefix and any trailing NULL keys.
JoinType::LeftMark => {
let mut mark = BooleanBufferBuilder::new(buffered_len);
mark.append_n(min_marked, false);
- mark.append_n(buffered_len - min_marked, true);
+ mark.append_n(non_null_end - min_marked, true);
+ mark.append_n(buffered_len - non_null_end, false);
let mut columns = buffered_batch.columns().to_vec();
columns.push(
@@ -434,11 +437,13 @@ impl ExistencePWMJStream {
);
(buffered_len, columns)
}
- // `LeftAnti`: the unmarked prefix, which includes every null-keyed row --
- // nulls sort first and the watermark never drops below the buffered null
- // count.
+ // `LeftAnti` includes the unmarked prefix and any trailing NULL keys.
JoinType::LeftAnti => {
- let sliced = buffered_batch.slice(0, min_marked);
+ let sliced = unmatched_buffered_batch(
+ buffered_batch,
+ min_marked,
+ non_null_end,
+ )?;
(sliced.num_rows(), sliced.columns().to_vec())
}
other => {
@@ -478,10 +483,36 @@ impl ExistencePWMJStream {
/// Ordered the same way as [`JoinKeyComparator`]: both use IEEE 754 totalOrder for floats,
/// and the comparator normalizes `-0.0` on either side of it.
///
-/// Numeric, temporal, string, binary and boolean keys get a typed arrow kernel -- a linear
-/// scan that allocates nothing. Dictionary and nested keys fall to `min_max_batch_generic`, a
-/// `ScalarValue`-per-row comparator loop; specializing those is left to a follow-up.
+/// Nested keys use Arrow's comparator, so inner NULLs have the same ordering as
+/// sorting and range comparisons. Other keys use the typed min/max kernels.
pub(super) fn extreme_key(values: &ArrayRef, descending: bool) -> Result<ArrayRef> {
+ if values.data_type().is_nested() {
+ // Reverse both value and inner-NULL ordering for max, matching SQL's
+ // ascending NULLS FIRST nested comparison order.
+ let cmp = make_comparator(
+ values.as_ref(),
+ values.as_ref(),
+ SortOptions::new(descending, !descending),
+ )?;
+ let nulls = values.logical_nulls();
+ let mut extreme = None;
+ for idx in 0..values.len() {
+ if nulls.as_ref().is_some_and(|n| n.is_null(idx)) {
+ continue;
+ }
+ if extreme.is_none_or(|current| cmp(idx, current) == Ordering::Less) {
+ extreme = Some(idx);
+ }
+ }
+ return match extreme {
+ Some(idx) => Ok(take(
+ values.as_ref(),
+ &UInt32Array::from(vec![idx as u32]),
+ None,
+ )?),
+ None => Ok(new_null_array(values.data_type(), 1)),
+ };
+ }
let extreme = if descending {
max_batch(values)?
} else {
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/utils.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/utils.rs
index 7f63e7eac5..2c0417aada 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/utils.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/utils.rs
@@ -15,8 +15,36 @@
// specific language governing permissions and limitations
// under the License.
+use arrow::array::{Array, RecordBatch};
+use arrow::compute::concat_batches;
+use arrow_schema::SortOptions;
+use datafusion_common::Result;
use datafusion_expr::JoinType;
+/// Bounds of the non-null keys in an array sorted with `options`.
+pub(super) fn non_null_range(values: &dyn Array, options: SortOptions) -> (usize, usize) {
+ let null_count = values.logical_null_count();
+ if options.nulls_first {
+ (null_count, values.len())
+ } else {
+ (0, values.len() - null_count)
+ }
+}
+
+/// Keep the unmatched prefix and any trailing NULL keys outside the matching range.
+pub(super) fn unmatched_buffered_batch(
+ batch: &RecordBatch,
+ min_marked: usize,
+ non_null_end: usize,
+) -> Result<RecordBatch> {
+ let prefix = batch.slice(0, min_marked.min(non_null_end));
+ if non_null_end == batch.num_rows() {
+ return Ok(prefix);
+ }
+ let nulls = batch.slice(non_null_end, batch.num_rows() - non_null_end);
+ Ok(concat_batches(&batch.schema(), [&prefix, &nulls])?)
+}
+
// Returns boolean for whether the join is a right existence join served by
// `RightExistencePWMJStream`, which reads nothing but a single min/max off the buffered side.
//
diff --git a/datafusion/sqllogictest/test_files/piecewise_merge_join_batches.slt b/datafusion/sqllogictest/test_files/piecewise_merge_join_batches.slt
index d8b3c16163..d68754c9d0 100644
--- a/datafusion/sqllogictest/test_files/piecewise_merge_join_batches.slt
+++ b/datafusion/sqllogictest/test_files/piecewise_merge_join_batches.slt
@@ -85,8 +85,8 @@ physical_plan
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]
+07)------------SortPreservingMergeExec: [CAST(v@0 AS Int64) DESC NULLS LAST]
+08)--------------SortExec: expr=[CAST(v@0 AS Int64) DESC NULLS LAST], 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]
diff --git a/datafusion/sqllogictest/test_files/piecewise_merge_join_nested_nulls.slt b/datafusion/sqllogictest/test_files/piecewise_merge_join_nested_nulls.slt
new file mode 100644
index 0000000000..e8f5007426
--- /dev/null
+++ b/datafusion/sqllogictest/test_files/piecewise_merge_join_nested_nulls.slt
@@ -0,0 +1,558 @@
+# 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.
+
+# configMatrix: datafusion.execution.batch_size=1,2,8192
+# configMatrix: datafusion.execution.target_partitions=1,4
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+statement ok
+CREATE TABLE l AS SELECT * FROM (VALUES (make_array(5), 'l5')) t(a, p);
+
+statement ok
+CREATE TABLE r AS SELECT * FROM (VALUES (make_array(arrow_cast(NULL, 'Int64')), 'rnull'), (make_array(7), 'r7'), (make_array(1), 'r1')) t(b, q);
+
+query TT
+SELECT l.p, r.q FROM l JOIN r ON l.a < r.b ORDER BY 1, 2;
+----
+l5 r7
+
+statement ok
+CREATE TABLE sl AS SELECT * FROM (VALUES (named_struct('x', 5), 'l5')) t(a, p);
+
+statement ok
+CREATE TABLE sr AS SELECT * FROM (VALUES (named_struct('x', 7), 'r7'), (named_struct('x', 1), 'r1')) t(b, q);
+
+query TT
+SELECT sl.p, sr.q FROM sl JOIN sr ON sl.a < sr.b ORDER BY 1, 2;
+----
+l5 r7
+
+
+statement ok
+CREATE TABLE list_left AS
+SELECT id, CASE WHEN id = 0 THEN NULL ELSE make_array(v) END AS k
+FROM (SELECT value AS id, CASE WHEN value < 2 THEN NULL WHEN value = 2 THEN 1 WHEN value < 5 THEN 5 ELSE 7 END AS v FROM range(0, 6)) AS t;
+
+statement ok
+CREATE TABLE list_right AS
+SELECT id, CASE WHEN id = 0 THEN NULL ELSE make_array(v) END AS k
+FROM (SELECT value AS id, CASE WHEN value < 2 THEN NULL WHEN value = 2 THEN 1 WHEN value < 5 THEN 5 ELSE 7 END AS v FROM range(0, 6)) AS t;
+
+# list keys, <: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_list_lt AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l INNER JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l LEFT JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l RIGHT JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l FULL JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT SEMI JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT ANTI JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT SEMI JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT ANTI JOIN list_right r ON l.k < r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM list_right r WHERE l.k < r.k) OR l.id = -1 AS INT) AS r_id FROM list_left l;
+
+statement ok
+CREATE TABLE ref_list_lt AS SELECT * FROM case_list_lt;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_list_lt), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_list_lt)
+UNION ALL
+(SELECT * FROM ref_list_lt EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# list keys, <=: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_list_le AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l INNER JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l LEFT JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l RIGHT JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l FULL JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT SEMI JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT ANTI JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT SEMI JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT ANTI JOIN list_right r ON l.k <= r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM list_right r WHERE l.k <= r.k) OR l.id = -1 AS INT) AS r_id FROM list_left l;
+
+statement ok
+CREATE TABLE ref_list_le AS SELECT * FROM case_list_le;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_list_le), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_list_le)
+UNION ALL
+(SELECT * FROM ref_list_le EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# list keys, >: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_list_gt AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l INNER JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l LEFT JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l RIGHT JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l FULL JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT SEMI JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT ANTI JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT SEMI JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT ANTI JOIN list_right r ON l.k > r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM list_right r WHERE l.k > r.k) OR l.id = -1 AS INT) AS r_id FROM list_left l;
+
+statement ok
+CREATE TABLE ref_list_gt AS SELECT * FROM case_list_gt;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_list_gt), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_list_gt)
+UNION ALL
+(SELECT * FROM ref_list_gt EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# list keys, >=: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_list_ge AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l INNER JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l LEFT JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l RIGHT JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM list_left l FULL JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT SEMI JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM list_left l LEFT ANTI JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT SEMI JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM list_left l RIGHT ANTI JOIN list_right r ON l.k >= r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM list_right r WHERE l.k >= r.k) OR l.id = -1 AS INT) AS r_id FROM list_left l;
+
+statement ok
+CREATE TABLE ref_list_ge AS SELECT * FROM case_list_ge;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_list_ge), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_list_ge)
+UNION ALL
+(SELECT * FROM ref_list_ge EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+statement ok
+CREATE TABLE struct_left AS
+SELECT id, CASE WHEN id = 0 THEN NULL ELSE named_struct('x', v) END AS k
+FROM (SELECT value AS id, CASE WHEN value < 2 THEN NULL WHEN value = 2 THEN 1 WHEN value < 5 THEN 5 ELSE 7 END AS v FROM range(0, 6)) AS t;
+
+statement ok
+CREATE TABLE struct_right AS
+SELECT id, CASE WHEN id = 0 THEN NULL ELSE named_struct('x', v) END AS k
+FROM (SELECT value AS id, CASE WHEN value < 2 THEN NULL WHEN value = 2 THEN 1 WHEN value < 5 THEN 5 ELSE 7 END AS v FROM range(0, 6)) AS t;
+
+# struct keys, <: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_struct_lt AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l INNER JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l LEFT JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l RIGHT JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l FULL JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT SEMI JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT ANTI JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT SEMI JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT ANTI JOIN struct_right r ON l.k < r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM struct_right r WHERE l.k < r.k) OR l.id = -1 AS INT) AS r_id FROM struct_left l;
+
+statement ok
+CREATE TABLE ref_struct_lt AS SELECT * FROM case_struct_lt;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_struct_lt), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_struct_lt)
+UNION ALL
+(SELECT * FROM ref_struct_lt EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# struct keys, <=: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_struct_le AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l INNER JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l LEFT JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l RIGHT JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l FULL JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT SEMI JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT ANTI JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT SEMI JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT ANTI JOIN struct_right r ON l.k <= r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM struct_right r WHERE l.k <= r.k) OR l.id = -1 AS INT) AS r_id FROM struct_left l;
+
+statement ok
+CREATE TABLE ref_struct_le AS SELECT * FROM case_struct_le;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_struct_le), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_struct_le)
+UNION ALL
+(SELECT * FROM ref_struct_le EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# struct keys, >: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_struct_gt AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l INNER JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l LEFT JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l RIGHT JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l FULL JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT SEMI JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT ANTI JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT SEMI JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT ANTI JOIN struct_right r ON l.k > r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM struct_right r WHERE l.k > r.k) OR l.id = -1 AS INT) AS r_id FROM struct_left l;
+
+statement ok
+CREATE TABLE ref_struct_gt AS SELECT * FROM case_struct_gt;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_struct_gt), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_struct_gt)
+UNION ALL
+(SELECT * FROM ref_struct_gt EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# struct keys, >=: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_struct_ge AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l INNER JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l LEFT JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l RIGHT JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM struct_left l FULL JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT SEMI JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM struct_left l LEFT ANTI JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT SEMI JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM struct_left l RIGHT ANTI JOIN struct_right r ON l.k >= r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM struct_right r WHERE l.k >= r.k) OR l.id = -1 AS INT) AS r_id FROM struct_left l;
+
+statement ok
+CREATE TABLE ref_struct_ge AS SELECT * FROM case_struct_ge;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_struct_ge), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_struct_ge)
+UNION ALL
+(SELECT * FROM ref_struct_ge EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+statement ok
+CREATE TABLE nested_left AS
+SELECT id, CASE WHEN id = 0 THEN NULL ELSE make_array(make_array(v)) END AS k
+FROM (SELECT value AS id, CASE WHEN value < 2 THEN NULL WHEN value = 2 THEN 1 WHEN value < 5 THEN 5 ELSE 7 END AS v FROM range(0, 6)) AS t;
+
+statement ok
+CREATE TABLE nested_right AS
+SELECT id, CASE WHEN id = 0 THEN NULL ELSE make_array(make_array(v)) END AS k
+FROM (SELECT value AS id, CASE WHEN value < 2 THEN NULL WHEN value = 2 THEN 1 WHEN value < 5 THEN 5 ELSE 7 END AS v FROM range(0, 6)) AS t;
+
+# nested keys, <: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_nested_lt AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l INNER JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l LEFT JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l RIGHT JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l FULL JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT SEMI JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT ANTI JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT SEMI JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT ANTI JOIN nested_right r ON l.k < r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM nested_right r WHERE l.k < r.k) OR l.id = -1 AS INT) AS r_id FROM nested_left l;
+
+statement ok
+CREATE TABLE ref_nested_lt AS SELECT * FROM case_nested_lt;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_nested_lt), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_nested_lt)
+UNION ALL
+(SELECT * FROM ref_nested_lt EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# nested keys, <=: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_nested_le AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l INNER JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l LEFT JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l RIGHT JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l FULL JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT SEMI JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT ANTI JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT SEMI JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT ANTI JOIN nested_right r ON l.k <= r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM nested_right r WHERE l.k <= r.k) OR l.id = -1 AS INT) AS r_id FROM nested_left l;
+
+statement ok
+CREATE TABLE ref_nested_le AS SELECT * FROM case_nested_le;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_nested_le), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_nested_le)
+UNION ALL
+(SELECT * FROM ref_nested_le EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# nested keys, >: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_nested_gt AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l INNER JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l LEFT JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l RIGHT JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l FULL JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT SEMI JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT ANTI JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT SEMI JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT ANTI JOIN nested_right r ON l.k > r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM nested_right r WHERE l.k > r.k) OR l.id = -1 AS INT) AS r_id FROM nested_left l;
+
+statement ok
+CREATE TABLE ref_nested_gt AS SELECT * FROM case_nested_gt;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_nested_gt), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_nested_gt)
+UNION ALL
+(SELECT * FROM ref_nested_gt EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+# nested keys, >=: compare all join variants with NestedLoopJoin.
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = false;
+
+statement ok
+CREATE VIEW case_nested_ge AS
+SELECT 'INNER' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l INNER JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l LEFT JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l RIGHT JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'FULL' AS kind, l.id AS l_id, r.id AS r_id FROM nested_left l FULL JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT SEMI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT SEMI JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'LEFT ANTI' AS kind, l.id AS l_id, NULL AS r_id FROM nested_left l LEFT ANTI JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT SEMI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT SEMI JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'RIGHT ANTI' AS kind, NULL AS l_id, r.id AS r_id FROM nested_left l RIGHT ANTI JOIN nested_right r ON l.k >= r.k
+UNION ALL
+SELECT 'MARK' AS kind, l.id AS l_id, CAST(EXISTS (SELECT 1 FROM nested_right r WHERE l.k >= r.k) OR l.id = -1 AS INT) AS r_id FROM nested_left l;
+
+statement ok
+CREATE TABLE ref_nested_ge AS SELECT * FROM case_nested_ge;
+
+statement ok
+SET datafusion.optimizer.enable_piecewise_merge_join = true;
+
+query I
+WITH actual AS (SELECT * FROM case_nested_ge), differences AS (
+(SELECT * FROM actual EXCEPT ALL SELECT * FROM ref_nested_ge)
+UNION ALL
+(SELECT * FROM ref_nested_ge EXCEPT ALL SELECT * FROM actual)
+)
+SELECT COUNT(*) FROM differences;
+----
+0
+
+statement ok
+RESET datafusion.optimizer.enable_piecewise_merge_join;
diff --git a/datafusion/sqllogictest/test_files/pwmj.slt b/datafusion/sqllogictest/test_files/pwmj.slt
index a15f4a3b82..104b38e1a5 100644
--- a/datafusion/sqllogictest/test_files/pwmj.slt
+++ b/datafusion/sqllogictest/test_files/pwmj.slt
@@ -175,7 +175,7 @@ physical_plan
01)SortPreservingMergeExec: [t1_id@0 ASC NULLS LAST, t2_id@1 ASC NULLS LAST]
02)--SortExec: expr=[t1_id@0 ASC NULLS LAST, t2_id@1 ASC NULLS LAST], preserve_partitioning=[true]
03)----PiecewiseMergeJoin: operator=Lt, join_type=Inner, on=(t1_id < t2_id)
-04)------SortExec: expr=[t1_id@0 DESC], preserve_partitioning=[false]
+04)------SortExec: expr=[t1_id@0 DESC NULLS LAST], preserve_partitioning=[false]
05)--------DataSourceExec: partitions=1, partition_sizes=[1]
06)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
07)--------FilterExec: t2_int@1 >= 3, projection=[t2_id@0]
@@ -222,7 +222,7 @@ physical_plan
01)SortPreservingMergeExec: [t1_id@0 ASC NULLS LAST, t2_id@1 ASC NULLS LAST]
02)--SortExec: expr=[t1_id@0 ASC NULLS LAST, t2_id@1 ASC NULLS LAST], preserve_partitioning=[true]
03)----PiecewiseMergeJoin: operator=Lt, join_type=Inner, on=(CAST(t1_id AS Int64) < CAST(t2_id AS Int64) + 1)
-04)------SortExec: expr=[CAST(t1_id@0 AS Int64) DESC], preserve_partitioning=[false]
+04)------SortExec: expr=[CAST(t1_id@0 AS Int64) DESC NULLS LAST], preserve_partitioning=[false]
05)--------DataSourceExec: partitions=1, partition_sizes=[1]
06)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
07)--------FilterExec: t2_int@1 >= 3, projection=[t2_id@0]
@@ -267,7 +267,7 @@ physical_plan
01)SortPreservingMergeExec: [t1_id@0 ASC NULLS LAST, t2_id@1 ASC NULLS LAST]
02)--SortExec: expr=[t1_id@0 ASC NULLS LAST, t2_id@1 ASC NULLS LAST], preserve_partitioning=[true]
03)----PiecewiseMergeJoin: operator=LtEq, join_type=Inner, on=(t1_id <= t2_id)
-04)------SortExec: expr=[t1_id@0 DESC], preserve_partitioning=[false]
+04)------SortExec: expr=[t1_id@0 DESC NULLS LAST], preserve_partitioning=[false]
05)--------FilterExec: t1_id@0 = 11 OR t1_id@0 = 44
06)----------DataSourceExec: partitions=1, partition_sizes=[1]
07)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
@@ -1167,7 +1167,7 @@ physical_plan
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
05)--------ProjectionExec: expr=[id@0 as id, mark@2 as mark]
06)----------PiecewiseMergeJoin: operator=Lt, join_type=LeftMark, on=(v < v)
-07)------------SortExec: expr=[v@1 DESC], preserve_partitioning=[false]
+07)------------SortExec: expr=[v@1 DESC NULLS LAST], preserve_partitioning=[false]
08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
09)------------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -1249,7 +1249,7 @@ physical_plan
04)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
05)--------ProjectionExec: expr=[id@0 as id, mark@2 as mark]
06)----------PiecewiseMergeJoin: operator=LtEq, join_type=LeftMark, on=(v <= v)
-07)------------SortExec: expr=[v@1 DESC], preserve_partitioning=[false]
+07)------------SortExec: expr=[v@1 DESC NULLS LAST], preserve_partitioning=[false]
08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
09)------------DataSourceExec: partitions=1, partition_sizes=[1]
@@ -1301,7 +1301,7 @@ logical_plan
physical_plan
01)SortExec: expr=[x@0 ASC NULLS LAST], preserve_partitioning=[false]
02)--PiecewiseMergeJoin: operator=Lt, join_type=Left, on=(x < y)
-03)----SortExec: expr=[x@0 DESC], preserve_partitioning=[false]
+03)----SortExec: expr=[x@0 DESC NULLS LAST], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]
05)----DataSourceExec: partitions=1, partition_sizes=[1]
@@ -2205,5 +2205,57 @@ physical_plan
04)----SortExec: TopK(fetch=5), expr=[id@0 DESC], preserve_partitioning=[false]
05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Nested range keys execute on PWMJ, including the classic streamed-side sort.
+
+query TT
+EXPLAIN SELECT l.k, r.k
+FROM (SELECT named_struct('x', value) AS k FROM range(0, 3)) l
+JOIN (SELECT named_struct('x', value) AS k FROM range(0, 3)) r ON l.k < r.k;
+----
+logical_plan
+01)Inner Join: Filter: l.k < r.k
+02)--SubqueryAlias: l
+03)----Projection: named_struct(Utf8("x"), range().value) AS k
+04)------TableScan: range() projection=[value]
+05)--SubqueryAlias: r
+06)----Projection: named_struct(Utf8("x"), range().value) AS k
+07)------TableScan: range() projection=[value]
+physical_plan
+01)PiecewiseMergeJoin: operator=Lt, join_type=Inner, on=(k < k)
+02)--SortPreservingMergeExec: [k@0 DESC NULLS LAST]
+03)----SortExec: expr=[k@0 DESC NULLS LAST], preserve_partitioning=[true]
+04)------ProjectionExec: expr=[named_struct(x, value@0) as k]
+05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+06)----------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=3, batch_size=8192]
+07)--ProjectionExec: expr=[named_struct(x, value@0) as k]
+08)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+09)------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=3, batch_size=8192]
+
+
+query TT
+EXPLAIN SELECT l.k, r.k
+FROM (SELECT make_array(make_array(value)) AS k FROM range(0, 3)) l
+JOIN (SELECT make_array(make_array(value)) AS k FROM range(0, 3)) r ON l.k < r.k;
+----
+logical_plan
+01)Inner Join: Filter: l.k < r.k
+02)--SubqueryAlias: l
+03)----Projection: make_array(make_array(range().value)) AS k
+04)------TableScan: range() projection=[value]
+05)--SubqueryAlias: r
+06)----Projection: make_array(make_array(range().value)) AS k
+07)------TableScan: range() projection=[value]
+physical_plan
+01)PiecewiseMergeJoin: operator=Lt, join_type=Inner, on=(k < k)
+02)--SortPreservingMergeExec: [k@0 DESC NULLS LAST]
+03)----SortExec: expr=[k@0 DESC NULLS LAST], preserve_partitioning=[true]
+04)------ProjectionExec: expr=[make_array(make_array(value@0)) as k]
+05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+06)----------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=3, batch_size=8192]
+07)--ProjectionExec: expr=[make_array(make_array(value@0)) as k]
+08)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+09)------LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=3, batch_size=8192]
+
statement ok
set datafusion.optimizer.enable_piecewise_merge_join = false;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org