hudi-agent commented on code in PR #20183:
URL: https://github.com/apache/hudi/pull/20183#discussion_r4235868604
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -346,10 +341,13 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
val dataStructTypeWithMandatoryPartitionFields =
StructType(dataStructType.fields ++ partitionSchema.fields.filter(f =>
mandatoryFields.contains(f.name) && !isNestedPartitionField(f.name)))
val dataSchema = HoodieSchemaUtils.pruneDataSchema(schema,
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(dataStructTypeWithMandatoryPartitionFields,
sanitizedTableName), exclusionFields)
-
spark.sessionState.conf.setConfString("spark.sql.parquet.enableVectorizedReader",
supportVectorizedRead.toString)
+ // Decide per scan: Spark skips supportBatch for scans wider than
spark.sql.codegen.maxFields.
+ // Never write it to spark.sql.parquet.enableVectorizedReader: Spark uses
that conf to decide whether to
+ // convert the scan output to UnsafeRow, and skipping the conversion for
vectorized rows fails.
+ val vectorizedRead = canReadVectorized(spark, outputSchema)
Review Comment:
🤖 It looks like this also moves wide scans from the row reader to the
vectorized one. On master a fresh relation never gets `supportBatch` for them,
so `select *` or a full-row UPDATE/MERGE on a table wider than
`spark.sql.codegen.maxFields` always read row-based. Now, if an older base file
still has a nested type promotion (e.g. `struct<a:int>` → `struct<a:long>`,
`array<int>` → `array<long>`), that read fails on the "Nested types with type
changes ... cannot be read in vectorized mode" check in
`ParquetSchemaEvolutionUtils.getHadoopConfClone` instead of succeeding. Would
it make sense to gate this the way Spark does, so wide scans stay row-based and
`supportsColumnar` still implies `vectorizedRead`?
```scala
val conf = spark.sessionState.conf
val vectorizedRead = canReadVectorized(spark, outputSchema) &&
conf.wholeStageEnabled && !WholeStageCodegenExec.isTooManyFields(conf,
outputSchema)
```
@yihua could you weigh in on which behavior we want for wide scans here?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -212,26 +193,40 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
s.dataType,
sparkSession.sessionState.conf.orcVectorizedReaderNestedColumnEnabled))
// TODO: Implement columnar batch reading
https://github.com/apache/hudi/issues/17736
val lanceBatchSupported = false
+ // The Vortex reader returns rows only
+ val vortexBatchSupported = false
- val supportBatch = if (isMultipleBaseFileFormatsEnabled) {
+ val formatSupportsBatch = if (isMultipleBaseFileFormatsEnabled) {
parquetBatchSupported && orcBatchSupported
} else if (hoodieFileFormat == HoodieFileFormat.PARQUET) {
parquetBatchSupported
} else if (hoodieFileFormat == HoodieFileFormat.ORC) {
orcBatchSupported
} else if (hoodieFileFormat == HoodieFileFormat.LANCE) {
lanceBatchSupported
+ } else if (hoodieFileFormat == HoodieFileFormat.VORTEX) {
+ vortexBatchSupported
} else {
throw new HoodieNotSupportedException("Unsupported file format: " +
hoodieFileFormat)
}
- supportVectorizedRead = !isIncremental && !isBootstrap && supportBatch
- supportReturningBatch = !isMOR && supportVectorizedRead
- logDebug(s"supportReturningBatch: $supportReturningBatch,
supportVectorizedRead: $supportVectorizedRead, isIncremental: $isIncremental, "
+
- s"isBootstrap: $isBootstrap, superSupportBatch: $supportBatch")
- supportReturningBatch
+ !isIncremental && !isBootstrap && formatSupportsBatch
}
}
+ /**
+ * Checks if the file format supports returning columnar batches, please
refer to SPARK-40918.
+ *
+ * NOTE: for MOR, a file slice with only a base file can still be read with
the vectorized reader,
+ * but the scan as a whole cannot return batches because slices with log
files are merged row by row.
+ */
+ override def supportBatch(sparkSession: SparkSession, schema: StructType):
Boolean = {
+ val vectorized = canReadVectorized(sparkSession, schema)
+ val returningBatch = !isMOR && vectorized
+ logDebug(s"supportReturningBatch: $returningBatch, supportVectorizedRead:
$vectorized, " +
Review Comment:
🤖 nit: could you rename the log keys to `returningBatch` and
`vectorizedRead` (or similar)? `supportReturningBatch` and
`supportVectorizedRead` are the names of the fields this PR removes, so anyone
grepping for them will be sent the wrong way.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
