prashantwason commented on code in PR #19914:
URL: https://github.com/apache/hudi/pull/19914#discussion_r4215676334
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala:
##########
@@ -145,6 +148,11 @@ private[parquet] class
HoodieFileGroupReaderFunction(baseFileReader: Broadcast[S
val fileGroupName = FSUtils.getFileIdFromFilePath(sparkAdapter
.getSparkPartitionedFileUtils.getPathFromPartitionedFile(file))
fileSliceMapping.getSlice(fileGroupName) match {
+ case Some(fileSlice) if state.readBaseOnlySlicesDirectly &&
!fileSlice.getLogFiles.findAny().isPresent =>
Review Comment:
You're right — `PartitionDirectoryConverter` only embeds a mapping for
slices with log files or a bootstrap base, so that branch was unreachable (and
so was the stock-reader bypass in the first revision). Removed the branch and
the `readBaseOnlySlicesDirectly` state in 0f3b14202463, and updated the title
and description to describe the change as vectorized reading + splitting for
COW incremental. The description now notes that the first-revision numbers came
from vectorization and splitting only; the benchmark will be re-run on the
current commit.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -262,8 +273,11 @@ 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
- logDebug(s"isSplitable: $splitable, super.isSplitable: $superSplitable,
isMOR: $isMOR, isIncremental: $isIncremental, isBootstrap: $isBootstrap")
+ // COW incremental reads have no log files to merge, so file splitting is
safe.
+ // MOR, CDC and bootstrap reads need to disable splitting.
+ val splitable = !isMOR && !isCDC && !isBootstrap && !isLance &&
superSplitable
Review Comment:
Added `testCowIncrementalReadWithMultipleSplitsPerFile` in 0f3b14202463,
parameterized by PARQUET and ORC. It writes four commits into one file group
with a 64KB block/stripe size, reads with
`spark.sql.files.maxPartitionBytes=65536`, asserts the scan has more input
partitions than files, and checks two ranges (one filtering out carried-over
rows of an earlier commit, one returning only the latest update) against both
the expected rows and the unsplit read.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala:
##########
@@ -349,6 +360,34 @@ private[parquet] object HoodieFileGroupReaderFunction {
iter.map(mapper.apply(_))
}
+ /**
+ * Evaluates `requiredFilters` on every row. Pushed-down filters only prune
row groups in the vectorized reader,
+ * and the plan-level filter for them is only added when the Hudi session
extension is registered, so the
+ * commit-time range of an incremental query is enforced here. The scan
returns rows rather than batches
+ * whenever required filters are present.
+ */
+ private def applyRequiredFilters(iter: Iterator[InternalRow],
Review Comment:
Done in 0f3b14202463. `HoodieFormatTrait` has a new
`requiredFiltersAppliedInPlan` flag, set alongside `isProjected` in
`applyNewFileFormatChanges` and in `createStreamingDataFrame` (Spark 3 and
Spark 4 adapters). When it is set, `supportBatch` keeps returning batches and
`readBaseFile` skips the per-row predicate; the required filters are still
pushed down for row-group pruning. When the predicate is evaluated, an `In`
over literals is rewritten to `InSet` before `Predicate.create`.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderFunction.scala:
##########
@@ -250,7 +260,7 @@ private[parquet] class
HoodieFileGroupReaderFunction(baseFileReader: Broadcast[S
//and we pass an empty schema for the partition schema
val iter = parquetFileReader.read(modifiedFile, readSchema, new
StructType(), state.internalSchemaOpt, filters, conf,
state.tableSchemaAsMessageType)
- projectIfNeeded(iter, readSchema, schemas.outputSchema)
+ projectIfNeeded(applyRequiredFilters(iter, readSchema, requiredFilters),
readSchema, schemas.outputSchema)
Review Comment:
Added `testCowIncrementalReadWithPartitionValuesReadFromFile` in
0f3b14202463, parameterized by key generator: `TimestampBasedKeyGenerator`
(every partition column read from the file, the second `readBaseFile` branch)
and `CustomKeyGenerator` with `pt:SIMPLE,event_ts:TIMESTAMP` (one partition
column read from the file, one appended from the path, the third branch). Both
run on COW without the session extension, over a base file that carries rows of
earlier commits, and check three ranges including the partition column values.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala:
##########
@@ -207,10 +210,18 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
} else {
throw new HoodieNotSupportedException("Unsupported file format: " +
hoodieFileFormat)
}
- supportVectorizedRead = !isIncremental && !isBootstrap && supportBatch
- supportReturningBatch = !isMOR && supportVectorizedRead
+ // MOR incremental embeds file slices that may contain log files
requiring row-level
+ // merging, and CDC rows are built by a row-based iterator that reads
base files through
+ // the same reader, so vectorized reading must be disabled for both. 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 = !(isIncremental && (isMOR || isCDC)) &&
!isBootstrap && supportBatch
Review Comment:
Parameterized `testIncrementalReadRanges` by base file format in
0f3b14202463 and added COW ORC rows for table versions 6 and 8; the new split
test also runs on ORC. One caveat: locally the ORC cases (like the existing
`TestHoodieMultipleBaseFileFormat.testMultiFileFormatForMORTableType`) fail
with `NoSuchMethodError: Reader$Options.useSelected` from a mismatched ORC jar
on my test classpath, so I'm relying on CI for the ORC cells.
--
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]
