sumitsingh-in opened a new pull request, #19822: URL: https://github.com/apache/hudi/pull/19822
### Describe the issue this Pull Request addresses
Hudi's legacy (pre-file-group-reader) Spark Parquet file format
(`Spark3LegacyHoodieParquetFileFormat` and its per-Spark-version
subclasses)
decides whether to hand back `ColumnarBatch` or row-based `InternalRow`
objects by recomputing `supportBatch(schema)` itself, ignoring
`FileFormat.OPTION_RETURNING_BATCH` — the option `FileSourceScanExec`
already threads down with the batching decision it made at plan time (and
which vanilla Spark's own `ParquetFileFormat` does honor).
The two answers disagree whenever `FileSourceScanExec.supportsColumnar`
depends on something `supportBatch` alone does not account for — most
commonly, `spark.sql.codegen.wholeStage` being disabled for the query (or
another operator's schema tripping `spark.sql.codegen.maxFields`). In that
case the planner marks the scan row-based ("Batched: false" in `explain`),
but the reader still returns `ColumnarBatch` objects. Any consumer that
reads the RDD as rows — most visibly `BroadcastExchangeExec`'s collection
path — then fails with:
java.lang.ClassCastException: class
org.apache.spark.sql.vectorized.ColumnarBatch
cannot be cast to class org.apache.spark.sql.catalyst.InternalRow
Closes #19817
### Summary and Changelog
Reading a Hudi table through the legacy Parquet read path inside a query
with whole-stage codegen disabled (e.g. because it exceeds
`spark.sql.codegen.maxFields`, or has it turned off explicitly) could
throw
a `ClassCastException` during execution, most commonly during a broadcast
join. This fixes it by making the reader honor
`FileFormat.OPTION_RETURNING_BATCH` when the plan supplies it, falling
back
to the existing `supportBatch`-based computation only when the option is
absent.
- `Spark3LegacyHoodieParquetFileFormat.buildReaderWithPartitionValues`:
read
`options.get(FileFormat.OPTION_RETURNING_BATCH)` first, matching the
precedent already set by Spark's own `ParquetFileFormat`.
- Added
`TestLegacyParquetReadPath#testBroadcastJoinHonorsPlanTimeBatchingDecision`,
a regression test that writes a flat, all-atomic-type COW table (so
`supportBatch` alone would trivially say yes), disables
`spark.sql.codegen.wholeStage`, and broadcast-joins the legacy-format
DataFrame — reproducing the exact crash and asserting it no longer
occurs.
No code was copied from elsewhere; the fix mirrors the existing
`OPTION_RETURNING_BATCH` handling already present in Spark's own
`ParquetFileFormat`.
### Impact
None to any public API, config, or storage format. Behavior-only fix to
the
legacy Parquet read path, and only in the previously-broken case
(whole-stage
codegen disabled or a wide-schema query alongside this scan). No change to
the common case where whole-stage codegen is enabled.
### Risk Level
low — the change is a single, narrowly-scoped computation gated behind a
Spark-supplied option key that is already part of Spark's public
`FileFormat` API, follows the exact precedent of vanilla Spark's own
`ParquetFileFormat`, and is covered by a new regression test reproducing
the
originally-reported crash.
### Documentation Update
none — internal bug fix, no new config, feature, or user-facing behavior
change.
### Contributor's checklist
- [ ] Read through [contributor's
guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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]
