ShreyeshArangath commented on code in PR #2395:
URL: https://github.com/apache/auron/pull/2395#discussion_r3746665671
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala:
##########
@@ -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[_] = {
Review Comment:
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?
##########
.github/workflows/tpcds-reusable.yml:
##########
@@ -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: |
Review Comment:
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?
##########
.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.
Review Comment:
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?
##########
.github/workflows/tpcds-reusable.yml:
##########
@@ -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
Review Comment:
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 ..
--
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]
