Fix Join output nullabilities.

ueshin · ueshin · commit f20f196e33e4 · 2014-06-30T18:30:37.000+09:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala
@@ -18,7 +18,7 @@
 package org.apache.spark.sql.catalyst.plans.logical
 
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.plans.{LeftSemi, JoinType}
+import org.apache.spark.sql.catalyst.plans._
 import org.apache.spark.sql.catalyst.types._
 
 case class Project(projectList: Seq[NamedExpression], child: LogicalPlan) extends UnaryNode {
@@ -81,11 +81,28 @@ case class Join(
   condition: Option[Expression]) extends BinaryNode {
 
   override def references = condition.map(_.references).getOrElse(Set.empty)
-  override def output = joinType match {
-    case LeftSemi =>
-      left.output
-    case _ =>
-      left.output ++ right.output
+  override def output = {
+    def nullabilize(output: Seq[Attribute]) = {
+      output.map {
+        case attr if !attr.nullable =>
+          AttributeReference(
+            attr.name, attr.dataType, nullable = true)(attr.exprId, attr.qualifiers)
+        case attr => attr
+      }
+    }
+
+    joinType match {
+      case LeftSemi =>
+        left.output
+      case LeftOuter =>
+        left.output ++ nullabilize(right.output)
+      case RightOuter =>
+        nullabilize(left.output) ++ right.output
+      case FullOuter =>
+        nullabilize(left.output) ++ nullabilize(right.output)
+      case _ =>
+        left.output ++ right.output
+    }
   }
 }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/joins.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/joins.scala
@@ -319,7 +319,26 @@ case class BroadcastNestedLoopJoin(
 
   override def otherCopyArgs = sqlContext :: Nil
 
-  def output = left.output ++ right.output
+  def output = {
+    def nullabilize(output: Seq[Attribute]) = {
+      output.map {
+        case attr if !attr.nullable =>
+          AttributeReference(attr.name, attr.dataType, nullable = true)(attr.exprId, attr.qualifiers)
+        case attr => attr
+      }
+    }
+
+    joinType match {
+      case LeftOuter =>
+        left.output ++ nullabilize(right.output)
+      case RightOuter =>
+        nullabilize(left.output) ++ right.output
+      case FullOuter =>
+        nullabilize(left.output) ++ nullabilize(right.output)
+      case _ =>
+        left.output ++ right.output
+    }
+  }
 
   /** The Streamed Relation */
   def left = streamed
