sumitsingh-in opened a new issue, #20182:
URL: https://github.com/apache/hudi/issues/20182
### Bug Description
**What happened:**
`HoodieFileGroupReaderBasedFileFormat` decides whether to build a
`VectorizedParquetRecordReader` from a mutable field, `supportVectorizedRead`,
that is written only inside `supportBatch`. Spark does not call `supportBatch`
for scans whose schema exceeds `spark.sql.codegen.maxFields` (default 100,
counting nested leaves), so a wide nested scan reuses whatever the last narrow
query on the same relation left in that field. `buildReaderWithPartitionValues`
also writes that stale value into the session conf
`spark.sql.parquet.enableVectorizedReader`, which Spark reads in
`FileSourceScanExec.needsUnsafeRowConversion`.
Two user-visible consequences:
1. A query whose plan says `Batched: false`, on a nested schema, with
`spark.sql.parquet.enableNestedColumnVectorizedReader=false`, is read with
`VectorizedParquetRecordReader` anyway. Whether that happens depends on which
statement ran earlier in the session.
2. When the session conf reads `false` at execute time (left there by an
earlier wide scan on another Hudi table, after `REFRESH TABLE`, or set by the
user) while the stale field is `true`, Spark skips the `UnsafeRow` conversion,
the reader emits `ColumnarBatchRow`, and `collect()` fails:
**What you expected:**
- The reader type is decided from the schema of the scan being built, as
`ParquetFileFormat.buildReaderWithPartitionValues` does with
`ParquetUtils.isBatchReadSupportedForSchema(sqlConf, resultSchema)` and as Hudi
0.14's `Spark33LegacyHoodieParquetFileFormat` did with
`resultSchema.forall(_.dataType.isInstanceOf[AtomicType])`. A nested schema
with `enableNestedColumnVectorizedReader=false` is never handed to the
vectorized reader, regardless of what ran earlier in the session.
- Hudi does not mutate `spark.sql.parquet.enableVectorizedReader` on the
session.
- No `ClassCastException`.
-
**Steps to reproduce:**
Spark 3.3, default `spark_catalog` (Hive Metastore or in-memory). Do **not**
set `spark.sql.catalog.spark_catalog=HoodieCatalog` (see "Why Hudi's own SQL
suite does not catch it" above).
```
spark-shell \
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
--conf
spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
--conf spark.sql.parquet.enableVectorizedReader=true \
--conf spark.sql.parquet.enableNestedColumnVectorizedReader=false \
--jars hudi-spark3.3-bundle_2.12-<version>.jar
```
1. Create a COW table whose nested leaf count exceeds
`spark.sql.codegen.maxFields` (100) and insert two rows:
```scala
val fields = (1 to 128).map(i => s"f$i int").mkString(", ")
spark.sql(s"create table t (id int, ts long, name string, payload
struct<$fields>) using hudi " +
s"tblproperties (primaryKey = 'id', preCombineField = 'ts')
location '/tmp/hudi_t'")
val ns = (1 to 128).map(i => s"'f$i', $i").mkString(", ")
spark.sql(s"insert into t select 1, 1L, 'a', named_struct($ns)")
spark.sql(s"insert into t select 2, 1L, 'b', named_struct($ns)")
```
2. Run a narrow query. Required schema is empty, `supportBatch` runs and
stores `supportVectorizedRead = true`:
```scala
spark.sql("select count(*) from t").show()
spark.conf.get("spark.sql.parquet.enableVectorizedReader") // "true"
```
3. Run the wide nested query. `isTooManyFields` is true, `supportBatch` is
skipped, the stale `true` is reused:
```scala
spark.sql("select * from t").explain() // Scan HudiFileGroup ...
Batched: false
spark.sql("select * from t").collect() // succeeds, but is read with
VectorizedParquetRecordReader
```
4. Run the same query with the conf at `false` when Spark evaluates
`needsUnsafeRowConversion`. The explicit `set` stands in for an earlier wide
scan on another Hudi table, or on this table after `REFRESH TABLE`, both of
which write `"false"` at L349:
```scala
spark.conf.set("spark.sql.parquet.enableVectorizedReader", "false")
spark.sql("select * from t").collect()
// java.lang.ClassCastException: class
org.apache.spark.sql.vectorized.ColumnarBatchRow cannot be cast to
// class org.apache.spark.sql.catalyst.expressions.UnsafeRow
```
5. Order control: in a fresh session run step 3 before step 2. Both queries
pass and `spark.conf.get("spark.sql.parquet.enableVectorizedReader")` reads
`"false"` afterwards, because the wide scan wrote the stored `false` into the
session.
To watch the decision being skipped, enable DEBUG for
`org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat`:
exactly one `supportReturningBatch: ... supportVectorizedRead: true` line
appears, for `count(*)`, and none for `select *`.
A ScalaTest that reproduces this deterministically is attached and can go
into the PR. It pulls the file format from the analyzed plan and checks
identity with `eq`, reads the stored field by reflection, checks
`FileSourceScanExec.supportsColumnar`, checks the conf after each statement,
runs the reverse-order control, and includes the `ClassCastException` step.
Observed values on release-1.2:
| Step | Check | Actual |
|---|---|---|
| same relation | file format object for `count(*)` eq file format object
for `select *` / relation present in `SessionCatalog.tableRelationCache` |
`true` / `true` |
| fresh | `supportVectorizedRead` before any execution | `false` |
| guard | `WholeStageCodegenExec.isTooManyFields` for `select *` (128 nested
fields, limit 100) | `true` |
| after `count(*)` | `supportVectorizedRead` / session conf | `true` /
`"true"` |
| `select *` | `FileSourceScanExec.supportsColumnar` | `false` |
| after `select *` | `supportVectorizedRead` / session conf | `true` /
`"true"` (succeeded only because Spark still converted rows) |
| conf set `"false"`, then `select *` | | `ClassCastException` |
### Environment
**Hudi version:** master
**Query engine:** (Spark/Flink/Trino etc)
Spark
**Relevant configs:**
### Logs and Stack Trace
_No response_
--
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]
