[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!