prashantwason commented on code in PR #19914:
URL: https://github.com/apache/hudi/pull/19914#discussion_r4210903253
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -347,6 +353,21 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
}
val broadcastedStorageConf = spark.sparkContext.broadcast(new
SerializableConfiguration(augmentedStorageConf.unwrap()))
+
+ // Build stock Spark ParquetFileFormat reader for COW base-file-only
bypass.
+ // When a file slice has only a base file (no log files), this avoids the
overhead of
+ // HoodieFileGroupReader (schema handler, record merger, row copy/seal) by
using Spark's
+ // native parquet reader directly.
+ val stockParquetReader: Option[PartitionedFile => Iterator[InternalRow]] =
if (!isMOR && !isBootstrap) {
Review Comment:
Addressed in b7a2388e7b94: the stock `ParquetFileFormat` bypass is removed.
No-log COW slices now go through `readBaseFile` with the Hudi `baseFileReader`,
which handles ORC, Lance, Vortex and multi-format tables. You're right that
this also covers COW snapshot reads; the PR description now calls that out.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -366,6 +387,12 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
.getSparkPartitionedFileUtils.getPathFromPartitionedFile(file))
fileSliceMapping.getSlice(fileGroupName) match {
case Some(fileSlice) if !isCount && (requiredSchema.nonEmpty ||
fileSlice.getLogFiles.findAny().isPresent) =>
+ if (!fileSlice.getLogFiles.findAny().isPresent && !isBootstrap
&& broadcastStockReader.isDefined) {
Review Comment:
Addressed in b7a2388e7b94: this case is now routed through
`readBaseFile(file, state, conf)` as suggested, so the Hudi reader's handling
of `internalSchemaOpt`, `tableSchemaAsMessageType` type promotion, mandatory
partition fields and VECTOR columns is kept. The benchmark will be re-run on
this path.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -218,7 +218,11 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
} else {
throw new HoodieNotSupportedException("Unsupported file format: " +
hoodieFileFormat)
}
- supportVectorizedRead = !isIncremental && !isBootstrap && supportBatch
+ // MOR incremental embeds file slices that may contain log files
requiring row-level
+ // merging, so vectorized reading must be disabled. All other
combinations (COW snapshot,
+ // COW incremental, MOR snapshot) either have no log merging or handle
it via a separate
+ // non-vectorized fileGroupBaseFileReader while the base file reader
stays vectorized.
+ supportVectorizedRead = !(isMOR && isIncremental) && !isBootstrap &&
supportBatch
Review Comment:
Fixed in b7a2388e7b94. The bypass is removed, and no-log COW slices are read
by `readBaseFile`, which reads `readRequiredSchema`, so `_hoodie_commit_time`
is always in the projection. Since the vectorized reader only uses pushed
filters for row-group pruning, `supportBatch` now returns rows rather than
batches whenever `requiredFilters` is non-empty (decoding stays vectorized),
and `readBaseFile` evaluates `requiredFilters` on every row before projecting.
`TestIncrementalReadWithFileGroupReader` (including
`testIncrementalReadRanges`, without the session extension) passes locally.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -273,7 +277,9 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
// For overly large single files, we can use multiple concurrent tasks to
read them, thereby reducing the overall job reading time consumption
val superSplitable = super.isSplitable(sparkSession, options, path)
val isLance = hoodieFileFormat == HoodieFileFormat.LANCE
- val splitable = !isMOR && !isIncremental && !isBootstrap && !isLance &&
superSplitable
+ // COW incremental reads have no log files to merge, so file splitting is
safe.
+ // Only MOR and bootstrap reads need to disable splitting.
+ val splitable = !isMOR && !isBootstrap && !isLance && superSplitable
Review Comment:
With b7a2388e7b94 a split file can no longer reach that branch: every COW
slice without log files (COW slices never have log files) is read by
`readBaseFile`, which honors `file.start`/`file.length`, and bootstrap, MOR and
CDC reads remain non-splittable. So the `HoodieFileGroupReader` branch only
sees unsplit files, and I left its length handling unchanged.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -218,7 +218,11 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
} else {
throw new HoodieNotSupportedException("Unsupported file format: " +
hoodieFileFormat)
}
- supportVectorizedRead = !isIncremental && !isBootstrap && supportBatch
+ // MOR incremental embeds file slices that may contain log files
requiring row-level
+ // merging, so vectorized reading must be disabled. All other
combinations (COW snapshot,
+ // COW incremental, MOR snapshot) either have no log merging or handle
it via a separate
+ // non-vectorized fileGroupBaseFileReader while the base file reader
stays vectorized.
+ supportVectorizedRead = !(isMOR && isIncremental) && !isBootstrap &&
supportBatch
Review Comment:
Good catch, thanks. Fixed in bc0ec0eb3b86 with a dedicated flag as you
suggested: `HoodieFileGroupReaderBasedFileFormat` takes a new `isCDC` argument
(default `false`), set to `true` by
`HoodieCopyOnWriteCDCHadoopFsRelationFactory` and
`HoodieMergeOnReadCDCHadoopFsRelationFactory`. CDC reads are excluded from both
the vectorized and the splittable gates. `TestCDCForSparkSQL`,
`TestCDCDataFrameSuite` and `TestCDCStreamingSuite` pass locally.
--
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]
