feat: Support RightMark join for NestedLoop and Hash join #16083

jonathanc-n · 2025-05-18T05:26:29Z

Which issue does this PR close?

Part of #13138 .

Rationale for this change

Revamp implementation of the previous stale implementation for RightMark

What changes are included in this PR?

Added support for right mark join functionality to NestedLoop + Hash join. A follow up pr will be made for supporting right mark in Sortmergejoin and sym hash join.

Are these changes tested?

Yes unit tests

jonathanc-n · 2025-05-18T06:04:49Z

cc @comphead

datafusion/physical-plan/src/joins/nested_loop_join.rs

2010YOUY01 · 2025-05-19T09:35:02Z

This is cool! I got some questions:

Can we test this feature through the SQL interface (some SQL with subqueries got optimized into RightMarkJoin)? Or maybe this feature is not supported yet.
Is RightMarkJoin equivalent to LeftMarkJoin with left and right child swapped? (just picturing how should mark joins be supported for memory-limited cases, it's fine for now since NLJ and HJ will always buffer all input from the build side)

jonathanc-n · 2025-05-20T21:10:50Z

@2010YOUY01 I think i'lll try to get sql queries optimized into a right mark join after support for symmetric hash join + sort merge join.

Right mark is equivalent to left mark however internally I didn't swap during planning time. I had just marked a column on the probe batch as probe batches were incoming.

jonathanc-n · 2025-05-26T17:01:45Z

@2010YOUY01 @Dandandan Is it possible to take a look? Thanks!

jonathanc-n · 2025-05-30T01:30:41Z

@alamb Are you able to add this pull request to here to get some eyes on it?

ctsk · 2025-05-30T20:05:51Z

Really cool work! I'll look over it in more detail over the weekend =)

ctsk · 2025-05-30T20:39:38Z

After updating JoinType::supports_swap to include LeftMark/RightMark join, the join_selection rule should already plan right joins where appropriate. Subsequently running the sqllogictests (cargo test --test sqllogictests) reveals some failures:

External error: 6 errors in file [...]/datafusion/datafusion/sqllogictest/test_files/subquery.slt

1. query failed: DataFusion error: join_selection
caused by
Internal error: Input field name t1_name does not match with the projection expression t1_id.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker
[SQL] select t1.t1_id,
       t1.t1_name,
       t1.t1_int
from t1
where t1.t1_id > 40 or t1.t1_id in (select t2.t2_id from t2 where t1.t1_int > 0)
at [...]/datafusion/datafusion/sqllogictest/test_files/subquery.slt:1108


2. query failed: DataFusion error: join_selection
caused by
Internal error: Input field name t1_name does not match with the projection expression t1_id.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker
[SQL] select t1.t1_id,
       t1.t1_name,
       t1.t1_int
from t1
where t1.t1_id = 11 or t1.t1_id + 12 not in (select t2.t2_id + 1 from t2 where t1.t1_int > 0)
at [...]/datafusion/datafusion/sqllogictest/test_files/subquery.slt:1136


3. query failed: DataFusion error: join_selection
caused by
Internal error: Input field name t1_name does not match with the projection expression t1_id.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker
[SQL] select t1.t1_id,
       t1.t1_name,
       t1.t1_int
from t1
where t1.t1_id > 40 or exists (select * from t2 where t1.t1_id = t2.t2_id)
at [...]/datafusion/datafusion/sqllogictest/test_files/subquery.slt:1162


4. query result mismatch:
[SQL] explain select t1.t1_id,
       t1.t1_name,
       t1.t1_int
from t1
where t1.t1_id > 40 or not exists (select * from t2 where t1.t1_id = t2.t2_id)
[Diff] (-expected|+actual)
    logical_plan
    01)Projection: t1.t1_id, t1.t1_name, t1.t1_int
    02)--Filter: t1.t1_id > Int32(40) OR NOT __correlated_sq_1.mark
    03)----LeftMark Join: t1.t1_id = __correlated_sq_1.t2_id
    04)------TableScan: t1 projection=[t1_id, t1_name, t1_int]
    05)------SubqueryAlias: __correlated_sq_1
-   06)--------TableScan: t2 projection=[t2_id]
-   physical_plan
-   01)CoalesceBatchesExec: target_batch_size=2
-   02)--FilterExec: t1_id@0 > 40 OR NOT mark@3, projection=[t1_id@0, t1_name@1, t1_int@2]
-   03)----CoalesceBatchesExec: target_batch_size=2
-   04)------HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(t1_id@0, t2_id@0)]
-   05)--------DataSourceExec: partitions=1, partition_sizes=[1]
-   06)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
-   07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+   06)--------TableScan: t2 projection=[t2_id]
at [...]/datafusion/datafusion/sqllogictest/test_files/subquery.slt:1177


5. query failed: DataFusion error: join_selection
caused by
Internal error: Input field name t1_name does not match with the projection expression t1_id.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker
[SQL] select t1.t1_id,
       t1.t1_name,
       t1.t1_int
from t1
where t1.t1_id > 40 or not exists (select * from t2 where t1.t1_id = t2.t2_id)
at [...]/datafusion/datafusion/sqllogictest/test_files/subquery.slt:1203


6. query failed: DataFusion error: join_selection
caused by
Internal error: Input field name t1_name does not match with the projection expression t1_id.
This was likely caused by a bug in DataFusion's code and we would welcome that you file an bug report in our issue tracker
[SQL] select t1.t1_id,
       t1.t1_name,
       t1.t1_int
from t1
where t1.t1_id > 40 or exists (select * from t2 cross join t3 where t1.t1_id = t2.t2_id)
at [...]/datafusion/datafusion/sqllogictest/test_files/subquery.slt:1244



External error: task 14575 panicked with message "index out of bounds: the len is 6 but the index is 6"
External error: task 17017 panicked with message "index out of bounds: the len is 6 but the index is 6"
Error: Execution("3 failures")
error: test failed, to rerun pass `-p datafusion-sqllogictest --test sqllogictests`

</details>

datafusion/sql/src/unparser/plan.rs

datafusion/physical-plan/src/joins/utils.rs

datafusion/physical-plan/src/joins/sort_merge_join.rs

comphead

Thanks @jonathanc-n I quickly went through the PR
It would be nice to have fuzz tests for RightMark join, I'm not sure 3 tests would cover all possible cases, WDYT?

datafusion/physical-plan/src/joins/utils.rs

ctsk · 2025-06-01T16:32:00Z

Alrighty!

Regarding the lack of support for RightMark joins in some join operators, I believe it would be best to return an error in the constructor of those operators if they do not support the JoinType. I believe the PR currently would just return a wrong result.

I don't (yet) know how much work it would be to adjust physical optimizer rules so that no bad JoinType x OperatorType combination gets constructed.

Algorithmically, what I found hardest to understand is the switcheroo between build and probe side - You've modified adjust_indices_by_join_type for then RightMark join so that left_indices holds the indices for the probe batch and right_indices holds the mark. Then you also switch probe_batch, build_input_buffer and the build_sidein the call tobuild_batch_from_indices` so that it all works out..... It's kind of brilliant, but would be best to at least document that this is going on.

I think a less confusing alternative would be to just add a mark_side: JoinSide parameter to build_batch_from_indices that lets it distinguish between LeftMark and RightMark join for the ColumnSide::None case.

jonathanc-n · 2025-06-02T02:30:49Z

Regarding the lack of support for RightMark joins in some join operators, I believe it would be best to return an error in the constructor of those operators if they do not support the JoinType. I believe the PR currently would just return a wrong result.

I think instead we should just not support swapping for now and support it after all the join operators are supported. Returning an error will cause some tests to fail.

ctsk · 2025-06-11T08:17:41Z

datafusion/optimizer/src/optimize_projections/mod.rs

-        | JoinType::LeftMark => {
+        | JoinType::LeftMark
+        | JoinType::RightMark => {


I admit that I don't understand why LeftMark/RightMark are different from their Semi/Anti counterparts. It seems like the documentation of this method is missing LeftMark / RightMark too

Added the doc! I'm not quite sure either, I think it has to do with the extra boolean column crossing the boundary into the other side, resulting in the mark column being missing from the requirements

ctsk · 2025-06-11T08:31:27Z

datafusion/physical-plan/src/joins/utils.rs

+    let mut bitmap = BooleanBufferBuilder::new(range.len());
+    bitmap.append_n(range.len(), false);
+    input_indices
+        .iter()
+        .flatten()
+        .map(|v| v.as_usize())
+        .filter(|v| range.contains(v))
+        .for_each(|v| {
+            bitmap.set_bit(v - range.start, true);
+        });


I noticed that these lines are the same for get_mark_indices, get_semi_indices and get_anti_indices. We could factor them out to a new method?

map.filter.foreach wondering can it be done in single iteration?

.for_each(|v| { let idx = v.as_usize(); if range.contains(&idx) { builder.set_bit(idx - range.start, true); } });

Could do something like this?

I havI have refactored to use this

datafusion/physical-plan/src/joins/utils.rs

Co-authored-by: Christian <[email protected]>

comphead · 2025-06-12T07:15:38Z

datafusion/core/tests/fuzz_cases/join_fuzz.rs

@@ -305,6 +305,31 @@ async fn test_left_mark_join_1k_filtered() {
    .await
 }

+// todo: add JoinTestType::HjSmj after Right mark SortMergeJoin support


It is needed to add RightMark Join support to SortMergeJoin execution as well

Oh I will close #16226 as it is a duplicate

comphead · 2025-06-15T04:44:27Z

datafusion/physical-plan/src/joins/symmetric_hash_join.rs

+                .collect::<PrimitiveArray<L>>();
+            let probe_indices = (0..prune_length)
+                .map(|idx| {
+                    // For mark join we output a dummy index 0 to indicate the row had a match


that would be nice to show as an example, it is challenging to read algorithms for upcoming contributors, but it can be done as follow up PR

Tracked in #16415

comphead

Thanks @jonathanc-n for bearing with this PR, I think it is good to go.

comphead · 2025-06-15T04:51:05Z

Thanks for this contribution, I'm planning to have this PR open for a little bit of more time to see if there are any other feedbacks

alamb · 2025-06-16T21:04:36Z

🚀 -- I am feeling physically nervous that there are so many PRs open so starting the merge train!

feat: Support RightMark join for NestedLoop and Hash join

d3ef287

jonathanc-n commented May 18, 2025

View reviewed changes

datafusion/physical-plan/src/joins/nested_loop_join.rs Show resolved Hide resolved

Dandandan reviewed May 30, 2025

View reviewed changes

datafusion/sql/src/unparser/plan.rs Outdated Show resolved Hide resolved

Dandandan reviewed May 30, 2025

View reviewed changes

datafusion/physical-plan/src/joins/utils.rs Outdated Show resolved Hide resolved

comphead reviewed May 31, 2025

View reviewed changes

datafusion/physical-plan/src/joins/sort_merge_join.rs Outdated Show resolved Hide resolved

comphead reviewed May 31, 2025

View reviewed changes

Dandandan reviewed Jun 1, 2025

View reviewed changes

datafusion/physical-plan/src/joins/utils.rs Outdated Show resolved Hide resolved

jonathanc-n mentioned this pull request Jun 1, 2025

Support RightMark join for SortMergeJoin #16226

Closed

fixes

87a058b

github-actions bot added the physical-plan Changes to the physical-plan crate label Jun 2, 2025

jonathanc-n added 3 commits June 1, 2025 22:46

Merge remote-tracking branch 'upstream/main' into right-mark

dd92f4e

producer fix

69b9dfb

fmt

8f7f429

jonathanc-n added 4 commits June 2, 2025 15:30

rem file

396a45b

Merge branch 'main' into right-mark

86e1c60

Merge branch 'main' into right-mark

3bb79b5

Merge branch 'main' into right-mark

0ae7414

jonathanc-n requested review from Dandandan, ctsk and comphead June 9, 2025 00:44

jonathanc-n added 3 commits June 8, 2025 20:46

Merge branch 'main' into right-mark

ecd904c

fix

b01e95d

fmt

1bef6dc

ctsk reviewed Jun 11, 2025

View reviewed changes

datafusion/physical-plan/src/joins/utils.rs Outdated Show resolved Hide resolved

jonathanc-n and others added 4 commits June 11, 2025 17:50

Update datafusion/physical-plan/src/joins/utils.rs

ed8cd56

Co-authored-by: Christian <[email protected]>

fixes

e9978b8

clippy

92ecd2b

Merge branch 'main' into right-mark

ae76056

comphead reviewed Jun 12, 2025

View reviewed changes

comphead mentioned this pull request Jun 12, 2025

RightMark Join support to SortMergeJoin execution #16385

Open

jonathanc-n added 3 commits June 12, 2025 04:30

refactor

c0a4185

Merge branch 'main' into right-mark

a58d516

Merge branch 'main' into right-mark

92a2b11

comphead reviewed Jun 15, 2025

View reviewed changes

comphead approved these changes Jun 15, 2025

View reviewed changes

Merge branch 'main' into right-mark

e9a8d48

jonathanc-n mentioned this pull request Jun 15, 2025

Add documentation to clarify algorithms for Mark Joins #16415

Closed

alamb merged commit 7d16764 into apache:main Jun 16, 2025
28 checks passed

jonathanc-n deleted the right-mark branch June 16, 2025 23:36

feat: Support RightMark join for NestedLoop and Hash join #16083

feat: Support RightMark join for NestedLoop and Hash join #16083

Uh oh!

Conversation

jonathanc-n commented May 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Uh oh!

jonathanc-n commented May 18, 2025

Uh oh!

Uh oh!

2010YOUY01 commented May 19, 2025

Uh oh!

jonathanc-n commented May 20, 2025

Uh oh!

jonathanc-n commented May 26, 2025

Uh oh!

jonathanc-n commented May 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ctsk commented May 30, 2025

Uh oh!

ctsk commented May 30, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ctsk commented Jun 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jonathanc-n commented Jun 2, 2025

Uh oh!

ctsk Jun 11, 2025 • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jonathanc-n Jun 12, 2025 • 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!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

comphead left a comment

Choose a reason for hiding this comment

Uh oh!

comphead commented Jun 15, 2025

Uh oh!

Uh oh!

alamb commented Jun 16, 2025

Uh oh!

Uh oh!

jonathanc-n commented May 18, 2025 •

edited

Loading

jonathanc-n commented May 30, 2025 •

edited

Loading

ctsk commented Jun 1, 2025 •

edited

Loading

ctsk Jun 11, 2025 •

edited

Loading

jonathanc-n Jun 12, 2025 •

edited

Loading