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/09/11 15:13:47 UTC
(datafusion) branch gh-readonly-queue/main/pr-24457-6f4cac32937e02ed51912bf73dc97f62ef77fa26 created (now 9e1e30d7cf)
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-24457-6f4cac32937e02ed51912bf73dc97f62ef77fa26
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 9e1e30d7cf feat(pwmj): support RightSemi/RightAnti existence joins (#24457)
This branch includes the following new commits:
new 9e1e30d7cf feat(pwmj): support RightSemi/RightAnti existence joins (#24457)
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: feat(pwmj): support RightSemi/RightAnti existence joins (#24457)
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-24457-6f4cac32937e02ed51912bf73dc97f62ef77fa26
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 9e1e30d7cfbacb975472569cedb051e76fb87e37
Author: Subham Singhal <Su...@users.noreply.github.com>
AuthorDate: Fri Sep 11 15:13:31 2026 +0000
feat(pwmj): support RightSemi/RightAnti existence joins (#24457)
## Which issue does this PR close?
Part of #17427 (the PiecewiseMergeJoin epic). This is the second half of
its existence-join item: #23870 added `LeftSemi`/`LeftAnti`, this adds
`RightSemi`/`RightAnti`. Only the Mark joins remain unsupported.
## Rationale for this change
A range-predicate right semi/anti join — `... RIGHT SEMI JOIN r ON l.v >
r.v`, or any query that decorrelates to one — currently falls back to
`NestedLoopJoinExec`.
## What changes are included in this PR?
New `RightExistencePWMJStream` (`right_existence_join.rs`), a 3-state
machine: await the buffered reduction, then per streamed batch evaluate
the key, compare it against the reduced key with a vectorized `cmp`
kernel, and filter. Plus:
* `exec.rs` — dispatch; `build_buffered_extreme`, which folds every
buffered partition on its own `SpawnedTask`; `BufferedExtreme` as the
shared state; relaxed ordering, distribution and partitioning
declarations for these two join types; removal of the dead
`sort_options` flip and the `unimplemented!()` placeholder that the swap
design had left behind.
* `physical_planner.rs` — the gate now excludes only the Mark joins.
* `utils.rs` — `is_supported_existence_join` widened; new
`is_supported_right_existence_join` (which deliberately excludes
`RightMark`, so a Mark join cannot inherit the relaxed input
requirements if that gate is ever loosened); no visited-indices bitmap
allocated for these join types.
* `extreme_key` made `pub(super)` so both existence streams share one
implementation.
## Are these changes tested?
Yes
## Benchmark
Microbenchmark, 20K × 20K rows, single inequality,
`enable_piecewise_merge_join` on vs off
| Case | PWMJ | NestedLoopJoin | Speedup |
|---|---|---|---|
| RightSemi, all match | ~0.023 ms | ~77 ms | ~3,300× |
| RightAnti, all match | ~0.024 ms | ~77 ms | ~3,200× |
| RightSemi, no match | ~0.023 ms | ~78 ms | ~3,300× |
| RightAnti, no match | ~0.024 ms | ~78 ms | ~3,300× |
| RightSemi, half match | ~0.036 ms | ~78 ms | ~2,200× |
| RightAnti, half match | ~0.036 ms | ~77 ms | ~2,100× |
## Are there any user-facing changes?
No
---------
Co-authored-by: Oleks V <co...@users.noreply.github.com>
---
datafusion/core/src/physical_planner.rs | 53 +-
datafusion/core/tests/fuzz_cases/join_fuzz.rs | 188 +++-
.../physical-plan/src/joins/nested_loop_join.rs | 18 +-
.../src/joins/piecewise_merge_join/exec.rs | 461 ++++++--
.../joins/piecewise_merge_join/existence_join.rs | 23 +-
.../src/joins/piecewise_merge_join/mod.rs | 1 +
.../piecewise_merge_join/right_existence_join.rs | 1177 ++++++++++++++++++++
.../src/joins/piecewise_merge_join/utils.rs | 28 +-
datafusion/physical-plan/src/joins/utils.rs | 14 +
datafusion/proto/tests/cases/plans/joins.rs | 42 +-
datafusion/sqllogictest/test_files/pwmj.slt | 798 +++++++++++++
11 files changed, 2570 insertions(+), 233 deletions(-)
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index f00a1b555a..77997c619e 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -1611,16 +1611,13 @@ impl DefaultPhysicalPlanner {
Arc::new(CrossJoinExec::new(physical_left, physical_right))
} else if num_range_filters == 1
&& total_filters == 1
- // PWMJ supports classic joins and Left Semi/Anti existence joins.
- // Right Semi/Anti and Mark joins are not implemented yet (they
- // would require swapping the inputs so the marked side is buffered),
- // so exclude them here and let them fall back to NestedLoopJoin.
+ // PWMJ supports classic joins and Semi/Anti existence joins. Mark
+ // joins are not implemented yet (they need an extra boolean column
+ // rather than a subset of one side's rows), so exclude them here
+ // and let them fall back to NestedLoopJoin.
&& !matches!(
join_type,
- JoinType::RightSemi
- | JoinType::RightAnti
- | JoinType::LeftMark
- | JoinType::RightMark
+ JoinType::LeftMark | JoinType::RightMark
)
&& session_state
.config_options()
@@ -1654,14 +1651,20 @@ impl DefaultPhysicalPlanner {
}
}
+ // `Neither` covers an operand that references no column from
+ // either side (e.g. a literal), and `Both` an operand that
+ // references columns from both. PWMJ needs one operand pinned
+ // to each side, so both fall back to NestedLoopJoin below rather
+ // than erroring or (for `Neither`) panicking.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Side {
Left,
Right,
Both,
+ Neither,
}
- let side_of = |e: &Expr| -> Result<Side> {
+ let side_of = |e: &Expr| -> Side {
let cols = e.column_refs();
let any_left = cols
.iter()
@@ -1670,20 +1673,29 @@ impl DefaultPhysicalPlanner {
.iter()
.any(|c| right_df_schema.index_of_column(c).is_ok());
- Ok(match (any_left, any_right) {
+ match (any_left, any_right) {
(true, false) => Side::Left,
(false, true) => Side::Right,
(true, true) => Side::Both,
- _ => unreachable!(),
- })
+ (false, false) => Side::Neither,
+ }
};
let mut lhs_logical = &be.left;
let mut rhs_logical = &be.right;
- let left_side = side_of(lhs_logical)?;
- let right_side = side_of(rhs_logical)?;
- if left_side == Side::Both || right_side == Side::Both {
+ let left_side = side_of(lhs_logical);
+ let right_side = side_of(rhs_logical);
+
+ if left_side == Side::Right && right_side == Side::Left {
+ std::mem::swap(&mut lhs_logical, &mut rhs_logical);
+ op = reverse_ineq(op);
+ } else if !(left_side == Side::Left && right_side == Side::Right)
+ {
+ // Anything other than a clean left/right split -- both
+ // operands on one side, one referencing neither side, or
+ // either referencing both -- isn't a range predicate PWMJ
+ // can plan, so let NestedLoopJoin evaluate it instead.
return Ok(Arc::new(NestedLoopJoinExec::try_new(
physical_left,
physical_right,
@@ -1693,17 +1705,6 @@ impl DefaultPhysicalPlanner {
)?));
}
- if left_side == Side::Right && right_side == Side::Left {
- std::mem::swap(&mut lhs_logical, &mut rhs_logical);
- op = reverse_ineq(op);
- } else if !(left_side == Side::Left && right_side == Side::Right)
- {
- return plan_err!(
- "Unsupported operator for PWMJ: {:?}. Expected one of <, <=, >, >=",
- op
- );
- }
-
let on_left = create_physical_expr(
lhs_logical,
left_df_schema,
diff --git a/datafusion/core/tests/fuzz_cases/join_fuzz.rs b/datafusion/core/tests/fuzz_cases/join_fuzz.rs
index cdd550013b..91a26d083d 100644
--- a/datafusion/core/tests/fuzz_cases/join_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/join_fuzz.rs
@@ -20,9 +20,9 @@ use std::time::SystemTime;
use crate::fuzz_cases::join_fuzz::JoinTestType::{HjSmj, NljHj};
-use arrow::array::{Array, ArrayRef, BinaryArray, Int32Array};
+use arrow::array::{Array, ArrayRef, BinaryArray, Float64Array, Int32Array};
use arrow::compute::SortOptions;
-use arrow::datatypes::Schema;
+use arrow::datatypes::{DataType, Schema};
use arrow::record_batch::RecordBatch;
use arrow::util::pretty::pretty_format_batches;
use datafusion::common::JoinSide;
@@ -1374,40 +1374,98 @@ fn make_staggered_batches_binary(
// streamed side below is spread round-robin over several partitions as one-row batches, so
// batches arrive in an order no static test pins down, and the counter that gates the final
// pass is seeded from a partition count that deliberately disagrees with the `num_partitions`
-// argument.
+// argument. `RightSemi`/`RightAnti` take the mirror path -- every partition emits its own
+// rows, decided against a single buffered key -- and are covered here too, over the same
+// inputs, so the two halves are held to the same oracle. Their buffered side is fanned out as
+// well, since it carries no single-partition requirement: each partition is folded to its own
+// extreme on a separate task and those are then combined, and only randomization varies which
+// partition holds the deciding key, or holds none at all.
+
+/// A key type the differential fuzz test can generate and array-ify. `i32` covers the plain
+/// ordering/duplicate/NULL dimensions; `f64` adds `-0.0` and `NaN`, whose comparison semantics
+/// arrow's `cmp` kernels and its `min`/`max` accumulators must agree on for
+/// `PiecewiseMergeJoinExec`'s buffered-side reduction to be sound (see `right_existence_join`'s
+/// module doc).
+trait PwmjFuzzKey: Copy + 'static {
+ fn data_type() -> DataType;
+ fn to_array(values: &[Option<Self>]) -> ArrayRef;
+ /// Draws a value for the narrow `[0, key_range)` range the rest of the test relies on to
+ /// force duplicates and equal-boundary cases, plus (for floats) the values a uniform draw
+ /// over that range would never produce on its own.
+ fn gen_value(key_range: i32, rng: &mut StdRng) -> Self;
+}
+
+impl PwmjFuzzKey for i32 {
+ fn data_type() -> DataType {
+ DataType::Int32
+ }
+
+ fn to_array(values: &[Option<Self>]) -> ArrayRef {
+ Arc::new(Int32Array::from(values.to_vec()))
+ }
+
+ fn gen_value(key_range: i32, rng: &mut StdRng) -> Self {
+ rng.random_range(0..key_range)
+ }
+}
+
+impl PwmjFuzzKey for f64 {
+ fn data_type() -> DataType {
+ DataType::Float64
+ }
-fn pwmj_kv_schema() -> Arc<Schema> {
+ fn to_array(values: &[Option<Self>]) -> ArrayRef {
+ Arc::new(Float64Array::from(values.to_vec()))
+ }
+
+ fn gen_value(key_range: i32, rng: &mut StdRng) -> Self {
+ // One in 8 draws hits `-0.0` or `NaN` instead of the uniform range: both must compare
+ // consistently between the `cmp` kernel used to filter the streamed side and the
+ // `min`/`max` accumulators used to reduce the buffered side, and a plain uniform draw
+ // over `[0, key_range)` would essentially never produce either on its own.
+ match rng.random_range(0..8u32) {
+ 0 => -0.0,
+ 1 => f64::NAN,
+ _ => f64::from(rng.random_range(0..key_range)),
+ }
+ }
+}
+
+fn pwmj_kv_schema<K: PwmjFuzzKey>() -> Arc<Schema> {
Arc::new(Schema::new(vec![
- arrow::datatypes::Field::new("id", arrow::datatypes::DataType::Int32, false),
- arrow::datatypes::Field::new("k", arrow::datatypes::DataType::Int32, true),
+ arrow::datatypes::Field::new("id", DataType::Int32, false),
+ arrow::datatypes::Field::new("k", K::data_type(), true),
]))
}
-fn pwmj_kv_batch(ids: &[i32], keys: &[Option<i32>]) -> RecordBatch {
+fn pwmj_kv_batch<K: PwmjFuzzKey>(ids: &[i32], keys: &[Option<K>]) -> RecordBatch {
RecordBatch::try_new(
- pwmj_kv_schema(),
- vec![
- Arc::new(Int32Array::from(ids.to_vec())),
- Arc::new(Int32Array::from(keys.to_vec())),
- ],
+ pwmj_kv_schema::<K>(),
+ vec![Arc::new(Int32Array::from(ids.to_vec())), K::to_array(keys)],
)
.unwrap()
}
/// Single-partition, single-batch input: the buffered side, and the oracle's probe side.
-fn pwmj_single_exec(ids: &[i32], keys: &[Option<i32>]) -> Arc<dyn ExecutionPlan> {
+fn pwmj_single_exec<K: PwmjFuzzKey>(
+ ids: &[i32],
+ keys: &[Option<K>],
+) -> Arc<dyn ExecutionPlan> {
MemorySourceConfig::try_new_exec(
&[vec![pwmj_kv_batch(ids, keys)]],
- pwmj_kv_schema(),
+ pwmj_kv_schema::<K>(),
None,
)
.unwrap()
}
-/// Streamed side spread round-robin across `nparts` partitions, one row per batch.
-fn pwmj_parts_exec(
+/// Rows spread round-robin across `nparts` partitions, one row per batch, with any partition
+/// that draws no row left holding a single empty batch. Used for the streamed side throughout,
+/// and for the buffered side of the right existence joins, which place no single-partition
+/// requirement on it.
+fn pwmj_parts_exec<K: PwmjFuzzKey>(
ids: &[i32],
- keys: &[Option<i32>],
+ keys: &[Option<K>],
nparts: usize,
) -> Arc<dyn ExecutionPlan> {
let nparts = nparts.max(1);
@@ -1417,10 +1475,10 @@ fn pwmj_parts_exec(
}
for p in partitions.iter_mut() {
if p.is_empty() {
- p.push(pwmj_kv_batch(&[], &[]));
+ p.push(pwmj_kv_batch::<K>(&[], &[]));
}
}
- MemorySourceConfig::try_new_exec(&partitions, pwmj_kv_schema(), None).unwrap()
+ MemorySourceConfig::try_new_exec(&partitions, pwmj_kv_schema::<K>(), None).unwrap()
}
fn pwmj_plan(
@@ -1430,37 +1488,43 @@ fn pwmj_plan(
join_type: JoinType,
) -> Arc<dyn ExecutionPlan> {
// Matches `PiecewiseMergeJoinExec::required_input_ordering`: descending for `<`/`<=`,
- // ascending for `>`/`>=`, NULLs first either way.
- let sort_options = match op {
- Operator::Lt | Operator::LtEq => SortOptions::new(true, true),
- Operator::Gt | Operator::GtEq => SortOptions::new(false, true),
- other => panic!("not a range operator: {other:?}"),
+ // ascending for `>`/`>=`, NULLs first either way. 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 => left,
+ _ => {
+ let sort_options = match op {
+ Operator::Lt | Operator::LtEq => SortOptions::new(true, true),
+ Operator::Gt | Operator::GtEq => SortOptions::new(false, true),
+ other => panic!("not a range operator: {other:?}"),
+ };
+ let ordering = LexOrdering::new(vec![PhysicalSortExpr::new(
+ Arc::new(Column::new("k", 1)),
+ sort_options,
+ )])
+ .unwrap();
+ Arc::new(SortExec::new(ordering, left))
+ }
};
- let ordering = LexOrdering::new(vec![PhysicalSortExpr::new(
- Arc::new(Column::new("k", 1)),
- sort_options,
- )])
- .unwrap();
- let sorted_left = Arc::new(SortExec::new(ordering, left));
let on: (PhysicalExprRef, PhysicalExprRef) =
(Arc::new(Column::new("k", 1)), Arc::new(Column::new("k", 1)));
// `num_partitions` is 1 while the streamed side has up to 3: the final-pass counter must
// come from the streamed side's partition count, not from this argument.
Arc::new(
- PiecewiseMergeJoinExec::try_new(sorted_left, right, on, op, join_type, 1)
- .unwrap(),
+ PiecewiseMergeJoinExec::try_new(buffered, right, on, op, join_type, 1).unwrap(),
)
}
-fn pwmj_nlj_oracle_plan(
+fn pwmj_nlj_oracle_plan<K: PwmjFuzzKey>(
left: Arc<dyn ExecutionPlan>,
right: Arc<dyn ExecutionPlan>,
op: Operator,
join_type: JoinType,
) -> Arc<dyn ExecutionPlan> {
let intermediate_schema = Schema::new(vec![
- arrow::datatypes::Field::new("k", arrow::datatypes::DataType::Int32, true),
- arrow::datatypes::Field::new("k", arrow::datatypes::DataType::Int32, true),
+ arrow::datatypes::Field::new("k", K::data_type(), true),
+ arrow::datatypes::Field::new("k", K::data_type(), true),
]);
let expr = Arc::new(BinaryExpr::new(
Arc::new(Column::new("k", 0)),
@@ -1486,7 +1550,10 @@ fn pwmj_nlj_oracle_plan(
/// Executes every output partition concurrently and returns the join's rows as
/// `(left id, right id)` pairs, sorted so partition interleaving does not affect the
/// comparison. `None` means the join filled that side with NULLs, or -- for the existence
-/// joins, whose output carries the left side only -- that the side is absent entirely.
+/// joins, whose output carries only the surviving side -- that the other side is absent
+/// entirely. Both halves of an existence join output that surviving `id` as their first
+/// column: the left side's for `LeftSemi`/`LeftAnti`, the right side's for
+/// `RightSemi`/`RightAnti`.
///
/// Concurrent rather than one partition at a time: the partitions share the watermark and race
/// to be the one that runs the final pass, which is the part a sequential drain cannot reach.
@@ -1534,7 +1601,8 @@ async fn pwmj_collect_id_pairs(
}
/// Differential test for every join type `PiecewiseMergeJoin` supports, against a
-/// `NestedLoopJoin` oracle.
+/// `NestedLoopJoin` oracle, run once over `i32` keys and once over `f64` keys via
+/// [`run_pwmj_fuzz`].
///
/// `Left`/`Full` are the ones with teeth: their unmatched buffered rows are derived from the
/// shared `min_marked` watermark rather than materialized per row, and that encoding is only
@@ -1542,7 +1610,8 @@ async fn pwmj_collect_id_pairs(
/// cheaper tests do not reach are the ones that matter here -- `pwmj.slt` and the unit tests
/// both run the streamed side at a single partition and the default batch size, so neither
/// covers several partitions racing to run the final pass, nor the mid-scan resume path a
-/// small batch size forces.
+/// small batch size forces. The `f64` pass adds `-0.0` and `NaN`, which `pwmj.slt` only
+/// exercises via a handful of fixed rows.
#[tokio::test(flavor = "multi_thread")]
async fn fuzz_pwmj_matches_nested_loop() {
// A small batch size splits output across several coalesced batches even for these tiny
@@ -1551,6 +1620,13 @@ async fn fuzz_pwmj_matches_nested_loop() {
TaskContext::default()
.with_session_config(SessionConfig::new().with_batch_size(3)),
);
+ // `f64` keys run every dimension below a second time with `-0.0` and `NaN` mixed in, on
+ // top of what `i32` already covers with plain duplicates and NULLs.
+ run_pwmj_fuzz::<i32>(&task_ctx).await;
+ run_pwmj_fuzz::<f64>(&task_ctx).await;
+}
+
+async fn run_pwmj_fuzz<K: PwmjFuzzKey + std::fmt::Debug>(task_ctx: &Arc<TaskContext>) {
let ops = [Operator::Lt, Operator::LtEq, Operator::Gt, Operator::GtEq];
let join_types = [
JoinType::Inner,
@@ -1559,6 +1635,8 @@ async fn fuzz_pwmj_matches_nested_loop() {
JoinType::Full,
JoinType::LeftSemi,
JoinType::LeftAnti,
+ JoinType::RightSemi,
+ JoinType::RightAnti,
];
for seed in 0..60u64 {
@@ -1568,10 +1646,14 @@ async fn fuzz_pwmj_matches_nested_loop() {
// A narrow key range forces duplicates and equal-boundary cases.
let key_range = rng.random_range(1..6i32);
let nparts = rng.random_range(1..4usize);
+ // Independent of the streamed side's, so the two counts disagree across seeds and the
+ // 1-partition buffered case is still drawn. Only the right existence joins can use it:
+ // the others require the buffered side coalesced and globally sorted.
+ let buffered_nparts = rng.random_range(1..4usize);
- let gen_keys = |n: usize, rng: &mut StdRng| -> Vec<Option<i32>> {
+ let gen_keys = |n: usize, rng: &mut StdRng| -> Vec<Option<K>> {
(0..n)
- .map(|_| (!rng.random_bool(0.2)).then(|| rng.random_range(0..key_range)))
+ .map(|_| (!rng.random_bool(0.2)).then(|| K::gen_value(key_range, rng)))
.collect()
};
@@ -1582,31 +1664,43 @@ async fn fuzz_pwmj_matches_nested_loop() {
for op in ops {
for join_type in join_types {
+ // Fanned out only for the right existence joins, whose buffered side is
+ // folded one partition per task and then combined -- a reduction the
+ // single-partition shape below never reaches.
+ let buffered = match join_type {
+ JoinType::RightSemi | JoinType::RightAnti => {
+ pwmj_parts_exec(&left_ids, &left_keys, buffered_nparts)
+ }
+ _ => pwmj_single_exec(&left_ids, &left_keys),
+ };
let got = pwmj_collect_id_pairs(
pwmj_plan(
- pwmj_single_exec(&left_ids, &left_keys),
+ buffered,
pwmj_parts_exec(&right_ids, &right_keys, nparts),
op,
join_type,
),
- Arc::clone(&task_ctx),
+ Arc::clone(task_ctx),
)
.await;
let want = pwmj_collect_id_pairs(
- pwmj_nlj_oracle_plan(
+ pwmj_nlj_oracle_plan::<K>(
pwmj_single_exec(&left_ids, &left_keys),
pwmj_single_exec(&right_ids, &right_keys),
op,
join_type,
),
- Arc::clone(&task_ctx),
+ Arc::clone(task_ctx),
)
.await;
assert_eq!(
- got, want,
- "mismatch seed={seed} op={op:?} join_type={join_type:?} \
- nparts={nparts} left_keys={left_keys:?} right_keys={right_keys:?}"
+ got,
+ want,
+ "mismatch key_type={} seed={seed} op={op:?} join_type={join_type:?} \
+ nparts={nparts} buffered_nparts={buffered_nparts} \
+ left_keys={left_keys:?} right_keys={right_keys:?}",
+ std::any::type_name::<K>(),
);
}
}
diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs
index 2fe48df474..462576b2b4 100644
--- a/datafusion/physical-plan/src/joins/nested_loop_join.rs
+++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs
@@ -32,8 +32,8 @@ use crate::execution_plan::{EmissionType, boundedness_from_children};
use crate::joins::SharedBitmapBuilder;
use crate::joins::utils::{
BuildProbeJoinMetrics, ColumnIndex, JoinFilter, OnceAsync, OnceFut,
- build_join_schema, check_join_is_valid, estimate_join_statistics,
- need_produce_right_in_final,
+ boolean_mask_from_filter, build_join_schema, check_join_is_valid,
+ estimate_join_statistics, need_produce_right_in_final,
};
use crate::metrics::{
Count, ExecutionPlanMetricsSet, MetricBuilder, MetricType, MetricsSet, RatioMetrics,
@@ -3798,20 +3798,6 @@ fn apply_filter_to_row_join_batch(
Ok(bitmap_combined)
}
-/// Convert a boolean filter array into a unified mask bitmap.
-///
-/// Caution: The filter result is NOT a bitmap; it contains true/false/null values.
-/// For example, `1 < NULL` evaluates to NULL. Therefore, we must combine (AND)
-/// the boolean array with its null bitmap to construct a unified bitmap.
-#[inline]
-fn boolean_mask_from_filter(filter_arr: &BooleanArray) -> BooleanArray {
- let (values, nulls) = filter_arr.clone().into_parts();
- match nulls {
- Some(nulls) => BooleanArray::new(nulls.inner() & &values, None),
- None => BooleanArray::new(values, None),
- }
-}
-
/// This function performs the following steps:
/// 1. Apply filter to probe-side batch
/// 2. Broadcast the left row (build_side_batch\[build_side_index\]) to the
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 efaf051f7a..07454dc50d 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs
@@ -18,15 +18,16 @@
use arrow::array::Array;
use arrow::{
array::{ArrayRef, RecordBatch},
- compute::concat_batches,
+ compute::{concat, concat_batches},
};
use arrow_schema::{SchemaRef, SortOptions};
use datafusion_common::not_impl_err;
use datafusion_common::tree_node::TreeNodeRecursion;
-use datafusion_common::{JoinSide, Result, internal_err};
+use datafusion_common::{JoinSide, Result, internal_datafusion_err, internal_err};
+use datafusion_common_runtime::SpawnedTask;
use datafusion_execution::{
SendableRecordBatchStream,
- memory_pool::{MemoryConsumer, MemoryReservation},
+ memory_pool::{MemoryConsumer, MemoryPool, MemoryReservation},
};
use datafusion_expr::{JoinType, Operator};
use datafusion_physical_expr::equivalence::join_equivalence_properties;
@@ -35,7 +36,7 @@ use datafusion_physical_expr::{
PhysicalSortExpr,
};
use datafusion_physical_expr_common::physical_expr::fmt_sql;
-use futures::TryStreamExt;
+use futures::{StreamExt, TryStreamExt};
use std::fmt::Formatter;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
@@ -45,9 +46,12 @@ use crate::execution_plan::{EmissionType, boundedness_from_children};
use crate::joins::piecewise_merge_join::classic_join::{
ClassicPWMJStream, PiecewiseMergeJoinStreamState,
};
-use crate::joins::piecewise_merge_join::existence_join::ExistencePWMJStream;
+use crate::joins::piecewise_merge_join::existence_join::{
+ ExistencePWMJStream, extreme_key,
+};
+use crate::joins::piecewise_merge_join::right_existence_join::RightExistencePWMJStream;
use crate::joins::piecewise_merge_join::utils::{
- is_existence_join, is_right_existence_join, is_supported_existence_join,
+ is_existence_join, is_supported_existence_join, is_supported_right_existence_join,
};
use crate::joins::utils::asymmetric_join_output_partitioning;
use crate::metrics::MetricsSet;
@@ -160,12 +164,11 @@ use crate::{
/// ```
///
/// ## Existence Joins (Semi, Anti, Mark)
-/// Currently only `LeftSemi` and `LeftAnti` are supported. For these the marked side is
-/// already the left (buffered) side, so no input swap is needed. The rest are rejected in
-/// [`Self::try_new`]: `RightSemi`/`RightAnti`/`RightMark` mark the right side and need an
-/// input swap, and `LeftMark` needs an extra boolean column rather than a filtered slice.
+/// Every Semi/Anti join is supported; the Mark joins are rejected in [`Self::try_new`], as they
+/// need an extra boolean column rather than a subset of one side's rows. The two sides are
+/// served by different streams, because a single range predicate makes them different problems.
///
-/// `LeftSemi`/`LeftAnti` are served by a dedicated stream, `ExistencePWMJStream` (see
+/// `LeftSemi`/`LeftAnti` mark the buffered (left) side, which is `ExistencePWMJStream` (see
/// `existence_join.rs`). Instead of materializing row pairs it records the matched set as a
/// single index -- the start of the matched suffix of the buffered side -- and slices the
/// buffered batch at that index once every streamed partition has been consumed.
@@ -210,8 +213,26 @@ use crate::{
/// min value: 200
/// ```
///
-/// For both types of joins, the buffered side must be sorted ascending for `Operator::Lt` (<) or
-/// `Operator::LtEq` (<=) and descending for `Operator::Gt` (>) or `Operator::GtEq` (>=).
+/// `RightSemi`/`RightAnti` mark the streamed (right) side, which is
+/// `RightExistencePWMJStream` (see `right_existence_join.rs`). Asking whether any buffered
+/// row matches a given streamed row is, for a single range predicate, decided by one buffered
+/// key -- the minimum for `<`/`<=`, the maximum for `>`/`>=`. So the buffered side is folded
+/// down to that key as it arrives and never materialized, and each streamed batch is compared
+/// against it, filtered, and emitted straight away rather than at the end. These are the only
+/// join types here that require no ordered input and hold `O(1)` state.
+///
+/// ```text
+/// // Using the example of a less than `<` operation
+/// let min = min_batch(buffered) // folded per batch, nothing retained
+///
+/// for stream_row in stream_batch:
+/// if min < stream_row: // some buffered row matches
+/// output stream_row
+/// ```
+///
+/// Except for `RightSemi`/`RightAnti`, the buffered side must be sorted ascending for
+/// `Operator::Lt` (<) or `Operator::LtEq` (<=) and descending for `Operator::Gt` (>) or
+/// `Operator::GtEq` (>=).
///
/// # Partitioning Logic
/// Piecewise Merge Join requires one buffered side partition + round robin partitioned stream side. A counter
@@ -219,6 +240,11 @@ use crate::{
/// for processing the rest of the unmatched rows for Left and Full joins. The last partition that finishes
/// execution will be responsible for outputting the unmatched rows.
///
+/// `RightSemi`/`RightAnti` need no such coordination: each streamed row is decided on its own,
+/// so every partition emits its own output and none has a final pass to run. They also place no
+/// single-partition requirement on the buffered side -- a min/max combines across partitions, so
+/// each is folded on its own task rather than funnelled through a `CoalescePartitionsExec`.
+///
/// # Performance Explanation (cost)
/// Piecewise Merge Join is used over Nested Loop Join due to its superior performance. Here is the breakdown:
///
@@ -260,8 +286,12 @@ pub struct PiecewiseMergeJoinExec {
pub join_type: JoinType,
/// The schema once the join is applied
schema: SchemaRef,
- /// Buffered data
+ /// Buffered data, collected once and shared by every streamed partition. Unused by right
+ /// existence joins, which take `buffered_extreme_fut` instead.
buffered_fut: OnceAsync<BufferedSideData>,
+ /// Right existence joins only: the buffered side folded down to one key, so that side is
+ /// never materialized. Unused by every other join type.
+ buffered_extreme_fut: OnceAsync<BufferedExtreme>,
/// Execution metrics
metrics: ExecutionPlanMetricsSet,
@@ -290,9 +320,8 @@ impl PiecewiseMergeJoinExec {
join_type: JoinType,
num_partitions: usize,
) -> Result<Self> {
- // Left Semi/Anti are handled by `ExistencePWMJStream` (the marked side is
- // already the buffered side, so no input swap is needed). Right existence joins
- // and Mark joins are not yet supported.
+ // Semi/Anti joins are handled by the existence streams; Mark joins are not
+ // supported yet.
if is_existence_join(join_type) && !is_supported_existence_join(join_type) {
return not_impl_err!(
"Existence join {join_type} is currently not supported for PiecewiseMergeJoin"
@@ -302,22 +331,8 @@ 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 => {
- // For left existence joins the inputs will be swapped so the sort
- // options are switched
- if is_right_existence_join(join_type) {
- SortOptions::new(false, true)
- } else {
- SortOptions::new(true, true)
- }
- }
- Operator::Gt | Operator::GtEq => {
- if is_right_existence_join(join_type) {
- SortOptions::new(true, true)
- } else {
- SortOptions::new(false, true)
- }
- }
+ Operator::Lt | Operator::LtEq => SortOptions::new(true, true),
+ Operator::Gt | Operator::GtEq => SortOptions::new(false, true),
_ => {
return internal_err!(
"Cannot contain non-range operator in PiecewiseMergeJoinExec"
@@ -367,6 +382,7 @@ impl PiecewiseMergeJoinExec {
join_type,
schema,
buffered_fut: Default::default(),
+ buffered_extreme_fut: Default::default(),
metrics: ExecutionPlanMetricsSet::new(),
left_child_plan_required_order,
right_batch_required_orders,
@@ -450,13 +466,20 @@ impl PiecewiseMergeJoinExec {
// more testing.
fn maintains_input_order(join_type: JoinType) -> Vec<bool> {
match join_type {
- // The existence side is expected to come in sorted
+ // One output batch per streamed batch, in arrival order, with rows only ever
+ // *removed* by a filter, so each output partition keeps its streamed partition's
+ // order. The buffered side contributes no output column, hence `false` there.
+ // `RightMark` is excluded: it adds a `mark` column, so its orderings would not
+ // map across unchanged.
+ JoinType::RightSemi | JoinType::RightAnti => vec![false, true],
+ // Unlike the right existence joins above, output here is gated on a watermark
+ // shared across every streamed partition (see `ExistencePWMJStream`) and emitted
+ // only from whichever partition finishes last, so no streamed partition's output
+ // order tracks its input order.
JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
vec![false, false]
}
- JoinType::RightSemi | JoinType::RightAnti | JoinType::RightMark => {
- vec![false, false]
- }
+ JoinType::RightMark => vec![false, false],
// Left, Right, Full, Inner Join is not guaranteed to maintain
// input order as the streamed side will be sorted during
// execution for `PiecewiseMergeJoin`
@@ -468,6 +491,38 @@ impl PiecewiseMergeJoinExec {
pub fn swap_inputs(&self) -> Result<Arc<dyn ExecutionPlan>> {
todo!()
}
+
+ /// Sets up the buffered-side collection used by the classic and left existence streams:
+ /// buffered partition 0 is executed here, and folded into one sorted batch when the
+ /// returned future is first polled.
+ ///
+ /// Right existence joins use `buffered_extreme_fut` instead, consuming every buffered
+ /// partition themselves.
+ fn buffered_side(
+ &self,
+ context: &Arc<datafusion_execution::TaskContext>,
+ on_buffered: &PhysicalExprRef,
+ metrics: &BuildProbeJoinMetrics,
+ streamed_partitions: usize,
+ ) -> Result<BufferedSide> {
+ let buffered_fut = self.buffered_fut.try_once(|| {
+ let reservation = MemoryConsumer::new("PiecewiseMergeJoinInput")
+ .register(context.memory_pool());
+
+ let buffered_stream = self.buffered.execute(0, Arc::clone(context))?;
+ Ok(build_buffered_data(
+ buffered_stream,
+ Arc::clone(on_buffered),
+ metrics.clone(),
+ reservation,
+ streamed_partitions,
+ ))
+ })?;
+
+ Ok(BufferedSide::Initial(BufferedSideInitialState {
+ buffered_fut,
+ }))
+ }
}
impl ExecutionPlan for PiecewiseMergeJoinExec {
@@ -496,23 +551,52 @@ impl ExecutionPlan for PiecewiseMergeJoinExec {
}
fn input_distribution_requirements(&self) -> crate::InputDistributionRequirements {
+ // Right existence joins reduce the buffered side to a min/max, which combines across
+ // partitions, so that side keeps whatever parallelism the plan gave it -- no
+ // `CoalescePartitionsExec` funnelling every buffered row through one thread. Every
+ // other join type walks the buffered side as a single sorted run and does need it.
+ let buffered = if is_supported_right_existence_join(self.join_type) {
+ Distribution::UnspecifiedDistribution
+ } else {
+ Distribution::SinglePartition
+ };
crate::InputDistributionRequirements::new(vec![
- Distribution::SinglePartition,
+ buffered,
Distribution::UnspecifiedDistribution,
])
}
+ fn benefits_from_input_partitioning(&self) -> Vec<bool> {
+ // Derived exactly as the default does, from this operator's own distribution
+ // requirements, so the two cannot drift apart as those change.
+ let mut benefits: Vec<bool> = self
+ .input_distribution_requirements()
+ .per_child_distributions()
+ .map(|dist| !matches!(dist, Distribution::SinglePartition))
+ .collect();
+
+ // One deviation: right existence joins ask for `UnspecifiedDistribution` on the buffered
+ // side, which that rule reads as "worth fanning out". Folding a batch is one linear scan,
+ // which does not pay for a channel hop, so decline the round-robin `RepartitionExec`.
+ if is_supported_right_existence_join(self.join_type)
+ && let Some(buffered) = benefits.first_mut()
+ {
+ *buffered = false;
+ }
+
+ benefits
+ }
+
fn required_input_ordering(&self) -> Vec<Option<OrderingRequirements>> {
- // Existence joins don't need to be sorted on one side.
- if is_right_existence_join(self.join_type) {
- // Unreachable: `try_new` rejects right existence joins, and this signature
- // cannot return `Result`. They swap the inputs, so whoever implements them
- // must require the order on the streamed side instead.
- unimplemented!(
- "required_input_ordering for right existence joins; guarded by try_new"
- )
+ // Right existence joins read nothing but a single min/max off the buffered side, which
+ // is `O(B)` from any order, so they require none. Every other join type walks the
+ // buffered side in order and does.
+ //
+ // The streamed side never carries a requirement: the classic and left existence
+ // streams sort each batch in memory, and the right existence stream needs no order.
+ if is_supported_right_existence_join(self.join_type) {
+ vec![None, None]
} else {
- // Sort the right side in memory, so we do not need to enforce any sorting
vec![
Some(OrderingRequirements::from(
self.left_child_plan_required_order.clone(),
@@ -550,6 +634,7 @@ impl ExecutionPlan for PiecewiseMergeJoinExec {
// Re-set state.
metrics: ExecutionPlanMetricsSet::new(),
buffered_fut: Default::default(),
+ buffered_extreme_fut: Default::default(),
}))
}
ChildrenPropertiesMode::Recompute => match &children[..] {
@@ -615,58 +700,99 @@ impl ExecutionPlan for PiecewiseMergeJoinExec {
// has a single partition), otherwise the counter never reaches 1 and the final
// pass is skipped.
let streamed_partitions = self.streamed.output_partitioning().partition_count();
- let buffered_fut = self.buffered_fut.try_once(|| {
- let reservation = MemoryConsumer::new("PiecewiseMergeJoinInput")
- .register(context.memory_pool());
-
- let buffered_stream = self.buffered.execute(0, Arc::clone(&context))?;
- Ok(build_buffered_data(
- buffered_stream,
- Arc::clone(&on_buffered),
- metrics.clone(),
- reservation,
- streamed_partitions,
- ))
- })?;
-
- let streamed = self.streamed.execute(partition, Arc::clone(&context))?;
let batch_size = context.session_config().batch_size();
-
- let buffered_side =
- BufferedSide::Initial(BufferedSideInitialState { buffered_fut });
-
- if is_supported_existence_join(self.join_type) {
- Ok(Box::pin(ExistencePWMJStream::try_new(
- Arc::clone(&self.schema),
- on_streamed,
- self.join_type,
- self.operator,
- streamed,
- buffered_side,
- self.sort_options,
- metrics,
- batch_size,
- )))
- } else if is_existence_join(self.join_type) {
- // Right existence joins and Mark joins are rejected in `try_new`.
- internal_err!(
+ match self.join_type {
+ // Right existence joins never read a buffered *row*, only a single min/max over
+ // the whole side, so they fold the buffered input away as it arrives instead of
+ // collecting it.
+ JoinType::RightSemi | JoinType::RightAnti => {
+ // `∃b. b < s` is decided by the smallest buffered key, `∃b. b > s` by the
+ // largest, so the operator alone picks the extreme.
+ let descending = matches!(self.operator, Operator::Gt | Operator::GtEq);
+ let extreme_fut = self.buffered_extreme_fut.try_once(|| {
+ let reservation =
+ MemoryConsumer::new("PiecewiseMergeJoinBufferedExtreme")
+ .register(context.memory_pool());
+
+ // Every buffered partition, not just partition 0: this join type does not
+ // require the buffered side coalesced, so it must consume all of it.
+ let buffered_partitions =
+ self.buffered.output_partitioning().partition_count();
+ let buffered_streams = (0..buffered_partitions)
+ .map(|p| self.buffered.execute(p, Arc::clone(&context)))
+ .collect::<Result<Vec<_>>>()?;
+ Ok(build_buffered_extreme(
+ buffered_streams,
+ self.buffered.schema(),
+ Arc::clone(&on_buffered),
+ metrics.clone(),
+ reservation,
+ Arc::clone(context.memory_pool()),
+ descending,
+ ))
+ })?;
+
+ let streamed = self.streamed.execute(partition, Arc::clone(&context))?;
+
+ Ok(Box::pin(RightExistencePWMJStream::try_new(
+ Arc::clone(&self.schema),
+ on_streamed,
+ self.join_type,
+ self.operator,
+ streamed,
+ extreme_fut,
+ metrics,
+ batch_size,
+ )))
+ }
+ JoinType::LeftSemi | JoinType::LeftAnti => {
+ let buffered_side = self.buffered_side(
+ &context,
+ &on_buffered,
+ &metrics,
+ streamed_partitions,
+ )?;
+ let streamed = self.streamed.execute(partition, Arc::clone(&context))?;
+
+ Ok(Box::pin(ExistencePWMJStream::try_new(
+ Arc::clone(&self.schema),
+ on_streamed,
+ self.join_type,
+ self.operator,
+ streamed,
+ buffered_side,
+ self.sort_options,
+ metrics,
+ batch_size,
+ )))
+ }
+ JoinType::LeftMark | JoinType::RightMark => internal_err!(
"PiecewiseMergeJoin does not support existence join {} (should have been rejected in try_new)",
self.join_type
- )
- } else {
- Ok(Box::pin(ClassicPWMJStream::try_new(
- Arc::clone(&self.schema),
- on_streamed,
- self.join_type,
- self.operator,
- streamed,
- buffered_side,
- PiecewiseMergeJoinStreamState::WaitBufferedSide,
- self.sort_options,
- metrics,
- batch_size,
- )))
+ ),
+ JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full => {
+ let buffered_side = self.buffered_side(
+ &context,
+ &on_buffered,
+ &metrics,
+ streamed_partitions,
+ )?;
+ let streamed = self.streamed.execute(partition, Arc::clone(&context))?;
+
+ Ok(Box::pin(ClassicPWMJStream::try_new(
+ Arc::clone(&self.schema),
+ on_streamed,
+ self.join_type,
+ self.operator,
+ streamed,
+ buffered_side,
+ PiecewiseMergeJoinStreamState::WaitBufferedSide,
+ self.sort_options,
+ metrics,
+ batch_size,
+ )))
+ }
}
}
@@ -694,6 +820,9 @@ impl ExecutionPlan for PiecewiseMergeJoinExec {
schema: _,
// buffered side collected at execution time, not part of the plan
buffered_fut: _,
+ // buffered-side extreme collected at execution time for right existence
+ // joins, not part of the plan
+ buffered_extreme_fut: _,
// runtime metrics, not part of the plan
metrics: _,
// recomputed from `on` and `sort_options` by `try_new` on decode
@@ -941,6 +1070,148 @@ impl BufferedSideData {
}
}
+/// The entire buffered side of a right existence join, reduced to the one key that decides
+/// every streamed row -- the minimum for `<`/`<=`, the maximum for `>`/`>=`.
+///
+/// Right existence joins never look at a buffered *row*, so unlike [`BufferedSideData`] this
+/// holds no batch: the buffered input is folded away as it streams in and the state is `O(1)`
+/// however large that side is. See `right_existence_join.rs`.
+pub(super) struct BufferedExtreme {
+ /// One-row array. The row is null exactly when no buffered key is non-null (an empty or
+ /// all-NULL buffered side), in which case nothing can ever match.
+ extreme: ArrayRef,
+ _reservation: MemoryReservation,
+}
+
+impl BufferedExtreme {
+ pub(super) fn extreme(&self) -> &ArrayRef {
+ &self.extreme
+ }
+}
+
+/// Folds `next` into `running`, re-reducing the pair rather than comparing them directly so
+/// the combined value is ordered exactly as each single reduction was. `extreme_key` ignores
+/// nulls, so a null from an all-NULL batch or partition never displaces a real key.
+fn fold_extreme(
+ running: Option<ArrayRef>,
+ next: ArrayRef,
+ descending: bool,
+) -> Result<ArrayRef> {
+ match running {
+ Some(running) => {
+ let pair = concat(&[running.as_ref(), next.as_ref()])?;
+ extreme_key(&pair, descending)
+ }
+ None => Ok(next),
+ }
+}
+
+/// Reduces one buffered partition to a single extreme key, dropping each batch as it goes.
+/// `None` when the partition produced no batch at all.
+async fn partition_extreme(
+ mut buffered: SendableRecordBatchStream,
+ on_buffered: PhysicalExprRef,
+ metrics: BuildProbeJoinMetrics,
+ reservation: MemoryReservation,
+ descending: bool,
+) -> Result<Option<ArrayRef>> {
+ let mut extreme: Option<ArrayRef> = None;
+
+ while let Some(batch) = buffered.next().await.transpose()? {
+ metrics.build_input_batches.add(1);
+ metrics.build_input_rows.add(batch.num_rows());
+
+ let keys = on_buffered.evaluate(&batch)?.into_array(batch.num_rows())?;
+
+ // Reduced and dropped within this iteration, but as wide as the batch, and every
+ // partition folds one of these at once. Resized rather than grown, so what the pool
+ // sees is the widest key array in flight here and not their sum.
+ reservation.try_resize(keys.get_array_memory_size())?;
+
+ let batch_extreme = extreme_key(&keys, descending)?;
+ extreme = Some(fold_extreme(extreme, batch_extreme, descending)?);
+ }
+
+ Ok(extreme)
+}
+
+/// Folds every buffered partition down to one extreme key. `O(B)` time, and the only state it
+/// retains is that one key -- no buffered batch is concatenated or held. The transient cost is
+/// one key array per folding partition, which each task accounts against the pool for as long
+/// as it holds it.
+///
+/// A min/max combines across partitions, so this side needs no single-partition funnel --
+/// `input_distribution_requirements` asks for `UnspecifiedDistribution` here and the partitions
+/// are folded independently, each in its own spawned task.
+async fn build_buffered_extreme(
+ buffered_streams: Vec<SendableRecordBatchStream>,
+ buffered_schema: SchemaRef,
+ on_buffered: PhysicalExprRef,
+ metrics: BuildProbeJoinMetrics,
+ reservation: MemoryReservation,
+ memory_pool: Arc<dyn MemoryPool>,
+ descending: bool,
+) -> Result<BufferedExtreme> {
+ let tasks: Vec<_> = buffered_streams
+ .into_iter()
+ .enumerate()
+ .map(|(partition, stream)| {
+ let on_buffered = Arc::clone(&on_buffered);
+ let metrics = metrics.clone();
+ // One reservation per task rather than a shared one: `MemoryReservation` is not
+ // shareable, and each task's transient is freed as soon as it finishes.
+ let reservation = MemoryConsumer::new(format!(
+ "PiecewiseMergeJoinBufferedFold[{partition}]"
+ ))
+ .register(&memory_pool);
+ SpawnedTask::spawn(partition_extreme(
+ stream,
+ on_buffered,
+ metrics,
+ reservation,
+ descending,
+ ))
+ })
+ .collect();
+
+ // The tasks run concurrently; this only collects them. Awaiting in order is fine, and a
+ // failure propagates after the rest have been joined, since `SpawnedTask` aborts on drop.
+ let mut extreme: Option<ArrayRef> = None;
+ for task in tasks {
+ let partition_extreme = task.join_unwind().await.map_err(|e| {
+ internal_datafusion_err!("buffered extreme task failed: {e}")
+ })??;
+ if let Some(partition_extreme) = partition_extreme {
+ // Same fold as within a partition, so a value accumulated across partitions is
+ // ordered by the same rule.
+ extreme = Some(fold_extreme(extreme, partition_extreme, descending)?);
+ }
+ }
+
+ // No partition produced a batch, so there was nothing to reduce. Evaluating the key
+ // expression over an empty batch gives a correctly typed empty array, whose reduction is
+ // the null that represents "no buffered key".
+ let extreme = match extreme {
+ Some(extreme) => extreme,
+ None => {
+ let empty = RecordBatch::new_empty(buffered_schema);
+ let keys = on_buffered.evaluate(&empty)?.into_array(0)?;
+ extreme_key(&keys, descending)?
+ }
+ };
+
+ // The one-row extreme is all this join type ever holds -- a few bytes, and unrelated to the
+ // size of the buffered input.
+ let size = extreme.get_array_memory_size();
+ reservation.try_grow(size)?;
+ metrics.build_mem_used.add(size);
+
+ Ok(BufferedExtreme {
+ extreme,
+ _reservation: reservation,
+ })
+}
+
pub(super) enum BufferedSide {
/// Indicates that build-side not collected yet
Initial(BufferedSideInitialState),
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 85eaf4b602..693e64a06f 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
@@ -18,9 +18,10 @@
//! PiecewiseMergeJoin stream specialized for existence joins.
//!
//! Instantiated by [`PiecewiseMergeJoinExec`] when the join type is `LeftSemi` or `LeftAnti`.
-//! The other existence joins are rejected in `PiecewiseMergeJoinExec::try_new`:
-//! `RightSemi`/`RightAnti`/`RightMark` mark the right side and so need an input swap, while
-//! `LeftMark` marks the left side but needs an extra boolean column rather than a slice.
+//! `RightSemi`/`RightAnti` mark the streamed side instead and are served by
+//! `RightExistencePWMJStream` (see `right_existence_join.rs`); the Mark joins are rejected in
+//! `PiecewiseMergeJoinExec::try_new`, as they need an extra boolean column rather than a
+//! subset of one side's rows.
//!
//! # Motivation
//!
@@ -452,7 +453,7 @@ impl ExistencePWMJStream {
/// 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.
-fn extreme_key(values: &ArrayRef, descending: bool) -> Result<ArrayRef> {
+pub(super) fn extreme_key(values: &ArrayRef, descending: bool) -> Result<ArrayRef> {
let extreme = if descending {
max_batch(values)?
} else {
@@ -777,11 +778,8 @@ mod tests {
Ok(())
}
- /// The unsupported existence joins must be rejected at construction, not deeper in.
- /// `required_input_ordering` still has an `unimplemented!()` for right existence joins
- /// and cannot return an error, so this test is what keeps that panic unreachable: if
- /// someone opens the gate for RightSemi/RightAnti without also supplying an ordering
- /// requirement, this fails instead of panicking the optimizer at runtime.
+ /// The Mark joins must be rejected at construction, not deeper in: `execute` has no way
+ /// to fall back, and the planner has already committed to this operator by then.
#[test]
fn try_new_rejects_unsupported_existence_joins() -> Result<()> {
let left = build_table(
@@ -799,12 +797,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- for join_type in [
- JoinType::RightSemi,
- JoinType::RightAnti,
- JoinType::LeftMark,
- JoinType::RightMark,
- ] {
+ for join_type in [JoinType::LeftMark, JoinType::RightMark] {
let err = PiecewiseMergeJoinExec::try_new(
Arc::clone(&left),
Arc::clone(&right),
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/mod.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/mod.rs
index 8c6815ad6c..23fa3261a9 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/mod.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/mod.rs
@@ -22,4 +22,5 @@ pub use exec::PiecewiseMergeJoinExec;
mod classic_join;
mod exec;
mod existence_join;
+mod right_existence_join;
mod utils;
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs
new file mode 100644
index 0000000000..483e828e41
--- /dev/null
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/right_existence_join.rs
@@ -0,0 +1,1177 @@
+// 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 stream specialized for right existence joins.
+//!
+//! Instantiated by [`PiecewiseMergeJoinExec`] when the join type is `RightSemi` or
+//! `RightAnti`. `LeftSemi`/`LeftAnti` are served by `ExistencePWMJStream` (see
+//! `existence_join.rs`); the Mark joins are still rejected in
+//! `PiecewiseMergeJoinExec::try_new`.
+//!
+//! # Algorithm
+//!
+//! Left and right existence joins mark opposite sides, and for a single range predicate
+//! that difference is not symmetric — it collapses the work.
+//!
+//! `LeftSemi`/`LeftAnti` ask, for each *buffered* row, whether any streamed row matches, so
+//! the answer depends on the whole streamed side and can only be emitted once it has all
+//! been read. `RightSemi`/`RightAnti` ask the mirror question — for each *streamed* row,
+//! does any buffered row match? — and with only `buffered_key OP streamed_key` to satisfy,
+//! that is decided by a single buffered key:
+//!
+//! ```text
+//! ∃b. b < s ⟺ min(b) < s ∃b. b > s ⟺ max(b) > s
+//! ∃b. b <= s ⟺ min(b) <= s ∃b. b >= s ⟺ max(b) >= s
+//! ```
+//!
+//! The buffered side is therefore reduced to that one key -- `min_batch`/`max_batch`, folded
+//! batch by batch as it streams in -- and each batch is dropped once folded. It is never
+//! concatenated or retained, so the state this join holds is a single row however large that
+//! side is. The reduction is shared, so every streamed partition reads the same key rather
+//! than repeating the pass:
+//!
+//! ```text
+//! operator `<`, buffered keys [NULL, 5, 9, 7] -> min = 5
+//! `b < s` holds for some b iff `5 < s`
+//! ```
+//!
+//! A min/max is `O(B)` from any order, so `required_input_ordering` returns nothing for either
+//! child and no `SortExec` is planned.
+//!
+//! Every streamed row is then decided by comparing it against that one key, which is a
+//! vectorized `cmp` kernel per batch and a filter. Nothing about a streamed row depends on
+//! any other, so:
+//!
+//! * output is produced per batch as it arrives — no watermark, no final pass, and no
+//! election among the streamed partitions,
+//! * all N streamed partitions produce output, rather than one non-empty partition,
+//! * the streamed side is never sorted, not even per batch.
+//!
+//! Rows whose join key is NULL never satisfy a comparison predicate. `min_batch`/`max_batch`
+//! ignore NULLs, so the reduced key is null only when *every* buffered key is (or the buffered
+//! side is empty) — and then no streamed row can match at all, which makes `RightSemi` empty
+//! without reading the streamed side and `RightAnti` a passthrough of it. A NULL streamed key
+//! makes its comparison NULL rather than false, which is "no match" — dropped by `RightSemi`,
+//! kept by `RightAnti`.
+//!
+//! Picking the extreme and comparing against it must agree on ordering. `min_batch`/`max_batch`
+//! order floats by `total_cmp` (`arrow-arith`'s `MinAccumulator` compares with
+//! `ArrowNativeTypeOp::is_lt`, seeded from `MAX_TOTAL_ORDER`), which puts `-0.0` strictly below
+//! `+0.0` -- but SQL comparisons treat them as equal, and that is what a real `k < r` predicate
+//! actually evaluates to: any `BinaryExpr` comparison, including the one the
+//! `NestedLoopJoinExec` oracle in the differential fuzz test builds its filter from, normalizes
+//! `-0.0` to `+0.0` first (see `apply_cmp` in `datafusion-physical-expr-common`, which
+//! [`filter_streamed_batch`](RightExistencePWMJStream::filter_streamed_batch) calls to compare
+//! the extreme against each streamed batch). `apply_cmp` normalizes both operands itself, so
+//! nothing here has to: the reduction can order `-0.0` below `+0.0` as `min`/`max` do, and the
+//! comparison it feeds into still agrees with SQL semantics. `NaN` needs no such fix-up: every
+//! kernel involved orders it as the maximum, so it is treated identically on both sides already.
+//! `apply_cmp` also dispatches to the nested-aware comparator for List/Struct keys, which the
+//! raw arrow `lt`/`lt_eq`/`gt`/`gt_eq` kernels reject outright.
+//!
+//! # Cost
+//!
+//! Let `B` be the buffered rows and `S` the streamed rows: `O(B)` to reduce the buffered side
+//! plus `O(S)` to filter, and no sort on either side. The state retained across batches is one
+//! row; while folding, each buffered partition also holds the key array of the batch it is
+//! reducing, which it accounts against the memory pool for that long.
+//!
+//! This is why the shared state is [`BufferedExtreme`] and not `BufferedSideData`: every field
+//! of the latter -- the concatenated batch, the key array, the visited-indices bitmap, the
+//! final-pass counter -- would be dead here.
+//!
+//! [`PiecewiseMergeJoinExec`]: super::PiecewiseMergeJoinExec
+
+use std::sync::Arc;
+use std::task::{Poll, ready};
+
+use arrow::array::{AsArray, RecordBatch};
+use arrow::compute::BatchCoalescer;
+use arrow::compute::filter_record_batch;
+use arrow::compute::kernels::boolean::not;
+use arrow_schema::SchemaRef;
+use datafusion_common::{Result, ScalarValue};
+use datafusion_execution::{RecordBatchStream, SendableRecordBatchStream};
+use datafusion_expr::{ColumnarValue, JoinType, Operator};
+use datafusion_physical_expr::PhysicalExprRef;
+use datafusion_physical_expr_common::datum::apply_cmp;
+use futures::{Stream, StreamExt};
+
+use crate::handle_state;
+use crate::joins::piecewise_merge_join::exec::BufferedExtreme;
+use crate::joins::utils::{
+ BuildProbeJoinMetrics, OnceFut, StatefulStreamResult, boolean_mask_from_filter,
+};
+use crate::stream::EmptyRecordBatchStream;
+
+pub(super) enum RightExistencePWMJStreamState {
+ /// Await the buffered side's reduction to a single key.
+ WaitBufferedExtreme,
+ /// Fetch streamed batches and emit the rows that do (`RightSemi`) or do not
+ /// (`RightAnti`) have a buffered match.
+ ScanStreamBatches,
+ /// Streamed side exhausted; drain whatever `output_batches` still holds.
+ Draining,
+ Completed,
+}
+
+pub(super) struct RightExistencePWMJStream {
+ /// Output schema, which for `RightSemi`/`RightAnti` is the streamed side's schema
+ schema: SchemaRef,
+ /// Physical expression evaluated on the streamed side. The buffered side's
+ /// equivalent is already evaluated when the buffered side is collected.
+ on_streamed: PhysicalExprRef,
+ /// `RightSemi` or `RightAnti`
+ join_type: JoinType,
+ /// Comparison operator
+ operator: Operator,
+ streamed: SendableRecordBatchStream,
+ /// Resolves to the whole buffered side reduced to one key. Shared with the other streamed
+ /// partitions, so that reduction happens exactly once.
+ buffered_extreme_fut: OnceFut<BufferedExtreme>,
+ state: RightExistencePWMJStreamState,
+ /// That key, held as a [`ScalarValue`] so `apply_cmp` broadcasts it against a streamed
+ /// batch in one call. `None` when the buffered side has no non-null key, i.e. nothing can
+ /// ever match. Only populated once `buffered_extreme_fut` has resolved.
+ buffered_extreme: Option<ScalarValue>,
+ join_metrics: BuildProbeJoinMetrics,
+ /// Chunks the per-streamed-batch output to `batch_size`, so a low-selectivity join
+ /// does not hand downstream operators one small batch per streamed batch.
+ output_batches: Box<BatchCoalescer>,
+}
+
+impl RightExistencePWMJStream {
+ #[expect(clippy::too_many_arguments)]
+ pub(super) fn try_new(
+ schema: SchemaRef,
+ on_streamed: PhysicalExprRef,
+ join_type: JoinType,
+ operator: Operator,
+ streamed: SendableRecordBatchStream,
+ buffered_extreme_fut: OnceFut<BufferedExtreme>,
+ join_metrics: BuildProbeJoinMetrics,
+ batch_size: usize,
+ ) -> Self {
+ Self {
+ output_batches: Box::new(BatchCoalescer::new(
+ Arc::clone(&schema),
+ batch_size,
+ )),
+ schema,
+ on_streamed,
+ join_type,
+ operator,
+ streamed,
+ buffered_extreme_fut,
+ state: RightExistencePWMJStreamState::WaitBufferedExtreme,
+ buffered_extreme: None,
+ join_metrics,
+ }
+ }
+
+ fn poll_next_impl(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Option<Result<RecordBatch>>> {
+ loop {
+ return match self.state {
+ RightExistencePWMJStreamState::WaitBufferedExtreme => {
+ handle_state!(ready!(self.collect_buffered_extreme(cx)))
+ }
+ RightExistencePWMJStreamState::ScanStreamBatches => {
+ handle_state!(ready!(self.scan_stream_batch(cx)))
+ }
+ RightExistencePWMJStreamState::Draining => {
+ handle_state!(self.drain_output())
+ }
+ RightExistencePWMJStreamState::Completed => Poll::Ready(None),
+ };
+ }
+ }
+
+ /// Picks up the buffered side's reduced key, which is all this join type needs from it.
+ fn collect_buffered_extreme(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>> {
+ let build_timer = self.join_metrics.build_time.timer();
+ let buffered_extreme = ready!(self.buffered_extreme_fut.get_shared(cx))?;
+ build_timer.done();
+
+ // Null exactly when no buffered key is non-null, and NULLs match nothing. `apply_cmp`
+ // normalizes `-0.0`/`+0.0` on both operands when it compares this against the streamed
+ // key in `filter_streamed_batch`, so no normalization is needed here.
+ let extreme = ScalarValue::try_from_array(buffered_extreme.extreme(), 0)?;
+ self.buffered_extreme = (!extreme.is_null()).then_some(extreme);
+
+ // With no non-null buffered key nothing matches, so `RightSemi` outputs nothing
+ // and does not need to read a single streamed batch. `RightAnti` still has to,
+ // since it outputs all of them.
+ self.state = match (&self.buffered_extreme, self.join_type) {
+ (None, JoinType::RightSemi) => {
+ let streamed_schema = self.streamed.schema();
+ self.streamed = Box::pin(EmptyRecordBatchStream::new(streamed_schema));
+ RightExistencePWMJStreamState::Completed
+ }
+ _ => RightExistencePWMJStreamState::ScanStreamBatches,
+ };
+
+ Poll::Ready(Ok(StatefulStreamResult::Continue))
+ }
+
+ /// Fetches one streamed batch, filters it, and pushes the survivors into
+ /// `output_batches`. Once the streamed side is exhausted, flushes the coalescer's
+ /// partial batch and hands off to [`Self::drain_output`].
+ fn scan_stream_batch(
+ &mut self,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>> {
+ if let Some(batch) = self.output_batches.next_completed_batch() {
+ return Poll::Ready(Ok(StatefulStreamResult::Ready(Some(batch))));
+ }
+
+ match ready!(self.streamed.poll_next_unpin(cx)) {
+ None => {
+ self.output_batches.finish_buffered_batch()?;
+ self.state = RightExistencePWMJStreamState::Draining;
+ }
+ Some(Ok(batch)) => {
+ self.join_metrics.input_batches.add(1);
+ self.join_metrics.input_rows.add(batch.num_rows());
+
+ let output = self.filter_streamed_batch(&batch)?;
+ if output.num_rows() > 0 {
+ self.output_batches.push_batch(output)?;
+ }
+ // Nothing pushed, or not enough to complete a batch yet: loop back
+ // through `poll_next_impl` to poll the next streamed batch.
+ }
+ Some(Err(err)) => return Poll::Ready(Err(err)),
+ }
+
+ Poll::Ready(Ok(StatefulStreamResult::Continue))
+ }
+
+ /// Drains whatever `output_batches` still holds once the streamed side is exhausted.
+ fn drain_output(&mut self) -> Result<StatefulStreamResult<Option<RecordBatch>>> {
+ match self.output_batches.next_completed_batch() {
+ Some(batch) => Ok(StatefulStreamResult::Ready(Some(batch))),
+ None => {
+ self.state = RightExistencePWMJStreamState::Completed;
+ Ok(StatefulStreamResult::Ready(None))
+ }
+ }
+ }
+
+ /// Keeps the streamed rows that have a buffered match (`RightSemi`) or that have none
+ /// (`RightAnti`), by comparing each against the single buffered extreme.
+ fn filter_streamed_batch(&self, batch: &RecordBatch) -> Result<RecordBatch> {
+ let columns = match &self.buffered_extreme {
+ // No non-null buffered key, so no streamed row matches and `RightAnti` keeps
+ // the batch whole. `RightSemi` never gets here: it completed without reading
+ // the streamed side.
+ None => batch.columns().to_vec(),
+ Some(extreme) => {
+ let stream_values = ColumnarValue::Array(
+ self.on_streamed
+ .evaluate(batch)?
+ .into_array(batch.num_rows())?,
+ );
+
+ // `extreme` is the buffered key, so it goes on the left of the operator,
+ // matching the `buffered OP streamed` orientation of the predicate. `apply_cmp`
+ // dispatches to the nested-aware comparator for List/Struct keys, unlike the
+ // raw arrow `lt`/`lt_eq`/`gt`/`gt_eq` kernels this used to call directly, which
+ // reject those types outright.
+ let matched = apply_cmp(
+ self.operator,
+ &ColumnarValue::Scalar(extreme.clone()),
+ &stream_values,
+ )?
+ .into_array(batch.num_rows())?;
+ let matched = matched.as_boolean();
+
+ let predicate = match self.join_type {
+ // A NULL streamed key compares NULL rather than false, and `filter`
+ // already treats NULL as "not selected" -- which is what a
+ // non-matching row is.
+ JoinType::RightSemi => matched.clone(),
+ // Anti needs the complement, so those NULLs have to be folded into
+ // false first: `not(NULL)` is NULL, which would drop a row that
+ // matched nothing.
+ _ => not(&boolean_mask_from_filter(matched))?,
+ };
+
+ filter_record_batch(batch, &predicate)?.columns().to_vec()
+ }
+ };
+
+ // Right existence joins output the streamed columns only. The streamed child's
+ // schema is field-for-field equal to the join's own, but rebuild against the
+ // latter so the stream's declared schema is what it yields.
+ Ok(RecordBatch::try_new(Arc::clone(&self.schema), columns)?)
+ }
+}
+
+impl RecordBatchStream for RightExistencePWMJStream {
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.schema)
+ }
+}
+
+impl Stream for RightExistencePWMJStream {
+ type Item = Result<RecordBatch>;
+
+ fn poll_next(
+ mut self: std::pin::Pin<&mut Self>,
+ cx: &mut std::task::Context<'_>,
+ ) -> Poll<Option<Self::Item>> {
+ // `record_poll` fills in `output_rows` and `end_time`; `elapsed_compute` is handled
+ // by `BuildProbeJoinMetrics::drop`.
+ let poll = self.poll_next_impl(cx);
+ self.join_metrics.baseline.record_poll(poll)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::sorts::sort::SortExec;
+ use crate::{
+ ExecutionPlan, ExecutionPlanProperties, common, joins::PiecewiseMergeJoinExec,
+ test::TestMemoryExec,
+ };
+ use arrow::array::{DictionaryArray, Int32Array, ListArray, StructArray};
+ use arrow::compute::SortOptions;
+ use arrow::datatypes::Int32Type;
+ use arrow_schema::{DataType, Field, Schema};
+ use datafusion_common::test_util::batches_to_string;
+ use datafusion_execution::TaskContext;
+ use datafusion_execution::config::SessionConfig;
+ use datafusion_execution::memory_pool::GreedyMemoryPool;
+ use datafusion_execution::runtime_env::RuntimeEnvBuilder;
+ use datafusion_physical_expr::expressions::Column;
+ use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr};
+ use insta::assert_snapshot;
+
+ // Coverage for right existence joins also lives in `pwmj.slt` (both correlation
+ // orientations, all four operators, NULLs, key types) and in the differential fuzz test
+ // `fuzz_pwmj_matches_nested_loop`, which checks them against
+ // `NestedLoopJoinExec` over randomized inputs. The tests here pin the parts SQL cannot
+ // observe: which streamed batches were read, and which partition emitted them.
+
+ fn kv_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("k", DataType::Int32, true),
+ ]))
+ }
+
+ fn kv_batch(rows: &[(i32, Option<i32>)]) -> RecordBatch {
+ let ids: Vec<i32> = rows.iter().map(|(id, _)| *id).collect();
+ let keys: Vec<Option<i32>> = rows.iter().map(|(_, k)| *k).collect();
+ RecordBatch::try_new(
+ kv_schema(),
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(keys)),
+ ],
+ )
+ .unwrap()
+ }
+
+ fn kv_exec(partitions: &[Vec<RecordBatch>]) -> Arc<dyn ExecutionPlan> {
+ TestMemoryExec::try_new_exec(partitions, kv_schema(), None).unwrap()
+ }
+
+ /// Right existence joins declare no input ordering requirement, so the buffered sides
+ /// below are deliberately **unsorted**: only their min/max matters. That is also why
+ /// building the exec directly is faithful here, where the other streams would need a
+ /// `SortExec` the tests have to supply by hand.
+ fn join(
+ buffered: Arc<dyn ExecutionPlan>,
+ streamed: Arc<dyn ExecutionPlan>,
+ operator: Operator,
+ join_type: JoinType,
+ ) -> Result<PiecewiseMergeJoinExec> {
+ let on = (
+ Arc::new(Column::new_with_schema("k", &buffered.schema())?) as _,
+ Arc::new(Column::new_with_schema("k", &streamed.schema())?) as _,
+ );
+ let probe = PiecewiseMergeJoinExec::try_new(
+ Arc::clone(&buffered),
+ Arc::clone(&streamed),
+ on.clone(),
+ operator,
+ join_type,
+ 1,
+ )?;
+ assert!(
+ probe.required_input_ordering().iter().all(Option::is_none),
+ "right existence joins must not require an input ordering"
+ );
+ assert!(
+ probe
+ .input_distribution_requirements()
+ .per_child_distributions()
+ .all(|d| matches!(
+ d,
+ datafusion_physical_expr::Distribution::UnspecifiedDistribution
+ )),
+ "right existence joins must not require the buffered side coalesced"
+ );
+ assert_eq!(
+ probe.benefits_from_input_partitioning(),
+ vec![false, true],
+ "the buffered side must not be fanned out just to fold it"
+ );
+ PiecewiseMergeJoinExec::try_new(buffered, streamed, on, operator, join_type, 1)
+ }
+
+ fn input_batches(join: &PiecewiseMergeJoinExec) -> usize {
+ join.metrics()
+ .unwrap()
+ .sum_by_name("input_batches")
+ .expect("input_batches metric")
+ .as_usize()
+ }
+
+ /// `RightSemi` keeps the streamed rows with at least one buffered match, and outputs
+ /// only the streamed columns. `>` needs the buffered maximum, 5, so of the streamed keys
+ /// {4, 5, 6} only 4 has some buffered key above it.
+ #[tokio::test]
+ async fn join_right_semi() -> Result<()> {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, Some(5)), (2, Some(1)), (3, Some(2))])]]),
+ kv_exec(&[vec![kv_batch(&[
+ (10, Some(4)),
+ (20, Some(5)),
+ (30, Some(6)),
+ ])]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+---+
+ | id | k |
+ +----+---+
+ | 10 | 4 |
+ +----+---+
+ ");
+ Ok(())
+ }
+
+ /// `RightAnti` is the complement: the streamed rows with no buffered match.
+ #[tokio::test]
+ async fn join_right_anti() -> Result<()> {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, Some(5)), (2, Some(1)), (3, Some(2))])]]),
+ kv_exec(&[vec![kv_batch(&[
+ (10, Some(4)),
+ (20, Some(5)),
+ (30, Some(6)),
+ ])]]),
+ Operator::Gt,
+ JoinType::RightAnti,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+---+
+ | id | k |
+ +----+---+
+ | 20 | 5 |
+ | 30 | 6 |
+ +----+---+
+ ");
+ Ok(())
+ }
+
+ fn kv_list_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "k",
+ DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true))),
+ true,
+ ),
+ ]))
+ }
+
+ fn kv_list_batch(rows: &[(i32, Option<Vec<i32>>)]) -> RecordBatch {
+ let ids: Vec<i32> = rows.iter().map(|(id, _)| *id).collect();
+ let keys = ListArray::from_iter_primitive::<Int32Type, _, _>(
+ rows.iter()
+ .map(|(_, k)| k.clone().map(|v| v.into_iter().map(Some))),
+ );
+ RecordBatch::try_new(
+ kv_list_schema(),
+ vec![Arc::new(Int32Array::from(ids)), Arc::new(keys)],
+ )
+ .unwrap()
+ }
+
+ fn kv_list_exec(partitions: &[Vec<RecordBatch>]) -> Arc<dyn ExecutionPlan> {
+ TestMemoryExec::try_new_exec(partitions, kv_list_schema(), None).unwrap()
+ }
+
+ /// List/Struct keys go through `apply_cmp`'s nested comparator, unlike arrow's raw
+ /// `lt`/`lt_eq`/`gt`/`gt_eq` kernels, which reject nested types outright -- this pins that
+ /// dispatch rather than any particular list-ordering rule. Same shape as `join_right_semi`,
+ /// with every key wrapped in a one-element list.
+ #[tokio::test]
+ async fn join_right_semi_supports_list_keys() -> Result<()> {
+ let join = join(
+ kv_list_exec(&[vec![kv_list_batch(&[
+ (1, Some(vec![5])),
+ (2, Some(vec![1])),
+ (3, Some(vec![2])),
+ ])]]),
+ kv_list_exec(&[vec![kv_list_batch(&[
+ (10, Some(vec![4])),
+ (20, Some(vec![5])),
+ (30, Some(vec![6])),
+ ])]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+-----+
+ | id | k |
+ +----+-----+
+ | 10 | [4] |
+ +----+-----+
+ ");
+ Ok(())
+ }
+
+ fn kv_struct_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "k",
+ DataType::Struct(vec![Field::new("v", DataType::Int32, true)].into()),
+ true,
+ ),
+ ]))
+ }
+
+ fn kv_struct_batch(rows: &[(i32, Option<i32>)]) -> RecordBatch {
+ let ids: Vec<i32> = rows.iter().map(|(id, _)| *id).collect();
+ let values: Vec<Option<i32>> = rows.iter().map(|(_, k)| *k).collect();
+ let is_valid: Vec<bool> = rows.iter().map(|(_, k)| k.is_some()).collect();
+ let keys = StructArray::new(
+ vec![Field::new("v", DataType::Int32, true)].into(),
+ vec![Arc::new(Int32Array::from(values))],
+ Some(is_valid.into()),
+ );
+ RecordBatch::try_new(
+ kv_struct_schema(),
+ vec![Arc::new(Int32Array::from(ids)), Arc::new(keys)],
+ )
+ .unwrap()
+ }
+
+ fn kv_struct_exec(partitions: &[Vec<RecordBatch>]) -> Arc<dyn ExecutionPlan> {
+ TestMemoryExec::try_new_exec(partitions, kv_struct_schema(), None).unwrap()
+ }
+
+ /// Struct keys, same shape and expectation as [`join_right_semi_supports_list_keys`].
+ #[tokio::test]
+ async fn join_right_semi_supports_struct_keys() -> Result<()> {
+ let join = join(
+ kv_struct_exec(&[vec![kv_struct_batch(&[
+ (1, Some(5)),
+ (2, Some(1)),
+ (3, Some(2)),
+ ])]]),
+ kv_struct_exec(&[vec![kv_struct_batch(&[
+ (10, Some(4)),
+ (20, Some(5)),
+ (30, Some(6)),
+ ])]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+--------+
+ | id | k |
+ +----+--------+
+ | 10 | {v: 4} |
+ +----+--------+
+ ");
+ Ok(())
+ }
+
+ fn kv_dict_schema() -> Arc<Schema> {
+ Arc::new(Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new(
+ "k",
+ DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
+ true,
+ ),
+ ]))
+ }
+
+ fn kv_dict_batch(rows: &[(i32, Option<&str>)]) -> RecordBatch {
+ let ids: Vec<i32> = rows.iter().map(|(id, _)| *id).collect();
+ let keys: DictionaryArray<Int32Type> = rows.iter().map(|(_, k)| *k).collect();
+ RecordBatch::try_new(
+ kv_dict_schema(),
+ vec![Arc::new(Int32Array::from(ids)), Arc::new(keys)],
+ )
+ .unwrap()
+ }
+
+ fn kv_dict_exec(partitions: &[Vec<RecordBatch>]) -> Arc<dyn ExecutionPlan> {
+ TestMemoryExec::try_new_exec(partitions, kv_dict_schema(), None).unwrap()
+ }
+
+ /// Dictionary-encoded keys, same shape as [`join_right_semi_supports_list_keys`] with
+ /// string values ordered the same as the int cases (`"4" < "5" < "6"`).
+ #[tokio::test]
+ async fn join_right_semi_supports_dictionary_keys() -> Result<()> {
+ let join = join(
+ kv_dict_exec(&[vec![kv_dict_batch(&[
+ (1, Some("5")),
+ (2, Some("1")),
+ (3, Some("2")),
+ ])]]),
+ kv_dict_exec(&[vec![kv_dict_batch(&[
+ (10, Some("4")),
+ (20, Some("5")),
+ (30, Some("6")),
+ ])]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+---+
+ | id | k |
+ +----+---+
+ | 10 | 4 |
+ +----+---+
+ ");
+ Ok(())
+ }
+
+ /// A NULL key satisfies no comparison predicate, on either side.
+ ///
+ /// The buffered side is sorted descending with NULLs first, as `<` requires, so the key
+ /// this operator needs -- the minimum, 5 -- is the last row and the NULL is nowhere near
+ /// it. On the streamed side the NULL-keyed row matches nothing, so `RightSemi` drops it
+ /// and `RightAnti` keeps it.
+ #[tokio::test]
+ async fn null_keys_match_nothing() -> Result<()> {
+ let buffered =
+ || kv_exec(&[vec![kv_batch(&[(1, Some(9)), (2, None), (3, Some(5))])]]);
+ let streamed = || {
+ kv_exec(&[vec![kv_batch(&[
+ (10, Some(5)),
+ (20, Some(6)),
+ (30, None),
+ (40, Some(100)),
+ ])]])
+ };
+
+ let semi = join(buffered(), streamed(), Operator::Lt, JoinType::RightSemi)?;
+ let batches =
+ common::collect(semi.execute(0, Arc::new(TaskContext::default()))?).await?;
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+-----+
+ | id | k |
+ +----+-----+
+ | 20 | 6 |
+ | 40 | 100 |
+ +----+-----+
+ ");
+
+ let anti = join(buffered(), streamed(), Operator::Lt, JoinType::RightAnti)?;
+ let batches =
+ common::collect(anti.execute(0, Arc::new(TaskContext::default()))?).await?;
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+---+
+ | id | k |
+ +----+---+
+ | 10 | 5 |
+ | 30 | |
+ +----+---+
+ ");
+ Ok(())
+ }
+
+ /// With no non-null buffered key nothing can match, so `RightSemi` is empty however many
+ /// streamed rows there are -- and it does not read a single one of them. Asserted through
+ /// the `input_batches` metric, which SQL cannot observe.
+ #[tokio::test]
+ async fn all_null_buffered_side_makes_right_semi_read_nothing() -> Result<()> {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, None), (2, None)])]]),
+ kv_exec(&[vec![
+ kv_batch(&[(10, Some(4))]),
+ kv_batch(&[(20, Some(5))]),
+ kv_batch(&[(30, Some(6))]),
+ ]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 0);
+ assert_eq!(input_batches(&join), 0, "no streamed batch should be read");
+ Ok(())
+ }
+
+ /// The same buffered side, here empty rather than all-NULL, sends every streamed row to
+ /// `RightAnti` instead. It has to read them all: they are the output.
+ #[tokio::test]
+ async fn empty_buffered_side_passes_right_anti_through() -> Result<()> {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[])]]),
+ kv_exec(&[vec![
+ kv_batch(&[(10, Some(4))]),
+ kv_batch(&[(20, None)]),
+ kv_batch(&[(30, Some(6))]),
+ ]]),
+ Operator::Gt,
+ JoinType::RightAnti,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+---+
+ | id | k |
+ +----+---+
+ | 10 | 4 |
+ | 20 | |
+ | 30 | 6 |
+ +----+---+
+ ");
+ assert_eq!(input_batches(&join), 3);
+ Ok(())
+ }
+
+ /// The streamed side's ordering survives this join -- one output batch per streamed batch, in
+ /// order, with rows only removed -- so `maintains_input_order` claims it and the operator
+ /// advertises the streamed child's ordering as its own. That lets a downstream operator skip
+ /// a re-sort, which means a wrong claim here would produce wrong results, not just a slow
+ /// plan. The runtime side of it is pinned by the row-order snapshots above.
+ #[test]
+ fn output_ordering_follows_the_streamed_side() -> Result<()> {
+ let streamed_input = kv_exec(&[vec![kv_batch(&[(20, Some(4)), (10, Some(9))])]]);
+ let ordering = LexOrdering::new(vec![PhysicalSortExpr::new(
+ Arc::new(Column::new_with_schema("id", &streamed_input.schema())?),
+ SortOptions::new(true, false),
+ )])
+ .unwrap();
+ let streamed = Arc::new(SortExec::new(ordering.clone(), streamed_input));
+
+ for join_type in [JoinType::RightSemi, JoinType::RightAnti] {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, Some(5))])]]),
+ Arc::clone(&streamed) as _,
+ Operator::Gt,
+ join_type,
+ )?;
+ assert_eq!(
+ join.properties().output_ordering(),
+ Some(&ordering),
+ "{join_type} should advertise the streamed side's ordering"
+ );
+ }
+
+ // The buffered side contributes no column, so its ordering must not leak out: a buffered
+ // child sorted on `k` cannot make the join claim anything.
+ let buffered_ordering = LexOrdering::new(vec![PhysicalSortExpr::new(
+ Arc::new(Column::new("k", 1)),
+ SortOptions::new(false, true),
+ )])
+ .unwrap();
+ let sorted_buffered = Arc::new(SortExec::new(
+ buffered_ordering,
+ kv_exec(&[vec![kv_batch(&[(1, Some(5))])]]),
+ ));
+ let join = join(
+ sorted_buffered as _,
+ kv_exec(&[vec![kv_batch(&[(20, Some(4))])]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+ assert_eq!(join.properties().output_ordering(), None);
+ Ok(())
+ }
+
+ /// A zero-row streamed batch still reaches the comparison kernel, which yields an empty
+ /// mask rather than erroring, so the batch is filtered away and the surrounding batches
+ /// are unaffected. Covered for both join types since only anti also runs `not` over it.
+ #[tokio::test]
+ async fn empty_streamed_batch_is_skipped() -> Result<()> {
+ for (join_type, expected) in [(JoinType::RightSemi, 1), (JoinType::RightAnti, 1)]
+ {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, Some(5)), (2, Some(1))])]]),
+ kv_exec(&[vec![
+ kv_batch(&[]),
+ kv_batch(&[(10, Some(4)), (20, Some(9))]),
+ kv_batch(&[]),
+ ]]),
+ Operator::Gt,
+ join_type,
+ )?;
+
+ let stream = join.execute(0, Arc::new(TaskContext::default()))?;
+ let batches = common::collect(stream).await?;
+
+ // Buffered maximum is 5, so `>` keeps 4 for semi and 9 for anti -- one row each,
+ // and no empty batch in between.
+ assert_eq!(
+ batches.iter().map(|b| b.num_rows()).sum::<usize>(),
+ expected,
+ "{join_type}"
+ );
+ assert!(
+ batches.iter().all(|b| b.num_rows() > 0),
+ "{join_type} should not yield an empty batch"
+ );
+ assert_eq!(input_batches(&join), 3, "{join_type}");
+ }
+ Ok(())
+ }
+
+ /// Each streamed batch here is one matching row, so without coalescing the stream would
+ /// hand back 5 one-row batches. `batch_size` 2 packs them into 3 instead, pinning that
+ /// `output_batches` actually merges across streamed-batch boundaries rather than just
+ /// forwarding `filter_streamed_batch`'s output directly.
+ #[tokio::test]
+ async fn output_is_coalesced_to_batch_size() -> Result<()> {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, Some(10))])]]),
+ kv_exec(&[vec![
+ kv_batch(&[(10, Some(1))]),
+ kv_batch(&[(20, Some(2))]),
+ kv_batch(&[(30, Some(3))]),
+ kv_batch(&[(40, Some(4))]),
+ kv_batch(&[(50, Some(5))]),
+ ]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let task_ctx = Arc::new(
+ TaskContext::default()
+ .with_session_config(SessionConfig::new().with_batch_size(2)),
+ );
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ assert_eq!(
+ batches.iter().map(|b| b.num_rows()).collect::<Vec<_>>(),
+ vec![2, 2, 1],
+ "5 matching rows over 5 input batches should pack into ceil(5/2) = 3 output \
+ batches of at most `batch_size` rows each"
+ );
+ assert_eq!(
+ input_batches(&join),
+ 5,
+ "every streamed batch is still read"
+ );
+ Ok(())
+ }
+
+ /// `output_is_coalesced_to_batch_size` above pins that coalescing happens; this pins that
+ /// it doesn't reorder rows while doing it. The streamed side is genuinely sorted (via
+ /// `SortExec`, the same claim `output_ordering_follows_the_streamed_side` checks at the
+ /// planning level), filtering drops rows unevenly across it, and a small `batch_size`
+ /// forces `output_batches` to repack across what were originally separate sorted
+ /// `SortExec` output batches. `BatchCoalescer`'s `completed` buffer is a plain FIFO
+ /// `VecDeque` in arrow-rs (`push_back`/`pop_front`, no sort), so this should hold
+ /// regardless -- the point is to check it at runtime, not just reason about it.
+ #[tokio::test]
+ async fn coalescing_does_not_reorder_a_sorted_streamed_side() -> Result<()> {
+ let streamed_input = kv_exec(&[vec![kv_batch(&[
+ (50, Some(5)),
+ (10, Some(199)),
+ (90, Some(6)),
+ (30, Some(200)),
+ (20, Some(7)),
+ (70, Some(8)),
+ (40, Some(199)),
+ (60, Some(9)),
+ ])]]);
+ let ordering = LexOrdering::new(vec![PhysicalSortExpr::new(
+ Arc::new(Column::new_with_schema("id", &streamed_input.schema())?),
+ SortOptions::new(false, false),
+ )])
+ .unwrap();
+ let streamed = Arc::new(SortExec::new(ordering, streamed_input));
+
+ // Buffered max is 100, so `>` keeps only streamed rows with k < 100 -- ids 10, 30, 40
+ // (k in the 100s) are filtered out, unevenly across the sorted id order, so the
+ // survivors (20, 50, 60, 70, 90) don't line up with any one original batch boundary.
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, Some(100))])]]),
+ streamed as _,
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let task_ctx = Arc::new(
+ TaskContext::default()
+ .with_session_config(SessionConfig::new().with_batch_size(2)),
+ );
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ let ids: Vec<i32> = batches
+ .iter()
+ .flat_map(|b| b.column(0).as_primitive::<Int32Type>().values().to_vec())
+ .collect();
+ assert_eq!(
+ ids,
+ vec![20, 50, 60, 70, 90],
+ "coalescing must not disturb the streamed side's sorted order"
+ );
+ assert_eq!(
+ batches.iter().map(|b| b.num_rows()).collect::<Vec<_>>(),
+ vec![2, 2, 1],
+ "5 surviving rows packed to batch_size = 2 confirms the coalescer, not a \
+ pass-through, produced this shape"
+ );
+ Ok(())
+ }
+
+ /// The buffered side is folded to one key as it streams in and never materialized, so this
+ /// join runs in a memory pool far smaller than that side. 20k buffered rows is >80 KB of
+ /// Int32 keys alone and ~160 KB of batches; the pool here is 64 KB, which the collecting
+ /// path cannot fit -- it reserves every batch, then the concatenation, then the key array.
+ /// The fold needs only the one batch's key array it is reducing (~8 KB), which it does
+ /// reserve, so the pool is sized above that and well below the collecting path.
+ ///
+ /// This is the assertion that pins the retained buffered state at one row. `build_mem_used`
+ /// is checked too, since a regression to collecting would show up there even under a
+ /// generous pool.
+ #[tokio::test]
+ async fn buffered_side_is_folded_not_collected() -> Result<()> {
+ let buffered: Vec<RecordBatch> = (0..10)
+ .map(|b| {
+ kv_batch(
+ &(0..2000)
+ .map(|i| (i, Some(b * 2000 + i)))
+ .collect::<Vec<_>>(),
+ )
+ })
+ .collect();
+
+ let join = join(
+ kv_exec(&[buffered]),
+ kv_exec(&[vec![kv_batch(&[(1, Some(0)), (2, Some(19_999))])]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let task_ctx = Arc::new(
+ TaskContext::default().with_runtime(Arc::new(
+ RuntimeEnvBuilder::new()
+ .with_memory_pool(Arc::new(GreedyMemoryPool::new(64 * 1024)))
+ .build()?,
+ )),
+ );
+ let stream = join.execute(0, task_ctx)?;
+ let batches = common::collect(stream).await?;
+
+ // Buffered maximum is 19999, so `>` keeps only the streamed 0.
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+---+
+ | id | k |
+ +----+---+
+ | 1 | 0 |
+ +----+---+
+ ");
+
+ // What is held is one row, not 20k.
+ let build_mem_used = join
+ .metrics()
+ .unwrap()
+ .sum_by_name("build_mem_used")
+ .expect("build_mem_used metric")
+ .as_usize();
+ assert!(
+ build_mem_used < 1024,
+ "buffered state should be a single key, got {build_mem_used} bytes"
+ );
+ Ok(())
+ }
+
+ /// The key array a partition is reducing is transient but as wide as the batch, and it is
+ /// reserved for as long as it is held. A pool that cannot fit one of them fails the join
+ /// rather than allocating outside the accounting -- which is what keeps the memory claim
+ /// honest, since without the reservation this fold would simply succeed.
+ #[tokio::test]
+ async fn folding_reserves_the_batch_key_array() -> Result<()> {
+ let join = join(
+ kv_exec(&[vec![kv_batch(
+ &(0..2000).map(|i| (i, Some(i))).collect::<Vec<_>>(),
+ )]]),
+ kv_exec(&[vec![kv_batch(&[(1, Some(0))])]]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ let task_ctx = Arc::new(
+ TaskContext::default().with_runtime(Arc::new(
+ RuntimeEnvBuilder::new()
+ .with_memory_pool(Arc::new(GreedyMemoryPool::new(1024)))
+ .build()?,
+ )),
+ );
+ let err = common::collect(join.execute(0, task_ctx)?)
+ .await
+ .expect_err("a 2000-row key array should not fit a 1 KB pool");
+ assert!(
+ err.to_string().contains("PiecewiseMergeJoinBufferedFold"),
+ "expected the fold's reservation to fail, got: {err}"
+ );
+ Ok(())
+ }
+
+ /// A min/max combines across partitions, so the buffered side is not required to be
+ /// coalesced -- and every partition of it has to be consumed. The deciding key here is in
+ /// the **last** buffered partition, so reading only partition 0 (which the other streams do,
+ /// since they require `SinglePartition`) silently loses it and changes the answer.
+ #[tokio::test]
+ async fn every_buffered_partition_is_folded() -> Result<()> {
+ // Buffered maxima per partition: 1, 3, 50. Only the last admits the streamed 40.
+ let buffered = vec![
+ vec![kv_batch(&[(1, Some(0)), (2, Some(1))])],
+ vec![kv_batch(&[(3, Some(3))])],
+ vec![kv_batch(&[(4, Some(2)), (5, Some(50))])],
+ ];
+ let streamed = || kv_exec(&[vec![kv_batch(&[(10, Some(2)), (20, Some(40))])]]);
+
+ let semi = join(
+ kv_exec(&buffered),
+ streamed(),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+ assert_eq!(semi.buffered().output_partitioning().partition_count(), 3);
+ let batches =
+ common::collect(semi.execute(0, Arc::new(TaskContext::default()))?).await?;
+ assert_snapshot!(batches_to_string(&batches), @r"
+ +----+----+
+ | id | k |
+ +----+----+
+ | 10 | 2 |
+ | 20 | 40 |
+ +----+----+
+ ");
+
+ let anti = join(
+ kv_exec(&buffered),
+ streamed(),
+ Operator::Gt,
+ JoinType::RightAnti,
+ )?;
+ let batches =
+ common::collect(anti.execute(0, Arc::new(TaskContext::default()))?).await?;
+ assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 0);
+
+ // All three partitions were read, not just the first.
+ assert_eq!(input_batches(&semi), 1, "streamed batches");
+ let build_batches = semi
+ .metrics()
+ .unwrap()
+ .sum_by_name("build_input_batches")
+ .expect("build_input_batches metric")
+ .as_usize();
+ assert_eq!(
+ build_batches, 3,
+ "every buffered partition should be folded"
+ );
+ Ok(())
+ }
+
+ /// Unlike `LeftSemi`/`LeftAnti`, where one elected partition emits everything at the end,
+ /// each streamed partition here emits its own rows as it reads them -- so the join's N
+ /// advertised output partitions are all live.
+ #[tokio::test]
+ async fn every_streamed_partition_emits() -> Result<()> {
+ let join = join(
+ kv_exec(&[vec![kv_batch(&[(1, Some(5)), (2, Some(1))])]]),
+ kv_exec(&[
+ vec![kv_batch(&[(10, Some(4)), (20, Some(9))])],
+ vec![kv_batch(&[(30, Some(0)), (40, Some(7))])],
+ ]),
+ Operator::Gt,
+ JoinType::RightSemi,
+ )?;
+
+ assert_eq!(join.properties().output_partitioning().partition_count(), 2);
+
+ let task_ctx = Arc::new(TaskContext::default());
+ let mut per_partition = Vec::new();
+ for partition in 0..2 {
+ let stream = join.execute(partition, Arc::clone(&task_ctx))?;
+ per_partition.push(common::collect(stream).await?);
+ }
+
+ // Buffered maximum is 5: partition 0 keeps 4, partition 1 keeps 0.
+ for (partition, batches) in per_partition.iter().enumerate() {
+ assert_eq!(
+ batches.iter().map(|b| b.num_rows()).sum::<usize>(),
+ 1,
+ "partition {partition} should have emitted its own row"
+ );
+ }
+ let out = arrow::compute::concat_batches(
+ &join.schema(),
+ per_partition.iter().flatten(),
+ )?;
+ assert_snapshot!(batches_to_string(&[out]), @r"
+ +----+---+
+ | id | k |
+ +----+---+
+ | 10 | 4 |
+ | 30 | 0 |
+ +----+---+
+ ");
+ Ok(())
+ }
+}
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 11ba328c01..7fef429f54 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/utils.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/utils.rs
@@ -17,12 +17,14 @@
use datafusion_expr::JoinType;
-// Returns boolean for whether the join is a right existence join
-pub(super) fn is_right_existence_join(join_type: JoinType) -> bool {
- matches!(
- join_type,
- JoinType::RightAnti | JoinType::RightSemi | JoinType::RightMark
- )
+// 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.
+//
+// `RightMark` is deliberately excluded even though it is a right existence join: it needs the
+// buffered side walked in order and an extra boolean column, so it must not inherit this
+// stream's relaxed input requirements if the `try_new` gate is ever loosened.
+pub(super) fn is_supported_right_existence_join(join_type: JoinType) -> bool {
+ matches!(join_type, JoinType::RightSemi | JoinType::RightAnti)
}
// Returns boolean for whether the join is an existence join
@@ -38,12 +40,16 @@ pub(super) fn is_existence_join(join_type: JoinType) -> bool {
)
}
-// Returns boolean for whether the join is a left existence join that is currently
-// supported by `PiecewiseMergeJoin`. These do not require swapping the inputs: the
-// marked (left) side is already the buffered side, so `ExistencePWMJStream` can track the
-// matched suffix and slice the buffered batch at its start.
+// Returns boolean for whether the join is an existence join that is currently supported by
+// `PiecewiseMergeJoin`, which is every one of them except the Mark joins
pub(super) fn is_supported_existence_join(join_type: JoinType) -> bool {
- matches!(join_type, JoinType::LeftSemi | JoinType::LeftAnti)
+ matches!(
+ join_type,
+ JoinType::LeftSemi
+ | JoinType::LeftAnti
+ | JoinType::RightSemi
+ | JoinType::RightAnti
+ )
}
// Returns boolean to check if the join type needs to record
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index 4e01588e5a..8cba438f55 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -1984,6 +1984,20 @@ pub(crate) fn symmetric_join_output_partitioning(
Ok(result)
}
+/// Convert a boolean filter array into a unified mask bitmap.
+///
+/// Caution: The filter result is NOT a bitmap; it contains true/false/null values.
+/// For example, `1 < NULL` evaluates to NULL. Therefore, we must combine (AND)
+/// the boolean array with its null bitmap to construct a unified bitmap.
+#[inline]
+pub(crate) fn boolean_mask_from_filter(filter_arr: &BooleanArray) -> BooleanArray {
+ let (values, nulls) = filter_arr.clone().into_parts();
+ match nulls {
+ Some(nulls) => BooleanArray::new(nulls.inner() & &values, None),
+ None => BooleanArray::new(values, None),
+ }
+}
+
pub(crate) fn asymmetric_join_output_partitioning(
left: &Arc<dyn ExecutionPlan>,
right: &Arc<dyn ExecutionPlan>,
diff --git a/datafusion/proto/tests/cases/plans/joins.rs b/datafusion/proto/tests/cases/plans/joins.rs
index 84b7b46b0f..3c0628f56c 100644
--- a/datafusion/proto/tests/cases/plans/joins.rs
+++ b/datafusion/proto/tests/cases/plans/joins.rs
@@ -572,9 +572,10 @@ fn piecewise_join(
/// orderings and plan properties inside `try_new`, so only the six constructor
/// arguments travel on the wire. Cover the full cartesian product of the two
/// enum-valued ones: every range operator against every supported join type --
-/// the four classic ones plus the two left existence joins, whose output schema is
-/// the buffered side alone. (Right existence joins and Mark joins are rejected by
-/// `try_new`, so this is the complete set.)
+/// the four classic ones plus the four existence joins, whose output schema is
+/// one side alone (buffered for `LeftSemi`/`LeftAnti`, streamed for
+/// `RightSemi`/`RightAnti`). Mark joins are rejected by `try_new`, so this is
+/// the complete set.
#[test]
fn roundtrip_piecewise_merge_join() -> Result<()> {
let (schema_buffered, schema_streamed) = piecewise_schemas();
@@ -591,6 +592,8 @@ fn roundtrip_piecewise_merge_join() -> Result<()> {
JoinType::Full,
JoinType::LeftSemi,
JoinType::LeftAnti,
+ JoinType::RightSemi,
+ JoinType::RightAnti,
] {
let result = roundtrip_test_and_return(
Arc::new(PiecewiseMergeJoinExec::try_new(
@@ -615,17 +618,17 @@ fn roundtrip_piecewise_merge_join() -> Result<()> {
// Both the name and the index have to survive on the correct side.
assert_eq!(result.on.0.to_string(), "a@2");
assert_eq!(result.on.1.to_string(), "b@0");
- // The existence joins output the buffered side alone (3 fields) while the
- // classic ones output both sides (3 + 2). The before/after comparison
+ // The left existence joins output the buffered side alone (3 fields),
+ // the right existence joins the streamed side alone (2 fields), and the
+ // classic joins output both sides (3 + 2). The before/after comparison
// inside the helper already covers the schema; this pins the expected
- // width absolutely, so the existence output contract is stated rather
+ // width absolutely, so each existence output contract is stated rather
// than merely preserved.
- let expected_fields =
- if matches!(join_type, JoinType::LeftSemi | JoinType::LeftAnti) {
- 3
- } else {
- 5
- };
+ let expected_fields = match join_type {
+ JoinType::LeftSemi | JoinType::LeftAnti => 3,
+ JoinType::RightSemi | JoinType::RightAnti => 2,
+ _ => 5,
+ };
assert_eq!(
result.schema().fields().len(),
expected_fields,
@@ -866,17 +869,10 @@ fn piecewise_merge_join_rejects_unsupported_join_type_on_the_wire() -> Result<()
.expect("a plan just encoded by try_to_proto must decode as a PhysicalPlanNode");
// (wire value, description, expected error fragment)
- let cases: [(i32, &str, &str); 5] = [
- (
- protobuf::JoinType::Rightsemi as i32,
- "RightSemi",
- "Existence join RightSemi is currently not supported",
- ),
- (
- protobuf::JoinType::Rightanti as i32,
- "RightAnti",
- "Existence join RightAnti is currently not supported",
- ),
+ // RightSemi/RightAnti are supported (see `piecewise_merge_join_existence_wire_layout`
+ // and `right_existence_join.rs`); only the Mark joins and unknown variants remain
+ // unsupported on the wire.
+ let cases: [(i32, &str, &str); 3] = [
(
protobuf::JoinType::Leftmark as i32,
"LeftMark",
diff --git a/datafusion/sqllogictest/test_files/pwmj.slt b/datafusion/sqllogictest/test_files/pwmj.slt
index f0753a47da..03670c06fd 100644
--- a/datafusion/sqllogictest/test_files/pwmj.slt
+++ b/datafusion/sqllogictest/test_files/pwmj.slt
@@ -1236,5 +1236,803 @@ physical_plan
05)--------DataSourceExec: partitions=1, partition_sizes=[1]
06)------DataSourceExec: partitions=1, partition_sizes=[1]
+# ------------------------------------------------------------------
+# Right existence joins (RightSemi / RightAnti) via PiecewiseMergeJoin
+# ------------------------------------------------------------------
+# These mark the streamed (right) side, which for a single range predicate is decided by one
+# buffered key: the minimum for `<`/`<=`, the maximum for `>`/`>=`. So unlike LeftSemi/LeftAnti
+# there is no watermark and no final pass -- each streamed batch is filtered and emitted as it
+# arrives -- and no sort on either side, since a min/max is `O(B)` from any order. `EXISTS`
+# always decorrelates to a *Left* existence join, so these are reached through the explicit
+# join syntax instead.
+
+statement ok
+CREATE TABLE rex_l(v INT);
+
+statement ok
+INSERT INTO rex_l VALUES (2), (4);
+
+statement ok
+CREATE TABLE rex_r(id INT, v INT);
+
+statement ok
+INSERT INTO rex_r VALUES (1, 1), (2, 2), (3, 4), (4, 5), (5, NULL);
+
+# `>` : the deciding buffered key is max = 4. 4>{1,2}; 4 is not > 4 or 5; NULL matches nothing.
+query I
+SELECT r.id FROM rex_l l RIGHT SEMI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_l l RIGHT ANTI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+5
+
+# `>=` additionally admits r.v = 4 via the equal buffered value, so it must differ from `>`.
+query I
+SELECT r.id FROM rex_l l RIGHT SEMI JOIN rex_r r ON l.v >= r.v ORDER BY 1;
+----
+1
+2
+3
+
+query I
+SELECT r.id FROM rex_l l RIGHT ANTI JOIN rex_r r ON l.v >= r.v ORDER BY 1;
+----
+4
+5
+
+# `<` : the deciding key flips to min = 2. 2<{4,5}; 2 is not < 1 or 2.
+query I
+SELECT r.id FROM rex_l l RIGHT SEMI JOIN rex_r r ON l.v < r.v ORDER BY 1;
+----
+3
+4
+
+query I
+SELECT r.id FROM rex_l l RIGHT ANTI JOIN rex_r r ON l.v < r.v ORDER BY 1;
+----
+1
+2
+5
+
+# `<=` additionally admits r.v = 2, so it must differ from `<`.
+query I
+SELECT r.id FROM rex_l l RIGHT SEMI JOIN rex_r r ON l.v <= r.v ORDER BY 1;
+----
+2
+3
+4
+
+query I
+SELECT r.id FROM rex_l l RIGHT ANTI JOIN rex_r r ON l.v <= r.v ORDER BY 1;
+----
+1
+5
+
+# **Neither** side is sorted: `required_input_ordering` is empty for both children, so the plan
+# has no `SortExec` at all. The buffered side only needs its min/max, which is `O(B)` from any
+# order -- every other PWMJ join type walks it and does require a global sort.
+query TT
+EXPLAIN SELECT r.id FROM rex_l l RIGHT SEMI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_l projection=[v]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--ProjectionExec: expr=[id@0 as id]
+03)----PiecewiseMergeJoin: operator=Gt, join_type=RightSemi, on=(v > v)
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# A min/max combines across partitions, so the buffered side carries no `SinglePartition`
+# requirement and every partition of it is folded. A `UNION ALL` gives it two, and the deciding
+# key -- max = 7 -- is in the second one, so folding only the first would return {1,2} here.
+statement ok
+CREATE TABLE rex_l2(v INT);
+
+statement ok
+INSERT INTO rex_l2 VALUES (3), (7);
+
+query I
+SELECT r.id FROM (SELECT v FROM rex_l UNION ALL SELECT v FROM rex_l2) l
+RIGHT SEMI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+3
+4
+
+query I
+SELECT r.id FROM (SELECT v FROM rex_l UNION ALL SELECT v FROM rex_l2) l
+RIGHT ANTI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+5
+
+# `UnionExec` feeds the join directly with both partitions -- no `CoalescePartitionsExec`
+# funnelling every buffered row through one thread, which is what `SinglePartition` would force.
+# Nor is a `RepartitionExec` added to split it: `benefits_from_input_partitioning` declines that
+# for this side, since folding a batch is one linear scan and not worth a channel hop.
+query TT
+EXPLAIN SELECT r.id FROM (SELECT v FROM rex_l UNION ALL SELECT v FROM rex_l2) l
+RIGHT SEMI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------Union
+06)----------TableScan: rex_l projection=[v]
+07)----------TableScan: rex_l2 projection=[v]
+08)------SubqueryAlias: r
+09)--------TableScan: rex_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--ProjectionExec: expr=[id@0 as id]
+03)----PiecewiseMergeJoin: operator=Gt, join_type=RightSemi, on=(v > v)
+04)------UnionExec
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
+07)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# An empty buffered side has no key at all, so nothing matches: RightSemi is empty and
+# RightAnti keeps every streamed row, NULL-keyed ones included.
+statement ok
+CREATE TABLE rex_empty(v INT);
+
+query I
+SELECT r.id FROM rex_empty l RIGHT SEMI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+
+query I
+SELECT r.id FROM rex_empty l RIGHT ANTI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+3
+4
+5
+
+# The same holds when the buffered side is present but every key is NULL, since a NULL key
+# matches nothing either.
+statement ok
+CREATE TABLE rex_null_l(v INT);
+
+statement ok
+INSERT INTO rex_null_l VALUES (NULL), (NULL);
+
+query I
+SELECT r.id FROM rex_null_l l RIGHT SEMI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+
+query I
+SELECT r.id FROM rex_null_l l RIGHT ANTI JOIN rex_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+3
+4
+5
+
+# An empty streamed side has nothing to filter, either way.
+statement ok
+CREATE TABLE rex_empty_r(id INT, v INT);
+
+query I
+SELECT r.id FROM rex_l l RIGHT SEMI JOIN rex_empty_r r ON l.v > r.v ORDER BY 1;
+----
+
+query I
+SELECT r.id FROM rex_l l RIGHT ANTI JOIN rex_empty_r r ON l.v > r.v ORDER BY 1;
+----
+
+# `count(*)` needs no column from the join, but the streamed side still has to project the
+# join key, so the output is never zero-column -- which `RecordBatch::try_new` would reject.
+query I
+SELECT count(*) FROM rex_l l RIGHT SEMI JOIN rex_r r ON l.v > r.v;
+----
+2
+
+# The planner's PWMJ gate only fires when the range predicate splits cleanly one operand per
+# side. A same-side-only operand (both columns from the streamed side, none from the buffered
+# side) or a literal-only operand (no column from either side) can't be split that way, so both
+# must fall back to NestedLoopJoin -- not error or panic, which they did before this predicate
+# shape was reachable for RightSemi/RightAnti (the gate previously excluded both join types
+# outright, so `side_of`'s classification of these shapes was never exercised for them).
+statement ok
+CREATE TABLE rex_gate_l(v INT);
+
+statement ok
+INSERT INTO rex_gate_l VALUES (1), (10);
+
+statement ok
+CREATE TABLE rex_gate_r(id INT, v INT);
+
+statement ok
+INSERT INTO rex_gate_r VALUES (1, 1), (2, 20);
+
+# Same-side operand: `pr.id`/`pr.v` are both columns of the streamed side alone, so the
+# predicate never references the buffered side at all -- equivalent to filtering the streamed
+# side by `id <= v` (the complement, since RightAnti keeps rows with no buffered match and no
+# buffered row can ever be a witness for a buffered-independent predicate once the buffered
+# side is non-empty).
+query I
+SELECT r.id FROM rex_gate_l l RIGHT ANTI JOIN rex_gate_r r ON r.id > r.v ORDER BY 1;
+----
+1
+2
+
+query TT
+EXPLAIN SELECT r.id FROM rex_gate_l l RIGHT ANTI JOIN rex_gate_r r ON r.id > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightAnti Join: Filter: r.id > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_gate_l projection=[]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_gate_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--ProjectionExec: expr=[id@0 as id]
+03)----NestedLoopJoinExec: join_type=RightAnti, filter=join_proj_push_down_1@0, projection=[id@0, v@1]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------ProjectionExec: expr=[id@0 as id, v@1 as v, id@0 > v@1 as join_proj_push_down_1]
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Literal-only operand: `5` references no column from either side, so it has no side to fold
+# into either. `5 > r.v` is buffered-independent the same way, so RightAnti keeps rows where
+# `NOT (5 > r.v)`, i.e. `r.v >= 5`.
+query I
+SELECT r.id FROM rex_gate_l l RIGHT ANTI JOIN rex_gate_r r ON 5 > r.v ORDER BY 1;
+----
+2
+
+query TT
+EXPLAIN SELECT r.id FROM rex_gate_l l RIGHT ANTI JOIN rex_gate_r r ON 5 > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightAnti Join: Filter: r.v < Int32(5)
+04)------SubqueryAlias: l
+05)--------TableScan: rex_gate_l projection=[]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_gate_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--ProjectionExec: expr=[id@0 as id]
+03)----NestedLoopJoinExec: join_type=RightAnti, filter=join_proj_push_down_1@0, projection=[id@0, v@1]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------ProjectionExec: expr=[id@0 as id, v@1 as v, v@1 < 5 as join_proj_push_down_1]
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Dictionary-encoded keys compare the buffered extreme as a dictionary `Scalar`, which is a
+# different arrow path from the primitive one above.
+statement ok
+CREATE TABLE rex_dict_l AS
+ SELECT arrow_cast(column1, 'Dictionary(Int32, Utf8)') AS v FROM (VALUES ('c'), (NULL));
+
+statement ok
+CREATE TABLE rex_dict_r AS
+ SELECT column1 AS id, arrow_cast(column2, 'Dictionary(Int32, Utf8)') AS v
+ FROM (VALUES (1, 'a'), (2, 'c'), (3, 'e'), (4, NULL));
+
+# `>` : the only buffered key is 'c', so only 'a' has something above it.
+query I
+SELECT r.id FROM rex_dict_l l RIGHT SEMI JOIN rex_dict_r r ON l.v > r.v ORDER BY 1;
+----
+1
+
+query I
+SELECT r.id FROM rex_dict_l l RIGHT ANTI JOIN rex_dict_r r ON l.v > r.v ORDER BY 1;
+----
+2
+3
+4
+
+# Float keys are the case where reducing the buffered side and comparing against the result must
+# agree on an ordering. Both use a *total* order (arrow's `cmp` kernels and `min`/`max` compare
+# floats with `total_cmp`), under which NaN is the largest value rather than incomparable. So the
+# buffered maximum below is NaN, and id 5 matches through `NaN > 2.0`; were the comparison IEEE
+# while the reduction stayed total, that would be false and no other buffered key is above 2.0,
+# so the row would be lost. id 4 is excluded for the mirror reason: nothing is above NaN.
+statement ok
+CREATE TABLE rex_f_l(v DOUBLE);
+
+statement ok
+INSERT INTO rex_f_l VALUES (1.0), (arrow_cast('NaN', 'Float64'));
+
+statement ok
+CREATE TABLE rex_f_r(id INT, v DOUBLE);
+
+statement ok
+INSERT INTO rex_f_r VALUES
+ (1, arrow_cast('-Inf', 'Float64')), (2, 0.0), (3, -0.0),
+ (4, arrow_cast('NaN', 'Float64')), (5, 2.0);
+
+query I
+SELECT r.id FROM rex_f_l l RIGHT SEMI JOIN rex_f_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+3
+5
+
+query I
+SELECT r.id FROM rex_f_l l RIGHT ANTI JOIN rex_f_r r ON l.v > r.v ORDER BY 1;
+----
+4
+
+# No `SortExec` on either side for a float key either -- the operator is chosen for these types,
+# not silently replaced by a NestedLoopJoin that would answer the queries above identically.
+query TT
+EXPLAIN SELECT r.id FROM rex_f_l l RIGHT SEMI JOIN rex_f_r r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_f_l projection=[v]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_f_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--ProjectionExec: expr=[id@0 as id]
+03)----PiecewiseMergeJoin: operator=Gt, join_type=RightSemi, on=(v > v)
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# String keys reduce through arrow's string kernels rather than the primitive ones: `Utf8` via
+# `min_string`/`max_string` and `Utf8View` via their view equivalents. Same shape as the
+# dictionary case above -- buffered 'c', so only 'a' has something above it.
+statement ok
+CREATE TABLE rex_s_l AS
+ SELECT arrow_cast(column1, 'Utf8') AS v FROM (VALUES ('c'), (NULL));
+
+statement ok
+CREATE TABLE rex_s_r AS
+ SELECT column1 AS id, arrow_cast(column2, 'Utf8') AS v
+ FROM (VALUES (1, 'a'), (2, 'c'), (3, 'e'), (4, NULL));
+
+query I
+SELECT r.id FROM rex_s_l l RIGHT SEMI JOIN rex_s_r r ON l.v > r.v ORDER BY 1;
+----
+1
+
+query I
+SELECT r.id FROM rex_s_l l RIGHT ANTI JOIN rex_s_r r ON l.v > r.v ORDER BY 1;
+----
+2
+3
+4
+
+statement ok
+CREATE TABLE rex_sv_l AS
+ SELECT arrow_cast(column1, 'Utf8View') AS v FROM (VALUES ('c'), (NULL));
+
+statement ok
+CREATE TABLE rex_sv_r AS
+ SELECT column1 AS id, arrow_cast(column2, 'Utf8View') AS v
+ FROM (VALUES (1, 'a'), (2, 'c'), (3, 'e'), (4, NULL));
+
+query I
+SELECT r.id FROM rex_sv_l l RIGHT SEMI JOIN rex_sv_r r ON l.v > r.v ORDER BY 1;
+----
+1
+
+query I
+SELECT r.id FROM rex_sv_l l RIGHT ANTI JOIN rex_sv_r r ON l.v > r.v ORDER BY 1;
+----
+2
+3
+4
+
+# ------------------------------------------------------------------
+# Right existence joins: nested and dictionary key regressions, PWMJ on vs. off
+# ------------------------------------------------------------------
+# The buffered-side reduction (`min_batch`/`max_batch`) handles List/Struct/Dictionary keys
+# through the same generic path the earlier Dictionary case above exercises, but comparing the
+# reduced extreme against the streamed batch is a separate step: it goes through `apply_cmp`,
+# which dispatches List/Struct (and a dictionary of either) to a nested-aware comparator instead
+# of arrow's `lt`/`lt_eq`/`gt`/`gt_eq` kernels, which reject those types outright. Each key type
+# below runs the same query with PWMJ enabled and disabled and checks the two agree, so a
+# regression in that dispatch fails a result comparison rather than only an internal error.
+
+statement ok
+CREATE TABLE rex_list_l(v INT[]);
+
+statement ok
+INSERT INTO rex_list_l VALUES ([2]), ([4]);
+
+statement ok
+CREATE TABLE rex_list_r(id INT, v INT[]);
+
+statement ok
+INSERT INTO rex_list_r VALUES (1, [1]), (2, [2]), (3, [4]), (4, [5]), (5, NULL);
+
+query I
+SELECT r.id FROM rex_list_l l RIGHT SEMI JOIN rex_list_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_list_l l RIGHT ANTI JOIN rex_list_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+5
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = false;
+
+query TT
+EXPLAIN SELECT r.id FROM rex_list_l l RIGHT SEMI JOIN rex_list_r r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_list_l projection=[v]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_list_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--NestedLoopJoinExec: join_type=RightSemi, filter=v@0 > v@1, projection=[id@0]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I
+SELECT r.id FROM rex_list_l l RIGHT SEMI JOIN rex_list_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_list_l l RIGHT ANTI JOIN rex_list_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+5
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = true;
+
+# `LargeList` and `FixedSizeList` go through the same nested-aware `apply_cmp` dispatch as
+# `List` above (`DataType::is_nested()` is true for all three), so this pins that the dispatch
+# isn't accidentally keyed on `List` specifically. Same shape and expected rows as the `List`
+# case, cast from the same underlying array.
+statement ok
+CREATE TABLE rex_largelist_l AS SELECT arrow_cast(v, 'LargeList(Int32)') AS v FROM rex_list_l;
+
+statement ok
+CREATE TABLE rex_largelist_r AS SELECT id, arrow_cast(v, 'LargeList(Int32)') AS v FROM rex_list_r;
+
+# Confirms PWMJ, not a silent NestedLoopJoin fallback, is what the flag-on queries below
+# actually exercise -- unlike the `List` block above, which never checks this for either type.
+query TT
+EXPLAIN SELECT r.id FROM rex_largelist_l l RIGHT SEMI JOIN rex_largelist_r r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_largelist_l projection=[v]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_largelist_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--ProjectionExec: expr=[id@0 as id]
+03)----PiecewiseMergeJoin: operator=Gt, join_type=RightSemi, on=(v > v)
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I
+SELECT r.id FROM rex_largelist_l l RIGHT SEMI JOIN rex_largelist_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_largelist_l l RIGHT ANTI JOIN rex_largelist_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+5
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = false;
+
+query I
+SELECT r.id FROM rex_largelist_l l RIGHT SEMI JOIN rex_largelist_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_largelist_l l RIGHT ANTI JOIN rex_largelist_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+5
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = true;
+
+# `FixedSizeList` additionally has a statically known width, which is a different arrow layout
+# from `List`/`LargeList`'s offsets buffer. Built from literal `VALUES` rather than cast from
+# `rex_list_l`/`rex_list_r` (unlike the `LargeList` case above): a `WHERE v IS NOT NULL` filter
+# to drop the source's NULL row -- needed because a NULL `FixedSizeList` value still has to
+# report the declared list size -- repartitions the result, which would make the `EXPLAIN`
+# below depend on `target_partitions` instead of pinning the join operator choice.
+statement ok
+CREATE TABLE rex_fsl_l AS
+ SELECT arrow_cast(column1, 'FixedSizeList(1, Int32)') AS v FROM (VALUES ([2]), ([4]));
+
+statement ok
+CREATE TABLE rex_fsl_r AS
+ SELECT column1 AS id, arrow_cast(column2, 'FixedSizeList(1, Int32)') AS v
+ FROM (VALUES (1, [1]), (2, [2]), (3, [4]), (4, [5]));
+
+query TT
+EXPLAIN SELECT r.id FROM rex_fsl_l l RIGHT SEMI JOIN rex_fsl_r r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_fsl_l projection=[v]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_fsl_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--ProjectionExec: expr=[id@0 as id]
+03)----PiecewiseMergeJoin: operator=Gt, join_type=RightSemi, on=(v > v)
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I
+SELECT r.id FROM rex_fsl_l l RIGHT SEMI JOIN rex_fsl_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_fsl_l l RIGHT ANTI JOIN rex_fsl_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = false;
+
+query I
+SELECT r.id FROM rex_fsl_l l RIGHT SEMI JOIN rex_fsl_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_fsl_l l RIGHT ANTI JOIN rex_fsl_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = true;
+
+statement ok
+CREATE TABLE rex_struct_l AS
+ SELECT named_struct('v', column1) AS v FROM (VALUES (2), (4)) t(column1);
+
+statement ok
+CREATE TABLE rex_struct_r AS
+ SELECT column1 AS id, named_struct('v', column2) AS v
+ FROM (VALUES (1, 1), (2, 2), (3, 4), (4, 5)) t(column1, column2);
+
+query I
+SELECT r.id FROM rex_struct_l l RIGHT SEMI JOIN rex_struct_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_struct_l l RIGHT ANTI JOIN rex_struct_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = false;
+
+query TT
+EXPLAIN SELECT r.id FROM rex_struct_l l RIGHT SEMI JOIN rex_struct_r r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_struct_l projection=[v]
+06)------SubqueryAlias: r
+07)--------TableScan: rex_struct_r projection=[id, v]
+physical_plan
+01)SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--NestedLoopJoinExec: join_type=RightSemi, filter=v@0 > v@1, projection=[id@0]
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----DataSourceExec: partitions=1, partition_sizes=[1]
+
+query I
+SELECT r.id FROM rex_struct_l l RIGHT SEMI JOIN rex_struct_r r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_struct_l l RIGHT ANTI JOIN rex_struct_r r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = true;
+
+# `Dictionary` of a non-nested value type is not itself `is_nested()`, so `apply_cmp` takes the
+# plain kernel path rather than the nested one above -- covered here for the same on/off
+# comparison, reusing the tables the earlier Dictionary case created.
+query I
+SELECT r.id FROM rex_dict_l l RIGHT SEMI JOIN rex_dict_r r ON l.v > r.v ORDER BY 1;
+----
+1
+
+query I
+SELECT r.id FROM rex_dict_l l RIGHT ANTI JOIN rex_dict_r r ON l.v > r.v ORDER BY 1;
+----
+2
+3
+4
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = false;
+
+query I
+SELECT r.id FROM rex_dict_l l RIGHT SEMI JOIN rex_dict_r r ON l.v > r.v ORDER BY 1;
+----
+1
+
+query I
+SELECT r.id FROM rex_dict_l l RIGHT ANTI JOIN rex_dict_r r ON l.v > r.v ORDER BY 1;
+----
+2
+3
+4
+
+statement ok
+set datafusion.optimizer.enable_piecewise_merge_join = true;
+
+# A temporal key checks that the reduced extreme keeps its own type: a `Date32` minimum that came
+# back as the underlying `Int32` would fail the comparison against a `Date32` streamed column
+# rather than answer it wrongly.
+statement ok
+CREATE TABLE rex_dt_l(v DATE);
+
+statement ok
+INSERT INTO rex_dt_l VALUES (DATE '2024-03-01'), (NULL);
+
+statement ok
+CREATE TABLE rex_dt_r(id INT, v DATE);
+
+statement ok
+INSERT INTO rex_dt_r VALUES
+ (1, DATE '2024-01-15'), (2, DATE '2024-03-01'), (3, DATE '2024-06-30'), (4, NULL);
+
+query I
+SELECT r.id FROM rex_dt_l l RIGHT SEMI JOIN rex_dt_r r ON l.v > r.v ORDER BY 1;
+----
+1
+
+query I
+SELECT r.id FROM rex_dt_l l RIGHT ANTI JOIN rex_dt_r r ON l.v > r.v ORDER BY 1;
+----
+2
+3
+4
+
+# Every streamed partition emits its own rows here, rather than one elected partition emitting
+# everything at the end as LeftSemi/LeftAnti do. The filter lets the streamed side be
+# repartitioned, and the EXPLAIN below asserts that it really is.
+query I
+SELECT r.id FROM rex_l l RIGHT SEMI JOIN (SELECT * FROM rex_r WHERE id > 0) r ON l.v > r.v ORDER BY 1;
+----
+1
+2
+
+query I
+SELECT r.id FROM rex_l l RIGHT ANTI JOIN (SELECT * FROM rex_r WHERE id > 0) r ON l.v > r.v ORDER BY 1;
+----
+3
+4
+5
+
+query TT
+EXPLAIN SELECT r.id FROM rex_l l RIGHT SEMI JOIN (SELECT * FROM rex_r WHERE id > 0) r ON l.v > r.v ORDER BY 1;
+----
+logical_plan
+01)Sort: r.id ASC NULLS LAST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_l projection=[v]
+06)------SubqueryAlias: r
+07)--------Filter: rex_r.id > Int32(0)
+08)----------TableScan: rex_r projection=[id, v]
+physical_plan
+01)SortPreservingMergeExec: [id@0 ASC NULLS LAST]
+02)--SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[true]
+03)----ProjectionExec: expr=[id@0 as id]
+04)------PiecewiseMergeJoin: operator=Gt, join_type=RightSemi, on=(v > v)
+05)--------DataSourceExec: partitions=1, partition_sizes=[1]
+06)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
+07)----------FilterExec: id@0 > 0
+08)------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# The streamed side's ordering survives this join -- one output batch per streamed batch, in
+# order, with rows only removed -- so `maintains_input_order` claims it. Here that claim is
+# load-bearing rather than merely advertised: the `LIMIT` pins an ordering below the join that the
+# optimizer cannot drop, the outer `ORDER BY` requires that same ordering above it, and the plan
+# below shows no `SortExec` above the join -- it was removed because the join carries the streamed
+# side's ordering through. A wrong claim would surface as wrong row order here, not a slow plan.
+query I
+SELECT r.id FROM rex_l l RIGHT SEMI JOIN (SELECT * FROM rex_r ORDER BY id DESC LIMIT 5) r ON l.v > r.v
+ORDER BY r.id DESC;
+----
+2
+1
+
+query I
+SELECT r.id FROM rex_l l RIGHT ANTI JOIN (SELECT * FROM rex_r ORDER BY id DESC LIMIT 5) r ON l.v > r.v
+ORDER BY r.id DESC;
+----
+5
+4
+3
+
+query TT
+EXPLAIN SELECT r.id FROM rex_l l RIGHT SEMI JOIN (SELECT * FROM rex_r ORDER BY id DESC LIMIT 5) r ON l.v > r.v
+ORDER BY r.id DESC;
+----
+logical_plan
+01)Sort: r.id DESC NULLS FIRST
+02)--Projection: r.id
+03)----RightSemi Join: Filter: l.v > r.v
+04)------SubqueryAlias: l
+05)--------TableScan: rex_l projection=[v]
+06)------SubqueryAlias: r
+07)--------Sort: rex_r.id DESC NULLS FIRST, fetch=5
+08)----------TableScan: rex_r projection=[id, v]
+physical_plan
+01)ProjectionExec: expr=[id@0 as id]
+02)--PiecewiseMergeJoin: operator=Gt, join_type=RightSemi, on=(v > v)
+03)----DataSourceExec: partitions=1, partition_sizes=[1]
+04)----SortExec: TopK(fetch=5), expr=[id@0 DESC], preserve_partitioning=[false]
+05)------DataSourceExec: partitions=1, partition_sizes=[1]
+
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