[SPARK-44388] Using an updated instance of ScalarUserDefinedFunction causes protobuf cast failures on server

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.5.0
• Fix Version/s: 3.5.0
• Component/s: Connect
• Epic Link: Spark Connect Scala Client - Basic
• Assignee: Venkata Sai Akhil Gudesa
• Reporter: Venkata Sai Akhil Gudesa
• Created: 12/Jul/23 15:13
• Updated: 13/Jul/23 09:55
• Resolved: 13/Jul/23 09:55

Description

When running the following code-

class A(x: Int) { def get = x * 7 }
val myUdf = udf((x: Int) => new A(x).get)
val modifiedUdf = myUdf.withName("myUdf")
spark.range(5).select(modifiedUdf(col("id"))).as[Int].collect()

which modifies the original myUdf instance through the `withName` method causes the following error to occur during execution:

java.lang.ClassCastException: org.apache.spark.connect.proto.ScalarScalaUDF cannot be cast to com.google.protobuf.MessageLite
    at com.google.protobuf.GeneratedMessageLite$SerializedForm.readResolve(GeneratedMessageLite.java:1462)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeReadResolve(ObjectStreamClass.java:1274)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2196)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2285)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
    at java.io.ObjectInputStream.readArray(ObjectInputStream.java:2093)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1655)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461)
    at org.apache.spark.util.Utils$.deserialize(Utils.scala:148)

Issue Links

• links to: [Github] Pull Request #41959 (vicennial)