yihua commented on code in PR #19898:
URL: https://github.com/apache/hudi/pull/19898#discussion_r4077686634
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala:
##########
@@ -124,11 +124,9 @@ case class MergeOnReadIncrementalRelationV2(override val
sqlContext: SQLContext,
val fsView = new HoodieTableFileSystemView(metaClient, timeline,
affectedFilesInCommits)
val modifiedPartitions = getWritePartitionPaths(commitsMetadata)
-
fileIndex.listMatchingPartitionPaths(HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient,
partitionFilters))
- .map(p => p.getPath).filter(p => modifiedPartitions.contains(p))
- .flatMap { relativePartitionPath =>
- fsView.getLatestMergedFileSlicesBeforeOrOn(relativePartitionPath,
latestCommit).iterator().asScala
- }
+ modifiedPartitions.asScala.flatMap { relativePartitionPath =>
Review Comment:
Confirmed this one by running it: on master an incremental read of two
commits (one per partition) with `.where("partition = '2022-01-02'")` returns
30 rows, on this branch it returns all 50 for both COW and MOR, and the plan
shows the predicate only under `PartitionFilters` with no post-scan `Filter`,
so Spark is trusting the index. Could we keep the predicate evaluation but
apply it to `modifiedPartitions` only (convert them to `PartitionPath`s and
reuse the same pruning that `listMatchingPartitionPaths` does), rather than
dropping it? That keeps the O(modified) listing this PR is after without
changing results.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieIncrementalFileIndex.scala:
##########
@@ -43,6 +43,25 @@ class HoodieIncrementalFileIndex(override val spark:
SparkSession,
spark, metaClient, schemaSpec, options, fileStatusCache, includeLogFiles,
shouldEmbedFileSlices = true
) with FileIndex {
+ // Skip the Spark optimizer's partition pruning rule (e.g.
Spark33HoodiePruneFileSourcePartitions)
+ // which would trigger a full-table partition listing via the base class.
The incremental file
+ // index already selects only modified file groups via listFileSplits().
+ hasPushedDownPartitionPredicates = true
+
+ override def filterFileSlices(dataFilters: Seq[Expression],
partitionFilters: Seq[Expression], isPartitionPruned: Boolean = false)
+ : Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])] = {
Review Comment:
This does not compile against master: `BaseHoodieTableFileIndex` is not
imported here (it lives in `org.apache.hudi.core.read`), which is what all the
CI jobs are failing on at lines 52 and 58. Adding `import
org.apache.hudi.core.read.BaseHoodieTableFileIndex` fixes the build locally.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala:
##########
@@ -124,11 +124,9 @@ case class MergeOnReadIncrementalRelationV2(override val
sqlContext: SQLContext,
val fsView = new HoodieTableFileSystemView(metaClient, timeline,
affectedFilesInCommits)
val modifiedPartitions = getWritePartitionPaths(commitsMetadata)
-
fileIndex.listMatchingPartitionPaths(HoodieFileIndex.convertFilterForTimestampKeyGenerator(metaClient,
partitionFilters))
- .map(p => p.getPath).filter(p => modifiedPartitions.contains(p))
- .flatMap { relativePartitionPath =>
- fsView.getLatestMergedFileSlicesBeforeOrOn(relativePartitionPath,
latestCommit).iterator().asScala
- }
+ modifiedPartitions.asScala.flatMap { relativePartitionPath =>
Review Comment:
non-blocking: the existing partition-filter checks in
TestCOWDataSource/TestMORDataSource only filter on a partition that is outside
the incremental range, so they pass either way. It would be worth adding a case
where the range touches two partitions and the query filters on one of them,
for both COW and MOR, so this path stays covered.
--
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]
