feat(pwmj): support LeftSemi/LeftAnti existence joins via classic scan

[SubhamSinghal]

Which issue does this close?

Part of #17427 (Make PiecewiseMergeJoin work in DataFusion). Adds LeftSemi / LeftAnti support, one of the epic's checklist items. Supersedes the stale #18392, taking the alternative approach that @2010YOUY01 suggested there (reuse the classic join path for generality) rather than a dedicated existence stream.

Rationale for this change

An inequality-correlated EXISTS / NOT EXISTS (e.g. WHERE EXISTS (SELECT 1 FROM r WHERE l.x < r.y)) has no equi-key, so it decorrelates to a LeftSemi / LeftAnti join with a single range predicate. Today PiecewiseMergeJoinExec rejects existence joins (not_impl_err!) and these queries fall back to NestedLoopJoinExec, which is O(n*m).

Microbenchmark (20K × 20K rows, single inequality, enable_piecewise_merge_join
on vs off), added in this PR as piecewise_merge_join_semi_anti:

Case	PWMJ	NestedLoopJoin	Speedup
LeftSemi, high selectivity	~0.50 ms	~75 ms	~150×
LeftAnti, high selectivity	~0.44 ms	~75 ms	~170×
LeftSemi, low selectivity	~0.48 ms	~76 ms	~158×
LeftAnti, low selectivity	~0.48 ms	~75 ms	~155×

What changes are included in this PR?

Existence joins (LeftSemi / LeftAnti) for PiecewiseMergeJoin:

• Route LeftSemi / LeftAnti with a single range predicate to PiecewiseMergeJoinExec in the physical planner (still gated behind enable_piecewise_merge_join, default false).
• Reuse the classic scan: on the first match, mark the matching buffered-row suffix in the visited bitmap and emit from it in the final pass (LeftSemi = marked rows, LeftAnti = unmarked; NULL join keys are never marked, so they are correctly excluded from Semi and included in Anti). Only left-side columns are produced.
• RightSemi / RightAnti / Mark remain unsupported (they require swapping the inputs); they are still rejected in try_new and excluded in the planner. Left as a follow-up.
• Two small optimizations so existence marking stays O(buffered), not O(n*m): stop scanning a batch after the first match (the marked suffix is maximal), and a cross-batch low-water mark so later batches only mark not-yet-marked rows.
• Updated the operator docstring to describe the implemented classic-reuse path and keep the min/max fast-path as a documented follow-up.

Are these changes tested?

Yes.

• 33 unit tests in classic_join.rs covering LeftSemi / LeftAnti across <, <=, >, >=; NULL join keys; all-null streamed side; empty inputs; Date32 keys; multi-batch and multi-partition streamed inputs; and the low-water-mark skip branch.
• End-to-end SLT coverage in pwmj.slt for EXISTS / NOT EXISTS (including NULLs) with EXPLAIN assertions confirming the plan uses PiecewiseMergeJoin.
• The multi-partition test also guards the final-pass counter fix.

Are there any user-facing changes?

No behaviour change by default: enable_piecewise_merge_join remains false. When enabled, single-range-predicate LeftSemi / LeftAnti joins are planned as PiecewiseMergeJoin instead of NestedLoopJoin. No API changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 88.35165% with 53 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.18%. Comparing base (1f0615a) to head (aba0325).

Files with missing lines	Patch %	Lines
...n/src/joins/piecewise_merge_join/existence_join.rs	88.13%	11 Missing and 40 partials ⚠️
...ysical-plan/src/joins/piecewise_merge_join/exec.rs	90.90%	2 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #23870    +/-   ##
========================================
  Coverage   81.17%   81.18%            
========================================
  Files        1109     1110     +1     
  Lines      388117   388566   +449     
  Branches   388117   388566   +449     
========================================
+ Hits       315071   315471   +400     
- Misses      54504    54513     +9     
- Partials    18542    18582    +40     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[SubhamSinghal]

@adriangb @comphead Can you please review this PR?

[comphead]

Oh nice! @coderfender FYI

[SubhamSinghal]

@kumarUjjawal Can you review this PR? Sorry for tagging you everywhere

[kumarUjjawal]

Sorry for tagging you everywhere

No worries. I'm happy to help. Should we split the benchmark in a new pr so we can assess easily?

[SubhamSinghal]

benchmark PR: #24160

[2010YOUY01]

Thank you for this really nice PR, I got some suggestions:

• (optional) Split the implementation for semi/anti to a new stream: although the high-level algorithm is similar, semi/anti and regular joins are still different relational operator, and have different algorithm/optimization details, separating them can make the code easier to maintain. (sorry I only figured this out after writing the original suggestion in feat: Add Semi/Anti join to PiecewiseMergeJoin #18392 (comment))
  If you opt to leave it to a follow-up, we might need some renaming/doc updates to avoid confusion. The existing term classic_join means non semi/anti joins, and now we're implementing semi/anti joins inside classic_join.rs
• Move most UTs to sqllogictests: as sqls are easier to maintain and stronger, they can also also exercise potential related optimizations like swap joins. I think for UT only basic/demo tests are needed.

This high-level approach LGTM, I might not be able to do detailed follow-up review timely, but it should be good to go if others can do the review.

[kumarUjjawal]

This high-level approach LGTM, I might not be able to do detailed follow-up review timely, but it should be good to go if others can do the review.

Thanks you @2010YOUY01 I will move this forward from here.

[viirya]

Took a focused correctness pass over the LeftSemi/LeftAnti path — not a line-by-line/differential review, but I traced the four spots most likely to hide the anti-vs-semi asymmetry class of bug (cf. #24002), and empirically checked one of them. Sharing what I verified and two small suggestions. The high-level approach looks sound to me.

What I checked (all held up):

• break 'stream_rows after the first match — the correctness of stopping the batch scan on the first match rests entirely on the sort invariant (first stream row yields the maximal matching buffered suffix, so later rows can only re-mark a subset). I built an adversarial single-batch, multi-row case (left.b1 > right.b1, buffered [2,4,6,8], streamed [1,3,5,7]) and confirmed it still emits all four expected rows. Correct — but the reasoning is subtle and load-bearing.
• Empty / all-null streamed side for LeftAnti — the "emit all buffered rows" case (the failure mode from #24002, where filtering the probe side of an anti join wrongly creates output). join_left_anti_empty_right and join_left_anti_all_right_nulls cover this and assert the full buffered set. Good.
• Multi-partition final pass — fetch_sub(1, SeqCst) == 1 correctly gates the final emit to the last partition, matching the existing classic Left/Full coordination.
• NULL join keys — never marked, so correctly excluded from Semi / included in Anti.

Two non-blocking suggestions:

1. The break 'stream_rows comment explains what it does but not that its correctness depends on the streamed side being sorted in the same direction as the buffered side. Since a future change to the sort logic could silently break this, it'd help to state that invariant explicitly at the break.
2. required_input_ordering does unimplemented!() for right-existence joins (exec.rs). That's a runtime panic guarded only by the planner gate (physical_planner.rs) not routing those types here — the two are far apart. When someone implements the RightSemi/RightAnti follow-up, the natural first step (opening the planner gate) would panic the optimizer if they forget this spot. Consider not_impl_err! here instead, so it degrades to an error rather than a panic.

This dovetails with @2010YOUY01's point about classic_join.rs now hosting semi/anti — if you do split the existence path into its own stream, suggestion (1)'s invariant and the naming confusion get resolved together.

For the RightSemi/RightAnti follow-up: the swap approach turns RightAnti into LeftAnti, which relocates the NULL-key handling into a new sort/operator-flip context that the current (all Left*) NULL tests don't exercise — worth dedicated null coverage there.

Scope caveat: this is a targeted correctness pass on the four points above, not a line-by-line audit or a differential (vs NestedLoopJoin) fuzz check — so treat it as "these specific high-risk areas look correct," not a full sign-off.

[SubhamSinghal]

Thanks @2010YOUY01 @viirya @kumarUjjawal for review. I have refactored code and split existence join in separate stream. This also enabled few more optimisations. PTAL.

[comphead]

Thanks @SubhamSinghal this PR is on my list today!

[comphead]

Thanks @SubhamSinghal here first pass AI review,

And if we planning to enable existence joins, we would need becnhes and fuzz tests to prevent regressions.

P1 — Classic Left/Full PiecewiseMergeJoin can silently drop unmatched rows pre-PR; the fix ships without a regression test

Where: datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs — diff hunk around line 108 (streamed_partitions = self.streamed.output_partitioning().partition_count()), then passed to build_buffered_data in place of self.num_partitions.

Description: Pre-PR, remaining_partitions was seeded from PiecewiseMergeJoinExec::num_partitions, which the physical planner captures from session_state.config().target_partitions() at datafusion/core/src/physical_planner.rs:1791. The classic stream at classic_join.rs:223 uses
remaining_partitions.fetch_sub(1, SeqCst) == 1 as the sole trigger for the ProcessUnmatched final pass — the only path that emits unmatched buffered rows for JoinType::Left/JoinType::Full. The streamed side has Distribution::UnspecifiedDistribution (exec.rs:490-493), so EnforceDistribution only
inserts a round-robin when enable_round_robin && roundrobin_beneficial && n_rows > batch_size && current < target. Whenever that fails (enable_round_robin_repartition = false, or a source whose exact stats are <= batch_size), the streamed side has fewer partitions than target_partitions, fetch_sub(1)
never returns 1, and unmatched left/full rows are silently dropped.

Reason: Wrong result on a common configuration.

Evidence — concrete repro: target_partitions = 4, enable_round_robin_repartition = false, tables L(x) = {1, 10}, R(y) = {5}, both single-partition sources. Query: SELECT * FROM L LEFT JOIN R ON L.x < R.y. Pre-PR PWMJ: counter starts at 4, only one execute(0) fires, ProcessUnmatched never runs,
L.x = 10 is silently dropped. Same for FULL JOIN.

Scope note: The PR fixes this in exec.rs but adds no test covering the classic Left/Full path — pwmj.slt covers only join_type=Inner for the classic side (grep confirms), and the unit tests hard-code num_partitions=1. A regression that reintroduces the mismatch would slip through today's tests.

---

P1 — Extreme-key extraction uses sort_to_indices(..., Some(1)) (O(N log N) full sort) instead of the O(N) MinAccumulator/MaxAccumulator pattern DataFusion already uses

Where: datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs:430-438.

Description: The call passes Some(1) as a limit, but with nulls_first: false arrow-rs's sort_impl sets v_limit = valids.len() (arrow-ord-59.1.0/src/sort.rs:637-643) and dispatches to sort_unstable_by at limit == len, calling stdlib array.sort_unstable_by(cmp) — a full O(N log N) unstable
sort. The PR's own inline comment acknowledges this. The identical pattern (find batch extreme for join bounds) lives in hash_join's CollectLeftAccumulator at datafusion/physical-plan/src/joins/hash_join/exec.rs:1999-2068, which uses MinAccumulator/MaxAccumulator. Arrow also exposes typed
arrow::compute::min::<T> / max::<T> and min_string / max_string (arrow-arith-59.1.0/src/aggregate.rs).

Reason: Per-batch cost is O(N log N) plus an index-vector allocation and a take, where an O(N) linear scan with no allocation is idiomatic in-tree.
Evidence — concrete repro: Streamed side of 8192-row batches with no early termination via the watermark (highly selective join where the extreme never crosses the current watermark). Every batch pays ~106k comparisons + one Vec<(u32, K)> allocation to keep one row. MinAccumulator::update_batch on the
same batch is ~8192 comparisons on the packed slice with zero auxiliary allocation. This is the hot path — the whole point of the PR.

---

P2 — No benchmark despite "magnitudes faster" performance claim

Where: datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs:165-175 (doc claim); whole PR (missing bench).

Description: grep -n -i 'bench' /tmp/23870.txt returns zero matches. datafusion/core/benches/ has no PWMJ or range-join bench, and none is added. The only PWMJ-related bench-adjacent code is the pre-existing enable_piecewise_merge_join flag in benchmarks/src/tpch/run.rs and
benchmarks/src/tpcds/run.rs, which does not target LeftSemi/LeftAnti inequality joins as a first-class case.

Reason: Per the pr-review skill's benchmark rubric, performance-motivated changes must have measured evidence. Reviewers cannot verify or regression-guard the "magnitudes faster" claim without a criterion harness or a targeted TPC comparison. Future refactors of ExistencePWMJStream (e.g., the sort-based
extreme-picking or the fetch_min contention pattern) can silently regress this without CI signal.

---

P2 — No fuzz coverage for PWMJ LeftSemi/LeftAnti in join_fuzz.rs

Where: datafusion/core/tests/fuzz_cases/join_fuzz.rs (unchanged by this PR).

Description: join_fuzz.rs (1349 lines) already parametrizes LeftSemi/LeftAnti (lines 208-311), and JoinFuzzTestCase at line 668 compares SortMergeJoinExec, HashJoinExec, and NestedLoopJoinExec on randomized inputs. Adding a piecewise_merge_join() builder that consumes a range-predicate filter
is a direct extension of the same pattern (compare PiecewiseMergeJoinExec vs NestedLoopJoinExec output). No such extension is in the diff. The AtomicUsize watermark on BufferedSideData::existence_min_marked updated via fetch_min, plus the "last streamed partition to finish emits" contract, is exactly
the concurrency logic that static SQL tests cannot stress.

Reason: Per the pr-review skill, complex join operators require fuzz coverage. The algorithm relies on a cross-partition emit contract that varying partition counts and batch boundaries would exercise.

Evidence — concrete repro: target_partitions = 4, enable_round_robin_repartition = false, tables L(x) = {1, 10}, R(y) = {5}, both single-partition sources. Query: SELECT * FROM L LEFT JOIN R ON L.x < R.y. Pre-PR PWMJ: counter starts at 4, only one execute(0) fires, ProcessUnmatched never runs,
L.x = 10 is silently dropped. Same for FULL JOIN.

Scope note: The PR fixes this in exec.rs but adds no test covering the classic Left/Full path — pwmj.slt covers only join_type=Inner for the classic side (grep confirms), and the unit tests hard-code num_partitions=1. A regression that reintroduces the mismatch would slip through today's tests.

[SubhamSinghal]

@comphead

• P1 — Classic Left/Full PiecewiseMergeJoin can silently drop unmatched rows pre-PR; the fix ships without a regression test -- Added UT in 2b73c41
• P1 — Extreme-key extraction uses sort_to_indices(..., Some(1)) (O(N log N) full sort) instead of the O(N) MinAccumulator/MaxAccumulator pattern DataFusion already uses - fixed in 2b73c41
• P2 — No benchmark despite "magnitudes faster" performance claim -- benchmark PR: bench: pwmj left semi/anti join #24160
• P2 — No fuzz coverage for PWMJ LeftSemi/LeftAnti in join_fuzz.rs -- added fuzz test in d3cec13

[comphead]

Thanks @SubhamSinghal I'll check it today

[kumarUjjawal]

Looks good!

[comphead]

Thnaks @SubhamSinghal we still gated by enable_piecewise_merge_join so the change is beta for users.

@viirya do you have anything to add?

lets have this PR open until next week to let other people a chance to be familiar with it

[viirya]

Came back for a closer look now that the existence path is its own ExistencePWMJStream — the split reads much cleaner, and the min/max-extreme + binary-search approach is a nice improvement over the earlier scan. LGTM.

I traced the new algorithm and then verified it with a differential fuzz against NestedLoopJoin (same SQL, enable_piecewise_merge_join on vs off).

Algorithm (traced, all correct):

• Extreme-key choice matches the operator: </<= → descending → max_batch, >/>= → ascending → min_batch. For l < r a buffered row matches iff l < max(r), so collapsing the whole streamed side to its extreme is sound.
• The compare == Less test in the binary search is correct precisely because JoinKeyComparator builds the comparator with the operator's SortOptions, so Less under a descending comparator means the streamed key sorts ahead of the buffered one — i.e. the value-level match condition. match_on_equal handles the <=/>= inclusive boundary.
• NULLs: the extreme key ignores nulls; buffered nulls sit at the front (nulls_first) and the scan starts past them, so they're correctly excluded from LeftSemi / included in LeftAnti.
• Cross-batch / cross-partition marking via the atomic existence_min_marked + fetch_min — the "marked set is always the contiguous suffix [min_marked, len)" invariant holds because each batch's extreme reaches the smallest matching index.

Differential fuzz (LeftSemi/LeftAnti vs NestedLoopJoin): 300 random datasets × {<,<=,>,>=} × {EXISTS, NOT EXISTS}, then a hardened pass with 4 partitions, up to 400 rows/side, 0–100% NULL density, and a tight value domain (many ties). 4800 checks, 0 mismatches. The existence path matches NestedLoopJoin exactly.