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 87.39206% with 73 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.03%. Comparing base (592eeab) to head (a16a489).
⚠️ Report is 177 commits behind head on main.

Files with missing lines	Patch %	Lines
...lan/src/joins/piecewise_merge_join/classic_join.rs	87.45%	8 Missing and 63 partials ⚠️
...ysical-plan/src/joins/piecewise_merge_join/exec.rs	71.42%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #23870      +/-   ##
==========================================
+ Coverage   80.71%   81.03%   +0.31%     
==========================================
  Files        1090     1105      +15     
  Lines      370339   381191   +10852     
  Branches   370339   381191   +10852     
==========================================
+ Hits       298925   308895    +9970     
- Misses      53605    54002     +397     
- Partials    17809    18294     +485     

☔ 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.