[DRAFT][SPARK-51705][CONNECT][SCALA] Support SparkSession.broadcast() for Scala UDFs over Spark Connect

[Tagar]

[DRAFT][POC] Stacks on #1 (broadcast-connect-python-v1). Base is the Python-v1 branch, so this diff shows only the Scala delta. SPARK-51705 Scala follow-up.

What changes were proposed in this pull request?

Extends broadcast variables over Spark Connect to the Scala ScalarScalaUDF path (Python v1 is #1). This is the path the Vizio "Cooker" workload needs — its broadcast is consumed inside a Scala UDF.

The crux: a Scala UDF is a Java-serialized closure that carries the Broadcast[T] object itself, so — unlike Python, where the id rides an out-of-band broadcast_ids list and the worker is handed a ready broadcastVars — the id must resolve during closure deserialization on the server.

client (JVM-less):
  spark.broadcast(v: T)
    -> SparkSerDeUtils.serialize(v) -> client.artifactManager.cacheArtifact(bytes) -> sha256   (same cache/ channel as Python)
    -> ExecutePlan(CreateBroadcastCommand{artifact_hash, value_type = JVM})
    <- CreateBroadcastResult{broadcast_id}    // = driver-side Broadcast.id
    -> new ConnectBroadcast[T](broadcast_id, v)     // retains v for driver-side .value; @transient
  capture in a UDF:
    ConnectBroadcast.writeReplace() -> ConnectBroadcastRef(id)   // value never serialized (already sent via artifact)
    + records id in ConnectBroadcastCapture; UdfToProtoUtils.toProto drains it into ScalarScalaUDF.broadcast_ids
server (SparkConnectPlanner.unpackScalaUDF):
    resolve broadcast_ids from the per-session SessionHolder registry (unknown id -> BROADCAST_NOT_FOUND)
    bind them in a thread-local around deserialization; ConnectBroadcastRef.readResolve swaps each token
    for the real driver-side Broadcast[_]
executor: UNCHANGED — classic TorrentBroadcast fan-out.

Module boundary — the spark-connect-shims mechanism (important)

The JVM-less client module sql/connect/common depends only on spark-sql-api, not spark-core, so org.apache.spark.broadcast.Broadcast is not normally on its classpath. The client compiles against spark-connect-shims — a module that hand-declares empty stub classes at real FQNs (org.apache.spark.rdd.RDD, SparkContext, …); modules with real spark-core (connect/server, catalyst, sql/core) exclude the shim so the identical FQN binds to the real class ("compile against stub, link against real").

Broadcast had no shim. This PR adds a Broadcast shim to spark-connect-shims mirroring the real abstract surface (getValue/doUnpersist/doDestroy + id/value). Result:

• ConnectBroadcast[T] extends Broadcast[T] and SparkSession.broadcast[T]: Broadcast[T] compile on the client (shim) and the server (real core, shim excluded).
• Full classic source-portability is preserved — the user writes the real org.apache.spark.broadcast.Broadcast[T] FQN; the shim binds on the client, the real class on the server/executor, so UDF source is byte-identical to classic.
• No spark-core is added to the thin client (charter intact); this rides the exact dependency swap the build already uses.
• The wire token ConnectBroadcastRef is core-free (id-only) and shared client↔server (connect/server depends on connect-common), so Java deserialization binds the same class on both sides.

Other pieces:

• CreateBroadcastCommand.value_type = {PYTHON (default), JVM}. JVM Java-deserializes the uploaded bytes (with the artifact classloader) and sc.broadcast(value) directly, so the executor reads Broadcast[T]. SessionHolder.broadcasts widened to Broadcast[_] to hold both Broadcast[PythonBroadcast] (Python) and Broadcast[T] (Scala).
• The client's own checkDeserializable round-trip in toUdfPacketBytes fires readResolve with no registry bound → returns a safe unresolved placeholder (never executed) rather than throwing.

Why are the changes needed?

Broadcast is a core primitive with no Scala equivalent on Spark Connect; the Vizio Cooker migration (and the RDD/SparkContext-heavy Scala install base) needs it. SPARK-51705.

Does this PR introduce any user-facing change?

Yes — SparkSession.broadcast[T](value): Broadcast[T] on the Connect Scala client. Additive proto only (ScalarScalaUDF.broadcast_ids, CreateBroadcastCommand.value_type + BroadcastValueType enum) plus the Broadcast compile-shim.

How was this patch tested?

Server unit tests added (registry holds Broadcast[T]; readResolve swap under withRegistry; client round-trip placeholder; capture round-trip). Locally: proto regenerated with the pinned toolchain (protobuf 6.33.5), scalafmt 3.8.6 --test clean. Full Scala compile + E2E run under CI.

Scope / follow-ups (DRAFT / DO-NOT-MERGE)

• Stacks on [DRAFT][SPARK-51705][CONNECT][PYTHON] Support SparkSession.broadcast() (broadcast variables over Spark Connect) #1 (Python v1); merge that first.
• Open architectural question (raised by Serverless eng): how this fits Databricks Lakeguard / the unified-UDF (BrickFunctions) isolation platform. The isolation-platform design is being discussed internally and gates any productionization.
• SPARK-46032 (Scala closure deserialization on Connect) is a related reliability risk.
• End-to-end Scala UDF test, UDAF coverage, and BROADCAST_VALUE_TOO_LARGE enforcement are follow-ups.

This pull request and its description were written by Isaac.