This is an automated email from the ASF dual-hosted git repository.
marin-ma pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new a5859848ed [GLUTEN-12719][VL] Fix ClassCastException in
ColumnarShuffleReader when Celeborn shuffle falls back to local
ColumnarShuffleManager (#12721)
a5859848ed is described below
commit a5859848ed285d173314c1732518dac0495ecf46
Author: Jiaan Geng <[email protected]>
AuthorDate: Fri Aug 7 22:29:46 2026 +0800
[GLUTEN-12719][VL] Fix ClassCastException in ColumnarShuffleReader when
Celeborn shuffle falls back to local ColumnarShuffleManager (#12721)
---
.../spark/shuffle/ColumnarShuffleReader.scala | 24 +++---
.../spark/shuffle/ColumnarShuffleReaderSuite.scala | 88 ++++++++++++++++++++++
2 files changed, 103 insertions(+), 9 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleReader.scala
b/backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleReader.scala
index 77b041fd29..c22daa09a7 100644
---
a/backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleReader.scala
+++
b/backends-velox/src/main/scala/org/apache/spark/shuffle/ColumnarShuffleReader.scala
@@ -73,9 +73,15 @@ class ColumnarShuffleReader[K, C](
/** Read the combined key-values for this reduce task */
override def read(): Iterator[Product2[K, C]] = {
- val recordIter = dep match {
- // If the dependency is a ColumnarShuffleDependency, we use the columnar
serializer.
- case columnarDep: ColumnarShuffleDependency[K, _, C] =>
+ // Dispatch on the concrete serializer instance type rather than the
dependency type.
+ // A ColumnarShuffleDependency does not always carry Gluten's
ColumnarBatchSerializerInstance:
+ // when the CelebornShuffleManager falls back to the local
ColumnarShuffleManager (celeborn
+ // service unavailable / fallback policy), the dependency was already
bound to the Celeborn
+ // serializer at plan build time, so testing the dependency type would
wrongly take the
+ // columnar path and crash with a ClassCastException.
+ val recordIter = dep.serializer.newInstance() match {
+ // Gluten's own columnar serializer deserializes all streams in one
batch.
+ case columnarSerializer: ColumnarBatchSerializerInstance =>
val shuffleBlockFetcherIterator =
SparkShimLoader.getSparkShims.getShuffleBlockFetcherIterator(
ShuffleBlockFetcherIteratorParams(
@@ -98,15 +104,17 @@ class ColumnarShuffleReader[K, C](
readMetrics,
fetchContinuousBlocksInBatch
))
- columnarDep.serializer
- .newInstance()
- .asInstanceOf[ColumnarBatchSerializerInstance]
+ columnarSerializer
.deserializeStreams(
shuffleBlockFetcherIterator,
shuffleBlockFetcherIterator.onComplete,
executionMode)
.asKeyValueIterator
- case _ =>
+ case serializerInstance =>
+ // The dependency's serializer is not Gluten's
ColumnarBatchSerializerInstance. This
+ // covers the CelebornShuffleManager fallback case (dependency bound
to the Celeborn
+ // serializer) as well as row-based dependencies. Fall back to the
per-stream path,
+ // which the standard SerializerInstance (including the Celeborn one)
supports.
val wrappedStreams = new ShuffleBlockFetcherIterator(
context,
blockManager.blockStoreClient,
@@ -128,8 +136,6 @@ class ColumnarShuffleReader[K, C](
fetchContinuousBlocksInBatch
).toCompletionIterator
- val serializerInstance = dep.serializer.newInstance()
-
// Create a key/value iterator for each stream
wrappedStreams.flatMap {
case (blockId, wrappedStream) =>
diff --git
a/backends-velox/src/test/scala/org/apache/spark/shuffle/ColumnarShuffleReaderSuite.scala
b/backends-velox/src/test/scala/org/apache/spark/shuffle/ColumnarShuffleReaderSuite.scala
new file mode 100644
index 0000000000..7209dab178
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/spark/shuffle/ColumnarShuffleReaderSuite.scala
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.shuffle
+
+import org.apache.gluten.execution.CPUStageMode
+import org.apache.gluten.vectorized.{ColumnarBatchSerializerInstance,
NativePartitioning}
+
+import org.apache.spark.{HashPartitioner, TaskContext}
+import org.apache.spark.executor.TempShuffleReadMetrics
+import org.apache.spark.serializer.{JavaSerializer, Serializer}
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * White-box regression test for the Celeborn-to-local shuffle fallback
ClassCastException.
+ *
+ * When [[org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleManager]]
falls back to the local
+ * [[org.apache.spark.shuffle.sort.ColumnarShuffleManager]] (Celeborn service
unavailable or a
+ * fallback policy is applied), the read side dispatches to Gluten's local
+ * [[ColumnarShuffleReader]]. The dependency is still a
[[ColumnarShuffleDependency]], but its
+ * serializer was bound at plan build time to the Celeborn serializer, which
is a plain
+ * `SerializerInstance` and not a [[ColumnarBatchSerializerInstance]].
+ *
+ * Before the fix, `read()` matched on the dependency type and hard-cast the
serializer with
+ * `asInstanceOf[ColumnarBatchSerializerInstance]`, throwing
`ClassCastException` before any block
+ * was fetched. After the fix, `read()` matches on the serializer instance
type and routes a
+ * non-columnar serializer through the per-stream `deserializeStream` path.
+ *
+ * The cast happens before the block-fetch iterator is constructed, so an
empty `blocksByAddress` is
+ * enough to distinguish the two behaviors without a real Celeborn cluster,
network, or on-disk
+ * shuffle data.
+ */
+class ColumnarShuffleReaderSuite extends SharedSparkSession {
+
+ private def newColumnarDep(
+ serializer: Serializer): ColumnarShuffleDependency[Int, Int, Int] = {
+ val rdd = spark.sparkContext.parallelize(Seq((0, 0)), 1)
+ new ColumnarShuffleDependency[Int, Int, Int](
+ rdd,
+ new HashPartitioner(1),
+ serializer = serializer,
+ nativePartitioning = new NativePartitioning("hash", 1),
+ metrics = Map.empty[String, SQLMetric]
+ )
+ }
+
+ private def readWith(serializer: Serializer): Seq[Product2[Int, Int]] = {
+ val dep = newColumnarDep(serializer)
+ val handle = new BaseShuffleHandle(dep.shuffleId, dep)
+ val reader = new ColumnarShuffleReader[Int, Int](
+ handle,
+ // Empty input: the serializer cast happens before any block is fetched,
so no real
+ // shuffle data is needed to exercise the dispatch decision.
+ Iterator.empty,
+ TaskContext.empty(),
+ new TempShuffleReadMetrics(),
+ CPUStageMode
+ )
+ reader.read().toSeq
+ }
+
+ test(
+ "read() tolerates a non-ColumnarBatchSerializerInstance on a
ColumnarShuffleDependency " +
+ "(Celeborn -> local fallback) instead of throwing ClassCastException") {
+ // A plain row-based serializer stands in for the Celeborn serializer left
on the
+ // dependency after CelebornShuffleManager falls back to the local
ColumnarShuffleManager.
+ val nonColumnarSerializer = new JavaSerializer(spark.sparkContext.getConf)
+
assert(!nonColumnarSerializer.newInstance().isInstanceOf[ColumnarBatchSerializerInstance])
+
+ // Must not throw ClassCastException; empty input yields no rows.
+ val rows = readWith(nonColumnarSerializer)
+ assert(rows.isEmpty)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
