fix: ship the functions of a native plan's ScalaUDFs once, in one serialization stream

spark/src/main/scala/org/apache/comet/CometExecIterator.scala

@@ -44,6 +44,7 @@ import org.apache.comet.exceptions.CometQueryExecutionException
 import org.apache.comet.parquet.CometFileKeyUnwrapper
 import org.apache.comet.serde.Config.ConfigMap
 import org.apache.comet.shuffle.ShufflePartitionPusher
+import org.apache.comet.udf.codegen.CometUdfFunctionPool
 import org.apache.comet.vector.NativeUtil
 
 /**
@@ -123,6 +124,8 @@ class CometExecIterator(
 
     val memoryConfig = CometExecIterator.getMemoryConfig(conf)
 
+    CometUdfFunctionPool.registerPlan(taskContext, protobufQueryPlan)
+
     val createdPlan = nativeLib.createPlan(
       id,
       inputObjects,

spark/src/main/scala/org/apache/comet/serde/CometScalaUDF.scala

@@ -32,7 +32,7 @@ import org.apache.comet.CometSparkSessionExtensions.{withCodegenDispatchExpr, wi
 import org.apache.comet.codegen.CometBatchKernelCodegen
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal, serializeDataType}
-import org.apache.comet.udf.codegen.CometScalaUDFCodegen
+import org.apache.comet.udf.codegen.{CometScalaUDFCodegen, CometUdfFunctionPool}
 
 /**
  * Routes scalar `ScalaUDF` (Scala and Java UDFs) through the codegen dispatcher.
@@ -124,13 +124,25 @@ object CometScalaUDF extends CometExpressionSerde[ScalaUDF] {
     // fails planning, which is a much worse outcome than falling the operator back to Spark.
     // `CometStaticInvoke` / `CometInvoke` route unrecognized nodes here as a catch-all, so the
     // trees reaching this point are arbitrary.
+    val functions = CometUdfFunctionPool.functionsOf(boundExpr)
+    functions.find(!CometUdfFunctionPool.isSerializable(_)).foreach { function =>
+      withFallbackReason(
+        expr,
+        s"$exprName: codegen dispatch: UDF function ${function.getClass.getName} could not " +
+          "be closure-serialized")
+      return None
+    }
     val bytes =
       try {
-        val serializer = SparkEnv.get.closureSerializer.newInstance()
-        val buffer = serializer.serialize(boundExpr)
-        val serialized = new Array[Byte](buffer.remaining())
-        buffer.get(serialized)
-        serialized
+        if (functions.nonEmpty) {
+          CometUdfFunctionPool.serializeExpression(boundExpr, functions)
+        } else {
+          val serializer = SparkEnv.get.closureSerializer.newInstance()
+          val buffer = serializer.serialize(boundExpr)
+          val serialized = new Array[Byte](buffer.remaining())
+          buffer.get(serialized)
+          serialized
+        }
       } catch {
         // `NonFatal` rather than `NotSerializableException`: Java serialization reports an
         // unserializable object graph as one of several exception types depending on where in

spark/src/main/scala/org/apache/comet/udf/codegen/CometScalaUDFCodegen.scala

@@ -155,9 +155,13 @@ class CometScalaUDFCodegen extends CometUDF with Logging {
           .getOrElse(classOf[Expression].getClassLoader)
         val boundExpr =
           try {
-            SparkEnv.get.closureSerializer
-              .newInstance()
-              .deserialize[Expression](ByteBuffer.wrap(bytes), loader)
+            if (CometUdfFunctionPool.isSharedExpression(bytes)) {
+              CometUdfFunctionPool.deserializeExpression(bytes, loader)
+            } else {
+              SparkEnv.get.closureSerializer
+                .newInstance()
+                .deserialize[Expression](ByteBuffer.wrap(bytes), loader)
+            }
           } catch {
             case NonFatal(t) =>
               logError(