[SPARK-32877][SQL] Fix Hive UDF not support decimal type in complex type

sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala

@@ -47,7 +47,8 @@ private[hive] case class HiveSimpleUDF(
   with HiveInspectors
   with CodegenFallback
   with Logging
-  with UserDefinedExpression {
+  with UserDefinedExpression
+  with ImplicitCastInputTypes {
 
   override lazy val deterministic: Boolean = isUDFDeterministic && children.forall(_.deterministic)
 
@@ -57,8 +58,19 @@ private[hive] case class HiveSimpleUDF(
   lazy val function = funcWrapper.createFunction[UDF]()
 
   @transient
-  private lazy val method =
-    function.getResolver.getEvalMethod(children.map(_.dataType.toTypeInfo).asJava)
+  private lazy val method = {
+    // the simple UDF method must be 'evaluate'
+    val methods = function.getClass.getMethods.filter(_.getName == "evaluate")
+    val passedMethod = methods.filter(_.getGenericParameterTypes.length == children.length)
+
+    // no matching parameter num for evaluate method
+    if (passedMethod.isEmpty) {
+      throw new NoMatchingMethodException(function.getClass,
+        children.map(_.dataType.toTypeInfo).asJava, methods.toSeq.asJava)
+    }
+    // if there exists many method, we choose the first
+    methods.head
+  }
 
   @transient
   private lazy val arguments = children.map(toInspector).toArray
@@ -69,6 +81,10 @@ private[hive] case class HiveSimpleUDF(
     udfType != null && udfType.deterministic() && !udfType.stateful()
   }
 
+  override def inputTypes: Seq[AbstractDataType] = {
+    method.getGenericParameterTypes.map(javaTypeToDataType)
+  }
+
   override def foldable: Boolean = isUDFDeterministic && children.forall(_.foldable)
 
   // Create parameter converters

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala

@@ -658,6 +658,24 @@ class HiveUDFSuite extends QueryTest with TestHiveSingleton with SQLTestUtils {
     }
   }
 
+  test("SPARK-32877: Fix Hive UDF not support decimal type in complex type") {
+    withUserDefinedFunction("testArraySum" -> false) {
+      sql(s"CREATE FUNCTION testArraySum AS '${classOf[ArraySumUDF].getName}'")
+      checkAnswer(
+        sql("SELECT testArraySum(array(1, 1.1, 1.2))"),
+        Seq(Row(3.3)))
+
+      val msg = intercept[AnalysisException] {
+        sql("SELECT testArraySum(1)")
+      }.getMessage
+      assert(msg.contains("cannot resolve 'default.testArraySum(1)' due to data type mismatch"))
+
+      val msg2 = intercept[AnalysisException] {
+        sql("SELECT testArraySum(1, 2)")
+      }.getMessage
+      assert(msg2.contains(s"No handler for UDF/UDAF/UDTF '${classOf[ArraySumUDF].getName}'"))
+    }
+  }
 }
 
 class TestPair(x: Int, y: Int) extends Writable with Serializable {
@@ -741,3 +759,14 @@ class StatelessUDF extends UDF {
     result
   }
 }
+
+class ArraySumUDF extends UDF {
+  import scala.collection.JavaConverters._
+  def evaluate(values: java.util.List[java.lang.Double]): java.lang.Double = {
+    var r = 0d
+    for (v <- values.asScala) {
+      r += v
+    }
+    r
+  }
+}