Fix: Sort Merge Join LeftSemi issues when JoinFilter is set #10304

comphead · 2024-04-29T23:56:52Z

Which issue does this PR close?

Closes #10379 .

Rationale for this change

Fixing some existing SMJ LeftSemi bugs when join filter is set. Currently the join either crashes or giving wrong results

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

comphead · 2024-04-29T23:57:31Z

I was able to fix initial query but now stuck on

Error: ArrowError(InvalidArgumentError("all columns in a record batch must have the same length"), None)

Likely related to nulls

comphead · 2024-05-07T14:26:50Z

fuzztests failing...

comphead · 2024-05-07T16:41:49Z

datafusion/core/tests/fuzz_cases/join_fuzz.rs

@@ -79,20 +79,20 @@ async fn test_full_join_1k() {
 }

 #[tokio::test]
-async fn test_semi_join_10k() {
+async fn test_semi_join_1k() {


Made it 1k to be consistent with other join fuzz tests

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

viirya · 2024-05-07T19:20:58Z

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

@@ -991,6 +992,9 @@ impl SMJStream {
            Ordering::Equal => {
                if matches!(self.join_type, JoinType::LeftSemi) {
                    join_streamed = !self.streamed_joined;
+                    // if the join filter specified there can be references to buffered columns
+                    // so its needed to join them
+                    join_buffered = self.filter.is_some();


LeftSemi doesn't join buffered side, why we want to do this?

Oh, I see. As the filter uses buffered columns, we need to access to it.

But the comment doesn't look correct as we don't actually join buffered columns.

Hmm, but if there are more rows at buffered side are matched on keys, won't it add additional joined pairs with nulls and buffered rows?

I hope I got the concern right and wrapped it into SQL query

query II select * from ( with t1 as ( select 11 a, 12 b union all select 11 a, 13 b), t2 as ( select 11 a, 12 b union all select 11 a, 12 b union all select 11 a, 14 b ) select t1.* from t1 where exists (select 1 from t2 where t2.a = t1.a and t2.b != t1.b) ) order by 1, 2; ---- 11 12 11 13

it passes, I can add it to slt file as well

I'm not able to run such test just because of other SMJ issue not related to this PR:
If the join filter is set and for the same streaming index there are matching rows more or equal to a batch size then the query just stuck. Likely the problem is in polling state and it can be easily reproduced on main branch.

#[tokio::test] async fn test_11() -> Result<()> { let ctx: SessionContext = SessionContext::new(); let sql = "set datafusion.optimizer.prefer_hash_join = false;"; let _ = ctx.sql(sql).await?.collect().await?; let sql = "set datafusion.execution.batch_size = 1"; let _ = ctx.sql(sql).await?.collect().await?; let sql = " select * from ( with t1 as ( select 12 a, 12 b ), t2 as ( select 12 a, 12 b ) select t1.* from t1 join t2 on t1.a = t2.b where t1.a > t2.b ) order by 1, 2; "; let actual = ctx.sql(sql).await?.collect().await?; Ok(()) }

I'll file a separate issue for this one, but perhaps we can go with this PR because potential problem you talking about cannot ever be hit because of the issue above

Thanks for creating the ticket.

I think that current fix to the LeftSemi isn't correct due to the additional joined pairs of nulls and buffered rows. I don't think we should move forward with it because of that.

I'm putting the PR to draft until we can check if PR requires modifications to avoid addition join pairs of nulls and buffered rows. It depends on #10491

I have checked the case, it doesn't fail, however it produces more rows than expected, looking into this

Yea, what I meant is actually it will produce some rows that are not correct results (due to the additional joined pairs of nulls and buffered rows).

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

comphead · 2024-05-13T14:40:37Z

@viirya @alamb can I get a review on this PR please?

alamb · 2024-05-13T15:22:48Z

I will review this today

viirya · 2024-05-13T15:30:18Z

I'll take another look today.

alamb

Thank you @comphead -- while I am not an expert in this code it looks like an improvement to me (less panic'ing and better test coverage all around) 👍

cc @richox who I think contributed the first version of this code in #2242 and @yjshen who reviewed, in case you would also like to review the change

alamb · 2024-05-13T18:18:59Z

datafusion/sqllogictest/test_files/sort_merge_join.slt

@@ -263,5 +263,139 @@ DROP TABLE t1;
 statement ok
 DROP TABLE t2;

+


To verify that these tests cover the code changes, I ran them locally without the code changes in this PR and they failed as expected 👍

Running "sort_merge_join.slt" thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/joins/sort_merge_join.rs:1356:22: index out of bounds: the len is 0 but the index is 1 note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace External error: task 17 panicked Error: Execution("1 failures") error: test failed, to rerun pass `-p datafusion-sqllogictest --test sqllogictests` Caused by: process didn't exit successfully: `/Users/andrewlamb/Software/datafusion/target/debug/deps/sqllogictests-ce3a36cfeab74789 sort_merge` (exit status: 1)

alamb · 2024-05-13T18:24:41Z

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

@@ -1161,6 +1162,15 @@ impl SMJStream {
            let filter_columns = if chunk.buffered_batch_idx.is_some() {
                if matches!(self.join_type, JoinType::Right) {
                    get_filter_column(&self.filter, &buffered_columns, &streamed_columns)
+                } else if matches!(self.join_type, JoinType::LeftSemi) {


I wonder if this should also check for JoinType::Left (and the clause above also check for JoinType::RightSemi 🤔

I have more to go, LeftAnti is first as it prevents TPCH to run and then double check RightSemi as well, good point

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

alamb · 2024-05-13T18:31:00Z

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

+        let streamed_indices_length = streamed_indices.len();
+        let mut corrected_mask: Vec<bool> = vec![false; streamed_indices_length];
+
+        #[allow(clippy::needless_range_loop)]


I wonder why ignore clippy here?

Clippy doesn't like for loops anymore ....

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

alamb · 2024-05-13T18:32:55Z

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

+        // have we seen a filter match for a streaming index before
+        let mut seen_as_true: bool = false;
+        let streamed_indices_length = streamed_indices.len();
+        let mut corrected_mask: Vec<bool> = vec![false; streamed_indices_length];


FWIW it might be faster / easier to followto create the BooleanArray directly using BooleanBuilder rather than Vec

done, btw, I'm thinking why BooleanArray, doesnt support capacity with default values to achieve the same as

vec![false; streamed_indices_length];

alamb · 2024-05-13T18:33:23Z

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

+    async fn left_semi_join_filtered_mask() -> Result<()> {
+        assert_eq!(
+            get_filtered_join_mask(
+                LeftSemi,


maybe we should test a type other than LeftSemi as negative test coverage 🤔

comphead · 2024-05-13T20:42:07Z

Thanks @alamb I'll add more docs and also added a task to check RightSemi join to #9846

I'll let it be opened a little longer to give @viirya more time to have a second eye on the PR

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

comphead · 2024-05-17T00:58:54Z

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

@@ -54,7 +45,17 @@ use datafusion_execution::TaskContext;
 use datafusion_physical_expr::equivalence::join_equivalence_properties;
 use datafusion_physical_expr::{PhysicalExprRef, PhysicalSortRequirement};

-use futures::{Stream, StreamExt};
+use crate::expressions::PhysicalSortExpr;


this change is by formatter

comphead · 2024-05-20T02:49:21Z

@viirya I'm planning to merge this PR soon as it fixes the crash, and addresses your concern (please see the slt test covering this specific case). All other improvements can be in follow up PR.

viirya · 2024-05-20T06:01:16Z

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

+                    join_streamed = !self
+                        .streamed_batch
+                        .join_filter_matched_idxs
+                        .contains(&(self.streamed_batch.idx as u64))
+                        && !self.streamed_joined;


If self.streamed_joined is false, join_filter_matched_idxs always doesn't contain self.streamed_batch.idx, so the two conditions are duplicated as they are both true.

If self.streamed_joined is true, this and check is failed, the another condition doesn't matter.

I'm not sure what this is added to check.

I don't see it addresses the issue in https://github.com/apache/datafusion/pull/10304/files#r1601943239.

Oh, I got what you want to do here. It is better to add a comment for later readers.

viirya · 2024-05-20T06:03:03Z

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

+                if matches!(self.join_type, JoinType::LeftSemi) && self.filter.is_none() {
                    join_streamed = !self.streamed_joined;
+                    // if the join filter specified there can be references to buffered columns


This block knows self.filter.is_none(), why you still do join_buffered = self.filter.is_some();?

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

viirya

I've seen some issues in this patch. It doesn't look like a correct fix.

comphead · 2024-05-20T15:09:34Z

I've seen some issues in this patch. It doesn't look like a correct fix.

The tests currently in sync with what hash join returns, is there a test showing the opposite?

viirya · 2024-05-20T15:40:20Z

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

+                    // if the join filter specified there can be references to buffered columns
+                    // so buffered columns are needed to access them
+                    join_buffered = self.filter.is_some();


Suggested change

// if the join filter specified there can be references to buffered columns

// so buffered columns are needed to access them

join_buffered = self.filter.is_some();

viirya · 2024-05-20T15:41:39Z

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

@@ -989,8 +996,21 @@ impl SMJStream {
                }
            }
            Ordering::Equal => {
-                if matches!(self.join_type, JoinType::LeftSemi) {
+                if matches!(self.join_type, JoinType::LeftSemi) && self.filter.is_some() {


You can combine this and below which both are for JoinType::LeftSemi under single if block.

viirya · 2024-05-20T15:51:22Z

I've seen some issues in this patch. It doesn't look like a correct fix.

Took another look. Looks okay to me.

alamb · 2024-05-21T00:33:55Z

🚀

comphead mentioned this pull request May 2, 2024

Document Sort Merge Join algorithm #10357

Open

comphead changed the title ~~Fix: Sort Merge Join crashes on TPCH Q21~~ Fix: Sort Merge Join LeftSemi issues May 4, 2024

github-actions bot added the sqllogictest label May 4, 2024

comphead force-pushed the dev0 branch from 8e730e4 to 0c8222a Compare May 4, 2024 16:23

comphead marked this pull request as ready for review May 7, 2024 04:18

comphead requested review from viirya, alamb and metesynnada May 7, 2024 04:18

github-actions bot added the core Core datafusion crate label May 7, 2024

comphead commented May 7, 2024

View reviewed changes

alamb mentioned this pull request May 7, 2024

DataFusion weekly project plan (Andrew Lamb) - May 6, 2024 #10395

Closed

7 tasks

comphead changed the title ~~Fix: Sort Merge Join LeftSemi issues~~ Fix: Sort Merge Join LeftSemi issues when JoinFilter is set May 7, 2024

viirya reviewed May 7, 2024

View reviewed changes

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

viirya reviewed May 7, 2024

View reviewed changes

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

viirya reviewed May 7, 2024

View reviewed changes

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

viirya reviewed May 7, 2024

View reviewed changes

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

comphead commented May 8, 2024

View reviewed changes

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

comphead requested a review from viirya May 9, 2024 15:04

alamb approved these changes May 13, 2024

View reviewed changes

comphead mentioned this pull request May 14, 2024

SortMergeJoin: The query stuck when join filter is set and more matched rows than batch size #10491

Closed

comphead marked this pull request as draft May 14, 2024 02:03

comphead added 2 commits May 16, 2024 17:49

Fix: Sort Merge Join crashes on TPCH Q21

ca564ce

Fix: Sort Merge Join crashes on TPCH Q21. LeftSemi

fd21ccf

comphead and others added 12 commits May 16, 2024 17:49

Fix: Sort Merge Join Left Semi crashes

4c2c8f3

Fix: Sort Merge Join Left Semi crashes. Comments

4052b0d

Fix: Sort Merge Join Left Semi crashes. Add tests

9da2c45

Fix: Sort Merge Join Left Semi crashes. Add tests

9c71eef

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

fe0bb60

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Fix: Sort Merge Join Left Semi crashes. Add tests

c0fd73e

Fix: Sort Merge Join crashes on TPCH Q21

f993b3c

Fix: Sort Merge Join crashes on TPCH Q21. LeftSemi

1354f83

Fix: Sort Merge Join Left Semi crashes

c129846

Fix: Sort Merge Join Left Semi crashes

22c61fc

Fix: Sort Merge Join Left Semi crashes

30f28fe

Fix: Sort Merge Join Left Semi crashes. Comments

823f396

comphead force-pushed the dev0 branch from 06760ca to 823f396 Compare May 17, 2024 00:56

github-actions bot removed the core Core datafusion crate label May 17, 2024

fmt

f0e60da

comphead commented May 17, 2024

View reviewed changes

comphead marked this pull request as ready for review May 17, 2024 04:18

viirya reviewed May 20, 2024

View reviewed changes

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

viirya reviewed May 20, 2024

View reviewed changes

viirya approved these changes May 20, 2024

View reviewed changes

fmt

a06acaa

comphead merged commit 94b5511 into apache:main May 20, 2024
23 checks passed

		@@ -263,5 +263,139 @@ DROP TABLE t1;
		statement ok
		DROP TABLE t2;

	// if the join filter specified there can be references to buffered columns
	// so buffered columns are needed to access them
	join_buffered = self.filter.is_some();

Fix: Sort Merge Join LeftSemi issues when JoinFilter is set #10304

Fix: Sort Merge Join LeftSemi issues when JoinFilter is set #10304

Conversation

comphead commented Apr 29, 2024 • edited

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?

comphead commented Apr 29, 2024

comphead commented May 7, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

comphead May 7, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya May 15, 2024 • edited

Choose a reason for hiding this comment

comphead commented May 13, 2024

alamb commented May 13, 2024

viirya commented May 13, 2024

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

comphead commented May 13, 2024

Choose a reason for hiding this comment

comphead commented May 20, 2024

viirya May 20, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya left a comment

Choose a reason for hiding this comment

comphead commented May 20, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya commented May 20, 2024

alamb commented May 21, 2024

comphead commented Apr 29, 2024 •

edited

comphead May 7, 2024 •

edited

viirya May 15, 2024 •

edited

viirya May 20, 2024 •

edited