Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/19656#discussion_r148933507
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastHashJoinExec.scala
---
@@ -76,6 +76,20 @@ case class BroadcastHashJoinExec(
streamedPlan.asInstanceOf[CodegenSupport].inputRDDs()
}
+ override def needCopyResult: Boolean = joinType match {
+ case _: InnerLike | LeftOuter | RightOuter =>
+ // For inner and outer joins, one row from the streamed side may
produce multiple result rows,
+ // if the build side has duplicated keys. Then we need to copy the
result rows before putting
+ // them in a buffer, because these result rows share one UnsafeRow
instance. Note that here
+ // we wait for the broadcast to be finished, which is a no-op
because it's already finished
+ // when we wait it in `doProduce`.
+ buildPlan.executeBroadcast[HashedRelation]().value.keyIsUnique
--- End diff --
Seems to be
`!buildPlan.executeBroadcast[HashedRelation]().value.keyIsUnique`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
