yihua commented on code in PR #19914:
URL: https://github.com/apache/hudi/pull/19914#discussion_r4077740886
##########
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:
Confirming this: CI has
`TestIncrementalReadWithFileGroupReader.testIncrementalReadRanges` failing for
all COW rows with records from earlier commits leaking into the range
(`expected Set((k3,2),(k4,2)) but was Set((k1,1),(k2,1),(k3,2),(k4,2))`), and
that suite deliberately runs without the extension, where the plan-level
commit-time Filter is never added. The vectorized/stock path only uses
`requiredFilters` for row-group and page pruning, so the row-level guarantee
the previous non-vectorized path gave is gone; note the stock reader is also
handed `requiredSchema` rather than `readRequiredSchema`, so
`_hoodie_commit_time` is not even guaranteed to be in the projection. Routing
this case through `readBaseFile(file, baseFileReader.value, ...)` as suggested
above, or wrapping the bypass output with the `requiredFilters` predicate,
would keep the semantics.
##########
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:
The COW CDC relation comes through here with `isMOR = false, isIncremental =
true` (its factory extends the COW incremental base factory), so this change
also turns on vectorized reading and batch output for CDC, whose
`CDCFileGroupIterator` reads base and native CDC files row by row. That looks
like what is behind the CDC failures in CI
(`TestCDCDataFrameSuite.testCOWDataSourceWrite` returns 0 of 100 rows, plus
`TestCDCStreamingSuite`, `TestCDCForSparkSQL` and the CDC variant tests), all
of which pass on master. Could the vectorized/splittable gates exclude the CDC
path explicitly, e.g. by keying off the CDC file index or a dedicated flag from
the factory, rather than inferring "COW incremental" from `!isMOR &&
isIncremental`?
--
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]
