[SPARK-42771][SQL] Refactor HiveGenericUDF

### What changes were proposed in this pull request?
The pr aims to refactor HiveGenericUDF.

### Why are the changes needed?
Following #39949.
Make the code more concise.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Pass GA.

Closes #40394 from panbingkun/refactor_HiveGenericUDF.

Lead-authored-by: panbingkun <[email protected]>
Co-authored-by: panbingkun <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: panbingkun

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

@@ -130,57 +130,32 @@ private[hive] case class HiveGenericUDF(
     name: String, funcWrapper: HiveFunctionWrapper, children: Seq[Expression])
   extends Expression
   with HiveInspectors
-  with Logging
   with UserDefinedExpression {
 
   override def nullable: Boolean = true
 
-  override lazy val deterministic: Boolean = isUDFDeterministic && children.forall(_.deterministic)
-
-  override def foldable: Boolean =
-    isUDFDeterministic && returnInspector.isInstanceOf[ConstantObjectInspector]
-
-  @transient
-  lazy val function = funcWrapper.createFunction[GenericUDF]()
+  override lazy val deterministic: Boolean =
+    isUDFDeterministic && children.forall(_.deterministic)
 
-  @transient
-  private lazy val argumentInspectors = children.map(toInspector)
+  override def foldable: Boolean = isUDFDeterministic &&
+    evaluator.returnInspector.isInstanceOf[ConstantObjectInspector]
 
-  @transient
-  private lazy val returnInspector = {
-    function.initializeAndFoldConstants(argumentInspectors.toArray)
-  }
+  override lazy val dataType: DataType = inspectorToDataType(evaluator.returnInspector)
 
-  // Visible for codegen
   @transient
-  lazy val unwrapper: Any => Any = unwrapperFor(returnInspector)
+  private lazy val evaluator = new HiveGenericUDFEvaluator(funcWrapper, children)
 
   @transient
-  private lazy val isUDFDeterministic = {
-    val udfType = function.getClass.getAnnotation(classOf[HiveUDFType])
+  private val isUDFDeterministic = {
+    val udfType = evaluator.function.getClass.getAnnotation(classOf[HiveUDFType])
     udfType != null && udfType.deterministic() && !udfType.stateful()
   }
 
-  // Visible for codegen
-  @transient
-  lazy val deferredObjects: Array[DeferredObject] = argumentInspectors.zip(children).map {
-    case (inspect, child) => new DeferredObjectAdapter(inspect, child.dataType)
-  }.toArray[DeferredObject]
-
-  override lazy val dataType: DataType = inspectorToDataType(returnInspector)
-
   override def eval(input: InternalRow): Any = {
-    returnInspector // Make sure initialized.
-
-    var i = 0
-    val length = children.length
-    while (i < length) {
-      val idx = i
-      deferredObjects(i).asInstanceOf[DeferredObjectAdapter]
-        .set(children(idx).eval(input))
-      i += 1
+    children.zipWithIndex.map {
+      case (child, idx) => evaluator.setArg(idx, child.eval(input))
     }
-    unwrapper(function.evaluate(deferredObjects))
+    evaluator.evaluate()
   }
 
   override def prettyName: String = name
@@ -191,18 +166,18 @@ private[hive] case class HiveGenericUDF(
 
   override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression =
     copy(children = newChildren)
-  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
-    val refTerm = ctx.addReferenceObj("this", this)
-    val childrenEvals = children.map(_.genCode(ctx))
 
-    val setDeferredObjects = childrenEvals.zipWithIndex.map {
+  protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val refEvaluator = ctx.addReferenceObj("evaluator", evaluator)
+    val evals = children.map(_.genCode(ctx))
+
+    val setValues = evals.zipWithIndex.map {
       case (eval, i) =>
-        val deferredObjectAdapterClz = classOf[DeferredObjectAdapter].getCanonicalName
         s"""
            |if (${eval.isNull}) {
-           |  (($deferredObjectAdapterClz) $refTerm.deferredObjects()[$i]).set(null);
+           |  $refEvaluator.setArg($i, null);
            |} else {
-           |  (($deferredObjectAdapterClz) $refTerm.deferredObjects()[$i]).set(${eval.value});
+           |  $refEvaluator.setArg($i, ${eval.value});
            |}
            |""".stripMargin
     }
@@ -211,13 +186,12 @@ private[hive] case class HiveGenericUDF(
     val resultTerm = ctx.freshName("result")
     ev.copy(code =
       code"""
-         |${childrenEvals.map(_.code).mkString("\n")}
-         |${setDeferredObjects.mkString("\n")}
+         |${evals.map(_.code).mkString("\n")}
+         |${setValues.mkString("\n")}
          |$resultType $resultTerm = null;
          |boolean ${ev.isNull} = false;
          |try {
-         |  $resultTerm = ($resultType) $refTerm.unwrapper().apply(
-         |    $refTerm.function().evaluate($refTerm.deferredObjects()));
+         |  $resultTerm = ($resultType) $refEvaluator.evaluate();
          |  ${ev.isNull} = $resultTerm == null;
          |} catch (Throwable e) {
          |  throw QueryExecutionErrors.failedExecuteUserDefinedFunctionError(
@@ -235,6 +209,36 @@ private[hive] case class HiveGenericUDF(
   }
 }
 
+class HiveGenericUDFEvaluator(
+    funcWrapper: HiveFunctionWrapper, children: Seq[Expression])
+  extends HiveInspectors
+  with Serializable {
+
+  @transient
+  lazy val function = funcWrapper.createFunction[GenericUDF]()
+
+  @transient
+  private lazy val argumentInspectors = children.map(toInspector)
+
+  @transient
+  lazy val returnInspector = {
+    function.initializeAndFoldConstants(argumentInspectors.toArray)
+  }
+
+  @transient
+  private lazy val deferredObjects: Array[DeferredObject] = argumentInspectors.zip(children).map {
+    case (inspect, child) => new DeferredObjectAdapter(inspect, child.dataType)
+  }.toArray[DeferredObject]
+
+  @transient
+  private lazy val unwrapper: Any => Any = unwrapperFor(returnInspector)
+
+  def setArg(index: Int, arg: Any): Unit =
+    deferredObjects(index).asInstanceOf[DeferredObjectAdapter].set(arg)
+
+  def evaluate(): Any = unwrapper(function.evaluate(deferredObjects))
+}
+
 /**
  * Converts a Hive Generic User Defined Table Generating Function (UDTF) to a
  * `Generator`. Note that the semantics of Generators do not allow