Skip to content

feat: Reenable tests for filtered SMJ anti join #1211

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
andygrove merged 12 commits into from
Jan 3, 2025
Merged

feat: Reenable tests for filtered SMJ anti join #1211

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
d40f816
feat: reenable filtered SMJ Anti join tests
comphead Jan 2, 2025
b813cb8
feat: reenable filtered SMJ Anti join tests
comphead Jan 2, 2025
8073591
feat: reenable filtered SMJ Anti join tests
comphead Jan 2, 2025
cd63de9
feat: reenable filtered SMJ Anti join tests
comphead Jan 2, 2025
f308335
Add CoalesceBatchesExec around SMJ with join filter
andygrove Jan 3, 2025
f4b0c08
Merge branch 'smj-coalesce' into coalesce
andygrove Jan 3, 2025
801760e
Merge pull request #1 from andygrove/coalesce
comphead Jan 3, 2025
d50d7e5
adding `CoalesceBatches`
comphead Jan 3, 2025
02ea910
adding `CoalesceBatches`
comphead Jan 3, 2025
b657551
adding `CoalesceBatches`
comphead Jan 3, 2025
0203636
feat: reenable filtered SMJ Anti join tests
comphead Jan 3, 2025
ab11578
feat: reenable filtered SMJ Anti join tests
comphead Jan 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -567,7 +567,7 @@ class CometSparkSessionExtensions

case op: SortMergeJoinExec
if CometConf.COMET_EXEC_SORT_MERGE_JOIN_ENABLED.get(conf) &&
op.children.forall(isCometNative(_)) =>
op.children.forall(isCometNative) =>
val newOp = transform1(op)
newOp match {
case Some(nativeOp) =>
Expand Down
6 changes: 1 addition & 5 deletions spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2859,11 +2859,7 @@ object QueryPlanSerde extends Logging with ShimQueryPlanSerde with CometExprShim
case RightOuter => JoinType.RightOuter
case FullOuter => JoinType.FullOuter
case LeftSemi => JoinType.LeftSemi
// TODO: DF SMJ with join condition fails TPCH q21
case LeftAnti if condition.isEmpty => JoinType.LeftAnti
case LeftAnti =>
withInfo(join, "LeftAnti SMJ join with condition is not supported")
return None
case LeftAnti => JoinType.LeftAnti
case _ =>
// Spark doesn't support other join types
withInfo(op, s"Unsupported join type ${join.joinType}")
Expand Down
6 changes: 1 addition & 5 deletions spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,7 @@ class CometJoinSuite extends CometTestBase {
v.toDouble,
v.toString,
v % 2 == 0,
v.toString().getBytes,
v.toString.getBytes,
Decimal(v))

withParquetTable((0 until 10).map(i => manyTypes(i, i % 5)), "tbl_a") {
Expand Down Expand Up @@ -391,9 +391,6 @@ class CometJoinSuite extends CometTestBase {
"AND tbl_a._2 >= tbl_b._1")
checkSparkAnswerAndOperator(df9)

// TODO: Enable these tests after fixing the issue:
// https://github.com/apache/datafusion-comet/issues/861
/*

@viirya viirya Jan 2, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not just remove the comment. This test is also ignored. We need to re-enable it too.

@comphead comphead Jan 3, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

indeed, thanks @viirya

val df10 = sql(
"SELECT * FROM tbl_a LEFT ANTI JOIN tbl_b ON tbl_a._2 = tbl_b._1 " +
"AND tbl_a._2 >= tbl_b._1")
Expand All @@ -403,7 +400,6 @@ class CometJoinSuite extends CometTestBase {
"SELECT * FROM tbl_b LEFT ANTI JOIN tbl_a ON tbl_a._2 = tbl_b._1 " +
"AND tbl_a._2 >= tbl_b._1")
checkSparkAnswerAndOperator(df11)
*/
}
}
}
Expand Down