Skip to content

Try to reproduce the issue in the GitHub CI pipeline#2395

Draft
xiaoyanxie wants to merge 3 commits into
apache:masterfrom
xiaoyanxie:fix/2386-runtime-bloom-filter-subquery-serialization
Draft

Try to reproduce the issue in the GitHub CI pipeline#2395
xiaoyanxie wants to merge 3 commits into
apache:masterfrom
xiaoyanxie:fix/2386-runtime-bloom-filter-subquery-serialization

Conversation

@xiaoyanxie

@xiaoyanxie xiaoyanxie commented Jul 19, 2026

Copy link
Copy Markdown

Which issue does this PR close?

Closes #2386

Note: This is currently a draft PR.

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_.

The serialized expression is later evaluated by SparkScalarSubqueryWrapperExpr, which delegates to the JVM expression wrapper. The JVM deserializes it using ObjectInputStream in NativeConverters.deserializeExpression(). On Scala 2.13, immutable collections are serialized through scala.collection.generic.DefaultSerializationProxy. During deserialization of the cyclic RDD object graph, a back-reference resolves to the serialization proxy before its readResolve() replacement is applied. Java consequently attempts to assign a DefaultSerializationProxy to the RDD.dependencies_: scala.collection.immutable.Seq field, producing:
ClassCastException: cannot assign instance of scala.collection.generic.DefaultSerializationProxy to field org.apache.spark.rdd.RDD.dependencies_.

Therefore, the runtime bloom-filter optimization is only the trigger. The underlying defect is that Auron serializes a plan-bearing execution expression after its value has already been computed. This unnecessarily captures the physical plan and RDD lineage.

The issue is specific to the Scala 2.13 path because those immutable collections use serialization proxies. The equivalent Scala 2.12 object graph does not use the same proxy mechanism and therefore happens to deserialize successfully.
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

(TODO)

What changes are included in this PR?

  • An integration test in the GitHub CI that reproduces the bug
  • A fix for the ScalarSubquery branch within the convertExprWithFallback function of NativeConverters.scala:
    Prevents serializing the entire RDD object graph for ScalarSubquery by exclusively serializing its Scala Literal value.

Are there any user-facing changes?

No

How was this patch tested?

Via integration tests.

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

  • Yes
  • No

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

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

@github-actions github-actions Bot added the infra label Jul 19, 2026
xiaoyanxie and others added 2 commits July 19, 2026 10:08
Disable broadcast joins (spark.sql.autoBroadcastJoinThreshold=-1) so the
runtime bloom-filter build side becomes a shuffle-backed plan. The
ClassCastException only triggers when the injected ScalarSubquery carries a
shuffle MapPartitionsRDD lineage; on the sf=1 CI fixture the joins otherwise
broadcast and the repro silently passes.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

ClassCastException (Scala 2.13 DefaultSerializationProxy) when deserializing ScalarSubquery injected by runtime bloom filter on Spark 4.1

1 participant