[BUG] Legacy Spark Parquet reader ignores plan-time batching decision, causing ClassCastException: ColumnarBatch cannot be cast to InternalRow
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
### Bug Description
**What happened:**
Reading a Hudi table through the legacy Spark Parquet reader
(`Spark3LegacyHoodieParquetFileFormat`, the shared base class used by
the Spark 3.3/3.4/3.5 "legacy" read path in `hudi-spark3-common`) can
throw at runtime:
java.lang.ClassCastException: class org.apache.spark.sql.vectorized.ColumnarBatch
cannot be cast to class org.apache.spark.sql.catalyst.InternalRow
This happens because `buildReaderWithPartitionValues` decides whether
to return columnar batches via:
```scala
val returningBatch = getReturningBatch(sparkSession, resultSchema)
and the Spark 3.3 implementation of that hook is:
override protected def getReturningBatch(sparkSession: SparkSession,
resultSchema: StructType): Boolean =
supportBatch(sparkSession, resultSchema)
```
This recomputes the decision from scratch based only on the scan's own
schema — it never looks at the options map that's passed into the
same method, even though Spark's planner (FileSourceScanExec) already
made this exact decision earlier and threads it down via
options(FileFormat.OPTION_RETURNING_BATCH) (this is how vanilla
Spark's own ParquetFileFormat reads it). FileSourceScanExec's real
decision is roughly:
```scala
supportsColumnar =
conf.wholeStageEnabled &&
!WholeStageCodegenExec.isTooManyFields(conf, this.schema) &&
fileFormat.supportBatch(sparkSession, this.schema)
```
So whenever whole-stage codegen is disabled for the query — either
explicitly (spark.sql.codegen.wholeStage=false) or because some
other node's schema trips spark.sql.codegen.maxFields — the
planner correctly marks the scan as row-based ("Batched: false" in
explain), but Hudi's reader still calls
vectorizedReader.enableReturningBatches() and hands back
ColumnarBatch objects. Anything downstream that consumes the RDD as
InternalRow (e.g. BroadcastExchangeExec's row-collection path)
then throws the ClassCastException above.
**What you expected:**
The legacy Parquet reader should honor the plan-time batching decision
via options.get(FileFormat.OPTION_RETURNING_BATCH) when Spark
provides it, falling back to the current getReturningBatch(...)
recomputation only when the option isn't set — matching how vanilla
Spark's ParquetFileFormat already does this.
**Steps to reproduce:**
1. spark.sql("set spark.sql.codegen.wholeStage=false") (standing in for the real-world trigger: whole-stage codegen getting disabled for a large/complex query).
2. Write a small Hudi COW table with a narrow, all-atomic-type schema and read it back:
```scala
spark.range(10).selectExpr("cast(id as string) as id", "cast(id as string) as name")
.write.format("hudi")
.option("hoodie.table.name", "narrow_tbl")
.option("hoodie.datasource.write.recordkey.field", "id")
.mode("overwrite").save("/tmp/narrow_tbl")
val small = spark.read.format("hudi").load("/tmp/narrow_tbl")
```
3. Use it as the broadcast side of a join and trigger an action:
```
val big = spark.range(1000).toDF("id2")
big.join(broadcast(small), $"id2" === $"id".cast("long"), "left").collect()
```
4. Observe java.lang.ClassCastException: ColumnarBatch cannot be cast to InternalRow.
### Environment
- Hudi version: master (confirmed 2026-09-02)
- Spark version: 3.3.x (bug is in the shared hudi-spark3-common base class, so 3.4.x/3.5.x are almost certainly affected too — not yet independently verified)
- Storage: any (not storage-specific)
### Logs and Stack Trace
```scala
java.lang.ClassCastException: class org.apache.spark.sql.vectorized.ColumnarBatch cannot be cast to class org.apache.spark.sql.catalyst.InternalRow (org.apache.spark.sql.vectorized.ColumnarBatch and
org.apache.spark.sql.catalyst.InternalRow are in unnamed module of loader 'app')
at scala.collection.Iterator$$anon$10.next(Iterator.scala:461)
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:514)
at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:364)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:899)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:899)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:368)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:332)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:138)
at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:563)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1668)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:566)
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.