Revert "[SPARK-26580][SQL] remove Scala 2.11 hack for Scala UDF"

This reverts commit 1f1d98c.

Author: Robert Kruszewski

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

@@ -960,6 +960,23 @@ trait ScalaReflection extends Logging {
     tpe.dealias.erasure.typeSymbol.asClass.fullName
   }
 
+  /**
+   * Returns the nullability of the input parameter types of the scala function object.
+   *
+   * Note that this only works with Scala 2.11, and the information returned may be inaccurate if
+   * used with a different Scala version.
+   */
+  def getParameterTypeNullability(func: AnyRef): Seq[Boolean] = {
+    if (!Properties.versionString.contains("2.11")) {
+      logWarning(s"Scala ${Properties.versionString} cannot get type nullability correctly via " +
+        "reflection, thus Spark cannot add proper input null check for UDF. To avoid this " +
+        "problem, use the typed UDF interfaces instead.")
+    }
+    val methods = func.getClass.getMethods.filter(m => m.getName == "apply" && !m.isBridge)
+    assert(methods.length == 1)
+    methods.head.getParameterTypes.map(!_.isPrimitive)
+  }
+
   /**
    * Returns the parameter names and types for the primary constructor of this type.
    *

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

@@ -54,6 +54,18 @@ case class ScalaUDF(
     udfDeterministic: Boolean = true)
   extends Expression with NonSQLExpression with UserDefinedExpression {
 
+  // The constructor for SPARK 2.1 and 2.2
+  def this(
+      function: AnyRef,
+      dataType: DataType,
+      children: Seq[Expression],
+      inputTypes: Seq[DataType],
+      udfName: Option[String]) = {
+    this(
+      function, dataType, children, ScalaReflection.getParameterTypeNullability(function),
+      inputTypes, udfName, nullable = true, udfDeterministic = true)
+  }
+
   override lazy val deterministic: Boolean = udfDeterministic && children.forall(_.deterministic)
 
   override def toString: String =

sql/core/src/main/scala/org/apache/spark/sql/expressions/UserDefinedFunction.scala

@@ -102,7 +102,17 @@ private[sql] case class SparkUserDefinedFunction(
     // It's possible that some of the inputs don't have a specific type(e.g. `Any`),  skip type
     // check and null check for them.
     val inputTypes = inputSchemas.map(_.map(_.dataType).getOrElse(AnyDataType))
-    val inputsNullSafe = inputSchemas.map(_.map(_.nullable).getOrElse(true))
+
+    val inputsNullSafe = if (inputSchemas.isEmpty) {
+      // This is for backward compatibility of `functions.udf(AnyRef, DataType)`. We need to
+      // do reflection of the lambda function object and see if its arguments are nullable or not.
+      // This doesn't work for Scala 2.12 and we should consider removing this workaround, as Spark
+      // uses Scala 2.12 by default since 3.0.
+      ScalaReflection.getParameterTypeNullability(f)
+    } else {
+      inputSchemas.map(_.map(_.nullable).getOrElse(true))
+    }
+
     ScalaUDF(
       f,
       dataType,

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

@@ -4355,13 +4355,6 @@ object functions {
    * By default the returned UDF is deterministic. To change it to nondeterministic, call the
    * API `UserDefinedFunction.asNondeterministic()`.
    *
-   * Note that, although the Scala closure can have primitive-type function argument, it doesn't
-   * work well with null values. Because the Scala closure is passed in as Any type, there is no
-   * type information for the function arguments. Without the type information, Spark may blindly
-   * pass null to the Scala closure with primitive-type argument, and the closure will see the
-   * default value of the Java type for the null argument, e.g. `udf((x: Int) => x, IntegerType)`,
-   * the result is 0 for null input.
-   *
    * @param f  A closure in Scala
    * @param dataType  The output data type of the UDF
    *

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

@@ -423,19 +423,6 @@ class UDFSuite extends QueryTest with SharedSQLContext {
     }
   }
 
-  test("SPARK-25044 Verify null input handling for primitive types - with udf(Any, DataType)") {
-    val f = udf((x: Int) => x, IntegerType)
-    checkAnswer(
-      Seq(new Integer(1), null).toDF("x").select(f($"x")),
-      Row(1) :: Row(0) :: Nil)
-
-    val f2 = udf((x: Double) => x, DoubleType)
-    checkAnswer(
-      Seq(new java.lang.Double(1.1), null).toDF("x").select(f2($"x")),
-      Row(1.1) :: Row(0.0) :: Nil)
-
-  }
-
   test("SPARK-26308: udf with decimal") {
     val df1 = spark.createDataFrame(
       sparkContext.parallelize(Seq(Row(new BigDecimal("2011000000000002456556")))),