take nullable into consideration

Author: cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -1083,9 +1083,12 @@ class Analyzer(
           assert(parameterTypes.length == inputs.length)
 
           val inputsNullCheck = parameterTypes.zip(inputs)
-            .filter(_._1.isPrimitive)
-            .map(i => IsNull(i._2))
-            .reduceLeftOption[Expression]((i1, i2) => Or(i1, i2))
+            // TODO: skip null handling for not-nullable primitive inputs after we can completely
+            // trust the `nullable` information.
+            // .filter { case (cls, expr) => cls.isPrimitive && expr.nullable }
+            .filter { case (cls, _) => cls.isPrimitive }
+            .map { case (_, expr) => IsNull(expr) }
+            .reduceLeftOption[Expression]((e1, e2) => Or(e1, e2))
           inputsNullCheck.map(If(_, Literal.create(null, udf.dataType), udf)).getOrElse(udf)
       }
     }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala

@@ -188,19 +188,34 @@ class AnalysisSuite extends AnalysisTest {
       )
     }
 
+    // non-primitive parameters do not need special null handling
     val udf1 = ScalaUDF((s: String) => "x", StringType, string :: Nil)
     val expected1 = udf1
     checkUDF(udf1, expected1)
 
+    // only primitive parameter needs special null handling
     val udf2 = ScalaUDF((s: String, d: Double) => "x", StringType, string :: double :: Nil)
     val expected2 = If(IsNull(double), nullResult, udf2)
     checkUDF(udf2, expected2)
 
+    // special null handling should apply to all primitive parameters
     val udf3 = ScalaUDF((s: Short, d: Double) => "x", StringType, short :: double :: Nil)
     val expected3 = If(
       IsNull(short) || IsNull(double),
       nullResult,
       udf3)
     checkUDF(udf3, expected3)
+
+    // we can skip special null handling for primitive parameters that are not nullable
+    // TODO: this is disabled for now as we can not completely trust `nullable`.
+    val udf4 = ScalaUDF(
+      (s: Short, d: Double) => "x",
+      StringType,
+      short :: double.withNullability(false) :: Nil)
+    val expected4 = If(
+      IsNull(short),
+      nullResult,
+      udf4)
+    // checkUDF(udf4, expected4)
   }
 }