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

[msafonov]

Problem

experiment-platform.ltv_flag on Comet 1.1-11 died with Java heap space in the 3G driver (vanilla passes). The codegen dispatcher (CometScalaUDF.emitJvmCodegenDispatch) closure-serialized every ScalaUDF expression in its own stream. The job's project has 10 UDFs, 9 of them capture one ~150 MB expResults map, so each plan conversion built ~1.35 GB of byte arrays (plus copies), and AQE re-converts the plan after every stage. Vanilla serializes the plan in one stream and pays 155 MiB once.

Fix

• emitJvmCodegenDispatch serializes the bound expression with each ScalaUDF.function replaced by a small arity-matching reference (CometSharedFunctionRef(id)); functions are checked for serializability once per function identity (null stream) so non-serializable UDFs still fall back to Spark at plan time.
• CometExec.serializeNativePlan (every native plan serialization) appends one function pool for the functions the plan's expressions reference (unknown Operator field 99, ignored by the native side): all functions in one Java-serialization stream, so objects they share are written once. The pool is cached per function set (soft values), so AQE re-conversions reuse the bytes instead of serializing again.
• CometExecIterator registers the pools of its plan for the task (root and nested children, e.g. the shuffle writer's inlined child); a pool is deserialized once per task on first use and dropped at task completion. CometScalaUDFCodegen resolves the references when it deserializes an expression.
• Expressions without ScalaUDFs (other dispatched built-ins) keep the old self-contained bytes.

Tests

• New CometCodegenSuite test "UDFs that capture one large object ship it once per serialized plan": 8 UDF expressions over 3 functions sharing a 4 MiB array. Serialized native plan < 2 x 4 MiB (the same test without the references measured 37.9 MB, i.e. one copy per expression); one pool build, pool deserialized on executors, a second run of the query reuses the cached pool (no rebuild); results match Spark.
• CometCodegenSuite, CometScalaUDFClassLoaderSuite, CometCodegenSourceSuite (Spark 3.5): 168 passed, 0 failed.

🤖 Generated with Claude Code