fix: CometShuffleExchangeExec logical link should be correct#324
Merged
viirya merged 8 commits intoapache:mainfrom Apr 30, 2024
Merged
fix: CometShuffleExchangeExec logical link should be correct#324viirya merged 8 commits intoapache:mainfrom
viirya merged 8 commits intoapache:mainfrom
Conversation
viirya
commented
Apr 25, 2024
Comment on lines
+336
to
+337
| "spark.sql.adaptive.autoBroadcastJoinThreshold" -> "-1", | ||
| "spark.sql.autoBroadcastJoinThreshold" -> "-1", |
Member
Author
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
| withSQLConf( | ||
| SQLConf.EXCHANGE_REUSE_ENABLED.key -> "true", | ||
| SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", | ||
| SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", |
Member
Author
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 | ||
| } | ||
| } |
Member
Author
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]") |
Member
Author
There was a problem hiding this comment.
Follow Spark Exchange.stringArgs.
kazuyukitanimura
approved these changes
Apr 25, 2024
viirya
commented
Apr 27, 2024
| 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") |
Member
Author
1 similar comment
Member
Author
Member
Author
|
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() |
Member
Author
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
Member
Author
|
Merged. Thanks @sunchao |
himadripal
pushed a commit
to himadripal/datafusion-comet
that referenced
this pull request
Sep 7, 2024
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
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?