feat(pwmj): support RightSemi/RightAnti existence joins

[SubhamSinghal]

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

[SubhamSinghal]

benchmark PR: #24458

[codecov-commenter]

Codecov Report

❌ Patch coverage is 87.96434% with 81 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.63%. Comparing base (d2b626c) to head (597b29e).
⚠️ Report is 9 commits behind head on main.

Files with missing lines	Patch %	Lines
...joins/piecewise_merge_join/right_existence_join.rs	88.08%	20 Missing and 36 partials ⚠️
...ysical-plan/src/joins/piecewise_merge_join/exec.rs	86.91%	6 Missing and 19 partials ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #24457    +/-   ##
========================================
  Coverage   81.63%   81.63%            
========================================
  Files        1123     1124     +1     
  Lines      409537   410470   +933     
  Branches   409537   410470   +933     
========================================
+ Hits       334308   335100   +792     
- Misses      55594    55642    +48     
- Partials    19635    19728    +93     

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

@comphead @kumarUjjawal PR for right semi/anti join. Please help in reviewing this PR. Thanks.

[comphead]

Thanks @SubhamSinghal I'll check this today

[comphead]

@SubhamSinghal sorry for the delay, last time we got some correctness issues for PWJM and Im planning to merge first #24390 to create a generic join test set and we can check this PR against it

[comphead]

Once #24863 merged we need to rebase and proceed with this PR

[comphead]

Findings

No P0/P1/P2 correctness, performance, memory, or concurrency issues found. All items below are P3 (documentation and defense-in-depth).

P3: Doc references a non-existent fuzz test name

Description: right_existence_join.rs:325 cites the differential fuzz test as fuzz_pwmj_existence_matches_nested_loop. The actual test is fuzz_pwmj_matches_nested_loop.

Reason: A maintainer searching for that name finds nothing. A repo-wide grep confirms the string exists only in this comment.

Evidence: join_fuzz.rs:1564 defines fuzz_pwmj_matches_nested_loop. No symbol fuzz_pwmj_existence_matches_nested_loop exists in datafusion/.

P3: Stale maintains_input_order TODO contradicts the code

Description: exec.rs:464-466 still reads "Now they're all false indicating it will not maintain the input order". The arm added right below returns vec![false, true] for RightSemi/RightAnti.

Reason: The comment is now factually wrong for the function it heads.

Evidence: exec.rs:468-474 returns true for the streamed side. That ordering is load-bearing, since the LIMIT SLT case relies on it.

P3: Fuzz coverage is Int32-only, so the float/NaN path is deterministic-only

Description: fuzz_pwmj_matches_nested_loop keys are Option<i32>. The subtlest correctness point (float total order, with NaN as max and -0.0 versus +0.0) is exercised only by fixed pwmj.slt
rows.

Reason: The deterministic SLT case (rex_f_*) is well-targeted and the reduce-then-con by the shared kernel, so this is defense-in-depth rather than a gap. A float key type inthe fuzz generator would add differential coverage of NaN, -0.0, and ±Inf against NestedLoopJoinExec.

Evidence: join_fuzz.rs gen_keys produces i32. Float edge cases appear only in pwmj.slt.

Note for the author (non-blocking)

-0.0 handling differs between the two existence-join families. The left stream normalizes -0.0 in JoinKeyComparator (existence_join.rs:446-447). The right stream uses the raw Arrow cmp kernels with no normalization, where -0.0 < +0.0 under total_cmp. This is correct for this PR, becamatches its NestedLoopJoinExec oracle exactly (same kernel) and stays internallyconsistent with the min/max reduction. Flagging it only because of the in-flight cross-DataFusion -0.0 equality effort. Was the asymmetry a conscious choice to match NLJ rather than the left stream?

[SubhamSinghal]

@comphead Addressed review comment in 597b29e commit.