[AURON #2386] Resolve deserialized expression classes with an explicit class loader

[xiaoyanxie]

Which issue does this PR close?

Closes #2386

Rationale for this change

This PR addresses the issue #2386, which reports a subquery serialization failure when the runtime bloom filter optimizer is enabled.

Root Cause Analysis:

Spark's InjectRuntimeFilter optimization rewrites eligible join filters into a
BloomFilterMightContain expression whose bloom-filter input is an execution-side

ScalarSubquery.
Auron converts the bloom-filter expression natively, including its ExecSubqueryExpression child. In NativeConverters.convertExprWithFallback,
Auron calls prepareExecSubquery() to materialize the subquery result, but then
serializes the entire ScalarSubquery object:

case subquery: ExecSubqueryExpression =>
  prepareExecSubquery(subquery)
  val serialized = serializeExpression(
    subquery.asInstanceOf[Expression with Serializable],
    StructType(Nil))

Although the result has already been materialized, the ScalarSubquery still retains its physical plan. Java serialization therefore traverses the plan and its RDD lineage, including MapPartitionsRDD.dependencies_. This over-capture is what places RDD objects in the payload at all, and it is a real defect in its own right.

The over-capture alone, however, does not raise the exception. The failure comes from how that payload is read back. The serialized expression is later evaluated by SparkScalarSubqueryWrapperExpr, which delegates to the JVM expression wrapper. The JVM deserializes it with a plain ObjectInputStream in NativeConverters.deserializeExpression(). The default ObjectInputStream.resolveClass resolves every class through VM.latestUserDefinedLoader(), which selects a class loader from the live call stack rather than from the thread context class loader. During a nested read the most recent user-defined frame is frequently a Spark or Scala class, whose loader cannot see Auron's classes when Auron is supplied through spark.jars and is therefore defined by Spark's MutableURLClassLoader.

The expression graph consequently resolves only partially. An un-readResolved scala.collection.generic.DefaultSerializationProxy is then assigned into the RDD.dependencies_: scala.collection.immutable.Seq field, producing:

java.lang.ClassCastException: cannot assign instance of
scala.collection.generic.DefaultSerializationProxy to field
org.apache.spark.rdd.RDD.dependencies_ of type scala.collection.immutable.Seq
in instance of org.apache.spark.rdd.MapPartitionsRDD

This explains two things that were previously puzzling. First, the failure depends on how Auron is deployed, not on the query alone: supplying the jar through --jars fails, while placing the same jar in $SPARK_HOME/jars (so it is defined by the application class loader, which can always see it) succeeds with byte-identical input. Second, it explains why the crash could never be reproduced in-process: in a SharedSparkSession test Auron sits on the application class loader, so the call-stack loader selection is harmless there.

DefaultSerializationProxy is specific to Scala 2.13 collections, so that is the shape the corruption takes on 2.13; the underlying class-resolution behaviour is not itself version-specific.

Therefore the runtime bloom-filter optimization is only the trigger, and the plan over-capture is only the precondition that puts an RDD graph in the payload. The exception itself is raised by call-stack-dependent class resolution during deserialization. This is also distinct from Auron's ordinary unsupported-expression fallback. Those expressions are converted into shallow bound expression trees and do not retain a physical plan or RDD lineage.

Supporting evidence

The defect reproduces in CI. The job added by this PR supplies Auron only through --jars (jar-on-system-classpath: false) and asserts no ClassCastException. On the commit preceding the fix, all ten TPC-DS shards fail with the exact exception above, raised through the JNI upcall at native-engine/datafusion-ext-exprs/src/spark_udf_wrapper.rs:97.

Class resolution is the operative cause, isolated from everything else. Replaying one captured payload in a single JVM, with identical bytes, identical thread context class loader, and only the resolveClass policy varying:

resolveClass policy	Result
pinned to the loader that defines Auron	succeeds
pinned to the application class loader	fails (cannot see Auron classes)
default (latestUserDefinedLoader())	fails with the reported ClassCastException

The default policy produces a third, distinct outcome that matches neither pinned policy — the signature of a loader that varies per call site.

Caching and object-graph shape are ruled out. A ten-order operation matrix run across fresh JVMs against two independently captured payloads is completely order-independent, which excludes ObjectStreamClass/field-reflector cache effects. The RDD graphs captured under both deployments are byte-identical (SHA-256 1178bc4a6af500b29e7c26503ff28d8cd538124bd497ecf1b54d6e846ee2d5f0), and cross-replay shows the outcome follows the reading environment rather than the payload writer.

What changes are included in this PR?

• An integration test in the GitHub CI that reproduces the bug
• A fix in NativeConverters.deserializeExpression: resolve classes against an explicit class loader (the context/Spark loader, falling back to the loader that defined Auron) instead of relying on VM.latestUserDefinedLoader(), so deserialization no longer depends on which frame happens to be on the call stack. Spark's own JavaDeserializationStream does the same; it is not reused directly because it is private[spark] and Auron builds against eight Spark versions. Note that Spark additionally overrides resolveProxyClass; that is deliberately not mirrored here, because the only non-deprecated way to obtain a proxy Class is Proxy.getProxyClass and the serialized expression graphs contain no dynamic proxies.
• Follow-up, not included here: stop serializing the physical plan for ScalarSubquery in convertExprWithFallback by serializing only its materialized Literal value. This removes the RDD over-capture and is worthwhile on correctness and payload-size grounds, but it is a separate change and is not required to fix this exception.

Are there any user-facing changes?

No

How was this patch tested?

Via integration tests.

Test matrix

Every row runs the TPC-DS harness with Auron supplied only through spark-submit --jars, so it is defined by Spark's MutableURLClassLoader. Environment is Spark 4.1 / Scala 2.13 / JDK 17. "Without the fix" always means a build of the commit preceding the fix on this branch, never a released Auron jar (see the note below).

#	Where	Data / bloom-filter threshold	Queries	Without the fix	With the fix
1	GitHub CI, job added by this PR	sf=1 / 1B	all 10 shards	10/10 shards fail with the ClassCastException	11/11 jobs pass, CCE assertion green in all 10 shards
2	Local, Spark 4.1.2	sf=1 / 1B	q1..q9	1515 ClassCastException	0
3	Local, Spark 4.1.3	sf=1 / 1B	q1..q9	850 ClassCastException	0
4	Local	sf=10 / 1GB	q2	42 ClassCastException	0
5	Local	sf=10 / 1B	q1..q9	not run	9/9 pass, 0

Row 1 is the primary regression guard. The CI job fails on the commit before the fix and passes on it, on the same runners and the same sf=1 dataset.

Rows with a pass compare query output against vanilla Spark through the harness, so they are correctness passes rather than merely the absence of an exception.

One trap worth flagging for anyone reproducing this. The released 8.0.0-incubating Auron jar does not reproduce the failure on the same dataset and configuration, so it is not a usable "before" baseline; an earlier revision of this description drew the wrong conclusion from it. Build the parent commit of the fix from this branch instead. The defect reproduces locally at sf=1 on an ordinary developer machine once the baseline is built correctly, and it does not depend on the Spark patch release, the dataset scale, or the number of cores.

Deterministic payload replay

The query runs above depend on scheduling and plan shape. To remove that variability, the two scalar-subquery payloads captured from a failing run were also replayed directly through NativeConverters.deserializeExpression, in a JVM where Auron is loaded by a MutableURLClassLoader exactly as spark-submit --jars arranges it.

Each payload was replayed twice, once with the thread context class loader set to the Auron loader and once with it set to the parent, giving four combinations. All four raise the ClassCastException without the fix and all four succeed with it. This is fully deterministic and needs no TPC-DS data, which is what made it usable for the isolation experiment in the Supporting evidence section above.

Exact local setup

# 1. Build the Auron jar for Spark 4.1 / Scala 2.13
./auron-build.sh --release --sparkver 4.1 --scalaver 2.13

# 2. Build the integration-test jar
#    (-Dscalafix.skip=true works around a pre-existing scalafix failure in this module)
cd dev/auron-it && ../../build/mvn -Pspark-4.1 -Pscala-2.13 -DskipTests -Dscalafix.skip=true package && cd ../..

# 3. TPC-DS data. This is the same sf=1 set CI uses, and it reproduces the
#    failure locally when the baseline is built from this branch.
git clone --depth 1 https://github.com/auron-project/tpcds_1g dev/tpcds_1g

# 4. Point at a Spark 4.1 distribution.
#    IMPORTANT: do NOT copy the Auron jar into $SPARK_HOME/jars. The defect only
#    appears when Auron is defined by MutableURLClassLoader rather than by the
#    application class loader.
export SPARK_HOME=/path/to/spark-4.1.2-bin-hadoop3

# 5. Run, mirroring the CI job's configuration
SPARK_VERSION=spark-4.1 SCALA_VERSION=2.13 \
AURON_SPARK_JAR=dev/mvn-build-helper/assembly/target/auron-spark-4.1_2.13-<version>.jar \
dev/auron-it/run-it.sh \
  --type tpcds \
  --data-location dev/tpcds_1g \
  --conf spark.sql.optimizer.runtime.bloomFilter.enabled=true \
  --conf spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1B \
  --conf spark.sql.autoBroadcastJoinThreshold=-1 \
  --query-filter q1,q2,q3,q4,q5,q6,q7,q8,q9

To observe the failure rather than the fix, build and run the same command against the commit preceding the fix on this branch. Do not substitute a released Auron jar for that baseline; as noted above, 8.0.0-incubating does not reproduce the failure.

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

If yes, include: Generated-by: <tool name and version>

Generated-by: Claude Code (Claude Opus 5)

ASF guidance: https://www.apache.org/legal/generative-tooling.html

[xiaoyanxie]

It took me a while to figure it out. Now I can successfully reproduce the bug in GitHub CI Test spark-4.1 JDK17 Scala-2.13 with bloomFilter optimizer enabled.

[xiaoyanxie]

Hi @ShreyeshArangath, could you please help to review this PR?

This PR fixes #2386. The problem is in NativeConverters.deserializeExpression. It used a plain ObjectInputStream. The default resolveClass resolves each class by VM.latestUserDefinedLoader(), which selects a class loader from the call stack. When Auron is provided by spark.jars, that loader cannot see the Auron classes. So the scalar subquery graph is only partly deserialized, and a ClassCastException is thrown.

The fix is to resolve the classes with an explicit class loader. Spark's own JavaDeserializationStream does the same thing.

This PR also adds a TPC-DS CI job for the runtime bloom filter. The job fails on the commit before the fix, and passes with the fix.

If you prefer, I can move the CI job into a separate PR.

Thank you!

[ShreyeshArangath]

Apologies for the delay, left a few comments here

[ShreyeshArangath]

Can you explain more of your rationale here for not overriding resolveProxyClass?  IIUC, the deserializeExpression is shared by UDF, UDAF, and UDTF expression graphs, which may contain serializable dynamic proxies...leaving proxy resolution on ObjectInputStream ’s default path reintroduces the stack-dependent class-loader behavior this change is intended to avoid, so proxy interfaces available only through Spark’s MutableURLClassLoader may still fail to deserialize when supplied through  --jars, right?

[xiaoyanxie]

Thanks for the review. You are right about the resolveProxyClass.

I checked Spark's JavaDeserializationStream. It overrides both resolveClass and resolveProxyClass, so the example I referenced actually supports your point. And "no dynamic proxies" is not something we can guarantee. deserializeExpression is called from SparkAuronUDFWrapperContext, SparkUDAFWrapperContext and SparkUDTFWrapperContext, which carry user payloads.

In the fix fa4047e I moved the pinned-loader -> Auron-loader -> default fallback into a helper method withLoaderFallback. Now both resolveClass and resolveProxyClass use it. Proxy.getProxyClass is deprecated since Java 9, and our build uses -Xfatal-warnings, so I also added a -Wconf suppression in pom.xml. This follows the same way we already handle Class.newInstance.

I also added a test for this. The test redefines the proxy interface in a child class loader, then checks which loader resolves the proxy after deserialization. When I run this test against the code before the fix, it fails:

- deserializeExpression resolves proxy interfaces with the pinned class loader *** FAILED ***
  jdk.internal.loader.ClassLoaders$AppClassLoader@5ffd2b27 was not the same instance as org.apache.auron.SingleClassRedefiningLoader@49353d43 proxy interface was resolved by a class loader taken from the call stack (jdk.internal.loader.ClassLoaders$AppClassLoader@5ffd2b27) instead of the pinned loader (NativeConvertersSuite.scala:124)

After the fix, the test passes.

One question I would like your opinion on. The fallback chain has one cost that the single-loader version did not have. If no loader can resolve a name, we now throw ClassNotFoundException on each level. I measured some cases, for example: a ScalaUDF with primitive type arguments will serialize ExpressionEncoder, and its ClassTag holds int.class, long.class and so on, with a payload that contains 8 primitive descriptors:

	µs/op
master (plain ObjectInputStream)	194
this PR	578
this PR + a 9-entry primitive lookup table	26

For normal payloads without primitives, both are about 11% faster than master, because pinning the loader avoids the stack walk in VM.latestUserDefinedLoader().

I already have the primitive lookup table change, but I did not include it here, because I want to keep this PR only for the correctness fix. Please let me know if you prefer to add it in this PR, or I can send a separate one.

[xiaoyanxie]

Correction to my comment above: I attributed the primitive descriptors to ScalaUDF's ExpressionEncoders, and that attribution is wrong.

I checked it properly this time. A ScalaUDF over (Int, Long, Double) taken from an analyzed plan carries three inputEncoders, and on Spark 4.1 they serialize as AgnosticEncoders$PrimitiveIntEncoder$, PrimitiveLongEncoder$ and PrimitiveDoubleEncoder$ — singleton module objects written through ModuleSerializationProxy. Deserializing that payload through a class loader that records every name requested gives 67 names, and not one primitive among them. scala.reflect.ClassTag.Int behaves the same way: it serializes as ManifestFactory$IntManifest, and runtimeClass is restored by readResolve rather than written to the stream.

So the 578µs number is real, but it was measured on a payload I built by hand with eight primitive descriptors in it, not on one that arises from this path. I have not found a real payload that carries them.

Unless you know of one, I would rather drop the primitive lookup table altogether than add an optimization I cannot justify. Sorry for the noise.

[ShreyeshArangath]

Could this run only the query or small query set known to reproduce the issue and match the specific deserialization failure signature? Running all 99 TPC-DS queries adds substantial CI cost, while grepping every ClassCastException  can attribute unrelated failures to expression deserialization. I think something more focused might be better here?

[xiaoyanxie]

Agree with both points.

The job did not set queries, so it used the default matrix and ran all queries. Since q1, q2 and q3 are sufficient to reproduce this problem issue (the plan contains scalar subqueries that trigger the bug), so I changed it to queries: '["q1,q2,q3"]' and now it uses only one runner.

I also made the pattern more specific. Before it matched any ClassCastException. Now it is ClassCastException.*DefaultSerializationProxy, so other unrelated failures will not be reported as expression deserialization problem.

The fix is in the commit e2dc0b3.

[ShreyeshArangath]

Could we add a positive check that the runtime bloom-filter ScalarSubquery was actually injected? Right now the job only checks that no ClassCastException occurred.

Also I think  --plan-check  is skipped for Spark 4.1, since PlanStabilityChecker currently supports only Spark 3.5. Without a positive assertion, the job could pass simply because the optimizer stopped producing the plan that triggers this path ..

[xiaoyanxie]

You are right about --plan-check. PlanStabilityChecker returns early for any version except spark-3.5. In my local run it just prints:

[PlanCheck] Unsupported Spark version: spark-4.1. Skipping.

So the job had no positive check at all. If the optimizer stops generating this plan, the job would still pass and we would not notice.

I added --print-plan to auron-it, and a new assert-log-matches input. For this job it is set to might_contain, which is the runtime bloom filter probe. In my local q1–q3 run it appears 10 times, and now the job will fail if it disappears.

This does not add extra cost. QueryRunner already builds the plan string every time, so the new flag only decides whether to print it.

See commit e2dc0b3.

[ShreyeshArangath]

I’m not sure this fully guarantees that Auron is loaded only through  --jars . The  auron-it  shaded JAR depends on the Auron uber JAR, so it looks like it may contain the same Auron classes itself. If that’s the case, removing the copy from  $SPARK_HOME/jars  may not reliably reproduce the original class-loader setup.

Would it make sense to exclude Auron from the integration-test fat JAR, or add a small runtime check that prints/asserts the actual class loader and code source?

[xiaoyanxie]

The auron-it jar does contain Auron classes. I checked locally, and Auron is actually loaded from that jar, not from --jars:

Auron Class Loader: org.apache.spark.util.MutableURLClassLoader
Auron Code Source: file:/.../auron-it-spark-4.1_2.13-...-jar-with-dependencies.jar
Auron On System Classpath: false

So my old comment was wrong. But the application jar is also loaded by MutableURLClassLoader, which is the real condition the reproduction needs, so it still reproduced the bug before the fix.

I tried your first suggestion but reverted it: the uber jar's dependency-reduced pom re-declares every Auron module, so excluding only the uber jar still pulls them in transitively, and org.apache.auron:* also removes our own integration test classes.

So I used your second suggestion. auron-it now prints the class loader and code source at startup, and the job asserts Auron On System Classpath: false. I verified it works: with the jar in $SPARK_HOME/jars it prints Auron On System Classpath: true and the step fails.