sumitsingh-in opened a new pull request, #20183:
URL: https://github.com/apache/hudi/pull/20183
…
Spark caches the resolved relation per session, so one file format instance
serves every statement on a table, and supportBatch is only called for scans
narrower than spark.sql.codegen.maxFields. The vectorized decision was kept in
a mutable field and written into spark.sql.parquet.enableVectorizedReader, so a
wide scan could inherit a narrow scan's answer and fail with
ClassCastException: ColumnarBatchRow cannot be cast to UnsafeRow. Evaluate
canReadVectorized against each scan's own schema in both supportBatch and
buildReaderWithPartitionValues, and stop mutating the session conf. Adds
TestVectorizedReadDecisionPerScan covering COW and MOR.
### Describe the issue this Pull Request addresses
`HoodieFileGroupReaderBasedFileFormat` kept its vectorized-read decision in
two mutable fields
(`supportVectorizedRead`, `supportReturningBatch`) that were only assigned
inside `supportBatch`.
`buildReaderWithPartitionValues` reused that stored value for every reader
it built and also wrote it into the
session conf `spark.sql.parquet.enableVectorizedReader`.
That decision is not per scan, which breaks in two ways:
1. Spark caches the resolved relation of a table per session
(`SessionCatalog.tableRelationCache`), so one file
format instance serves every statement on that table. `HoodieCatalog`
returns the V1 relation unless
`hoodie.schema.on.read.enable=true`, so the default Spark SQL path shares
the instance too.
2. `FileSourceScanExec.supportsColumnar` only calls `supportBatch` when the
scan schema has fewer than
`spark.sql.codegen.maxFields` (default 100) leaf fields. A wider scan
never calls it, so it inherits whatever the
previous statement on the table decided.
Consequences:
- `select count(*)` sets the flag to `true`. A following `select *` on a
table with more than 100 nested leaf fields
is planned row-based but is handed a vectorized base-file reader. When
`spark.sql.parquet.enableVectorizedReader`
is `false` for that query, Spark skips the UnsafeRow conversion of the
scan output
(`FileSourceScanExec.needsUnsafeRowConversion`) and the query fails with
`java.lang.ClassCastException:
org.apache.spark.sql.vectorized.ColumnarBatchRow cannot be cast to
org.apache.spark.sql.catalyst.expressions.UnsafeRow`
at `SparkPlan.getByteArrayRdd`.
- The reverse order is a silent performance regression: a wide scan first
(never asked, flag still `false`)
disables vectorized reading for every later narrow scan on the table in
that session.
- Every Hudi scan overwrote the user's
`spark.sql.parquet.enableVectorizedReader` session setting.
### Summary and Changelog
Vectorized and batch reading now behave the same regardless of which
statements ran before on the same table in
the session: wide nested scans no longer fail with the `ColumnarBatchRow`
`ClassCastException`, narrow scans stay
vectorized after a wide scan, and Hudi no longer modifies
`spark.sql.parquet.enableVectorizedReader`.
Changes in `HoodieFileGroupReaderBasedFileFormat` (`hudi-spark-common`):
- Removed the `supportVectorizedRead` and `supportReturningBatch` fields.
- Extracted the decision into `canReadVectorized(sparkSession, schema)`, a
pure function of the session conf, the
scan schema and the table properties. It keeps the existing early exits
(vector columns, Spark 4.1 variant
projections and variant types) and the per-format checks; the local
`supportBatch` value is renamed
`formatSupportsBatch`.
- `supportBatch` returns `!isMOR && canReadVectorized(...)`, unchanged
semantics for the scans Spark asks about.
- `buildReaderWithPartitionValues` evaluates `canReadVectorized(spark,
outputSchema)` for the scan it is building
and uses that for both `baseFileReader` and `fileGroupBaseFileReader`. It
cannot simply call `supportBatch`:
for MOR that returns `false` while file slices without log files are still
read with the vectorized reader.
- Removed the `setConfString("spark.sql.parquet.enableVectorizedReader",
...)` call.
**Tests:**
- New `TestVectorizedReadDecisionPerScan` (`hudi-spark`,
`org.apache.spark.sql.hudi.common`), run for COW and MOR
on a table whose `select *` has 120 nested leaf fields, i.e. wider than
`spark.sql.codegen.maxFields`:
- narrow `count(*)` followed by wide `select *`: the wide scan is not
columnar, returns the right rows, also with
`spark.sql.parquet.enableVectorizedReader=false` (the
`ClassCastException` case), and the session conf is left
as the user set it;
- wide `select *` followed by narrow `count(*)`: the later COW scan is
still columnar and the conf is unchanged.
Both fail without the fix and pass with it.
### Impact
- No public API or configuration change.
- Hudi scans no longer rewrite `spark.sql.parquet.enableVectorizedReader` in
the Spark session.
- Performance: the vectorized decision is made from each scan's own schema.
Narrow scans after a wide scan are
vectorized again; wide scans use the vectorized reader when the format and
the conf allow it, with Spark
converting the output to rows as it does for any non-batched vectorized
scan.
- The class lives in `hudi-spark-common`, so all Spark profiles are covered.
The legacy
`Spark*LegacyHoodieParquetFileFormat` classes already decide per call and
are not touched.
### Risk Level
low. The change is confined to one class and evaluates the same predicate
from the same inputs as before, only per
scan instead of once per cached instance; the readers themselves are
unchanged. Covered by the new regression test
(fails before the change, passes after) and the existing Spark SQL test
suites in CI.
### Documentation Update
none. No new config or user-facing feature
### Contributor's checklist
- [x] 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
https://github.com/apache/hudi/issues/20182
--
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]
