You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yi...@apache.org on 2026/09/28 18:55:04 UTC
(spark) branch master updated: [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads
This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new a648b6b3467f [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads
a648b6b3467f is described below
commit a648b6b3467f1910e07b586896a59b7a4106d0c1
Author: Yicong Huang <17...@users.noreply.github.com>
AuthorDate: Mon Sep 28 18:54:47 2026 +0000
[SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads
### What changes were proposed in this pull request?
This PR makes Spark Connect Scala UDF payload deserialization accept audited `serialVersionUID` changes of `org.apache.spark.sql.types` classes between Spark versions.
- Adds `UdfSerialization`, an `ObjectInputStream`-based reader with an exact table of audited `(class, stream SUID, local SUID)` transitions. A class descriptor is rebound to the local class only when its triple is in the table, the local class uses default field serialization (no custom `readObject`/`writeObject`, not `Externalizable`) and does not declare a `serialVersionUID`, and the complete persistent field layout of the stream and local descriptors is identical (including Scala [...]
- The table is audited against the published `spark-sql-api_2.13` jars of 4.0.0 to 4.0.4, 4.1.0 to 4.1.3 and 4.2.0: 24 classes (the `DataType` singletons, `StringType`, `Decimal`, `UserDefinedType`, ...) changed their computed `serialVersionUID` on master while keeping the same default serialized form.
- Routes the server-side decode paths `SparkConnectPlanner.unpackScalaUDF`, the Scala `foreachBatch` path `StreamingForeachBatchHelper.scalaForeachBatchWrapper`, and `UdfPacket.apply(InputStream)` through it.
`StructField`, `StructType`, `CharType` and `VarcharType` are not in the table: their persistent field layout differs from the 4.x releases (new private lazy-val cache fields, and a new `collation` field for `CharType`/`VarcharType`), so their payloads are still rejected rather than read through a mismatched layout.
### Why are the changes needed?
Many `org.apache.spark.sql.types` classes, e.g. the `DataType` singletons, `StringType` and `Decimal`, do not declare a `serialVersionUID`, so the JVM computes it from the whole class shape, including members that are irrelevant to serialization such as the `$anonfun$` methods Scala emits for lambdas. Such a class can change its `serialVersionUID` between releases without changing its serialized form. A Scala UDF payload references these classes through its encoders, so a plain `Objec [...]
### Does this PR introduce _any_ user-facing change?
Yes. A Scala UDF payload produced by one of the audited releases, whose `org.apache.spark.sql.types` classes differ from this build only in an audited `serialVersionUID` change, now deserializes instead of failing with `InvalidClassException`. Any other `serialVersionUID` mismatch, and any change of the persistent field layout, still fails as before.
### How was this patch tested?
New `UdfSerializationSuite`:
- Fixture pairs with same-length class names plus an in-place class-name patch of the serialized stream simulate a producer and a consumer with different `serialVersionUID`s. An audited change is accepted through both the `(bytes, loader)` and `(InputStream)` overloads, while a plain `ObjectInputStream` rejects the same stream. It is rejected when the change is not audited, and even when audited if the field layout differs, the consumer declares a `serialVersionUID`, the consumer has [...]
- Every table entry is checked to target this build's `serialVersionUID` and a class with default serialization, so a future change to these classes forces a re-audit.
- Every table entry is backed by a released producer descriptor: test resources hold, for 4.0.0, 4.0.4, 4.1.3 and 4.2.0, a stream of the `Class` objects of the audited classes as loaded from that release's published jar. The suite checks that each entry's `(className, streamSuid)` appears in them, that a plain `ObjectInputStream` rejects them, and that `UdfSerialization` reads them.
- A `UdfPacket` payload with a production-audited `serialVersionUID` change is accepted through `UdfPacket.apply`, `SparkConnectPlanner.unpackScalaUDF` and `StreamingForeachBatchHelper.scalaForeachBatchWrapper`, so reverting any of these call sites fails the suite.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
This pull request and its description were written by Isaac.
Closes #58962 from Yicong-Huang/suid-tolerant-udf-deser.
Authored-by: Yicong Huang <17...@users.noreply.github.com>
Signed-off-by: Yicong-Huang <17...@users.noreply.github.com>
---
.../spark/sql/connect/common/UdfPacket.scala | 7 +-
.../sql/connect/common/UdfSerialization.scala | 267 +++++++++++++++++++
.../sql/connect/planner/SparkConnectPlanner.scala | 7 +-
.../planner/StreamingForeachBatchHelper.scala | 5 +-
.../udf-serialization/sql-types-classes-4.0.0.bin | Bin 0 -> 173 bytes
.../udf-serialization/sql-types-classes-4.0.4.bin | Bin 0 -> 163 bytes
.../udf-serialization/sql-types-classes-4.1.3.bin | Bin 0 -> 1117 bytes
.../udf-serialization/sql-types-classes-4.2.0.bin | Bin 0 -> 438 bytes
.../sql/connect/common/UdfSerializationSuite.scala | 284 +++++++++++++++++++++
.../common/UdfSerializationTestClasses.scala | 82 ++++++
10 files changed, 644 insertions(+), 8 deletions(-)
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfPacket.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfPacket.scala
index e46bc134b73a..7fa46a514842 100644
--- a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfPacket.scala
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfPacket.scala
@@ -16,7 +16,7 @@
*/
package org.apache.spark.sql.connect.common
-import java.io.{InputStream, ObjectInputStream, ObjectOutputStream, OutputStream}
+import java.io.{InputStream, ObjectOutputStream, OutputStream}
import com.google.protobuf.ByteString
@@ -56,9 +56,10 @@ case class UdfPacket(
}
object UdfPacket {
+ // Uses UdfSerialization, which accepts a sql.types serialVersionUID change only for the exact
+ // (class, stream SUID, local SUID) transitions it audits; any other mismatch still fails.
def apply(in: InputStream): UdfPacket = {
- val ois = new ObjectInputStream(in)
- ois.readObject().asInstanceOf[UdfPacket]
+ UdfSerialization.deserialize[UdfPacket](in)
}
def apply(bytes: ByteString): UdfPacket = {
diff --git a/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala
new file mode 100644
index 000000000000..2ab1005a6363
--- /dev/null
+++ b/sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala
@@ -0,0 +1,267 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.common
+
+import java.io.{ByteArrayInputStream, Externalizable, InputStream, ObjectInputStream}
+import java.io.{ObjectOutputStream, ObjectStreamClass}
+import java.lang.reflect.Modifier
+
+/**
+ * Java deserialization for Scala UDF payloads that tolerates audited `serialVersionUID` changes
+ * of `org.apache.spark.sql.types` classes between Spark versions.
+ *
+ * A Scala UDF payload embeds `org.apache.spark.sql.types` classes (the UDF's input/output
+ * schema). Many of them, e.g. the `DataType` singletons, `StringType` and `Decimal`, do not
+ * declare a `serialVersionUID`, so the JVM computes it from the whole class shape, including
+ * members that are irrelevant to serialization such as the `$anonfun$` methods Scala emits for
+ * lambdas. Such a class can change its `serialVersionUID` between releases without changing its
+ * serialized form, and a plain [[ObjectInputStream]] then rejects a payload produced by another
+ * Spark version with an `InvalidClassException`.
+ *
+ * A stream descriptor flags custom `writeObject` data, but its numeric SUID does not reveal
+ * whether the producer declared or computed that SUID. Tolerance is therefore limited to the
+ * exact transitions in [[auditedTransitions]]. A class descriptor is rebound to the local class
+ * only when its (class, stream SUID, local SUID) triple is audited, the local class uses default
+ * field serialization with a computed SUID, and the complete persistent field layout of the
+ * stream and local descriptors is identical. Anything else keeps the stream descriptor, so the
+ * standard `serialVersionUID` check applies.
+ */
+private[spark] object UdfSerialization {
+
+ /** An audited `serialVersionUID` change of one class from a producer build to this build. */
+ private[connect] case class SuidTransition(className: String, streamSuid: Long, localSuid: Long)
+
+ /**
+ * `serialVersionUID` changes from released Spark versions to this build, audited against the
+ * published `spark-sql-api_2.13` jars of 4.0.0 to 4.0.4, 4.1.0 to 4.1.3 and 4.2.0. For every
+ * entry, the released and the local class both use default field serialization (no
+ * `readObject`/`writeObject`, not `Externalizable`), neither declares a `serialVersionUID`, and
+ * their `ObjectStreamClass.getFields` are identical. `localSuid` must match this build, which
+ * `UdfSerializationSuite` enforces: any change to these classes' computed `serialVersionUID`
+ * requires re-auditing the affected entries.
+ */
+ private[connect] val auditedTransitions: Set[SuidTransition] = Set(
+ SuidTransition(
+ "org.apache.spark.sql.types.BinaryType$",
+ 4981368476672807243L,
+ 8371054193783994026L),
+ SuidTransition(
+ "org.apache.spark.sql.types.BooleanType$",
+ 5066630839557347008L,
+ 7944921333908111375L),
+ SuidTransition(
+ "org.apache.spark.sql.types.ByteType$",
+ -2622274212739253403L,
+ -758215751304624994L),
+ SuidTransition(
+ "org.apache.spark.sql.types.CalendarIntervalType$",
+ -2877907547147508957L,
+ 3785141433624333846L),
+ SuidTransition(
+ "org.apache.spark.sql.types.CharType$",
+ 2241144916566246639L,
+ 7142653180585234499L),
+ SuidTransition(
+ "org.apache.spark.sql.types.DateType$",
+ -5061141662885574084L,
+ 4147040835737091283L),
+ SuidTransition(
+ "org.apache.spark.sql.types.Decimal",
+ 1715621871942419369L,
+ 6398599065815978361L),
+ SuidTransition(
+ "org.apache.spark.sql.types.Decimal$",
+ 4103410110050351305L,
+ 622493820172246208L),
+ SuidTransition(
+ "org.apache.spark.sql.types.DoubleType$",
+ 8550059415794444422L,
+ 8798493769972832591L),
+ SuidTransition(
+ "org.apache.spark.sql.types.FloatType$",
+ -6368152288801752152L,
+ 6520180645217752115L),
+ SuidTransition(
+ "org.apache.spark.sql.types.IntegerType$",
+ 5392303096310563711L,
+ 7538148419353647761L),
+ SuidTransition(
+ "org.apache.spark.sql.types.LongType$",
+ 5556619015085473997L,
+ -1069125617143205720L),
+ SuidTransition(
+ "org.apache.spark.sql.types.NullType$",
+ 4636924136831884463L,
+ 5274842102097366055L),
+ SuidTransition(
+ "org.apache.spark.sql.types.ShortType$",
+ 4196314124164507241L,
+ -1640260345880352154L),
+ SuidTransition(
+ "org.apache.spark.sql.types.StringHelper$",
+ -2950363081197529498L,
+ 5684776218413467458L),
+ SuidTransition(
+ "org.apache.spark.sql.types.StringType",
+ 2313622123462496566L,
+ -7943010735412336143L),
+ SuidTransition(
+ "org.apache.spark.sql.types.StringType$",
+ -6884398962421839287L,
+ -3522724403804736810L),
+ SuidTransition(
+ "org.apache.spark.sql.types.StructType$",
+ 4125835014175492466L,
+ -2680503192028675703L),
+ SuidTransition(
+ "org.apache.spark.sql.types.TimestampNTZType$",
+ -1053634932597855477L,
+ 6615591828519817938L),
+ SuidTransition(
+ "org.apache.spark.sql.types.TimestampType$",
+ -6671223943884622784L,
+ 3549865156948637452L),
+ SuidTransition(
+ "org.apache.spark.sql.types.UDTRegistration$",
+ -4424248461314799595L,
+ -3417430582631602433L),
+ SuidTransition(
+ "org.apache.spark.sql.types.UserDefinedType",
+ 941169123764985454L,
+ 7081765465869405043L),
+ SuidTransition(
+ "org.apache.spark.sql.types.VarcharType$",
+ 1412037499997274523L,
+ 1465122347601003917L),
+ SuidTransition(
+ "org.apache.spark.sql.types.VariantType$",
+ 3207924491175853678L,
+ -8985635318007546424L))
+
+ private val suidTolerantPackagePrefix = "org.apache.spark.sql.types."
+
+ /** Deserialize `bytes` resolving classes with `loader`, tolerating audited SUID changes. */
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T =
+ deserialize(bytes, loader, auditedTransitions)
+
+ /** Deserialize from `in` with default class resolution, tolerating audited SUID changes. */
+ def deserialize[T](in: InputStream): T = deserialize(in, auditedTransitions)
+
+ private[connect] def deserialize[T](
+ bytes: Array[Byte],
+ loader: ClassLoader,
+ transitions: Set[SuidTransition]): T = {
+ val ois =
+ new SuidTolerantObjectInputStream(new ByteArrayInputStream(bytes), loader, transitions)
+ try ois.readObject().asInstanceOf[T]
+ finally ois.close()
+ }
+
+ private[connect] def deserialize[T](in: InputStream, transitions: Set[SuidTransition]): T = {
+ new SuidTolerantObjectInputStream(in, null, transitions).readObject().asInstanceOf[T]
+ }
+
+ /**
+ * The complete serialized field layout of a descriptor: the set of persistent field name + JVM
+ * type signature. Every persistent slot is included, in particular the Scala lazy-val init
+ * `bitmap$*` slots: rebinding to a local descriptor whose slot shape differs would misalign the
+ * stream, so any slot difference must block rebinding.
+ */
+ private[connect] def fieldSignature(desc: ObjectStreamClass): Set[String] = {
+ // getTypeString is null for primitives, where the single-char type code is the signature.
+ desc.getFields
+ .map(f => s"${f.getName}:${Option(f.getTypeString).getOrElse(f.getTypeCode.toString)}")
+ .toSet
+ }
+
+ /**
+ * Whether the local class can consume a rebound descriptor: it must use default field-based
+ * serialization (no custom `readObject`/`writeObject` and not `Externalizable`, whose protocol
+ * `ObjectInputStream` takes from the descriptor) and must not declare a `serialVersionUID`.
+ */
+ private[connect] def isRebindSafe(clazz: Class[_]): Boolean = {
+ !classOf[Externalizable].isAssignableFrom(clazz) &&
+ !hasExplicitSerialVersionUID(clazz) &&
+ !hasCustomSerialization(clazz)
+ }
+
+ private def hasExplicitSerialVersionUID(clazz: Class[_]): Boolean = {
+ try {
+ val field = clazz.getDeclaredField("serialVersionUID")
+ Modifier.isStatic(field.getModifiers) && field.getType == java.lang.Long.TYPE
+ } catch {
+ case _: NoSuchFieldException => false
+ }
+ }
+
+ private def hasCustomSerialization(clazz: Class[_]): Boolean = {
+ def declares(name: String, paramType: Class[_]): Boolean = {
+ try {
+ clazz.getDeclaredMethod(name, paramType)
+ true
+ } catch {
+ case _: NoSuchMethodException => false
+ }
+ }
+ declares("readObject", classOf[ObjectInputStream]) ||
+ declares("writeObject", classOf[ObjectOutputStream])
+ }
+
+ private class SuidTolerantObjectInputStream(
+ in: InputStream,
+ loader: ClassLoader,
+ transitions: Set[SuidTransition])
+ extends ObjectInputStream(in) {
+
+ override def resolveClass(desc: ObjectStreamClass): Class[_] = {
+ if (loader != null) {
+ // scalastyle:off classforname
+ Class.forName(desc.getName, false, loader)
+ // scalastyle:on classforname
+ } else {
+ super.resolveClass(desc)
+ }
+ }
+
+ override def readClassDescriptor(): ObjectStreamClass = {
+ val streamDesc = super.readClassDescriptor()
+ val name = streamDesc.getName
+ if (!name.startsWith(suidTolerantPackagePrefix)) {
+ return streamDesc
+ }
+ val localClass =
+ try {
+ resolveClass(streamDesc)
+ } catch {
+ case _: ClassNotFoundException => return streamDesc
+ }
+ val localDesc = ObjectStreamClass.lookup(localClass)
+ if (localDesc != null &&
+ transitions.contains(
+ SuidTransition(name, streamDesc.getSerialVersionUID, localDesc.getSerialVersionUID)) &&
+ isRebindSafe(localClass) &&
+ fieldSignature(streamDesc) == fieldSignature(localDesc)) {
+ localDesc
+ } else {
+ // Not an audited transition (including an unchanged SUID), or a local class or layout
+ // that cannot safely consume the stream: keep the stream descriptor so the standard
+ // serialVersionUID check applies.
+ streamDesc
+ }
+ }
+ }
+}
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
index 0d11148a2441..faa70c099c2c 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
+++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/SparkConnectPlanner.scala
@@ -59,7 +59,7 @@ import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils}
import org.apache.spark.sql.classic.{Catalog, DataFrameWriter, Dataset, MergeIntoWriter, RelationalGroupedDataset, SparkSession, TypedAggUtils, UserDefinedFunctionUtils}
import org.apache.spark.sql.classic.ClassicConversions._
import org.apache.spark.sql.connect.client.arrow.ArrowSerializer
-import org.apache.spark.sql.connect.common.{DataTypeProtoConverter, ForeachWriterPacket, LiteralValueProtoConverter, StorageLevelProtoConverter, StreamingListenerPacket, UdfPacket}
+import org.apache.spark.sql.connect.common.{DataTypeProtoConverter, ForeachWriterPacket, LiteralValueProtoConverter, StorageLevelProtoConverter, StreamingListenerPacket, UdfPacket, UdfSerialization}
import org.apache.spark.sql.connect.config.Connect.CONNECT_GRPC_ARROW_MAX_BATCH_SIZE
import org.apache.spark.sql.connect.ml.MLHandler
import org.apache.spark.sql.connect.pipelines.PipelinesHandler
@@ -2151,10 +2151,11 @@ class SparkConnectPlanner(
unpackScalaUDF[ForeachWriterPacket](fun)
}
- private def unpackScalaUDF[T](fun: proto.ScalarScalaUDF): T = {
+ private[connect] def unpackScalaUDF[T](fun: proto.ScalarScalaUDF): T = {
try {
logDebug(s"Unpack using class loader: ${Utils.getContextOrSparkClassLoader}")
- Utils.deserialize[T](fun.getPayload.toByteArray, Utils.getContextOrSparkClassLoader)
+ UdfSerialization
+ .deserialize[T](fun.getPayload.toByteArray, Utils.getContextOrSparkClassLoader)
} catch {
case t: Throwable =>
Utils.getRootCause(t) match {
diff --git a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelper.scala b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelper.scala
index 425a583e10c5..29086b182c4f 100644
--- a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelper.scala
+++ b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/planner/StreamingForeachBatchHelper.scala
@@ -31,7 +31,7 @@ import org.apache.spark.internal.LogKeys.{DATAFRAME_ID, PYTHON_EXEC, QUERY_ID, R
import org.apache.spark.sql.{DataFrame, Dataset}
import org.apache.spark.sql.catalyst.encoders.{AgnosticEncoder, AgnosticEncoders}
import org.apache.spark.sql.connect.IllegalStateErrors
-import org.apache.spark.sql.connect.common.ForeachWriterPacket
+import org.apache.spark.sql.connect.common.{ForeachWriterPacket, UdfSerialization}
import org.apache.spark.sql.connect.config.Connect
import org.apache.spark.sql.connect.service.SessionHolder
import org.apache.spark.sql.connect.service.SparkConnectService
@@ -101,7 +101,8 @@ object StreamingForeachBatchHelper extends Logging {
payloadBytes: Array[Byte],
sessionHolder: SessionHolder): ForeachBatchFnType = {
val foreachBatchPkt =
- Utils.deserialize[ForeachWriterPacket](payloadBytes, Utils.getContextOrSparkClassLoader)
+ UdfSerialization
+ .deserialize[ForeachWriterPacket](payloadBytes, Utils.getContextOrSparkClassLoader)
val fn = foreachBatchPkt.foreachWriter.asInstanceOf[(Dataset[Any], Long) => Unit]
val encoder = foreachBatchPkt.datasetEncoder.asInstanceOf[AgnosticEncoder[Any]]
// TODO(SPARK-44462): Set up Spark Connect session.
diff --git a/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.0.0.bin b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.0.0.bin
new file mode 100644
index 000000000000..a8ab6b620665
Binary files /dev/null and b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.0.0.bin differ
diff --git a/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.0.4.bin b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.0.4.bin
new file mode 100644
index 000000000000..783f808b0420
Binary files /dev/null and b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.0.4.bin differ
diff --git a/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.1.3.bin b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.1.3.bin
new file mode 100644
index 000000000000..7abbb39f68aa
Binary files /dev/null and b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.1.3.bin differ
diff --git a/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.2.0.bin b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.2.0.bin
new file mode 100644
index 000000000000..29c8176f8315
Binary files /dev/null and b/sql/connect/server/src/test/resources/udf-serialization/sql-types-classes-4.2.0.bin differ
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationSuite.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationSuite.scala
new file mode 100644
index 000000000000..335c296d0990
--- /dev/null
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationSuite.scala
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.connect.common
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InvalidClassException}
+import java.io.{ObjectInputStream, ObjectOutputStream, ObjectStreamClass, ObjectStreamConstants}
+import java.nio.ByteBuffer
+import java.nio.charset.StandardCharsets
+
+import scala.collection.mutable
+
+import com.google.protobuf.ByteString
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.connect.proto
+import org.apache.spark.sql.Dataset
+import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.PrimitiveIntEncoder
+import org.apache.spark.sql.connect.SparkConnectTestUtils
+import org.apache.spark.sql.connect.common.UdfSerialization.SuidTransition
+import org.apache.spark.sql.connect.planner.{SparkConnectPlanner, StreamingForeachBatchHelper}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{SuidCompatV1, SuidCompatV2, SuidCustomV1, SuidCustomV2}
+import org.apache.spark.sql.types.{SuidExplicitV1, SuidExplicitV2, SuidExternalV1, SuidExternalV2}
+import org.apache.spark.sql.types.{SuidLayoutV1, SuidLayoutV2}
+import org.apache.spark.sql.types.{SuidProducerCustomV1, SuidProducerCustomV2}
+import org.apache.spark.sql.types.{SuidProducerExplicitV1, SuidProducerExplicitV2}
+import org.apache.spark.sql.types.{SuidWriteObjectV1, SuidWriteObjectV2}
+import org.apache.spark.util.Utils
+
+/**
+ * Tests for [[UdfSerialization]]'s tolerance of audited `serialVersionUID` changes of
+ * `org.apache.spark.sql.types` classes.
+ *
+ * The fixture pairs have same-length class names, so serializing a `*V1` instance and patching
+ * the class name in the stream to `*V2` yields what a `*V2` consumer sees from a `*V1` producer
+ * with a different `serialVersionUID`, without needing two builds.
+ */
+class UdfSerializationSuite extends SparkFunSuite with SharedSparkSession {
+
+ private val loader = getClass.getClassLoader
+
+ private def serialize(o: AnyRef): Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ val oos = new ObjectOutputStream(bos)
+ oos.writeObject(o)
+ oos.close()
+ bos.toByteArray
+ }
+
+ private def plainDeserialize(bytes: Array[Byte]): AnyRef =
+ new ObjectInputStream(new ByteArrayInputStream(bytes)).readObject()
+
+ private def suid(cls: Class[_]): Long = ObjectStreamClass.lookup(cls).getSerialVersionUID
+
+ /** Serialize `producer` and rename its class in the stream to `consumer`. */
+ private def drifted(producer: AnyRef, consumer: Class[_]): Array[Byte] = {
+ val from = producer.getClass.getName.getBytes(StandardCharsets.UTF_8)
+ val to = consumer.getName.getBytes(StandardCharsets.UTF_8)
+ assert(from.length == to.length, "class names must have the same length")
+ val bytes = serialize(producer)
+ val idx = bytes.indexOfSlice(from)
+ assert(idx >= 0, s"${producer.getClass.getName} not found in the stream")
+ val patched = bytes.clone()
+ System.arraycopy(to, 0, patched, idx, to.length)
+ patched
+ }
+
+ /** The transition that lets a `consumer` reader accept a stream produced by `producer`. */
+ private def transition(producer: Class[_], consumer: Class[_]): SuidTransition =
+ SuidTransition(consumer.getName, suid(producer), suid(consumer))
+
+ test("accepts an audited serialVersionUID change (bytes overload)") {
+ val stream = drifted(SuidCompatV1(7, "hi"), classOf[SuidCompatV2])
+ // A plain ObjectInputStream rejects the SUID mismatch: this is the failure being fixed.
+ intercept[InvalidClassException](plainDeserialize(stream))
+ val audited = Set(transition(classOf[SuidCompatV1], classOf[SuidCompatV2]))
+ assert(
+ UdfSerialization
+ .deserialize[SuidCompatV2](stream, loader, audited) === SuidCompatV2(7, "hi"))
+ }
+
+ test("accepts an audited serialVersionUID change (InputStream overload)") {
+ val stream = drifted(SuidCompatV1(9, "yo"), classOf[SuidCompatV2])
+ val audited = Set(transition(classOf[SuidCompatV1], classOf[SuidCompatV2]))
+ val result =
+ UdfSerialization.deserialize[SuidCompatV2](new ByteArrayInputStream(stream), audited)
+ assert(result === SuidCompatV2(9, "yo"))
+ }
+
+ test("rejects a serialVersionUID change that is not audited") {
+ val stream = drifted(SuidCompatV1(1, "x"), classOf[SuidCompatV2])
+ intercept[InvalidClassException](
+ UdfSerialization.deserialize[AnyRef](stream, loader, Set.empty[SuidTransition]))
+ intercept[InvalidClassException](UdfSerialization.deserialize[AnyRef](stream, loader))
+ }
+
+ test("rejects an audited change when the field layout differs") {
+ val stream = drifted(SuidLayoutV1(1, "x"), classOf[SuidLayoutV2])
+ val audited = Set(transition(classOf[SuidLayoutV1], classOf[SuidLayoutV2]))
+ intercept[InvalidClassException](
+ UdfSerialization.deserialize[AnyRef](stream, loader, audited))
+ }
+
+ test("rejects an audited change when the consumer declares a serialVersionUID") {
+ val stream = drifted(SuidExplicitV1(1, "x"), classOf[SuidExplicitV2])
+ val audited = Set(transition(classOf[SuidExplicitV1], classOf[SuidExplicitV2]))
+ intercept[InvalidClassException](
+ UdfSerialization.deserialize[AnyRef](stream, loader, audited))
+ }
+
+ test("rejects an audited change when the consumer has a custom readObject") {
+ val stream = drifted(SuidCustomV1(1, "x"), classOf[SuidCustomV2])
+ val audited = Set(transition(classOf[SuidCustomV1], classOf[SuidCustomV2]))
+ intercept[InvalidClassException](
+ UdfSerialization.deserialize[AnyRef](stream, loader, audited))
+ }
+
+ test("rejects an audited change when the consumer has a custom writeObject") {
+ val stream = drifted(SuidWriteObjectV1(1, "x"), classOf[SuidWriteObjectV2])
+ val audited = Set(transition(classOf[SuidWriteObjectV1], classOf[SuidWriteObjectV2]))
+ intercept[InvalidClassException](
+ UdfSerialization.deserialize[AnyRef](stream, loader, audited))
+ }
+
+ test("rejects an audited change when the consumer is Externalizable") {
+ val stream = drifted(new SuidExternalV1(1), classOf[SuidExternalV2])
+ val audited = Set(transition(classOf[SuidExternalV1], classOf[SuidExternalV2]))
+ intercept[InvalidClassException](
+ UdfSerialization.deserialize[AnyRef](stream, loader, audited))
+ }
+
+ test("rejects an unaudited change from a producer with a custom writeObject") {
+ val stream = drifted(SuidProducerCustomV1(1, "x"), classOf[SuidProducerCustomV2])
+ intercept[InvalidClassException](UdfSerialization.deserialize[AnyRef](stream, loader))
+ }
+
+ test("rejects an unaudited change from a producer that declares a serialVersionUID") {
+ val stream = drifted(SuidProducerExplicitV1(1, "x"), classOf[SuidProducerExplicitV2])
+ intercept[InvalidClassException](UdfSerialization.deserialize[AnyRef](stream, loader))
+ }
+
+ test("rejects an audited change outside sql.types") {
+ val stream = drifted(SuidUntolerantV1(1, "x"), classOf[SuidUntolerantV2])
+ val audited = Set(transition(classOf[SuidUntolerantV1], classOf[SuidUntolerantV2]))
+ intercept[InvalidClassException](
+ UdfSerialization.deserialize[AnyRef](stream, loader, audited))
+ }
+
+ test("audited transitions target this build") {
+ assert(UdfSerialization.auditedTransitions.nonEmpty)
+ UdfSerialization.auditedTransitions.foreach { t =>
+ assert(t.className.startsWith("org.apache.spark.sql.types."), t)
+ assert(t.streamSuid != t.localSuid, t)
+ val cls = Utils.classForName(t.className)
+ assert(
+ suid(cls) == t.localSuid,
+ s"The serialVersionUID of ${t.className} is now ${suid(cls)}, not ${t.localSuid}. " +
+ "Re-audit its entries in UdfSerialization.auditedTransitions against the released " +
+ "versions before updating localSuid.")
+ assert(UdfSerialization.isRebindSafe(cls), t)
+ }
+ }
+
+ // Streams written by released builds: each resource serializes the Class objects of audited
+ // classes as loaded from the published spark-sql-api_2.13 jar of that release, so it carries
+ // that release's class descriptors (serialVersionUID, flags, fields and superclasses).
+ private val releasedDescriptorResources =
+ Seq("4.0.0", "4.0.4", "4.1.3", "4.2.0").map(v =>
+ s"udf-serialization/sql-types-classes-$v.bin")
+
+ private def resourceBytes(name: String): Array[Byte] = {
+ val in = getClass.getClassLoader.getResourceAsStream(name)
+ assert(in != null, s"missing test resource $name")
+ try in.readAllBytes()
+ finally in.close()
+ }
+
+ /**
+ * The (class name, serialVersionUID) of every class descriptor in a stream of Class objects.
+ */
+ private def descriptorsIn(bytes: Array[Byte]): Set[(String, Long)] = {
+ val seen = mutable.Set[(String, Long)]()
+ val ois = new ObjectInputStream(new ByteArrayInputStream(bytes)) {
+ // A stream of Class objects has no instance data, so substituting the local descriptor
+ // cannot misread it; this only walks the stream to collect the producer descriptors.
+ override def readClassDescriptor(): ObjectStreamClass = {
+ val desc = super.readClassDescriptor()
+ seen += ((desc.getName, desc.getSerialVersionUID))
+ Option(ObjectStreamClass.lookup(resolveClass(desc))).getOrElse(desc)
+ }
+ }
+ try ois.readObject()
+ finally ois.close()
+ seen.toSet
+ }
+
+ test("audited transitions are backed by released producer descriptors") {
+ val released = releasedDescriptorResources.map(name => name -> resourceBytes(name))
+ val producerDescriptors = released.flatMap { case (_, bytes) => descriptorsIn(bytes) }.toSet
+ UdfSerialization.auditedTransitions.foreach { t =>
+ assert(
+ producerDescriptors.contains((t.className, t.streamSuid)),
+ s"no released descriptor backs $t")
+ }
+ released.foreach { case (name, bytes) =>
+ intercept[InvalidClassException](plainDeserialize(bytes))
+ val classes = UdfSerialization.deserialize[Array[AnyRef]](bytes, loader)
+ assert(classes.nonEmpty && classes.forall(_.isInstanceOf[Class[_]]), name)
+ }
+ }
+
+ // An audited production transition of a Scala object. The object is serialized through its
+ // ModuleSerializationProxy, whose Class field writes the object's class descriptor, and hence
+ // its serialVersionUID, into the stream.
+ private lazy val objectTransition: SuidTransition =
+ UdfSerialization.auditedTransitions.toSeq
+ .sortBy(_.className)
+ .find(_.className.endsWith("$"))
+ .getOrElse(fail("expected an audited transition of a Scala object"))
+
+ private def objectOf(t: SuidTransition): AnyRef =
+ Utils.classForName(t.className).getField("MODULE$").get(null)
+
+ /** Rewrite `t.className`'s serialVersionUID in `bytes` from `t.localSuid` to `t.streamSuid`. */
+ private def asProducedBy(bytes: Array[Byte], t: SuidTransition): Array[Byte] = {
+ val name = t.className.getBytes(StandardCharsets.UTF_8)
+ val header =
+ Array(ObjectStreamConstants.TC_CLASSDESC, (name.length >> 8).toByte, name.length.toByte) ++
+ name
+ val idx = bytes.indexOfSlice(header)
+ assert(idx >= 0, s"no class descriptor for ${t.className} in the stream")
+ val patched = ByteBuffer.wrap(bytes.clone())
+ val suidOffset = idx + header.length
+ assert(patched.getLong(suidOffset) == t.localSuid)
+ patched.putLong(suidOffset, t.streamSuid)
+ patched.array()
+ }
+
+ /** A UdfPacket referencing `t`'s object, as produced by a build with `t.streamSuid`. */
+ private def driftedUdfPacket(t: SuidTransition): (AnyRef, Array[Byte]) = {
+ val obj = objectOf(t)
+ val bytes = UdfPacket(obj, Seq.empty, PrimitiveIntEncoder).toByteString.toByteArray
+ (obj, asProducedBy(bytes, t))
+ }
+
+ test("UdfPacket.apply accepts an audited serialVersionUID change") {
+ val (obj, bytes) = driftedUdfPacket(objectTransition)
+ intercept[InvalidClassException](plainDeserialize(bytes))
+ assert(UdfPacket(ByteString.copyFrom(bytes)).function eq obj)
+ }
+
+ test("SparkConnectPlanner.unpackScalaUDF accepts an audited serialVersionUID change") {
+ val (obj, bytes) = driftedUdfPacket(objectTransition)
+ val planner =
+ new SparkConnectPlanner(SparkConnectTestUtils.createDummySessionHolder(spark))
+ val udf = proto.ScalarScalaUDF.newBuilder().setPayload(ByteString.copyFrom(bytes)).build()
+ assert(planner.unpackScalaUDF[UdfPacket](udf).function eq obj)
+ }
+
+ test("StreamingForeachBatchHelper.scalaForeachBatchWrapper accepts an audited change") {
+ val obj = objectOf(objectTransition)
+ val fn: (Dataset[Any], Long) => Unit = (_, _) => obj.hashCode()
+ val bytes =
+ asProducedBy(serialize(ForeachWriterPacket(fn, PrimitiveIntEncoder)), objectTransition)
+ intercept[InvalidClassException](plainDeserialize(bytes))
+ StreamingForeachBatchHelper.scalaForeachBatchWrapper(
+ bytes,
+ SparkConnectTestUtils.createDummySessionHolder(spark))
+ }
+}
diff --git a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationTestClasses.scala b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationTestClasses.scala
new file mode 100644
index 000000000000..8c8b1b81cfa2
--- /dev/null
+++ b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationTestClasses.scala
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Deterministic fixtures for UdfSerializationSuite. Each pair has same-length class names, so a
+// serialized stream of the *V1 class can be turned into a *V2 stream by an in-place class-name
+// patch, letting a single build simulate a producer (*V1) and a consumer (*V2) whose
+// serialVersionUIDs differ.
+
+package org.apache.spark.sql.types {
+ // Default field serialization, computed SUID, identical layout.
+ case class SuidCompatV1(a: Int, b: String)
+ case class SuidCompatV2(a: Int, b: String)
+
+ // Different persistent field layout.
+ case class SuidLayoutV1(a: Int, b: String)
+ case class SuidLayoutV2(a: Int)
+
+ // The consumer declares a serialVersionUID.
+ @SerialVersionUID(4242L) case class SuidExplicitV1(a: Int, b: String)
+ @SerialVersionUID(4243L) case class SuidExplicitV2(a: Int, b: String)
+
+ // The consumer has a custom readObject.
+ case class SuidCustomV1(a: Int, b: String) {
+ private def readObject(in: java.io.ObjectInputStream): Unit = in.defaultReadObject()
+ }
+ case class SuidCustomV2(a: Int, b: String) {
+ private def readObject(in: java.io.ObjectInputStream): Unit = in.defaultReadObject()
+ }
+
+ // The consumer has a custom writeObject.
+ case class SuidWriteObjectV1(a: Int, b: String) {
+ private def writeObject(out: java.io.ObjectOutputStream): Unit = out.defaultWriteObject()
+ }
+ case class SuidWriteObjectV2(a: Int, b: String) {
+ private def writeObject(out: java.io.ObjectOutputStream): Unit = out.defaultWriteObject()
+ }
+
+ // The consumer is Externalizable.
+ class SuidExternalV1(var a: Int) extends java.io.Externalizable {
+ def this() = this(0)
+ override def writeExternal(out: java.io.ObjectOutput): Unit = out.writeInt(a)
+ override def readExternal(in: java.io.ObjectInput): Unit = a = in.readInt()
+ }
+ class SuidExternalV2(var a: Int) extends java.io.Externalizable {
+ def this() = this(0)
+ override def writeExternal(out: java.io.ObjectOutput): Unit = out.writeInt(a)
+ override def readExternal(in: java.io.ObjectInput): Unit = a = in.readInt()
+ }
+
+ // Only the producer writes custom class data; the consumer uses default serialization.
+ case class SuidProducerCustomV1(a: Int, b: String) {
+ private def writeObject(out: java.io.ObjectOutputStream): Unit = {
+ out.defaultWriteObject()
+ out.writeInt(42)
+ }
+ }
+ case class SuidProducerCustomV2(a: Int, b: String)
+
+ // Only the producer declares a serialVersionUID; the consumer's is computed.
+ @SerialVersionUID(4244L) case class SuidProducerExplicitV1(a: Int, b: String)
+ case class SuidProducerExplicitV2(a: Int, b: String)
+}
+
+package org.apache.spark.sql.connect.common {
+ // Outside the tolerant `sql.types` package.
+ case class SuidUntolerantV1(a: Int, b: String)
+ case class SuidUntolerantV2(a: Int, b: String)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org