hudi-agent commented on code in PR #19898:
URL: https://github.com/apache/hudi/pull/19898#discussion_r3983854441
##########
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
Review Comment:
🤖 Setting this in the constructor means
`Spark3xHoodiePruneFileSourcePartitions` never runs for this index, so
`sizeInBytes` is now computed from an empty `cachedAllInputFileSlices`
(`getTotalCachedFilesSize` == 0) at planning time — with the Hudi extension
enabled that reports a 0-byte relation, which can make the optimizer broadcast
an arbitrarily large incremental scan. Have you checked what stats the planner
sees for an incremental read after this change?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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:
🤖 Dropping `listMatchingPartitionPaths(...)` here also drops the
partition-predicate filtering it was doing (the old code intersected the pruned
partitions with `modifiedPartitions`), so `listFileSplits` now returns every
modified partition regardless of `partitionFilters`. Since
`HoodieIncrementalFileIndex.partitionSchema` is non-empty for a partitioned
table, Spark's `FileSourceStrategy` removes partition-only predicates from
`afterScanFilters` and trusts the index to prune — so an incremental read like
`.load().where("partition_col = 'x'")` could return rows from other partitions.
Would it work to keep the predicate evaluation but apply it only to
`modifiedPartitions` (no full listing) rather than removing it? @yihua could
you sanity-check this one?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
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)
Review Comment:
🤖 This override returns `Option.empty` for the partition path, but the
base-class contract treats it as present (`HoodieFileIndex.listFiles` does
`partitionOpt.get.getValues`), so any caller other than the prune rules would
hit a `NoSuchElementException`. It also silently ignores the third parameter
(renamed here from `isPartitionPruneOnly` to `isPartitionPruned`), meaning the
"safety net" does a full `listFileSplits` during optimization. Would it be
cleaner to preserve the `PartitionPath` from the grouping, or keep the base
name/semantics of the flag?
<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]
