Add nullabilization to Generate of SparkPlan.

Author: ueshin

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala

@@ -50,7 +50,7 @@ case class Generate(
     val output = alias
       .map(a => generator.output.map(_.withQualifiers(a :: Nil)))
       .getOrElse(generator.output)
-    if (outer) {
+    if (join && outer) {
       output.map {
         case attr if !attr.nullable =>
           AttributeReference(

sql/core/src/main/scala/org/apache/spark/sql/execution/Generate.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.execution
 
 import org.apache.spark.annotation.DeveloperApi
-import org.apache.spark.sql.catalyst.expressions.{Generator, JoinedRow, Literal, Projection}
+import org.apache.spark.sql.catalyst.expressions._
 
 /**
  * :: DeveloperApi ::
@@ -39,8 +39,21 @@ case class Generate(
     child: SparkPlan)
   extends UnaryNode {
 
+  protected def generatorOutput: Seq[Attribute] = {
+    if (join && outer) {
+      generator.output.map {
+        case attr if !attr.nullable =>
+          AttributeReference(
+            attr.name, attr.dataType, nullable = true)(attr.exprId, attr.qualifiers)
+        case attr => attr
+      }
+    } else {
+      generator.output
+    }
+  }
+
   override def output =
-    if (join) child.output ++ generator.output else generator.output
+    if (join) child.output ++ generatorOutput else generatorOutput
 
   override def execute() = {
     if (join) {