Github user mgaido91 commented on a diff in the pull request:
https://github.com/apache/spark/pull/22957#discussion_r238642801
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala
---
@@ -145,9 +145,14 @@ case class EnsureRequirements(conf: SQLConf) extends
Rule[SparkPlan] {
assert(requiredChildDistributions.length == children.length)
assert(requiredChildOrderings.length == children.length)
+ val aliasMap =
AttributeMap[Expression](children.flatMap(_.expressions.collect {
+ case a: Alias => (a.toAttribute, a)
+ }))
+
// Ensure that the operator's children satisfy their output
distribution requirements.
children = children.zip(requiredChildDistributions).map {
- case (child, distribution) if
child.outputPartitioning.satisfies(distribution) =>
+ case (child, distribution) if child.outputPartitioning.satisfies(
+ distribution.mapExpressions(replaceAlias(_, aliasMap))) =>
--- End diff --
I don't think that is right: that would cause the shuffle to happen for
every plan which is hashed by both `[hash part c, hash part b]` and `[hash part
d, hash part b]` (and also `[hash part a, hash part b]`). I think that if we
want to go that way, we'd need a set of equivalent outputPatitioning
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
