feat: Add Semi/Anti join to PiecewiseMergeJoin #18392

jonathanc-n · 2025-10-30T22:43:19Z

Which issue does this PR close?

Part of [EPIC]: Make PiecewiseMergeJoin work in Datafusion #17427 .

Rationale for this change

Adds Semi/Anti join to piecewisemerge join.

What changes are included in this PR?

Adds swap inputs. refactored a bit to hold both sides of the partition in case of swap so that the correct number of partitions is handed the buffer side data.
Created new file for ExistenceJoin stream implementation.

Are these changes tested?

Slt tests + unit tests

Are there any user-facing changes?

…usion into pwmj-existence

jonathanc-n · 2025-10-30T22:57:58Z

@2010YOUY01 I noticed after coding that the collect buffered batch can just be loaded after when it is needed for the ProcessMatched however this would involve refactoring the num_partitions to be stored in PiecewiseMergeJoinExec instead of Buffered side data since the classic join file needs it.

I think I'm going to do this in a follow up instead to keep the current changes be only relevant for the existence join + to reduce noise.

jonathanc-n

@2010YOUY01 @comphead PR is ready for review, i added some review comments that you can read first to help with the whole review process.

Here are how the benchmarks are:

Queries:

Click to expand


r#"
        SELECT t1.*
        FROM range(30000) AS t1
        LEFT SEMI JOIN range(30000) AS t2
        ON t1.value < t2.value;
    "#,
    r#"
        SELECT t1.*
        FROM range(30000) AS t1
        LEFT ANTI JOIN range(30000) AS t2
        ON t1.value < t2.value;
    "#,
    r#"
        SELECT t1.*
        FROM range(30000) AS t2
        RIGHT SEMI JOIN range(30000) AS t1
        ON t2.value < t1.value;
    "#,
    r#"
        SELECT t1.*
        FROM range(30000) AS t2
        RIGHT ANTI JOIN range(30000) AS t1
        ON t2.value < t1.value;
    "#,
    r#"
        set datafusion.optimizer.enable_piecewise_merge_join = true;
    "#,
    r#"
        SELECT t1.*
        FROM range(30000) AS t1
        LEFT SEMI JOIN range(30000) AS t2
        ON t1.value < t2.value;
    "#,
    r#"
        SELECT t1.*
        FROM range(30000) AS t1
        LEFT ANTI JOIN range(30000) AS t2
        ON t1.value < t2.value;
    "#,
    r#"
        SELECT t1.*
        FROM range(30000) AS t2
        RIGHT SEMI JOIN range(30000) AS t1
        ON t2.value < t1.value;
    "#,
    r#"
        SELECT t1.*
        FROM range(30000) AS t2
        RIGHT ANTI JOIN range(30000) AS t1
        ON t2.value < t1.value;
    "#,

Benchmark results (~100-400 x faster) 🔥

Click to expand


Query 1 iteration 0 returned 29999 rows in 92.105833ms
Query 1 iteration 1 returned 29999 rows in 80.096958ms
Query 1 iteration 2 returned 29999 rows in 77.611ms
Query 1 iteration 3 returned 29999 rows in 80.158458ms
Query 1 iteration 4 returned 29999 rows in 84.724ms
Query 2 iteration 0 returned 1 rows in 80.679084ms
Query 2 iteration 1 returned 1 rows in 92.0405ms
Query 2 iteration 2 returned 1 rows in 115.497709ms
Query 2 iteration 3 returned 1 rows in 143.337208ms
Query 2 iteration 4 returned 1 rows in 79.985583ms
Query 3 iteration 0 returned 29999 rows in 80.447292ms
Query 3 iteration 1 returned 29999 rows in 86.385708ms
Query 3 iteration 2 returned 29999 rows in 84.197166ms
Query 3 iteration 3 returned 29999 rows in 83.120125ms
Query 3 iteration 4 returned 29999 rows in 83.668042ms
Query 4 iteration 0 returned 1 rows in 85.033042ms
Query 4 iteration 1 returned 1 rows in 80.9975ms
Query 4 iteration 2 returned 1 rows in 85.006625ms
Query 4 iteration 3 returned 1 rows in 81.335333ms
Query 4 iteration 4 returned 1 rows in 81.747542ms
using pwmj
Query 5 iteration 0 returned 0 rows in 25.792µs
Query 5 iteration 1 returned 0 rows in 21.792µs
Query 5 iteration 2 returned 0 rows in 20.083µs
Query 5 iteration 3 returned 0 rows in 19.208µs
Query 5 iteration 4 returned 0 rows in 18.708µs
using pwmj
Query 6 iteration 0 returned 29999 rows in 518.542µs
Query 6 iteration 1 returned 29999 rows in 283.834µs
Query 6 iteration 2 returned 29999 rows in 322.459µs
Query 6 iteration 3 returned 29999 rows in 299.458µs
Query 6 iteration 4 returned 29999 rows in 291.084µs
using pwmj
Query 7 iteration 0 returned 1 rows in 291.875µs
Query 7 iteration 1 returned 1 rows in 217.792µs
Query 7 iteration 2 returned 1 rows in 213.459µs
Query 7 iteration 3 returned 1 rows in 297.416µs
Query 7 iteration 4 returned 1 rows in 296.375µs
using pwmj
Query 8 iteration 0 returned 29999 rows in 266.041µs
Query 8 iteration 1 returned 29999 rows in 258.416µs
Query 8 iteration 2 returned 29999 rows in 231.209µs
Query 8 iteration 3 returned 29999 rows in 204.875µs
Query 8 iteration 4 returned 29999 rows in 318.417µs
using pwmj
Query 9 iteration 0 returned 1 rows in 224.042µs
Query 9 iteration 1 returned 1 rows in 221.208µs
Query 9 iteration 2 returned 1 rows in 271.584µs
Query 9 iteration 3 returned 1 rows in 225.375µs
Query 9 iteration 4 returned 1 rows in 270µs

jonathanc-n · 2025-10-30T22:59:06Z

datafusion/physical-optimizer/src/join_selection.rs

-                    collect_threshold_byte_size,
-                    collect_threshold_num_rows,
-                )?
+    let transformed = if let Some(hash_join) =


I don't know why these showed a diff but there isn't anything different here except for the addition of the PiecewiseMergeJoin branch for swapping.

jonathanc-n · 2025-10-30T23:00:20Z

datafusion/physical-optimizer/src/join_selection.rs

+                // PiecewiseMergeJoin wants the left side to be the larger one, so only swap if 
+                // left < right
+                && (!should_swap_join_order(&**left, &**right)?
+                || matches!(pwmj.join_type(), JoinType::RightSemi | JoinType::RightAnti))


LeftSemi and Left Anti do not swap and only right existence joins do. This is explained above swap_inputs in PiecewiseMergeJoinExec

jonathanc-n · 2025-10-30T23:00:55Z

datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs

 use crate::joins::utils::{BuildProbeJoinMetrics, StatefulStreamResult};

-pub(super) enum PiecewiseMergeJoinStreamState {
+pub(super) enum ClassicPWMJStreamState {


Renamed to make it more distinct between Existence join streamstate and Classic join stream state.

jonathanc-n · 2025-10-30T23:03:19Z

datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs

-                } else {
-                    SortOptions::new(false, true)
-                }
+                SortOptions::new(true, true)


This was changed because for whatever reason I had thought that swapping happened before join creation 😆

jonathanc-n · 2025-10-30T23:03:42Z

datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs

                "PiecewiseMergeJoinExec requires valid sort expressions for its left side"
            );
        };
-        let Some(right_batch_required_orders) =


No longer needed, only the left side needs to be sorted.

jonathanc-n · 2025-10-30T23:04:09Z

datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs


-    // TODO
+    // Inner, Outer, Left, Right joins are swapped so that left side is larger than right
+    // Left Semi/Anti joins are not swapped while Right Semi/Anti joins are always swapped. This


This will explain the logic for why only Right semi/anti joins are swapped.

datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs

jonathanc-n · 2025-10-30T23:54:43Z

datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs

+    }
+}
+
+// `ExistencePWMJStreamState` is separated into `CollectBufferedSide`, `FetchAndProcessStreamBatch`,


Read the main logic here, after the min//max value is found for the streamed side, then the buffered side is probed. After it finds the first match it will emit all rows from matched_idx - buffered_batch_length for Semi joins (matches) and 0 - matched_idx for anti joins (non-matches)

You can refer back to the PiecewiseMergeJoinExec docs if need a refresher.

feat: Add Semi/Anti join to PiecewiseMergeJoin

af4f5e7

github-actions bot added optimizer Optimizer rules core Core DataFusion crate sqllogictest SQL Logic Tests (.slt) physical-plan Changes to the physical-plan crate labels Oct 30, 2025

Merge branch 'main' into pwmj-existence

233d7a3

jonathanc-n mentioned this pull request Oct 30, 2025

[EPIC]: Make PiecewiseMergeJoin work in Datafusion #17427

Open

jonathanc-n added 3 commits October 30, 2025 18:47

Clarify swapping

6861080

Merge branch 'pwmj-existence' of https://github.com/jonathanc-n/dataf…

ca86eaa

…usion into pwmj-existence

clippy

8bd2a1b

jonathanc-n commented Oct 31, 2025

View reviewed changes

jonathanc-n added 2 commits October 30, 2025 20:29

Update datafusion/physical-plan/src/joins/piecewise_merge_join/exec.rs

9055043

fix test

89a8b2b

2010YOUY01 self-requested a review November 2, 2025 01:48

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat: Add Semi/Anti join to PiecewiseMergeJoin #18392

feat: Add Semi/Anti join to PiecewiseMergeJoin #18392

jonathanc-n commented Oct 30, 2025

Uh oh!

jonathanc-n commented Oct 30, 2025

Uh oh!

jonathanc-n left a comment •

edited

Loading

Uh oh!

jonathanc-n Oct 30, 2025

Uh oh!

jonathanc-n Oct 30, 2025

Uh oh!

jonathanc-n Oct 30, 2025

Uh oh!

jonathanc-n Oct 30, 2025

Uh oh!

jonathanc-n Oct 30, 2025

Uh oh!

jonathanc-n Oct 30, 2025

Uh oh!

Uh oh!

jonathanc-n Oct 30, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

feat: Add Semi/Anti join to PiecewiseMergeJoin #18392

Are you sure you want to change the base?

feat: Add Semi/Anti join to PiecewiseMergeJoin #18392

Conversation

jonathanc-n commented Oct 30, 2025

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

jonathanc-n commented Oct 30, 2025

Uh oh!

jonathanc-n left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jonathanc-n Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

jonathanc-n Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

jonathanc-n Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

jonathanc-n Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

jonathanc-n Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

jonathanc-n Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jonathanc-n Oct 30, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jonathanc-n left a comment •

edited

Loading