[SPARK-12335][SPARK-12336][SPARK-12341][SPARK-12342][SQL] Fixes several expression nullablility bugs

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/BoundAttribute.scala

@@ -69,10 +69,29 @@ case class BoundReference(ordinal: Int, dataType: DataType, nullable: Boolean)
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = {
     val javaType = ctx.javaType(dataType)
     val value = ctx.getValue(ctx.INPUT_ROW, dataType, ordinal.toString)
-    s"""
-      boolean ${ev.isNull} = ${ctx.INPUT_ROW}.isNullAt($ordinal);
-      $javaType ${ev.value} = ${ev.isNull} ? ${ctx.defaultValue(dataType)} : ($value);
-    """
+
+    if (nullable) {
+      s"""
+        boolean ${ev.isNull} = ${ctx.INPUT_ROW}.isNullAt($ordinal);
+        $javaType ${ev.value} = ${ev.isNull} ? ${ctx.defaultValue(dataType)} : ($value);
+       """
+    } else {
+      s"""
+        boolean ${ev.isNull} = ${ctx.INPUT_ROW}.isNullAt($ordinal);
+        $javaType ${ev.value};
+        if (!${ev.isNull}) {
+          ${ev.value} = ($value);
+        } else {
+          throw new RuntimeException(
+            "Null value appeared in non-nullable field: " +
+            "ordinal=$ordinal, dataType=${dataType.simpleString}. " +
+            "If the schema is inferred from a Scala tuple/case class, or a Java bean, " +
+            "please try to use scala.Option[_] or other nullable types " +
+            "(e.g. java.lang.Integer instead of int/scala.Int)."
+          );
+        }
+       """
+    }
   }
 }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/CentralMomentAgg.scala

@@ -53,7 +53,7 @@ abstract class CentralMomentAgg(child: Expression) extends ImperativeAggregate w
 
   override def children: Seq[Expression] = Seq(child)
 
-  override def nullable: Boolean = false
+  override def nullable: Boolean = true
 
   override def dataType: DataType = DoubleType
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Corr.scala

@@ -20,7 +20,6 @@ package org.apache.spark.sql.catalyst.expressions.aggregate
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.expressions._
-import org.apache.spark.sql.catalyst.util.TypeUtils
 import org.apache.spark.sql.types._
 
 /**
@@ -42,7 +41,7 @@ case class Corr(
 
   override def children: Seq[Expression] = Seq(left, right)
 
-  override def nullable: Boolean = false
+  override def nullable: Boolean = true
 
   override def dataType: DataType = DoubleType
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateProjection.scala

@@ -42,7 +42,7 @@ object GenerateProjection extends CodeGenerator[Seq[Expression], Projection] {
   protected def bind(in: Seq[Expression], inputSchema: Seq[Attribute]): Seq[Expression] =
     in.map(BindReferences.bindReference(_, inputSchema))
 
-  // Make Mutablility optional...
+  // Make Mutability optional...
   protected def create(expressions: Seq[Expression]): Projection = {
     val ctx = newCodeGenContext()
     val columns = expressions.zipWithIndex.map {
@@ -65,7 +65,7 @@ object GenerateProjection extends CodeGenerator[Seq[Expression], Projection] {
         """
     }.mkString("\n")
 
-    val getCases = (0 until expressions.size).map { i =>
+    val getCases = expressions.indices.map { i =>
       s"case $i: return c$i;"
     }.mkString("\n")
 

sql/catalyst/src/main/scala/org/apache/spark/sql/types/StructType.scala

@@ -358,10 +358,10 @@ object StructType extends AbstractDataType {
           case leftField @ StructField(leftName, leftType, leftNullable, _) =>
             rightMapped.get(leftName)
               .map { case rightField @ StructField(_, rightType, rightNullable, _) =>
-              leftField.copy(
-                dataType = merge(leftType, rightType),
-                nullable = leftNullable || rightNullable)
-            }
+                leftField.copy(
+                  dataType = merge(leftType, rightType),
+                  nullable = leftNullable || rightNullable)
+              }
               .orElse(Some(leftField))
               .foreach(newFields += _)
         }

sql/core/src/main/scala/org/apache/spark/sql/DataFrame.scala

@@ -499,10 +499,16 @@ class DataFrame private[sql](
     // Analyze the self join. The assumption is that the analyzer will disambiguate left vs right
     // by creating a new instance for one of the branch.
     val joined = sqlContext.executePlan(
-      Join(logicalPlan, right.logicalPlan, joinType = Inner, None)).analyzed.asInstanceOf[Join]
+      Join(logicalPlan, right.logicalPlan, joinType = JoinType(joinType), None)
+    ).analyzed.asInstanceOf[Join]
 
     // Project only one of the join columns.
-    val joinedCols = usingColumns.map(col => withPlan(joined.right).resolve(col))
+    //
+    // SPARK-12336: For outer joins, attributes of at least one child plan output will be forced to
+    // be nullable. An `AttributeSet` is necessary so that we are not affected by different
+    // nullability values.
+    val joinedCols = AttributeSet(usingColumns.map(col => withPlan(joined.right).resolve(col)))
+
     val condition = usingColumns.map { col =>
       catalyst.expressions.EqualTo(
         withPlan(joined.left).resolve(col),
@@ -514,12 +520,7 @@ class DataFrame private[sql](
     withPlan {
       Project(
         joined.output.filterNot(joinedCols.contains(_)),
-        Join(
-          joined.left,
-          joined.right,
-          joinType = JoinType(joinType),
-          condition)
-      )
+        joined.copy(condition = condition))
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ddl.scala

@@ -42,7 +42,7 @@ case class DescribeCommand(
       new MetadataBuilder().putString("comment", "name of the column").build())(),
     AttributeReference("data_type", StringType, nullable = false,
       new MetadataBuilder().putString("comment", "data type of the column").build())(),
-    AttributeReference("comment", StringType, nullable = false,
+    AttributeReference("comment", StringType, nullable = true,
       new MetadataBuilder().putString("comment", "comment of the column").build())()
   )
 }

sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala

@@ -23,6 +23,7 @@ import scala.language.postfixOps
 
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.test.SharedSQLContext
+import org.apache.spark.sql.types._
 
 
 class DatasetSuite extends QueryTest with SharedSQLContext {
@@ -489,12 +490,117 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
     }
     assert(e.getMessage.contains("cannot resolve 'c' given input columns a, b"), e.getMessage)
   }
+
+  def testNonNullable[T: Encoder](name: String, schema: StructType, rows: Row*): Unit = {
+    test(s"non-nullable field - $name") {
+      val rowRDD = sqlContext.sparkContext.parallelize(rows)
+      val ds = sqlContext.createDataFrame(rowRDD, schema).as[T]
+      val message = intercept[RuntimeException](ds.collect()).getMessage
+      assert(message.contains("Null value appeared in non-nullable field"))
+    }
+  }
+
+  testNonNullable[ClassData](
+    "scala.Int",
+    StructType(Seq(
+      StructField("a", StringType, nullable = true),
+      StructField("b", IntegerType, nullable = false)
+    )),
+    Row("hello", 1: Integer),
+    Row("world", null)
+  )
+
+  testNonNullable[NestedClassData](
+    "struct",
+    StructType(Seq(
+      StructField("a", StructType(Seq(
+        StructField("a", StringType, nullable = true),
+        StructField("b", IntegerType, nullable = false)
+      )), nullable = false)
+    )),
+    Row(Row("hello", 1: Integer)),
+    Row(null)
+  )
+
+  ignore("non-nullable field in nested struct") {
+    testNonNullable[NestedClassData](
+      "non-nullable field in nested struct",
+      StructType(Seq(
+        StructField("a", StructType(Seq(
+          StructField("a", StringType, nullable = true),
+          StructField("b", IntegerType, nullable = false)
+        )), nullable = false)
+      )),
+      Row(Row("hello", 1: Integer)),
+      Row(Row("hello", null))
+    )
+  }
+
+  testNonNullable[NestedNonNullableArray](
+    "array",
+    StructType(Seq(
+      StructField("a", ArrayType(StructType(Seq(
+        StructField("a", StringType, nullable = true),
+        StructField("b", IntegerType, nullable = false)
+      ))), nullable = false)
+    )),
+    Row(Seq(Row("hello", 1))),
+    Row(null)
+  )
+
+  ignore("non-nullable element in nested array") {
+    testNonNullable[NestedNonNullableArray](
+      "non-nullable element in nested array",
+      StructType(Seq(
+        StructField("a", ArrayType(StructType(Seq(
+          StructField("a", StringType, nullable = true),
+          StructField("b", IntegerType, nullable = false)
+        )), containsNull = false), nullable = false)
+      )),
+      Row(Seq(Row("hello", 1), null))
+    )
+  }
+
+  testNonNullable[NestedNonNullableMap](
+    "map",
+    StructType(Seq(
+      StructField("a", MapType(
+        IntegerType,
+        StructType(Seq(
+          StructField("a", StringType, nullable = true),
+          StructField("b", IntegerType, nullable = false)
+        ))
+      ), nullable = false)
+    )),
+    Row(Map(1 -> Row("hello", 1))),
+    Row(null)
+  )
+
+  ignore("non-nullable value in nested map") {
+    testNonNullable[NestedNonNullableMap](
+      "non-nullable value in nested map",
+      StructType(Seq(
+        StructField("a", MapType(
+          IntegerType,
+          StructType(Seq(
+            StructField("a", StringType, nullable = true),
+            StructField("b", IntegerType, nullable = false)
+          ))
+        ), nullable = false)
+      )),
+      Row(Map(1 -> Row("hello", 1), 2 -> null))
+    )
+  }
 }
 
 case class ClassData(a: String, b: Int)
 case class ClassData2(c: String, d: Int)
 case class ClassNullableData(a: String, b: Integer)
 
+case class NestedClassData(a: ClassData)
+case class NestedNonNullableArray(a: Array[ClassData])
+case class NestedNonNullableMap(a: scala.collection.Map[Int, ClassData])
+
 /**
  * A class used to test serialization using encoders. This class throws exceptions when using
  * Java serialization -- so the only way it can be "serialized" is through our encoders.