Tagar opened a new pull request, #57573: URL: https://github.com/apache/spark/pull/57573
### What changes were proposed in this pull request? This PR adds **`SparkSession.broadcast()` support for Scala UDFs over Spark Connect** (`SPARK-51705`), building on the Python-side support in #57385. Spark Connect's thin client has no `SparkContext`, so the classic `sc.broadcast(value)` API is unavailable — attempting to close over a broadcast in a Connect UDF fails. This change introduces a server-mediated broadcast lane so Scala closures can capture and reuse a broadcast value on the server side, reusing the classic `Broadcast[T]` / `TorrentBroadcast` machinery once the value reaches the driver. Key pieces: - `SparkSession.broadcast[T](value: T): Broadcast[T]` on the Connect client — serializes the value, uploads it via the artifact/cache lane, and registers it through a new `CreateBroadcastCommand` (`value_type = JVM`). - A `Broadcast[T]` shim in `spark-connect-shims` so `connect-common` can reference the real FQN without depending on `spark-core`. - `ConnectBroadcast` / `ConnectBroadcastRef` / thread-local resolver so a Java-serialized Scala UDF closure (`UdfPacket`) can resolve broadcast ids **during** deserialization on the server (`writeReplace` / `readResolve`). - Server planner (`SparkConnectPlanner`) resolves registered broadcasts and rebinds them into the deserialized closure; `SessionHolder` owns the per-session broadcast registry and its lifecycle. This is marked **POC** and stacks on #57385 (the first commit here is that PR's Python change; the second is the Scala work). It is intended to accompany the internal design discussion referenced in SPARK-51705. ### Why are the changes needed? Broadcast variables are the single largest feature-parity gap for migrating large Scala/UDF pipelines from classic Spark to Spark Connect / serverless. Without them, a large lookup table cannot be shared once per executor and must be re-materialized per record, causing OOM and cost blow-ups. See SPARK-51705. ### Does this PR introduce _any_ user-facing change? Yes — a new `SparkSession.broadcast()` method usable from the Scala Connect client, plus `unpersistBroadcast`. Behavior mirrors the classic broadcast API from the user's perspective. ### How was this patch tested? - New `SparkConnectSessionHolderSuite` coverage for the broadcast registry and lifecycle. - Full `build_and_test` matrix green on the author's fork (all connect / avro unit tests, linters, protobuf/Python codegen check, precompile). ### Was this patch authored or co-authored using generative AI tooling? Yes, drafted with assistance from generative AI tooling and reviewed by the author. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
