sumitsingh-in opened a new issue, #19817: URL: https://github.com/apache/hudi/issues/19817
### 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)
--
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]
