From 7f0eea2b02de9eb495623dbb5fd5ef9ca30f9da3 Mon Sep 17 00:00:00 2001 From: chloex <122268864+xiaoyanxie@users.noreply.github.com> Date: Sat, 18 Jul 2026 22:08:01 -0400 Subject: [PATCH 1/3] [AURON #2386] Add a TPC-DS CI job with the runtime bloom filter enabled Adds a spark-4.1 / JDK17 / Scala-2.13 TPC-DS job that turns on Spark's runtime bloom filter optimization: spark.sql.optimizer.runtime.bloomFilter.enabled=true spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1B spark.sql.autoBroadcastJoinThreshold=-1 The low scan-size threshold makes InjectRuntimeFilter eligible on the TPC-DS queries, and disabling broadcast joins forces sort-merge joins so the runtime filter is injected as an execution-side ScalarSubquery rather than being folded into a broadcast exchange. --- .github/workflows/tpcds.yml | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/.github/workflows/tpcds.yml b/.github/workflows/tpcds.yml index d2711dee5..479cacf85 100644 --- a/.github/workflows/tpcds.yml +++ b/.github/workflows/tpcds.yml @@ -112,3 +112,17 @@ jobs: scalaver: '2.13' hadoop-profile: 'hadoop3' sparktests: 'true' + + test-spark-41-jdk17-scala-2-13-with-bloomfilter-optimizer-enabled: + name: Test spark-4.1 JDK17 Scala-2.13 with bloomFilter optimizer enabled + uses: ./.github/workflows/tpcds-reusable.yml + with: + sparkver: spark-4.1 + javaver: '17' + scalaver: '2.13' + hadoop-profile: 'hadoop3' + sparktests: 'true' + extrasparkconf: >- + --conf spark.sql.optimizer.runtime.bloomFilter.enabled=true + --conf spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1B + --conf spark.sql.autoBroadcastJoinThreshold=-1 From cb5244e4765da37e3d163b8c6140c81edc54c4db Mon Sep 17 00:00:00 2001 From: chloex <122268864+xiaoyanxie@users.noreply.github.com> Date: Tue, 4 Aug 2026 02:46:43 -0700 Subject: [PATCH 2/3] [AURON #2386] Make the bloom-filter CI job able to reproduce the deserialization defect The bloom-filter TPC-DS job could never have caught AURON #2386. tpcds-reusable.yml copies the Auron jar into $SPARK_HOME/jars in addition to passing it through spark-submit --jars. MutableURLClassLoader is parent-first, so the $SPARK_HOME/jars copy wins and NativeConverters is defined by the application class loader, which can see every Auron class. The default resolveClass then resolves the whole expression graph correctly no matter which frame VM.latestUserDefinedLoader() selects, and the ClassCastException cannot occur. Disabling broadcast joins was necessary but not sufficient; both conditions have to hold at once. Add a jar-on-system-classpath input, default 'true' so every other job is unchanged, and set it to 'false' for the bloom-filter job so Auron reaches the JVM only through --jars. That alone still would not fail the build. Task-level deserialization failures are absorbed by Spark's task retries: at sf=1 with the job's own confs and the jar off the system classpath, the run logs 213 ClassCastExceptions while every query still reports PASS and run-it.sh exits 0. Add an assert-no-classcastexception input that greps the run log and fails the job, and tee the run output so it can be inspected. Verified locally at sf=1 against dev/tpcds_1g with the job's exact configuration, q1,q2,q3, Spark 4.1.2 / Scala 2.13 / JDK 17, Auron supplied only through --jars: without the fix in f9b49c1a the run logs 213 ClassCastExceptions and the new assertion exits 1; with the fix it logs 0 and the assertion exits 0. Co-Authored-By: Claude Opus 5 --- .github/workflows/tpcds-reusable.yml | 44 ++++++++++++++++++++++++++-- .github/workflows/tpcds.yml | 6 ++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/.github/workflows/tpcds-reusable.yml b/.github/workflows/tpcds-reusable.yml index ee176d0d1..672a58b44 100644 --- a/.github/workflows/tpcds-reusable.yml +++ b/.github/workflows/tpcds-reusable.yml @@ -64,6 +64,23 @@ on: required: false type: string default: '' + assert-no-classcastexception: + description: >- + Whether to fail the job if the TPC-DS run logs any ClassCastException. Spark retries + absorb task-level deserialization failures, so queries can report PASS while throwing + hundreds of exceptions; this asserts on the log instead. + required: false + type: string + default: 'false' + jar-on-system-classpath: + description: >- + Whether to also copy the Auron jar into $SPARK_HOME/jars. When true the jar is + loaded by the application class loader; when false it reaches the JVM only through + spark-submit --jars, i.e. Spark's MutableURLClassLoader. Some class-loading defects + only reproduce in the latter configuration. + required: false + type: string + default: 'true' queries: description: 'Optional list of queries to run' required: false @@ -257,12 +274,18 @@ jobs: path: dev/tpcds_1g - name: Install Auron JAR + env: + JAR_ON_SYSTEM_CLASSPATH: ${{ inputs.jar-on-system-classpath }} run: | ls -la jar=$(ls -1 auron-${{ inputs.sparkver }}_${{ inputs.scalaver }}*.jar | head -n1) [ -n "$jar" ] || { echo "No jar matched: auron-${{ inputs.sparkver }}_${{ inputs.scalaver }}*.jar"; exit 1; } echo "AURON_SPARK_JAR=$jar" >> "$GITHUB_ENV" - cp "$jar" spark-bin-${{ inputs.sparkver }}_${{ inputs.scalaver }}/jars/ + if [ "$JAR_ON_SYSTEM_CLASSPATH" = "false" ]; then + echo "Auron reaches the JVM only through spark-submit --jars (MutableURLClassLoader)" + else + cp "$jar" spark-bin-${{ inputs.sparkver }}_${{ inputs.scalaver }}/jars/ + fi - name: Setup Java and Maven cache uses: actions/setup-java@v5 @@ -380,13 +403,30 @@ jobs: SPARK_HOME: spark-bin-${{ inputs.sparkver }}_${{ inputs.scalaver }} run: | ls -la + set -o pipefail dev/auron-it/run-it.sh \ ${{ inputs.extrasparkconf }} \ --type tpcds \ --data-location dev/tpcds_1g \ --query-filter ${{ matrix.query }} \ --result-check \ - --plan-check + --plan-check 2>&1 | tee tpcds-run-${{ matrix.query }}.log + + # Task-level deserialization failures are absorbed by Spark's task retries, so the + # queries can still report PASS while throwing hundreds of exceptions. Assert on the + # log directly, otherwise a regression of AURON #2386 goes unnoticed. + - name: Assert no deserialization ClassCastException + if: ${{ inputs.assert-no-classcastexception == 'true' }} + env: + QUERY_LOG: tpcds-run-${{ matrix.query }}.log + run: | + count=$(grep -c 'ClassCastException' "$QUERY_LOG" || true) + if [ "$count" -gt 0 ]; then + echo "::error::$count ClassCastException(s) during expression deserialization" + grep -m5 'ClassCastException' "$QUERY_LOG" || true + exit 1 + fi + echo "No ClassCastException found." - name: Upload RSS log if: ${{ failure() && (inputs.celebornver != '' || inputs.unifflever != '') }} diff --git a/.github/workflows/tpcds.yml b/.github/workflows/tpcds.yml index 479cacf85..33787fc28 100644 --- a/.github/workflows/tpcds.yml +++ b/.github/workflows/tpcds.yml @@ -122,6 +122,12 @@ jobs: scalaver: '2.13' hadoop-profile: 'hadoop3' sparktests: 'true' + # Supply Auron only through spark-submit --jars, so it is defined by Spark's + # MutableURLClassLoader rather than the application class loader. The runtime + # bloom-filter ScalarSubquery deserialization defect (AURON #2386) cannot reproduce + # when the jar is also on $SPARK_HOME/jars. + jar-on-system-classpath: 'false' + assert-no-classcastexception: 'true' extrasparkconf: >- --conf spark.sql.optimizer.runtime.bloomFilter.enabled=true --conf spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1B From af806de31254b268ea0253d68e15abe271065b62 Mon Sep 17 00:00:00 2001 From: chloex <122268864+xiaoyanxie@users.noreply.github.com> Date: Tue, 4 Aug 2026 01:08:54 -0700 Subject: [PATCH 3/3] [AURON #2386] Resolve deserialized expression classes with an explicit class loader NativeConverters.deserializeExpression built a plain ObjectInputStream, whose default resolveClass resolves each class through VM.latestUserDefinedLoader() -- a loader selected from the live call stack rather than the context class loader. During a nested read the most recent user-defined frame is often a Spark or Scala class, whose loader cannot see Auron classes when Auron is supplied through spark.jars and therefore loaded by MutableURLClassLoader. The expression graph then resolves only partially and an un-readResolve'd DefaultSerializationProxy is assigned into RDD.dependencies_, raising: 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 also explains why the crash never reproduced in-process: in a test session Auron sits on the application class loader, which can always see its own classes, so the call-stack loader selection is harmless there. Pin resolution to an explicit loader -- the context/Spark loader, falling back to the loader that defined Auron, then to the default -- so class resolution no longer depends on which frame happens to be on the stack. Spark's own JavaDeserializationStream does the same; it is not reused here because it is private[spark] and Auron builds against eight Spark versions. Verified on TPC-DS with Spark 4.1.2 / Scala 2.13 / JDK 17, sf=10 with spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1GB and Auron supplied through --jars. A controlled A/B over identical builds differing only in this change gives 42 ClassCastExceptions before and 0 after; q1, q2 and q3 pass 3/3 with results validated against vanilla Spark. Co-Authored-By: Claude Opus 5 --- .../spark/sql/auron/NativeConverters.scala | 67 +++++++++++++++---- 1 file changed, 54 insertions(+), 13 deletions(-) diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala index 95f42e0fd..ed4d952ed 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala @@ -18,8 +18,10 @@ package org.apache.spark.sql.auron import java.io.ByteArrayInputStream import java.io.ByteArrayOutputStream +import java.io.InputStream import java.io.ObjectInputStream import java.io.ObjectOutputStream +import java.io.ObjectStreamClass import scala.collection.mutable import scala.jdk.CollectionConverters._ @@ -1505,24 +1507,63 @@ object NativeConverters extends Logging { } } + /** + * ObjectInputStream that resolves classes against an explicit class loader. + * + * The default ObjectInputStream.resolveClass resolves each class through + * VM.latestUserDefinedLoader(), which selects a loader from the live call stack rather than the + * context class loader. During a nested read the most recent user-defined frame is often a + * Spark or Scala class, whose loader cannot see Auron classes when Auron is supplied through + * spark.jars and therefore loaded by MutableURLClassLoader. The expression graph then resolves + * only partially and an un-readResolve'd DefaultSerializationProxy is assigned into + * RDD.dependencies_, raising a ClassCastException. Pinning the loader keeps resolution + * independent of the call stack. Spark's own JavaDeserializationStream does the same. + */ + private class AuronObjectInputStream(in: InputStream, loader: ClassLoader) + extends ObjectInputStream(in) { + + // scalastyle:off classforname + private def load(name: String, cl: ClassLoader): Class[_] = Class.forName(name, false, cl) + // scalastyle:on classforname + + // resolveProxyClass is deliberately not overridden: the only non-deprecated way to obtain a + // proxy Class is Proxy.getProxyClass, and serialized expressions contain no dynamic proxies. + override def resolveClass(desc: ObjectStreamClass): Class[_] = { + val name = desc.getName + try { + load(name, loader) + } catch { + case _: ClassNotFoundException => + // the loader that defined Auron always resolves Auron's own classes even when the + // context loader cannot, and its parent chain still covers Spark and Scala classes + try { + load(name, getClass.getClassLoader) + } catch { + case _: ClassNotFoundException => super.resolveClass(desc) + } + } + } + } + def deserializeExpression[E <: Expression, S <: Serializable]( serialized: Array[Byte]): (E with Serializable, S) = { Utils.tryWithResource(new ByteArrayInputStream(serialized)) { bis => - Utils.tryWithResource(new ObjectInputStream(bis)) { ois => - def read(): (E with Serializable, S) = { - val expr = ois.readObject().asInstanceOf[E with Serializable] - val payload = ois.readObject().asInstanceOf[S with Serializable] - (expr, payload) - } - // Spark TaskMetrics#externalAccums is not thread-safe - val taskContext = TaskContext.get() - if (taskContext != null) { - taskContext.taskMetrics().synchronized { + Utils.tryWithResource(new AuronObjectInputStream(bis, Utils.getContextOrSparkClassLoader)) { + ois => + def read(): (E with Serializable, S) = { + val expr = ois.readObject().asInstanceOf[E with Serializable] + val payload = ois.readObject().asInstanceOf[S with Serializable] + (expr, payload) + } + // Spark TaskMetrics#externalAccums is not thread-safe + val taskContext = TaskContext.get() + if (taskContext != null) { + taskContext.taskMetrics().synchronized { + read() + } + } else { read() } - } else { - read() - } } } }