beliefer opened a new issue, #12719:
URL: https://github.com/apache/gluten/issues/12719
### Backend
VL (Velox)
### Bug description
When running with CelebornShuffleManager (RSS), a task can crash with:
java.lang.ClassCastException: class
org.apache.spark.shuffle.CelebornColumnarBatchSerializerInstance
cannot be cast to class
org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
at
org.apache.spark.shuffle.ColumnarShuffleReader.read(ColumnarShuffleReader.scala:97)
The task fails 4 times and the whole SQL aborts.
Expected behavior: when the Celeborn shuffle falls back to the local
ColumnarShuffleManager, the shuffle read should succeed (Celeborn's serializer
implements the standard deserializeStream).
Actual behavior: the local ColumnarShuffleReader unconditionally casts the
dependency's serializer to ColumnarBatchSerializerInstance and throws
ClassCastException, killing the task.
Root cause:
1. At plan-build time, VeloxSparkPlanExecApi.createColumnarBatchSerializer
sees the shuffle manager is a NeedCustomColumnarBatchSerializer and binds the
Celeborn serializer (CelebornColumnarBatchSerializerInstance, which extends
SerializerInstance — not ColumnarBatchSerializerInstance) into the
ColumnarShuffleDependency.
2. On the driver, CelebornShuffleManager.registerShuffle runs
applyFallbackPolicies. When it returns true (Celeborn unavailable / fallback
threshold / quota) and
spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled=true (default), it
falls back to registering on the local ColumnarShuffleManager, returning a
plain BaseShuffleHandle.
3. On read, CelebornShuffleManager.getReader sees a
non-CelebornShuffleHandle and dispatches to the local ColumnarShuffleReader.
Its read() does
dep.serializer.newInstance().asInstanceOf[ColumnarBatchSerializerInstance], but
the serializer is still the Celeborn one from step 1 → ClassCastException.
The exchange's serializer is a lazy val instantiated on the executor; the
fallback decision is made on the driver. The executor cannot observe the
fallback, so the local reader receives a serializer it does not expect.
Proposed fix: in ColumnarShuffleReader.read(), dispatch on the serializer
instance type instead of the dependency type — use the batched
deserializeStreams path only when the serializer really is a
ColumnarBatchSerializerInstance, otherwise use the per-stream deserializeStream
path. This keeps the single-stream path that PR #12403 was closed for removing
("deserializeStream is still used by uniffle").
### Gluten version
main branch
### Spark version
Spark-3.5.x
### Spark configurations
spark.shuffle.manager=org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleManager
spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled=true # default
### System information
_No response_
### Relevant logs
```bash
java.lang.ClassCastException: class
org.apache.spark.shuffle.CelebornColumnarBatchSerializerInstance
cannot be cast to class
org.apache.gluten.vectorized.ColumnarBatchSerializerInstance
at
org.apache.spark.shuffle.ColumnarShuffleReader.read(ColumnarShuffleReader.scala:97)
```
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
