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

**Describe the bug**

On Spark 4.1 (Scala 2.13), queries fail at runtime whenever Spark's runtime bloom filter optimization (`InjectRuntimeFilter`) is applied. Every affected task dies with:

```
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
    at java.base/java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2096)
    ...
    at org.apache.spark.sql.auron.NativeConverters$.deserializeExpression(NativeConverters.scala:1495)
    at org.apache.auron.spark.sql.SparkAuronUDFWrapperContext.<init>(SparkAuronUDFWrapperContext.scala:44)
    at org.apache.auron.jni.SparkAuronAdaptor.getAuronUDFWrapperContext(SparkAuronAdaptor.java:103)
    at org.apache.auron.jni.JniBridge.getAuronUDFWrapperContext(JniBridge.java:113)
```

followed by a native panic (`output_with_sender[Filter]: output() returns error: External error: Java exception thrown at native-engine/datafusion-ext-exprs/src/spark_udf_wrapper.rs:97`) and job abort.

Root cause (verified, see repro below):

1. `InjectRuntimeFilter` rewrites join filters into `might_contain(scalar-subquery(bloom_agg), xxhash64(key))`. Auron converts `BloomFilterMightContain` natively (`ShimsImpl.convertBloomFilterMightContain`), and its `bloomFilterExpression` — an `ExecSubqueryExpression` — is converted at `NativeConverters.scala:462`, which Java-serializes **the entire execution-side `ScalarSubquery` object, including its `plan: BaseSubqueryExec` reference**. An executed plan transitively references RDDs (`MapPartitionsRDD` in the trace above).
2. On the native side, `SparkScalarSubqueryWrapperExpr` evaluates by delegating to `SparkUDFWrapperExpr` (`native-engine/datafusion-ext-exprs/src/spark_scalar_subquery_wrapper.rs`), which calls back into the JVM and deserializes the bytes with a plain `java.io.ObjectInputStream` (`NativeConverters.deserializeExpression`). This is why the stack shows `SparkAuronUDFWrapperContext` although no UDF fallback is involved.
3. On Scala 2.13, immutable collections serialize through `scala.collection.generic.DefaultSerializationProxy`, which is only replaced by the real collection in `readResolve()`. Per the Java serialization spec, `readResolve` substitution does not propagate to back-references, so in a cyclic graph (RDD lineage) the *unresolved proxy* gets assigned into `RDD.dependencies_` → `ClassCastException`. Scala 2.12 has no serialization proxies for these collections, which is why the identical graph deserializes fine on 2.12.

Note this is **not** the `SparkUDFWrapper` expression-fallback path: ordinary fallback expressions (e.g. `UnscaledValue` / `MakeDecimal`, 167 instances in these runs) serialize shallow bound expression trees with no plan references and round-trip fine on Scala 2.13.

**To Reproduce**

Minimal repro (small dataset, threshold lowered so the runtime bloom filter is injected):

1. Build Auron for Spark 4.1 / Scala 2.13 (`./auron-build.sh --pre --sparkver 4.1 --scalaver 2.13`).
2. Generate TPC-DS parquet data at sf=10 (~5 GB).
3. Run:
   ```
   SPARK_HOME=<spark-4.1.2-bin-hadoop3> SPARK_VERSION=spark-4.1 SCALA_VERSION=2.13 \
     dev/auron-it/run-it.sh --type tpcds --data-location <tpcds_10g> \
       --conf spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1GB \
       --query-filter q1,q2,q3
   ```
4. q2 and q3 fail with the exception above (the Auron run produces 0 rows, reported as a result mismatch vs the vanilla Spark baseline). q1 passes — its scanned tables stay below even the lowered threshold, so no runtime filter is injected there.

With default Spark settings the bug reproduces at sf=100, where the relevant fact-table scans cross the default `spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold` (10GB):

| per-table scan size (uncompressed parquet) | sf=30 | sf=60 | sf=100 |
|---|---|---|---|
| `store_sales` (q3 application side) | 4.7 GB | 9.5 GB | **16 GB** |
| `catalog_sales` (q2 application side) | 3.3 GB | 6.9 GB | **12 GB** |

Verification matrix (Spark 4.1.2 / Scala 2.13, same host, JDK 21; CCE = `ClassCastException` count in log):

| | sf=1 | sf=10 | sf=10, threshold=1GB | sf=30 | sf=60 | sf=100 | sf=100, bloomFilter.enabled=false |
|---|---|---|---|---|---|---|---|
| result | PASS 3/3 | PASS 3/3 | **FAIL q2,q3** | PASS 3/3 | PASS 3/3 | **FAIL q2,q3** | PASS 3/3 |
| CCE | 0 | 0 | 83 | 0 | 0 | 52 | 0 |

Spark 3.5.8 / Scala 2.12 passes 3/3 at every scale factor with default settings (same code path serializes the same plan-carrying subquery, but Scala 2.12 collections have no serialization proxies).

The two toggles isolate the cause in both directions: disabling the runtime bloom filter at sf=100 eliminates the crash with everything else unchanged (including all 167 fallback-expression serializations), and lowering the threshold at sf=10 introduces it.

Independently re-verified on a second machine/OS (sf=10, threshold=1GB case): macOS aarch64, JDK 17.0.16, Scala 2.13, Spark 4.1.2, Auron 8.0.0-incubating built from source (`./auron-build.sh --pre --sparkver 4.1 --scalaver 2.13`). Result: `q1 PASS, q2 FAIL, q3 FAIL`, 66 `ClassCastException` occurrences, 25 ordinary fallback-expression serializations (`UnscaledValue`/`MakeDecimal`) — same qualitative signature as the Linux/JDK21 run (CCE count differs, as expected, since partitioning/task counts differ across environments/runs). The captured stack matches frame-for-frame:

```
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
    ...
    at org.apache.spark.sql.auron.NativeConverters$.read$1(NativeConverters.scala:1498)
    at org.apache.spark.sql.auron.NativeConverters$.$anonfun$deserializeExpression$4(NativeConverters.scala:1506)
    at org.apache.spark.sql.auron.NativeConverters$.deserializeExpression(NativeConverters.scala:1495)
    at org.apache.auron.spark.sql.SparkAuronUDFWrapperContext.<init>(SparkAuronUDFWrapperContext.scala:44)
    at org.apache.auron.jni.SparkAuronAdaptor.getAuronUDFWrapperContext(SparkAuronAdaptor.java:103)
    at org.apache.auron.jni.JniBridge.getAuronUDFWrapperContext(JniBridge.java:113)

thread 'auron-native-stage-141-part-0-tid-251' panicked at native-engine/datafusion-ext-plans/src/common/execution_context.rs:573:21:
output_with_sender[Filter]: output() returns error: External error: Java exception thrown at native-engine/datafusion-ext-exprs/src/spark_udf_wrapper.rs:97: java.lang.ClassCastException: ...
```

This rules out a JDK-version-specific or OS/arch-specific explanation: the bug reproduces identically on Linux x86_64/JDK21 and macOS aarch64/JDK17, consistent with the root cause being purely about the Scala 2.13 collections runtime (`DefaultSerializationProxy`), not the JVM or platform.

**Expected behavior**

Queries with runtime-bloom-filter-injected scalar subqueries should run correctly on Spark 4.1 / Scala 2.13, as they do on Spark 3.5 / Scala 2.12. TPC-DS q2/q3 should pass the consistency check at sf=100 with default Spark settings.

**Screenshots**

N/A (full logs available on request).

**Additional context**

- Environment: Spark 4.1.2, Scala 2.13, JDK 21 (GraalVM 21.0.8), Auron master (8.0.0-incubating), local mode, Linux x86_64.
- Suggested fix direction: `NativeConverters.prepareExecSubquery` already calls `updateResult()` before serialization, so the subquery result is materialized on the driver at that point. Serializing the *evaluated result* (e.g. as a `Literal`) instead of the `ScalarSubquery` object would avoid capturing the physical plan entirely — fixing this crash and also removing plan/RDD payloads from task binaries (related to #2351 / #2307). Alternatively/additionally, a fail-fast check that serialized expressions contain no plan-referencing nodes would turn future regressions of this class into clear errors instead of Scala-version-dependent `ClassCastException`s.
- Related issues:
  - #1953 — Spark 4.1 support epic (this only bites on Scala 2.13, hence only Spark 4.x).
  - #2351 / #2307 — oversized task serialization from over-capture; same underlying over-capture, different symptom.
  - #369 / #648 — earlier `ClassCastException`s in the same `deserializeExpression` path (classloader-related), showing this boundary is sensitive to the serialization environment.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

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

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

per-table scan size (uncompressed parquet)	sf=30	sf=60	sf=100
`store_sales` (q3 application side)	4.7 GB	9.5 GB	16 GB
`catalog_sales` (q2 application side)	3.3 GB	6.9 GB	12 GB

	sf=1	sf=10	sf=10, threshold=1GB	sf=30	sf=60	sf=100	sf=100, bloomFilter.enabled=false
result	PASS 3/3	PASS 3/3	FAIL q2,q3	PASS 3/3	PASS 3/3	FAIL q2,q3	PASS 3/3
CCE	0	0	83	0	0	52	0

Uh oh!

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

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions