Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/19327#discussion_r140900883
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamingSymmetricHashJoinExec.scala
---
@@ -157,11 +164,20 @@ case class StreamingSymmetricHashJoinExec(
override def requiredChildDistribution: Seq[Distribution] =
ClusteredDistribution(leftKeys) :: ClusteredDistribution(rightKeys) ::
Nil
- override def output: Seq[Attribute] = left.output ++ right.output
+ override def output: Seq[Attribute] = joinType match {
+ case _: InnerLike => left.output ++ right.output
+ case LeftOuter => left.output ++
right.output.map(_.withNullability(true))
+ case RightOuter => left.output.map(_.withNullability(true)) ++
right.output
+ case _ =>
+ throwBadJoinTypeException()
+ Seq()
+ }
override def outputPartitioning: Partitioning = joinType match {
case _: InnerLike =>
PartitioningCollection(Seq(left.outputPartitioning,
right.outputPartitioning))
+ case LeftOuter => PartitioningCollection(Seq(right.outputPartitioning))
--- End diff --
shouldnt this be `left.outputPartitioning` for LeftOuter? The the
attributes from the right side can become null, hence the right side
partitioning will not be preserved.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
