andygrove opened a new issue, #5281:
URL: https://github.com/apache/datafusion-comet/issues/5281
### Describe the bug
When a `ScalaUDF` (or Java UDF) is dispatched through the codegen dispatcher
(`spark.comet.exec.scalaUDF.codegen.enabled=true`) and the class that captured
the UDF closure comes from a user jar (`--jars` / `spark.jars`), the query
fails at execution time:
```
java.lang.ClassCastException: cannot assign instance of
java.lang.invoke.SerializedLambda to field
org.apache.spark.sql.catalyst.expressions.ScalaUDF.f of type
scala.Function1
in instance of org.apache.spark.sql.catalyst.expressions.ScalaUDF
at
java.base/java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2096)
at
java.base/java.io.ObjectStreamClass$FieldReflector.checkObjectFieldValueTypes(ObjectStreamClass.java:2060)
at
java.base/java.io.ObjectStreamClass.checkObjFieldValueTypes(ObjectStreamClass.java:1347)
at
java.base/java.io.ObjectInputStream$FieldValues.defaultCheckFieldValues(ObjectInputStream.java:2679)
...
at
org.apache.comet.udf.codegen.CometScalaUDFCodegen.lookupOrCompile(CometScalaUDFCodegen.scala:160)
at
org.apache.comet.udf.codegen.CometScalaUDFCodegen.evaluate(CometScalaUDFCodegen.scala:123)
at
org.apache.comet.udf.CometUdfBridge.evaluateInternal(CometUdfBridge.java:203)
at org.apache.comet.udf.CometUdfBridge.evaluate(CometUdfBridge.java:122)
```
### Root cause
The `ClassCastException` is a **masked `ClassNotFoundException`**. When the
classloader used for deserialization cannot resolve a lambda's *capturing
class*, `ObjectInputStream` records the CNFE against the object handle,
therefore skips `SerializedLambda.readResolve()`, and the raw
`SerializedLambda` then fails the field-type check in
`defaultCheckFieldValues`. The CNFE never surfaces.
The loader is wrong because of where the deserialization runs.
`CometScalaUDFCodegen.lookupOrCompile` deserializes the bound expression with
the calling thread's context classloader:
```scala
val loader = Option(Thread.currentThread().getContextClassLoader)
.getOrElse(classOf[Expression].getClassLoader)
```
`CometUdfBridge.evaluate` is invoked from a Tokio worker thread, which
attaches to the JVM through JNI. **An attached thread has no context
classloader**, so this falls back to `classOf[Expression].getClassLoader` — the
loader that loaded spark-catalyst and Comet, which never contains user jars.
Spark's executor classloader (`MutableURLClassLoader`, which does contain them)
is only installed on Spark task threads.
Measured, by returning thread identity from inside a UDF body:
| stage leaf | thread running the UDF | context classloader |
| --- | --- | --- |
| `CometNativeScan` | `Thread-28..32` (Tokio workers) | **null** |
| shuffle read / join build side | `Executor task launch worker …` |
`MutableURLClassLoader` |
| native scan disabled | `Executor task launch worker …` |
`MutableURLClassLoader` |
So any plan whose stage leaf is a native scan (`CometNativeScan`,
`CometIcebergNativeScan`) evaluates UDFs on a Tokio worker and hits this.
`CometUdfBridge` already propagates `TaskContext` across the JNI boundary for
exactly this reason, but the classloader is not propagated.
This is invisible to Comet's own test suites because they register UDFs from
test classes that are already on the application classpath.
`CometUdfBridge.java:180` has the same latent problem: a user-supplied
`CometUDF` implementation shipped in a user jar would fail the same way at
`Class.forName`.
### Steps to reproduce
Requires two conditions together:
1. The class capturing the UDF lambda is reachable only from Spark's
executor classloader (a `--jars` submission), not from Comet's own classloader.
2. The UDF is evaluated on a Tokio worker, i.e. the stage's leaf is a native
scan.
A self-contained failing test compiles a class holding a serializable
`scala.Function1` lambda into a jar at test time and wires it in via
`spark.executor.extraClassPath` (the local-mode equivalent of `--jars`:
`LocalSchedulerBackend` feeds it into the executor's `MutableURLClassLoader`),
then runs `SELECT hiddenUdf(s) FROM t`. It reproduces the trace above exactly.
The same query with the native scan disabled passes, which isolates the failure
to classloader propagation rather than class availability.
Note that `sparkContext.addJar` is *not* sufficient to set this up in local
mode — the jar does not reach the local executor's classloader, and the job
then fails earlier, in Spark's own task deserialization.
### Expected behavior
A UDF whose closure class lives in a user jar should execute regardless of
which thread the native side dispatches from. The Spark task thread's context
classloader should be captured at plan-creation time and propagated to the
Tokio worker, alongside the `TaskContext` that is already propagated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
