fix: CometShuffleExchangeExec logical link should be correct #324
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
viirya
commented
Apr 25, 2024
Comment on lines
+336
to
+337
| "spark.sql.adaptive.autoBroadcastJoinThreshold" -> "-1", | ||
| "spark.sql.autoBroadcastJoinThreshold" -> "-1", |
There was a problem hiding this comment.
Query plan is changed due to the logical link fix. In order to have CometSortMergeJoin, we need to disable broadcast join and AQE broadcast join.
viirya
force-pushed
the
fix_shuffle_logical_link
branch
from
3a9fdb9
to
aa100c6
Compare
viirya
commented
Apr 25, 2024
| @@ -373,6 +405,7 @@ class CometExecSuite extends CometTestBase { | |||
| withSQLConf( | |||
| SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true", | |||
| SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", | |||
| SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", | |||
There was a problem hiding this comment.
Similar reason here. The query plan is changed and AQE interferes with a broadcast join.
viirya
force-pushed
the
fix_shuffle_logical_link
branch
from
aa100c6
to
96b9826
Compare
viirya
commented
Apr 25, 2024
Comment on lines
+199
to
+209
| override def equals(obj: Any): Boolean = { | ||
| obj match { | ||
| case other: CometShuffleExchangeExec => | ||
| this.outputPartitioning == other.outputPartitioning && | ||
| this.shuffleOrigin == other.shuffleOrigin && this.child == other.child && | ||
| this.shuffleType == other.shuffleType && | ||
| this.advisoryPartitionSize == other.advisoryPartitionSize | ||
| case _ => | ||
| false | ||
| } | ||
| } |
There was a problem hiding this comment.
Because CometShuffleExchangeExec adds originalPlan parameter which is not covered by canonicalization in Spark, we need to exclude it when compare two CometShuffleExchangeExec to make sure Spark reuse shuffle rule work.
viirya
commented
Apr 25, 2024
Comment on lines
+214
to
+215
| override def stringArgs: Iterator[Any] = | ||
| Iterator(outputPartitioning, shuffleOrigin, shuffleType, child) ++ Iterator(s"[plan_id=$id]") |
There was a problem hiding this comment.
Follow Spark Exchange.stringArgs.
kazuyukitanimura
approved these changes
Apr 25, 2024
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
Outdated
Show resolved
Hide resolved
viirya
commented
Apr 27, 2024
| @@ -91,6 +91,7 @@ class CometTPCHQuerySuite extends QueryTest with CometTPCBase with SQLQueryTestH | |||
| conf.set(CometConf.COMET_EXEC_ALL_OPERATOR_ENABLED.key, "true") | |||
| conf.set(CometConf.COMET_EXEC_ALL_EXPR_ENABLED.key, "true") | |||
| conf.set(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "true") | |||
| conf.set(CometConf.COMET_COLUMNAR_SHUFFLE_ENABLED.key, "true") | |||
1 similar comment
|
cc @andygrove Can you take a look if you have time? Thanks. |
viirya
commented
Apr 29, 2024
Comment on lines
+267
to
+273
| case ReusedExchangeExec(_, c: CometBroadcastExchangeExec) => | ||
| inputs += c.setNumPartitions(firstNonBroadcastPlanNumPartitions).executeColumnar() | ||
| case BroadcastQueryStageExec( | ||
| _, | ||
| ReusedExchangeExec(_, c: CometBroadcastExchangeExec), | ||
| _) => | ||
| inputs += c.setNumPartitions(firstNonBroadcastPlanNumPartitions).executeColumnar() |
There was a problem hiding this comment.
These are necessary but missed to add previously. This fix exposes that.
sunchao
approved these changes
Apr 30, 2024
|
Merged. Thanks @sunchao |
himadripal
pushed a commit
to himadripal/datafusion-comet
that referenced
this pull request
Sep 7, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #323.
Rationale for this change
What changes are included in this PR?
How are these changes tested?