feat: Support sort merge join with a join condition #553

viirya · 2024-06-11T07:31:43Z

Which issue does this PR close?

Closes #398.

Rationale for this change

What changes are included in this PR?

How are these changes tested?

comphead · 2024-06-11T18:34:21Z

Some tests doesn't pass, its good to check them in DF, SMJ joined filter is still in progress

viirya · 2024-06-11T18:43:27Z

Yea, that's why I marked this as a draft PR now.

comphead · 2024-06-11T19:04:16Z

just checked DF on one of the failed queries

#[tokio::test]
    async fn test_smj_left_filtered() -> 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 = 100";
        let _ = ctx.sql(sql).await?.collect().await?;

        let sql = "
            select * from (
            with t as (
                select id, id % 5 id1 from (select unnest(range(0,10)) id)
            ), t1 as (
                select id % 10 id, id + 2 id1 from (select unnest(range(0,10)) id)
            )
            select * from  t left join  t1 on t.id1 = t1.id and t.id > t1.id1
            ) order by 1, 2, 3, 4
        ";

        let actual = ctx.sql(sql).await?.collect().await?;

        let expected: Vec<&str> = vec![
            "+----+-----+----+-----+",
            "| id | id1 | id | id1 |",
            "+----+-----+----+-----+",
            "| 0  | 0   |    |     |",
            "| 1  | 1   |    |     |",
            "| 2  | 2   |    |     |",
            "| 3  | 3   |    |     |",
            "| 4  | 4   |    |     |",
            "| 5  | 0   | 0  | 2   |",
            "| 6  | 1   | 1  | 3   |",
            "| 7  | 2   | 2  | 4   |",
            "| 8  | 3   | 3  | 5   |",
            "| 9  | 4   | 4  | 6   |",
            "+----+-----+----+-----+",
        ];
        datafusion_common::assert_batches_eq!(expected, &actual);

        Ok(())
    }

it passes

comphead · 2024-06-11T19:04:52Z

looks like Comet produces duplicates

comphead · 2024-06-11T20:32:07Z

Right join fails in DF

#[tokio::test]
    async fn test_smj_left_filtered() -> 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 = 100";
        let _ = ctx.sql(sql).await?.collect().await?;

        let sql = "
            select * from (
            with t as (
                select id, id % 5 id1 from (select unnest(range(0,10)) id)
            ), t1 as (
                select id % 10 id, id + 2 id1 from (select unnest(range(0,10)) id)
            )
            select * from  t right join  t1 on t.id1 = t1.id and t.id > t1.id1
            ) order by 1, 2, 3, 4
        ";

        let actual = ctx.sql(sql).await?.collect().await?;

        let expected: Vec<&str> = vec![
            "+----+-----+----+-----+",
            "| id | id1 | id | id1 |",
            "+----+-----+----+-----+",
            "| 5  | 0   | 0  | 2   |",
            "| 6  | 1   | 1  | 3   |",
            "| 7  | 2   | 2  | 4   |",
            "| 8  | 3   | 3  | 5   |",
            "| 9  | 4   | 4  | 6   |",
            "|    |     | 5  | 7   |",
            "|    |     | 6  | 8   |",
            "|    |     | 7  | 9   |",
            "|    |     | 8  | 10  |",
            "|    |     | 9  | 11  |",
            "+----+-----+----+-----+",
        ];
        datafusion_common::assert_batches_eq!(expected, &actual);

        Ok(())
    }

comphead · 2024-06-12T00:05:26Z

Filed apache/datafusion#10882

viirya · 2024-08-21T06:15:46Z

native/Cargo.toml

-datafusion-physical-plan = { git = "https://github.com/apache/datafusion.git", rev = "41.0.0-rc1", default-features = false }
-datafusion-physical-expr-common = { git = "https://github.com/apache/datafusion.git", rev = "41.0.0-rc1", default-features = false }
-datafusion-physical-expr = { git = "https://github.com/apache/datafusion.git", rev = "41.0.0-rc1", default-features = false }
+datafusion-common = { git = "https://github.com/viirya/arrow-datafusion.git", rev = "f98693e" }


Use the commit of the PR apache/datafusion#12090. When the PR is merged, we can change back to DataFusion repo.

viirya · 2024-08-21T06:16:09Z

native/core/src/execution/datafusion/planner.rs

+                AggregateExprBuilder::new(sum_udaf(), vec![child])
+                    .schema(schema)
+                    .alias("count")
+                    .with_ignore_nulls(false)
+                    .with_distinct(false)
+                    .build().map_err(|e| ExecutionError::DataFusionError(e.to_string()))


Again, DataFusion API changes.

viirya · 2024-08-21T06:16:28Z

native/core/src/execution/operators/filter.rs

@@ -126,7 +126,7 @@ impl FilterExec {
        let schema = input.schema();
        if !check_support(predicate, &schema) {
            let selectivity = default_selectivity as f64 / 100.0;
-            let mut stats = input_stats.into_inexact();
+            let mut stats = input_stats.to_inexact();


DataFusion API change.

comphead · 2024-08-21T16:15:28Z

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

+          val left = sql("SELECT * FROM tbl_a")
+          val right = sql("SELECT * FROM tbl_b")
+
+          val df8 =


I feel we can also use SQL for Anti, Semi joins?

Changed to SQL syntax.

comphead · 2024-08-21T21:03:41Z

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -2860,7 +2860,11 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
          case RightOuter => JoinType.RightOuter
          case FullOuter => JoinType.FullOuter
          case LeftSemi => JoinType.LeftSemi
-          case LeftAnti => JoinType.LeftAnti
+          // TODO: DF SMJ with join condition fails TPCH q21


let me have a look on q21. I remember Anti join had issues with TPCH in DF but it was fixed

But it might be also related to apache/datafusion#11555

codecov-commenter · 2024-08-21T23:09:02Z

Codecov Report

Attention: Patch coverage is 36.36364% with 7 lines in your changes missing coverage. Please review.

Project coverage is 34.20%. Comparing base (9d8730d) to head (b9cd9da).
Report is 6 commits behind head on main.

Files	Patch %	Lines
.../scala/org/apache/comet/serde/QueryPlanSerde.scala	36.36%	5 Missing and 2 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##               main     #553       +/-   ##
=============================================
- Coverage     55.16%   34.20%   -20.96%     
- Complexity      857      888       +31     
=============================================
  Files           109      112        +3     
  Lines         10542    43071    +32529     
  Branches       2010     9509     +7499     
=============================================
+ Hits           5815    14733     +8918     
- Misses         3714    25352    +21638     
- Partials       1013     2986     +1973

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

viirya · 2024-08-26T05:40:49Z

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

+    }
+  }
+
+  test("full outer join") {


This test fails currently. It needs the fix at DataFusion apache/datafusion#12159

The fix was merged at DataFusion. I updated Comet to use the latest commit.

viirya · 2024-08-26T05:41:49Z

spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala

@@ -75,7 +75,6 @@ abstract class CometTestBase
    conf.set(MEMORY_OFFHEAP_SIZE.key, "2g")
    conf.set(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key, "1g")
    conf.set(SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key, "1g")
-    conf.set(SQLConf.COALESCE_PARTITIONS_ENABLED.key, "false")


We don't need to set SQLConf.COALESCE_PARTITIONS_ENABLED.key false now.

We also need to remove this to trigger test failure https://github.com/apache/datafusion-comet/pull/553/files#r1730694210.

viirya · 2024-08-26T22:02:54Z

native/Cargo.toml

-datafusion-physical-plan = { version = "41.0.0", default-features = false }
-datafusion-physical-expr-common = { version = "41.0.0", default-features = false }
-datafusion-physical-expr = { version = "41.0.0", default-features = false }
+datafusion-common = { git = "https://github.com/apache/datafusion.git", rev = "dff590b" }


Updated to latest DataFusion to use the two commits including bug fixes.

viirya · 2024-08-26T22:03:45Z

native/core/src/execution/datafusion/expressions/avg_decimal.rs

@@ -142,6 +142,22 @@ impl AggregateExpr for AvgDecimal {
            ),
        }
    }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {


New DataFusion API required for AggregateExpr trait.

comphead · 2024-08-28T23:17:18Z

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

+            withInfo(join, cond)
+            return None
+          }
+          condProto.get


is there any scenario of None.get?

if (condProto.isEmpty) { withInfo(join, cond) return None }

If it is None, it will return None earlier.

comphead · 2024-08-28T23:22:13Z

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

+          checkSparkAnswerAndOperator(df7)
+
+          val df8 = sql(
+            "SELECT * FROM tbl_a LEFT SEMI JOIN tbl_b ON tbl_a._2 = tbl_b._1 " +


is there a RIGHT SEMI in Spark? afair there is still no proper support in DF for RightSemi

apache/datafusion#9846

comphead · 2024-08-28T23:25:40Z

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

+      val left = UnresolvedRelation(TableIdentifier("left"))
+      val right = UnresolvedRelation(TableIdentifier("right"))
+
+      checkSparkAnswer(left.join(right, $"left.N" === $"right.N", "full"))


is it possible to rephrase it in SQL as well?

This test is copied from Spark. I think it is good to keep it as the same.

comphead

lgtm thanks @viirya

comphead · 2024-08-29T17:25:54Z

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

+          checkSparkAnswerAndOperator(df9)
+
+          // TODO: Enable these tests after fixing the issue:
+          // https://github.com/apache/datafusion-comet/issues/861


we probably can create a separate github ticket on this to not forget enabling tests

viirya · 2024-08-29T17:50:35Z

Thanks @comphead

viirya · 2024-08-29T17:50:46Z

I need to update plan stability results...

* Init * test * test * test * Use specified commit to test * Fix format * fix clippy * fix * fix * Fix * Change to SQL syntax * Disable SMJ LeftAnti with join filter * Fix * Add test * Add test * Update to last DataFusion commit * fix format * fix * Update diffs (cherry picked from commit e57ead4)

viirya marked this pull request as draft June 11, 2024 07:31

comphead mentioned this pull request Jun 12, 2024

SMJ: incorrect result for filtered right outer join apache/datafusion#10882

Closed

viirya force-pushed the smj_filter branch from 40bed97 to 069762d Compare July 11, 2024 05:24

This was referenced Aug 19, 2024

feat: Add configs to disable specified operator #850

Closed

fix: Produce buffered null join row only if all joined rows are failed on join filter in SMJ full join apache/datafusion#12090

Merged

viirya force-pushed the smj_filter branch from 069762d to 5421382 Compare August 21, 2024 06:14

viirya commented Aug 21, 2024

View reviewed changes

viirya force-pushed the smj_filter branch from 5421382 to 936b9ae Compare August 21, 2024 06:23

viirya marked this pull request as ready for review August 21, 2024 06:23

viirya force-pushed the smj_filter branch 2 times, most recently from c27f1f6 to bb7586f Compare August 21, 2024 06:55

comphead reviewed Aug 21, 2024

View reviewed changes

viirya force-pushed the smj_filter branch from 95e29cb to cc041fc Compare August 21, 2024 18:59

comphead reviewed Aug 21, 2024

View reviewed changes

comphead mentioned this pull request Aug 21, 2024

[BUG]: SortMergeJoin filtered LeftAnti fails on TPCH Q21 #861

Open

viirya mentioned this pull request Aug 25, 2024

fix: Skip buffered rows which are not joined with streamed side when checking join filter results apache/datafusion#12159

Merged

viirya commented Aug 26, 2024

View reviewed changes

viirya force-pushed the smj_filter branch from 2c6c616 to 1d655cd Compare August 26, 2024 22:02

viirya commented Aug 26, 2024

View reviewed changes

comphead reviewed Aug 28, 2024

View reviewed changes

comphead approved these changes Aug 29, 2024

View reviewed changes

comphead mentioned this pull request Aug 29, 2024

Reenable SMJ Antijoin Spark tests when upstream issue resolved #891

Open

viirya added 19 commits August 29, 2024 14:46

Init

7e8af66

test

77f3dde

test

7f5f50d

test

e9147bd

Use specified commit to test

2bcb004

Fix format

1f82e8a

fix clippy

e401a62

fix

445afab

fix

d201a19

Fix

5037b97

Change to SQL syntax

90c0020

Disable SMJ LeftAnti with join filter

0743df5

Fix

5b8a430

Add test

28e3bf9

Add test

c362f4c

Update to last DataFusion commit

97781d1

fix format

8b6ca06

fix

a2f3049

Update diffs

4a05b02

viirya force-pushed the smj_filter branch from 5f5d174 to 4a05b02 Compare August 29, 2024 23:07

viirya merged commit e57ead4 into apache:main Aug 30, 2024
74 checks passed

viirya deleted the smj_filter branch August 30, 2024 00:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Support sort merge join with a join condition #553

feat: Support sort merge join with a join condition #553

viirya commented Jun 11, 2024

comphead commented Jun 11, 2024

viirya commented Jun 11, 2024

comphead commented Jun 11, 2024

comphead commented Jun 11, 2024

comphead commented Jun 11, 2024

comphead commented Jun 12, 2024

viirya Aug 21, 2024

viirya Aug 21, 2024

viirya Aug 21, 2024

comphead Aug 21, 2024

viirya Aug 21, 2024

viirya Aug 21, 2024

comphead Aug 21, 2024

comphead Aug 21, 2024

codecov-commenter commented Aug 21, 2024 •

edited

Loading

viirya Aug 26, 2024

viirya Aug 27, 2024

viirya Aug 26, 2024

viirya Aug 26, 2024

viirya Aug 26, 2024

comphead Aug 28, 2024

viirya Aug 29, 2024

comphead Aug 28, 2024

viirya Aug 29, 2024

comphead Aug 28, 2024

viirya Aug 29, 2024

comphead left a comment

comphead Aug 29, 2024

comphead Aug 29, 2024

viirya commented Aug 29, 2024

viirya commented Aug 29, 2024

feat: Support sort merge join with a join condition #553

feat: Support sort merge join with a join condition #553

Conversation

viirya commented Jun 11, 2024

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

comphead commented Jun 11, 2024

viirya commented Jun 11, 2024

comphead commented Jun 11, 2024

comphead commented Jun 11, 2024

comphead commented Jun 11, 2024

comphead commented Jun 12, 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Aug 21, 2024 • edited Loading

Codecov Report

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

comphead left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya commented Aug 29, 2024

viirya commented Aug 29, 2024

codecov-commenter commented Aug 21, 2024 •

edited

Loading