[SPARK-20897][SQL] cached self-join should not fail #18121
Closed
Conversation
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
viirya
reviewed
May 26, 2017
| @@ -58,6 +59,10 @@ case class ReusedExchangeExec(override val output: Seq[Attribute], child: Exchan | |||
| override protected[sql] def doExecuteBroadcast[T](): broadcast.Broadcast[T] = { | |||
| child.executeBroadcast() | |||
| } | |||
|
|
|||
| override def outputPartitioning: Partitioning = child.outputPartitioning | |||
There was a problem hiding this comment.
ReusedExchangeExec can have distinct sets of output attribute ids. Shall we also update outputPartitioning and outputOrdering, if its output is different to child.output.
|
LGTM except one comment. |
|
Test build #77426 has finished for PR 18121 at commit
|
|
Test build #77445 has finished for PR 18121 at commit
|
|
Test build #77454 has finished for PR 18121 at commit
|
|
Thanks! Merging to master/2.2. |
asfgit
pushed a commit
that referenced
this pull request
May 27, 2017
## What changes were proposed in this pull request? The failed test case is, we have a `SortMergeJoinExec` for a self-join, which means we have a `ReusedExchange` node in the query plan. It works fine without caching, but throws an exception in `SortMergeJoinExec.outputPartitioning` if we cache it. The root cause is, `ReusedExchange` doesn't propagate the output partitioning from its child, so in `SortMergeJoinExec.outputPartitioning` we create `PartitioningCollection` with a hash partitioning and an unknown partitioning, and fail. This bug is mostly fine, because inserting the `ReusedExchange` is the last step to prepare the physical plan, we won't call `SortMergeJoinExec.outputPartitioning` anymore after this. However, if the dataframe is cached, the physical plan of it becomes `InMemoryTableScanExec`, which contains another physical plan representing the cached query, and it has gone through the entire planning phase and may have `ReusedExchange`. Then the planner call `InMemoryTableScanExec.outputPartitioning`, which then calls `SortMergeJoinExec.outputPartitioning` and trigger this bug. ## How was this patch tested? a new regression test Author: Wenchen Fan <wenchen@databricks.com> Closes #18121 from cloud-fan/bug. (cherry picked from commit 08ede46) Signed-off-by: Xiao Li <gatorsmile@gmail.com>
yucai
reviewed
Jun 14, 2018
| case h: HashPartitioning => h.copy(expressions = h.expressions.map(updateAttr)) | ||
| case other => other | ||
| } | ||
|
|
There was a problem hiding this comment.
@cloud-fan @viirya Could you help explain why we only consider HashPartitioning here?
How about RangePartitioning?
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.
What changes were proposed in this pull request?
The failed test case is, we have a
SortMergeJoinExecfor a self-join, which means we have aReusedExchangenode in the query plan. It works fine without caching, but throws an exception inSortMergeJoinExec.outputPartitioningif we cache it.The root cause is,
ReusedExchangedoesn't propagate the output partitioning from its child, so inSortMergeJoinExec.outputPartitioningwe createPartitioningCollectionwith a hash partitioning and an unknown partitioning, and fail.This bug is mostly fine, because inserting the
ReusedExchangeis the last step to prepare the physical plan, we won't callSortMergeJoinExec.outputPartitioninganymore after this.However, if the dataframe is cached, the physical plan of it becomes
InMemoryTableScanExec, which contains another physical plan representing the cached query, and it has gone through the entire planning phase and may haveReusedExchange. Then the planner callInMemoryTableScanExec.outputPartitioning, which then callsSortMergeJoinExec.outputPartitioningand trigger this bug.How was this patch tested?
a new regression test