initial commit

Author: kiszk

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

@@ -63,7 +63,11 @@ case class BoundReference(ordinal: Int, dataType: DataType, nullable: Boolean)
     val value = ctx.getValue(ctx.INPUT_ROW, dataType, ordinal.toString)
     if (ctx.currentVars != null && ctx.currentVars(ordinal) != null) {
       val oev = ctx.currentVars(ordinal)
-      ev.isNull = oev.isNull
+      ev.isNull = if (!nullable || ctx.INPUT_ROW == null) {
+        oev.isNull
+      } else {
+        s"((${oev.isNull}) || ${ctx.INPUT_ROW}.isNullAt($ordinal))"
+      }
       ev.value = oev.value
       val code = oev.code
       oev.code = ""

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

@@ -1722,4 +1722,15 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {
         "Cannot have map type columns in DataFrame which calls set operations"))
     }
   }
+
+  test("SPARK-19959: df[java.lang.Long].collect includes null throws NullPointerException") {
+    val dfInt = sparkContext.parallelize(Seq[java.lang.Integer](0, null, 2), 1).toDF
+    assert(dfInt.collect === Array(Row(0), Row(null), Row(2)))
+    val dfLong = sparkContext.parallelize(Seq[java.lang.Long](0L, null, 2L), 1).toDF
+    assert(dfLong.collect === Array(Row(0L), Row(null), Row(2L)))
+    val dfFloat = sparkContext.parallelize(Seq[java.lang.Float](0.0F, null, 2.0F), 1).toDF
+    assert(dfFloat.collect === Array(Row(0.0F), Row(null), Row(2.0F)))
+    val dfDouble = sparkContext.parallelize(Seq[java.lang.Double](0.0D, null, 2.0D), 1).toDF
+    assert(dfDouble.collect === Array(Row(0.0D), Row(null), Row(2.0D)))
+  }
 }