beliefer opened a new pull request, #12721:
URL: https://github.com/apache/gluten/pull/12721
## What changes are proposed in this pull request?
When Gluten runs with `CelebornShuffleManager` (RSS), a task can crash on
the shuffle read side 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)
```
The task fails 4 times and the whole SQL aborts.
### Root cause
The crash comes from Celeborn's *fallback* path, not from any particular
query shape:
1. At plan build time, `VeloxSparkPlanExecApi.createColumnarBatchSerializer`
sees the shuffle manager is a `NeedCustomColumnarBatchSerializer` and binds the
**Celeborn** serializer (`CelebornColumnarBatchSerializerInstance`) into the
`ColumnarShuffleDependency`. That class `extends SerializerInstance` — it is
**not** a `ColumnarBatchSerializerInstance`.
2. On the driver, `CelebornShuffleManager.registerShuffle` runs
`applyFallbackPolicies`. When it returns `true` (Celeborn service unavailable /
fallback policy threshold / quota) and
`spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled=true` (its
default), it **falls back to registering the shuffle on the local
`ColumnarShuffleManager`**, returning a plain `BaseShuffleHandle` rather than a
`CelebornShuffleHandle`.
3. On read, `CelebornShuffleManager.getReader` sees a
non-`CelebornShuffleHandle` and dispatches to the local
`ColumnarShuffleReader`. Its `read()` matched on the **dependency type** and,
for a `ColumnarShuffleDependency`, hard-cast the serializer with
`.asInstanceOf[ColumnarBatchSerializerInstance]`. But the dependency's
serializer is still the Celeborn one from step 1 → `ClassCastException`.
The exchange node's `serializer` is a `lazy val` instantiated on the
**executor** after plan deserialization, while the fallback decision is made on
the **driver**. The executor cannot observe that the driver fell back, so the
local reader is handed a serializer it does not expect. The dependency stays a
`ColumnarShuffleDependency`, so testing the dependency type is the wrong
discriminator.
## How was this patch tested?
- **New unit test** `ColumnarShuffleReaderSuite` (`backends-velox`): builds a
`ColumnarShuffleDependency` whose serializer is a plain (non-columnar)
`SerializerInstance` —
exactly the state left on the dependency after `CelebornShuffleManager`
falls back to the local
`ColumnarShuffleManager` — and asserts that `ColumnarShuffleReader.read()`
returns an empty
iterator instead of throwing `ClassCastException`. The cast/dispatch
decision happens before any
block is fetched, so an empty `blocksByAddress` deterministically
exercises the fix without a
real Celeborn cluster, network, or on-disk shuffle data. Before the patch
this test fails with
the `ClassCastException` above; after the patch it passes.
- The fix is a read-side dispatch change with an equivalence argument: the
batched arm is now
guarded by an `isInstanceOf` check that is strictly narrower than the
previous unconditional
cast, so the normal columnar path is byte-for-byte identical, and the
previously-crashing
fallback path now takes the existing, already-exercised per-stream path.
- End-to-end reproduction is environment-dependent because it requires
`applyFallbackPolicies` to
fire at runtime (Celeborn unavailable / threshold / quota), which is not
deterministic from SQL
alone. Verified on a production cluster (Spark 3.5 + Gluten Velox +
Celeborn): before the patch
the affected SQL aborts with the `ClassCastException` above; after the
patch the same SQL
completes and returns correct results.
The workaround before this fix is to set
`spark.gluten.sql.columnar.shuffle.celeborn.fallback.enabled=false`, which
turns the fallback into a clear `GlutenException: The Celeborn service is
unavailable` instead of the confusing `ClassCastException` — but it disables
the fallback entirely. This PR fixes the fallback itself.
## Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude-opus-4-8.
--
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]
