LuciferYang commented on code in PR #57790:
URL: https://github.com/apache/spark/pull/57790#discussion_r3727249079
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownJoinThroughUnion.scala:
##########
@@ -79,53 +75,67 @@ case class PushDownJoinThroughUnion(override val conf:
SQLConf)
val deduped = dedupRight(right)
(deduped, AttributeMap(right.output.zip(deduped.output)))
}
- val leftRewrites = AttributeMap(unionHeadOutput.zip(child.output))
- val newCond = joinCond.map(_.transform {
- case a: Attribute if leftRewrites.contains(a) => leftRewrites(a)
- case a: Attribute if rightRewrites.contains(a) => rightRewrites(a)
- })
- Join(child, newRight, joinType, newCond, hint)
+ branchJoin(join, unionHeadOutput, child, newRight, rightRewrites)
}
u.withNewChildren(newChildren)
}
}
+ /**
+ * The join for one `Union` branch: the branch on the left, `newRight` on
the right, and the
+ * condition rewritten from the `Union` output to the outputs of both new
children.
+ */
+ private def branchJoin(
+ join: Join,
+ unionHeadOutput: Seq[Attribute],
+ child: LogicalPlan,
+ newRight: LogicalPlan,
+ rightRewrites: AttributeMap[Attribute]): Join = {
+ val leftRewrites = AttributeMap(unionHeadOutput.zip(child.output))
+ val newCond = join.condition.map(_.transform {
+ case a: Attribute if leftRewrites.contains(a) => leftRewrites(a)
+ case a: Attribute if rightRewrites.contains(a) => rightRewrites(a)
+ })
+ Join(child, newRight, join.joinType, newCond, join.hint)
+ }
+
/**
* Whether every join produced by the rewrite is expected to broadcast its
right side.
*
- * `canPlanAsBroadcastHashJoin` is not enough. It holds when either side is
broadcastable, and
- * for an inner join the planner may build from either side, choosing the
smaller one when both
- * qualify. The rewrite replaces the `Union` on the left with one of its
children, so the build
- * side is decided per branch against a smaller left. Any branch that ends
up building from the
- * left leaves its copy of the right side as a plain probe input, which is
not reused, so the
- * right side is read once per such branch instead of once in total.
+ * Asking whether a broadcast hash join is possible is not enough: for an
inner join the planner
+ * may build from either side, choosing the smaller one when both qualify.
The rewrite replaces
+ * the `Union` on the left with one of its children, so the build side is
decided per branch
+ * against a smaller left. Any branch that ends up building from the left
leaves its copy of the
+ * right side as a plain probe input, which is not reused, so the right side
is read once per such
+ * branch instead of once in total.
+ *
+ * `getBroadcastHashJoinBuildSide` returning `None` also covers the joins
where no broadcast hash
Review Comment:
[529d10a](https://github.com/apache/spark/pull/57790/commits/529d10a273b775a2e915b1c196f7ca5f329ce029)
address this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
