You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "Yicong-Huang (via GitHub)" <gi...@apache.org> on 2026/09/22 04:57:23 UTC
[PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Yicong-Huang opened a new pull request, #58962:
URL: https://github.com/apache/spark/pull/58962
### What changes were proposed in this pull request?
This PR makes Spark Connect Scala UDF payload deserialization tolerant of `serialVersionUID` drift for `org.apache.spark.sql.types` classes.
- Adds `UdfSerialization`, an `ObjectInputStream`-based reader that, for classes under `org.apache.spark.sql.types`, rebinds a stream class descriptor to the local class when their `serialVersionUID`s differ but their **serialized field layouts are identical** (the complete persistent field set, including Scala lazy-val `bitmap$*` slots). Any field-layout difference is left untouched so the standard `serialVersionUID` check still fails fast. Classes outside `org.apache.spark.sql.types` keep the standard check unchanged.
- Routes `UdfPacket.apply(InputStream)` through it.
### Why are the changes needed?
`org.apache.spark.sql.types` classes (which make up a UDF's input/output schema) carry no explicit `@SerialVersionUID`, so the JVM auto-computes it from the whole class shape. That computation folds in compiler-synthesized members that are irrelevant to serialization -- most notably the `$anonfun$` public static methods Scala emits for lambdas. A source change that only reshapes a lambda (for example, rewriting a helper to use `existsRecursively { ... }`) changes the auto-computed `serialVersionUID` without changing any serialized field. A plain `ObjectInputStream` then rejects a UDF payload produced by a different Spark version with `java.io.InvalidClassException: ... local class incompatible`, even though the payload is field-compatible. This breaks the ability to run a Scala UDF serialized by one version against another.
### Does this PR introduce _any_ user-facing change?
Yes. A Scala UDF payload whose `org.apache.spark.sql.types` classes differ from the executing side only in `serialVersionUID` (identical serialized fields) now deserializes successfully instead of failing with `InvalidClassException`. Behavior for a genuinely incompatible serialized field layout is unchanged: it still fails fast.
### How was this patch tested?
New `UdfSerializationSuite`. Using same-length sibling fixture classes plus an in-place class-name patch of the serialized stream (a deterministic stand-in for a cross-version producer/consumer pair), it covers:
- SUID-only drift of a `sql.types` class deserializes and recovers the value, through both the `(bytes, loader)` and `(InputStream)` overloads (and a plain `ObjectInputStream` is asserted to reject the same stream).
- a genuine serialized field-layout change fails fast with `InvalidClassException`.
- SUID drift outside `org.apache.spark.sql.types` is not tolerated (fails fast).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Isaac (Claude Code)
This pull request and its description were written by Isaac.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on PR #58962:
URL: https://github.com/apache/spark/pull/58962#issuecomment-5819685910
Included the Scala `foreachBatch` route: `StreamingForeachBatchHelper.scalaForeachBatchWrapper` now deserializes through `UdfSerialization` as well, with an entry-point test that fails if it goes back to `Utils.deserialize`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #58962:
URL: https://github.com/apache/spark/pull/58962#discussion_r4096978184
##########
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(
Review Comment:
Added released producer descriptors as test resources. For 4.0.0, 4.0.4, 4.1.3 and 4.2.0, each resource is a stream of the `Class` objects of the audited classes as loaded from that release's published `spark-sql-api_2.13` jar, so it carries that release's descriptors (SUID, flags, fields, superclasses). The suite checks that every entry's `(className, streamSuid)` appears in these streams, that a plain `ObjectInputStream` rejects them, and that `UdfSerialization` reads them. Flipping one digit of an entry's `streamSuid` now fails the suite.
##########
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) &&
Review Comment:
Added audited-transition cases with a consumer that has a custom `writeObject` and with an `Externalizable` consumer; both are rejected. Removing either `isRebindSafe` branch now fails the suite.
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfPacket.scala:
##########
@@ -56,9 +56,10 @@ case class UdfPacket(
}
object UdfPacket {
+ // Uses the SUID-tolerant reader so a UdfPacket stays deserializable across Spark versions even
Review Comment:
Reworded: it now says only the audited `(class, stream SUID, local SUID)` transitions are accepted and any other mismatch still fails.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on PR #58962:
URL: https://github.com/apache/spark/pull/58962#issuecomment-5876471464
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/a648b6b3467f1910e07b586896a59b7a4106d0c1
- merged into branch-4.x https://github.com/apache/spark/commit/45a4562462fe4af54e451e1dc9be5ce6d57d16ab
- merged into branch-4.3 https://github.com/apache/spark/commit/c33a3b87d990ffee03b84e388babddeac9861e6d
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58962:
URL: https://github.com/apache/spark/pull/58962#discussion_r4095510838
##########
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(
Review Comment:
**Non-blocking (P2):** The table promises compatibility with released producer descriptors, but the suite checks only each current `localSuid` and deserializes one locally generated descriptor after patching its SUID. A mistyped `streamSuid` or a released field-layout mismatch for the other entries would leave the advertised client transition broken without failing this suite.
##########
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) &&
Review Comment:
**Nit (P3):** The negative fixtures do not exercise either a local `writeObject` consumer or an `Externalizable` consumer under an otherwise audited transition. Removing either corresponding `isRebindSafe` guard leaves the suite green, so these two fail-closed protocol branches lack a regression signal.
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfPacket.scala:
##########
@@ -56,9 +56,10 @@ case class UdfPacket(
}
object UdfPacket {
+ // Uses the SUID-tolerant reader so a UdfPacket stays deserializable across Spark versions even
Review Comment:
**Nit (P3):** This sentence omits the essential allowlist restriction: an auto-computed SUID drift is tolerated only when the exact class/from-SUID/to-SUID triple is in `auditedTransitions`; every unlisted drift still fails. Please qualify the local promise as applying to audited transitions.
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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, InputStream, ObjectInputStream, ObjectStreamClass}
+
+/**
+ * Java deserialization for Scala UDF payloads that tolerates the `serialVersionUID` drift 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).
+ * Those classes carry no explicit `@SerialVersionUID`, so the JVM auto-computes it from the whole
+ * class shape, which folds in compiler-synthesized members that are irrelevant to serialization --
+ * most notably the `$anonfun$` public static methods Scala emits for lambdas. A source change that
+ * only reshapes a lambda (e.g. rewriting a helper to use `existsRecursively { ... }`) changes the
+ * auto-computed `serialVersionUID` without changing any serialized field, which makes a plain
+ * [[ObjectInputStream]] reject a payload produced by a different Spark version with an
+ * `InvalidClassException`, even though the payload is field-compatible.
+ *
+ * The property that actually governs compatibility is the serialized field layout. When a
+ * `sql.types` class arrives with a mismatched SUID but an identical serialized field layout, this
+ * reader rebinds the stream descriptor to the local class; any field-layout difference is left
+ * untouched so the standard SUID check still fails fast rather than misreading the stream.
+ *
+ * Only `org.apache.spark.sql.types` descriptors are treated tolerantly; every other class keeps
+ * the standard `serialVersionUID` compatibility check.
+ */
+private[spark] object UdfSerialization {
+
+ private val suidTolerantPackagePrefix = "org.apache.spark.sql.types."
+
+ /** Deserialize `bytes` resolving classes with `loader`, tolerating `sql.types` SUID drift. */
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ val ois = new SuidTolerantObjectInputStream(new ByteArrayInputStream(bytes), loader)
+ try ois.readObject().asInstanceOf[T]
+ finally ois.close()
+ }
+
+ /** Deserialize from `in` with default class resolution, tolerating `sql.types` SUID drift. */
+ def deserialize[T](in: InputStream): T = {
+ new SuidTolerantObjectInputStream(in, null).readObject().asInstanceOf[T]
+ }
+
+ /**
+ * The complete serialized field layout of a descriptor: the set of persistent field name + JVM
+ * type signature. This is what governs whether [[ObjectInputStream]] can consume the producer's
+ * class-data block through the local descriptor. 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 a bitmap difference must also 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
+ }
+
+ private class SuidTolerantObjectInputStream(in: InputStream, loader: ClassLoader)
+ 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()
+ if (!streamDesc.getName.startsWith(suidTolerantPackagePrefix)) {
+ return streamDesc
+ }
+ val localClass =
+ try {
+ resolveClass(streamDesc)
+ } catch {
+ case _: ClassNotFoundException => return streamDesc
+ }
+ val localDesc = ObjectStreamClass.lookup(localClass)
+ if (localDesc == null ||
+ localDesc.getSerialVersionUID == streamDesc.getSerialVersionUID ||
+ fieldSignature(streamDesc) != fieldSignature(localDesc)) {
Review Comment:
Thanks, the exact transition table closes the asymmetric producer-side hole. One smaller gap remains in the negative coverage: please add audited local-consumer cases for both `writeObject` and `Externalizable`, since neither `isRebindSafe` branch is exercised by the current fixtures.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4068944306","thread_id":"inline:4068944306","verdict_sha256":"89077c4c5d2a0b8c2766e5a3c4664ab51a3fcd80a2dd2cb2b022a5fdbcfa56b6"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58962:
URL: https://github.com/apache/spark/pull/58962#discussion_r4068944306
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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, InputStream, ObjectInputStream, ObjectStreamClass}
+
+/**
+ * Java deserialization for Scala UDF payloads that tolerates the `serialVersionUID` drift 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).
+ * Those classes carry no explicit `@SerialVersionUID`, so the JVM auto-computes it from the whole
+ * class shape, which folds in compiler-synthesized members that are irrelevant to serialization --
+ * most notably the `$anonfun$` public static methods Scala emits for lambdas. A source change that
+ * only reshapes a lambda (e.g. rewriting a helper to use `existsRecursively { ... }`) changes the
+ * auto-computed `serialVersionUID` without changing any serialized field, which makes a plain
+ * [[ObjectInputStream]] reject a payload produced by a different Spark version with an
+ * `InvalidClassException`, even though the payload is field-compatible.
+ *
+ * The property that actually governs compatibility is the serialized field layout. When a
+ * `sql.types` class arrives with a mismatched SUID but an identical serialized field layout, this
+ * reader rebinds the stream descriptor to the local class; any field-layout difference is left
+ * untouched so the standard SUID check still fails fast rather than misreading the stream.
+ *
+ * Only `org.apache.spark.sql.types` descriptors are treated tolerantly; every other class keeps
+ * the standard `serialVersionUID` compatibility check.
+ */
+private[spark] object UdfSerialization {
+
+ private val suidTolerantPackagePrefix = "org.apache.spark.sql.types."
+
+ /** Deserialize `bytes` resolving classes with `loader`, tolerating `sql.types` SUID drift. */
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ val ois = new SuidTolerantObjectInputStream(new ByteArrayInputStream(bytes), loader)
+ try ois.readObject().asInstanceOf[T]
+ finally ois.close()
+ }
+
+ /** Deserialize from `in` with default class resolution, tolerating `sql.types` SUID drift. */
+ def deserialize[T](in: InputStream): T = {
+ new SuidTolerantObjectInputStream(in, null).readObject().asInstanceOf[T]
+ }
+
+ /**
+ * The complete serialized field layout of a descriptor: the set of persistent field name + JVM
+ * type signature. This is what governs whether [[ObjectInputStream]] can consume the producer's
+ * class-data block through the local descriptor. 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 a bitmap difference must also 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
+ }
+
+ private class SuidTolerantObjectInputStream(in: InputStream, loader: ClassLoader)
+ 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()
+ if (!streamDesc.getName.startsWith(suidTolerantPackagePrefix)) {
+ return streamDesc
+ }
+ val localClass =
+ try {
+ resolveClass(streamDesc)
+ } catch {
+ case _: ClassNotFoundException => return streamDesc
+ }
+ val localDesc = ObjectStreamClass.lookup(localClass)
+ if (localDesc == null ||
+ localDesc.getSerialVersionUID == streamDesc.getSerialVersionUID ||
+ fieldSignature(streamDesc) != fieldSignature(localDesc)) {
Review Comment:
**Non-blocking (P2):** Persistent-field equality is not sufficient to replace the entire stream descriptor. `Metadata` is already a `sql.types` class with an explicit SUID, and `ObjectInputStream.readSerialData` also uses descriptor state such as `hasReadObjectMethod` and `hasWriteObjectData`. Returning `localDesc` here can therefore suppress a deliberate compatibility break or consume producer data under the local class's protocol, leading to a later stream error or incorrectly populated encoder state. This path should remain fail-closed unless the complete descriptor transition is known to be safe.
**Recommended change:** Replace package-wide field-only eligibility with an exact audited class/from-SUID/to-SUID transition policy, retain the field-layout check as an additional guard, and add negative compatibility coverage for explicit-SUID classes and custom class-data protocols.
**Why this works:** Key eligibility by class name plus the exact stream and local SUID pair for transitions whose producer and consumer are known to use the same default field-only Java serialization protocol. Require the existing persistent-field equality check as defense in depth. For any unlisted pair, including explicitly versioned production classes such as Metadata and classes with custom serialization hooks, return streamDesc so ObjectInputStream performs its standard compatibility rejection.
**Scope:** Constrain whole-descriptor substitution to transitions whose SUID provenance and default serialization protocol have been audited, with regression-sensitive positive and negative fixtures.
**Compatibility:** Classes outside sql.types, genuine field-layout mismatches, and authorized auto-computed-SUID drift retain their current helper outcomes.
**Risks:** An incomplete transition set may reject a harmless drift until that exact pair is audited. A transition entry that omits either endpoint SUID could accidentally authorize a later incompatible version.
**Constraints:** Do not depend on unsupported reflective access to private JDK descriptor flags. Keep standard ObjectInputStream rejection for every transition not affirmatively authorized.
**Success:** A production sql.types class with an explicit SUID is not tolerated merely because its package and persistent fields match. A custom writeObject/readObject protocol difference cannot be hidden by local descriptor substitution. Each audited auto-SUID-only transition with an identical default field layout still deserializes successfully. Every unknown or incompatible transition retains streamDesc and uses standard ObjectInputStream checks.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #58962:
URL: https://github.com/apache/spark/pull/58962#discussion_r4081256376
##########
sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationSuite.scala:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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}
+
+import org.apache.spark.sql.connect.test.ConnectFunSuite
+import org.apache.spark.sql.types.{SuidCompatV1, SuidCompatV2, SuidCustomV1, SuidExplicitV1}
+import org.apache.spark.sql.types.{SuidLayoutV1, SuidLayoutV2}
+
+/**
+ * Tests for [[UdfSerialization]]'s tolerance of `serialVersionUID` drift for
+ * `org.apache.spark.sql.types` classes.
+ *
+ * Each fixture pair has same-length class names, so serializing the `*V1` instance and patching
+ * the class name in the byte stream to `*V2` yields exactly what a `*V2` reader would see from a
+ * `*V1` producer whose `serialVersionUID` differs -- a deterministic stand-in for a cross-version
+ * payload without needing two builds.
+ */
+class UdfSerializationSuite extends ConnectFunSuite {
+
+ private def serialize(o: AnyRef): Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ val oos = new ObjectOutputStream(bos)
+ oos.writeObject(o)
+ oos.close()
+ bos.toByteArray
+ }
+
+ /** Replace the ASCII `from` class name with the same-length `to` in a serialized stream. */
+ private def patchClassName(bytes: Array[Byte], from: String, to: String): Array[Byte] = {
+ assert(from.length == to.length, "class names must be the same length for an in-place patch")
+ val fromBytes = from.getBytes("US-ASCII")
+ val toBytes = to.getBytes("US-ASCII")
+ val idx = bytes.indexOfSlice(fromBytes)
+ assert(idx >= 0, s"$from not found in the serialized stream")
+ val patched = bytes.clone()
+ System.arraycopy(toBytes, 0, patched, idx, toBytes.length)
+ patched
+ }
+
+ private val loader = getClass.getClassLoader
+
+ test("tolerates serialVersionUID drift for a sql.types class (bytes overload)") {
Review Comment:
**Nit (P3):** Every new drift assertion calls `UdfSerialization` directly, so reverting either `UdfPacket.apply` or `unpackScalaUDF` to its previous deserializer would leave this suite green. Please exercise a deterministic drifted payload through each production entrypoint so removal of either delegation fails a test.
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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, InputStream, ObjectInputStream, ObjectStreamClass}
+
+/**
+ * Java deserialization for Scala UDF payloads that tolerates the `serialVersionUID` drift 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).
+ * Those classes carry no explicit `@SerialVersionUID`, so the JVM auto-computes it from the whole
+ * class shape, which folds in compiler-synthesized members that are irrelevant to serialization --
+ * most notably the `$anonfun$` public static methods Scala emits for lambdas. A source change that
+ * only reshapes a lambda (e.g. rewriting a helper to use `existsRecursively { ... }`) changes the
+ * auto-computed `serialVersionUID` without changing any serialized field, which makes a plain
+ * [[ObjectInputStream]] reject a payload produced by a different Spark version with an
+ * `InvalidClassException`, even though the payload is field-compatible.
+ *
+ * The property that actually governs compatibility is the serialized field layout. When a
+ * `sql.types` class arrives with a mismatched SUID but an identical serialized field layout, this
+ * reader rebinds the stream descriptor to the local class; any field-layout difference is left
+ * untouched so the standard SUID check still fails fast rather than misreading the stream.
+ *
+ * Only `org.apache.spark.sql.types` descriptors are treated tolerantly; every other class keeps
+ * the standard `serialVersionUID` compatibility check.
+ */
+private[spark] object UdfSerialization {
+
+ private val suidTolerantPackagePrefix = "org.apache.spark.sql.types."
+
+ /** Deserialize `bytes` resolving classes with `loader`, tolerating `sql.types` SUID drift. */
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ val ois = new SuidTolerantObjectInputStream(new ByteArrayInputStream(bytes), loader)
+ try ois.readObject().asInstanceOf[T]
+ finally ois.close()
+ }
+
+ /** Deserialize from `in` with default class resolution, tolerating `sql.types` SUID drift. */
+ def deserialize[T](in: InputStream): T = {
+ new SuidTolerantObjectInputStream(in, null).readObject().asInstanceOf[T]
+ }
+
+ /**
+ * The complete serialized field layout of a descriptor: the set of persistent field name + JVM
+ * type signature. This is what governs whether [[ObjectInputStream]] can consume the producer's
+ * class-data block through the local descriptor. 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 a bitmap difference must also 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
+ }
+
+ private class SuidTolerantObjectInputStream(in: InputStream, loader: ClassLoader)
+ 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()
+ if (!streamDesc.getName.startsWith(suidTolerantPackagePrefix)) {
+ return streamDesc
+ }
+ val localClass =
+ try {
+ resolveClass(streamDesc)
+ } catch {
+ case _: ClassNotFoundException => return streamDesc
+ }
+ val localDesc = ObjectStreamClass.lookup(localClass)
+ if (localDesc == null ||
+ localDesc.getSerialVersionUID == streamDesc.getSerialVersionUID ||
+ fieldSignature(streamDesc) != fieldSignature(localDesc)) {
Review Comment:
Thanks for tightening the local guard. There is still an asymmetric hole: if the producer had custom writeObject or an explicit SUID and the local version no longer does, isRebindSafe(localClass) passes and returning localDesc discards the producer protocol/provenance. Please switch to exact audited class/from-SUID/to-SUID transitions (or another mechanism that proves both endpoints) and add asymmetric negative fixtures; otherwise these cases should retain streamDesc and fail normally.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4068944306","thread_id":"inline:4068944306","verdict_sha256":"2ae3a2ae3e55fd3d436c676ebc57241a76489670d5b49df7d1ba348e0eed52a7"} -->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang closed pull request #58962: [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads
URL: https://github.com/apache/spark/pull/58962
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #58962:
URL: https://github.com/apache/spark/pull/58962#discussion_r4078096252
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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, InputStream, ObjectInputStream, ObjectStreamClass}
+
+/**
+ * Java deserialization for Scala UDF payloads that tolerates the `serialVersionUID` drift 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).
+ * Those classes carry no explicit `@SerialVersionUID`, so the JVM auto-computes it from the whole
+ * class shape, which folds in compiler-synthesized members that are irrelevant to serialization --
+ * most notably the `$anonfun$` public static methods Scala emits for lambdas. A source change that
+ * only reshapes a lambda (e.g. rewriting a helper to use `existsRecursively { ... }`) changes the
+ * auto-computed `serialVersionUID` without changing any serialized field, which makes a plain
+ * [[ObjectInputStream]] reject a payload produced by a different Spark version with an
+ * `InvalidClassException`, even though the payload is field-compatible.
+ *
+ * The property that actually governs compatibility is the serialized field layout. When a
+ * `sql.types` class arrives with a mismatched SUID but an identical serialized field layout, this
+ * reader rebinds the stream descriptor to the local class; any field-layout difference is left
+ * untouched so the standard SUID check still fails fast rather than misreading the stream.
+ *
+ * Only `org.apache.spark.sql.types` descriptors are treated tolerantly; every other class keeps
+ * the standard `serialVersionUID` compatibility check.
+ */
+private[spark] object UdfSerialization {
+
+ private val suidTolerantPackagePrefix = "org.apache.spark.sql.types."
+
+ /** Deserialize `bytes` resolving classes with `loader`, tolerating `sql.types` SUID drift. */
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ val ois = new SuidTolerantObjectInputStream(new ByteArrayInputStream(bytes), loader)
+ try ois.readObject().asInstanceOf[T]
+ finally ois.close()
+ }
+
+ /** Deserialize from `in` with default class resolution, tolerating `sql.types` SUID drift. */
+ def deserialize[T](in: InputStream): T = {
+ new SuidTolerantObjectInputStream(in, null).readObject().asInstanceOf[T]
+ }
+
+ /**
+ * The complete serialized field layout of a descriptor: the set of persistent field name + JVM
+ * type signature. This is what governs whether [[ObjectInputStream]] can consume the producer's
+ * class-data block through the local descriptor. 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 a bitmap difference must also 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
+ }
+
+ private class SuidTolerantObjectInputStream(in: InputStream, loader: ClassLoader)
+ 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()
+ if (!streamDesc.getName.startsWith(suidTolerantPackagePrefix)) {
+ return streamDesc
+ }
+ val localClass =
+ try {
+ resolveClass(streamDesc)
+ } catch {
+ case _: ClassNotFoundException => return streamDesc
+ }
+ val localDesc = ObjectStreamClass.lookup(localClass)
+ if (localDesc == null ||
+ localDesc.getSerialVersionUID == streamDesc.getSerialVersionUID ||
+ fieldSignature(streamDesc) != fieldSignature(localDesc)) {
Review Comment:
Good point, tightened in ddbbf69. Rebinding is now gated on an `isRebindSafe` check in addition to complete field-layout equality: the local class must
- declare **no explicit `serialVersionUID`** -- so a deliberate explicit-SUID change (e.g. `Metadata`) is never silently tolerated, and only auto-computed SUID drift is bridged; and
- define **no custom `readObject`/`writeObject`** -- so a class whose `readSerialData` protocol keys off the descriptor is left fail-closed.
The complete persistent field-layout equality (including `bitmap$*`) is kept as defense in depth. Added negative tests covering an explicit-SUID class and a custom-`readObject` class (both fail fast).
I kept a package-scoped predicate ("auto-SUID, default-field-serialized `sql.types` class with identical layout") rather than an exact from/to-SUID transition allowlist, since that predicate captures exactly the safe set without a hand-maintained SUID table. Happy to switch to an explicit transition list if you'd prefer.
Also wired `SparkConnectPlanner.unpackScalaUDF` through `UdfSerialization` so the production ScalarScalaUDF path is actually covered (previously it went through `Utils.deserialize`); updated the PR description accordingly.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59708][CONNECT] Tolerate serialVersionUID drift of sql.types classes when deserializing Scala UDF payloads [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #58962:
URL: https://github.com/apache/spark/pull/58962#discussion_r4086705231
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/UdfSerialization.scala:
##########
@@ -0,0 +1,108 @@
+/*
+ * 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, InputStream, ObjectInputStream, ObjectStreamClass}
+
+/**
+ * Java deserialization for Scala UDF payloads that tolerates the `serialVersionUID` drift 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).
+ * Those classes carry no explicit `@SerialVersionUID`, so the JVM auto-computes it from the whole
+ * class shape, which folds in compiler-synthesized members that are irrelevant to serialization --
+ * most notably the `$anonfun$` public static methods Scala emits for lambdas. A source change that
+ * only reshapes a lambda (e.g. rewriting a helper to use `existsRecursively { ... }`) changes the
+ * auto-computed `serialVersionUID` without changing any serialized field, which makes a plain
+ * [[ObjectInputStream]] reject a payload produced by a different Spark version with an
+ * `InvalidClassException`, even though the payload is field-compatible.
+ *
+ * The property that actually governs compatibility is the serialized field layout. When a
+ * `sql.types` class arrives with a mismatched SUID but an identical serialized field layout, this
+ * reader rebinds the stream descriptor to the local class; any field-layout difference is left
+ * untouched so the standard SUID check still fails fast rather than misreading the stream.
+ *
+ * Only `org.apache.spark.sql.types` descriptors are treated tolerantly; every other class keeps
+ * the standard `serialVersionUID` compatibility check.
+ */
+private[spark] object UdfSerialization {
+
+ private val suidTolerantPackagePrefix = "org.apache.spark.sql.types."
+
+ /** Deserialize `bytes` resolving classes with `loader`, tolerating `sql.types` SUID drift. */
+ def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
+ val ois = new SuidTolerantObjectInputStream(new ByteArrayInputStream(bytes), loader)
+ try ois.readObject().asInstanceOf[T]
+ finally ois.close()
+ }
+
+ /** Deserialize from `in` with default class resolution, tolerating `sql.types` SUID drift. */
+ def deserialize[T](in: InputStream): T = {
+ new SuidTolerantObjectInputStream(in, null).readObject().asInstanceOf[T]
+ }
+
+ /**
+ * The complete serialized field layout of a descriptor: the set of persistent field name + JVM
+ * type signature. This is what governs whether [[ObjectInputStream]] can consume the producer's
+ * class-data block through the local descriptor. 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 a bitmap difference must also 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
+ }
+
+ private class SuidTolerantObjectInputStream(in: InputStream, loader: ClassLoader)
+ 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()
+ if (!streamDesc.getName.startsWith(suidTolerantPackagePrefix)) {
+ return streamDesc
+ }
+ val localClass =
+ try {
+ resolveClass(streamDesc)
+ } catch {
+ case _: ClassNotFoundException => return streamDesc
+ }
+ val localDesc = ObjectStreamClass.lookup(localClass)
+ if (localDesc == null ||
+ localDesc.getSerialVersionUID == streamDesc.getSerialVersionUID ||
+ fieldSignature(streamDesc) != fieldSignature(localDesc)) {
Review Comment:
Agreed, a local-only predicate cannot prove the producer side. Switched to an exact audited table of `(className, streamSuid, localSuid)` transitions. Rebinding now requires the triple to be audited, and additionally that the local class passes `isRebindSafe` (now also excluding `Externalizable`) and that the complete persistent field layout matches. Any unlisted pair, including an unchanged SUID, keeps `streamDesc`.
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 against this build: 24 classes whose computed SUID changed while both sides use default serialization, declare no `serialVersionUID`, and have identical `ObjectStreamClass.getFields`. A test checks that every entry's `localSuid` matches this build, so any later change to these classes forces a re-audit. `StructField`, `StructType`, `CharType` and `VarcharType` are not listed because their persistent layout differs from 4.x.
Added the asymmetric fixtures: a producer with a custom `writeObject` and a producer with an explicit `serialVersionUID`, each read by a default-serialized consumer, are both rejected. The audited cases are also rejected when the layout differs, the consumer declares a SUID or custom `readObject`, or the class is outside `sql.types`.
##########
sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/common/UdfSerializationSuite.scala:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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}
+
+import org.apache.spark.sql.connect.test.ConnectFunSuite
+import org.apache.spark.sql.types.{SuidCompatV1, SuidCompatV2, SuidCustomV1, SuidExplicitV1}
+import org.apache.spark.sql.types.{SuidLayoutV1, SuidLayoutV2}
+
+/**
+ * Tests for [[UdfSerialization]]'s tolerance of `serialVersionUID` drift for
+ * `org.apache.spark.sql.types` classes.
+ *
+ * Each fixture pair has same-length class names, so serializing the `*V1` instance and patching
+ * the class name in the byte stream to `*V2` yields exactly what a `*V2` reader would see from a
+ * `*V1` producer whose `serialVersionUID` differs -- a deterministic stand-in for a cross-version
+ * payload without needing two builds.
+ */
+class UdfSerializationSuite extends ConnectFunSuite {
+
+ private def serialize(o: AnyRef): Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ val oos = new ObjectOutputStream(bos)
+ oos.writeObject(o)
+ oos.close()
+ bos.toByteArray
+ }
+
+ /** Replace the ASCII `from` class name with the same-length `to` in a serialized stream. */
+ private def patchClassName(bytes: Array[Byte], from: String, to: String): Array[Byte] = {
+ assert(from.length == to.length, "class names must be the same length for an in-place patch")
+ val fromBytes = from.getBytes("US-ASCII")
+ val toBytes = to.getBytes("US-ASCII")
+ val idx = bytes.indexOfSlice(fromBytes)
+ assert(idx >= 0, s"$from not found in the serialized stream")
+ val patched = bytes.clone()
+ System.arraycopy(toBytes, 0, patched, idx, toBytes.length)
+ patched
+ }
+
+ private val loader = getClass.getClassLoader
+
+ test("tolerates serialVersionUID drift for a sql.types class (bytes overload)") {
Review Comment:
Added `UdfPacket.apply` and `SparkConnectPlanner.unpackScalaUDF` cases. Each builds a `UdfPacket` that references a production-audited object, rewrites that object's SUID in the stream to the audited release value, and checks that a plain `ObjectInputStream` rejects the stream while the entry point accepts it, so reverting either delegation fails. `unpackScalaUDF` is now `private[connect]` for this. The suite moved to the server module so it can reach the planner.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org