hudi-agent commented on code in PR #19914:
URL: https://github.com/apache/hudi/pull/19914#discussion_r3992078084
##########
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:
π€ Going straight to stock `ParquetFileFormat` drops everything Hudi's own
parquet reader layers on top: `internalSchemaOpt` (schema-on-read renames/type
changes), `tableSchemaAsMessageType`-driven type promotion (intβlong etc. for
files written with an older schema), `mandatoryFields` partition columns that
must be read from the file rather than the path (timestamp/custom keygen,
precombine-as-partition β and `vectorTypes` still declares those as
`OnHeapColumnVector` while the stock reader emits `ConstantColumnVector`, so
codegen would CCE), and the VECTOR BinaryType rewrite. Have you considered
routing this case through `readBaseFile(file, baseFileReader.value, ...)`
instead? That already skips `HoodieFileGroupReader` while keeping the Hudi
reader semantics β it'd be useful to know how much of the benchmark win is from
bypassing the FG reader vs. bypassing the Hudi parquet reader.
<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:
##########
@@ -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:
π€ One thing to double-check: the incremental `requiredFilters`
(`_hoodie_commit_time IN (...)`) only get a plan-level `Filter` via
`HoodieAnalysis.maybeApplyForNewFileFormat`, i.e. when
`HoodieSparkSessionExtension` is registered. Without the extension the
row-based `ParquetRecordReader` + `FilterCompat` was doing record-level
filtering, but the vectorized reader only prunes row groups/pages β so a
rewritten COW base file containing rows from several commits could leak
out-of-range rows. Is DataFrame incremental read without the extension still a
supported path? @yihua might be able to confirm.
<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:
##########
@@ -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:
π€ This gate doesn't check `hoodieFileFormat`, so a COW table with ORC (or
Lance/Vortex/multi-format) base files will be handed to `ParquetFileFormat` and
fail. Also note `HoodieCopyOnWriteSnapshotHadoopFsRelationFactory` builds its
`HoodieFileIndex` with `shouldEmbedFileSlices = true`, so
`file.partitionValues` is a `HoodiePartitionFileSliceMapping` for COW snapshot
reads as well β this bypass ends up on every COW snapshot scan, not just
incremental. Could you scope it to `hoodieFileFormat == PARQUET &&
!isMultipleBaseFileFormatsEnabled` at minimum, and update the description to
reflect the snapshot impact?
<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:
##########
@@ -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 splitting now enabled for incremental, the `HoodieFileGroupReader`
branch below still does `.withStart(file.start).withLength(baseFileLength)`
(full file size, not `file.length`), so if a split ever lands there (e.g. once
the bypass is scoped to Parquet and an ORC/Lance COW incremental read falls
through) the reader range would be `[start, start + fileSize)` and overlap the
next split. Would it make sense to pass `file.length` there, or keep
incremental non-splittable unless the bypass is taken?
<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]
