You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "prashantwason (via GitHub)" <gi...@apache.org> on 2026/10/06 18:05:17 UTC
[PR] perf(spark): check file existence in parallel in MOR incremental fallback scan [hudi]
prashantwason opened a new pull request, #20211:
URL: https://github.com/apache/hudi/pull/20211
### Describe the issue this Pull Request addresses
With `hoodie.datasource.read.incr.fallback.fulltablescan.enable=true`, the `fullTableScan` lazy val in `HoodieIncrementalRelationV1Trait` and `HoodieIncrementalRelationV2Trait` checks whether any file touched by the incremental commits has been deleted. It does this by calling `storage.exists()` on each file, one at a time, on the driver:
```scala
affectedFilesInCommits.asScala.exists(fileStatus => !metaClient.getStorage.exists(fileStatus.getPath))
```
These traits back the MOR incremental relations, and also COW tables with table version < 8, which are read through `MergeOnReadIncrementalRelationV1` via `HoodieCopyOnWriteIncrementalHadoopFsRelationFactory`. In a production pipeline reading 10 commits that touched about 51K files on GCS, this made about 51K sequential HEAD requests (about 20 ms each), so the driver stalled for about 17 minutes with no log output.
`IncrementalRelationV1` / `IncrementalRelationV2` already run this check in parallel with `sparkContext.parallelize`, but the MOR traits never got that change.
### Summary and Changelog
The existence check in the MOR incremental fallback scan now runs across executors instead of one file at a time on the driver.
- Add `HoodieBaseRelation.checkIfAnyFilesMissing`, which spreads the `exists()` calls over `min(fileCount, defaultParallelism)` partitions and gives the Spark job a description (`Checking existence of N files for incremental fallback scan`) so it is easy to find in the Spark UI. The previous job description is restored afterwards.
- `HoodieIncrementalRelationV1Trait.fullTableScan` and `HoodieIncrementalRelationV2Trait.fullTableScan` use it. They skip the check when fallback is disabled or the start/end instant is archived, and log when they fall back because files are missing.
### Impact
Performance only, with no API or behavior change. The fallback check now takes about as long as a single Spark job with `defaultParallelism` tasks, instead of growing with the number of affected files on the driver.
### Risk Level
low. Only how the existence check is computed changes; when the check runs and how its result is used stay the same. Verified against the production pipeline above: the check now shows up as one Spark job and the 17-minute driver stall is gone.
### Documentation Update
none
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Enough context is provided in the sections above
- [ ] Adequate tests were added if applicable
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(spark): check file existence in parallel in MOR incremental fallback scan [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua merged PR #20211:
URL: https://github.com/apache/hudi/pull/20211
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(spark): check file existence in parallel in MOR incremental fallback scan [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #20211:
URL: https://github.com/apache/hudi/pull/20211#discussion_r4198981763
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ filePaths: java.util.List[StoragePathInfo]): Boolean = {
+ val sc = sqlContext.sparkContext
+ val allFilesToCheck = filePaths.asScala.map(_.getPath.toString)
+ val storageConf = HadoopFSUtils.getStorageConfWithCopy(sc.hadoopConfiguration)
+ val localBasePathStr = metaClient.getBasePath.toString
+ val numPartitions = Math.max(1, Math.min(allFilesToCheck.size, sc.defaultParallelism))
+ val previousJobDescription = sc.getLocalProperty("spark.job.description")
+ try {
+ sc.setJobDescription(s"Checking existence of ${allFilesToCheck.size} files for incremental fallback scan")
+ val missingFileFound = sc.parallelize(allFilesToCheck.toSeq, numPartitions)
+ .map(path => {
+ val storage = HoodieStorageUtils.getStorage(localBasePathStr, storageConf)
Review Comment:
Should we use `mapPartitions` so that file paths in the same Spark partition trigger a single storage instantiation call?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(spark): check file existence in parallel in MOR incremental fallback scan [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #20211:
URL: https://github.com/apache/hudi/pull/20211#discussion_r4198933284
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ filePaths: java.util.List[StoragePathInfo]): Boolean = {
+ val sc = sqlContext.sparkContext
+ val allFilesToCheck = filePaths.asScala.map(_.getPath.toString)
+ val storageConf = HadoopFSUtils.getStorageConfWithCopy(sc.hadoopConfiguration)
Review Comment:
Could this use `metaClient.getStorageConf` rather than `sc.hadoopConfiguration`? The streaming source builds its meta client from `sessionState.newHadoopConf()`, so session-scoped `fs.*` settings that the old driver-side `metaClient.getStorage.exists` saw would not reach the executors here.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
Review Comment:
non-blocking: `IncrementalRelationV1` and `IncrementalRelationV2` still carry an inline copy of this check using `parallelize(allFilesToCheck.toSeq, allFilesToCheck.size)`, i.e. one task per file. It might be worth pointing them at `checkIfAnyFilesMissing` so they get the capped parallelism too.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(spark): check file existence in parallel in MOR incremental fallback scan [hudi]
Posted by "prashantwason (via GitHub)" <gi...@apache.org>.
prashantwason commented on code in PR #20211:
URL: https://github.com/apache/hudi/pull/20211#discussion_r4199277191
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ filePaths: java.util.List[StoragePathInfo]): Boolean = {
+ val sc = sqlContext.sparkContext
+ val allFilesToCheck = filePaths.asScala.map(_.getPath.toString)
+ val storageConf = HadoopFSUtils.getStorageConfWithCopy(sc.hadoopConfiguration)
Review Comment:
Done in 47c4fd1cb40d. `checkIfAnyFilesMissing` now ships `metaClient.getStorageConf` to the executors instead of `sc.hadoopConfiguration`, so session-scoped `fs.*` settings are honored.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
Review Comment:
Done in 47c4fd1cb40d. `IncrementalRelationV1` and `IncrementalRelationV2` now call `HoodieBaseRelation.checkIfAnyFilesMissing`, so they get the capped parallelism, the `mapPartitions` change and the meta client storage conf. The helper now takes path strings, so it works for both the MOR relations and these.
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala:
##########
@@ -796,4 +796,32 @@ object HoodieBaseRelation extends SparkAdapterSupport {
DataSourceReadOptions.SCHEMA_EVOLUTION_ENABLED.defaultValue.toString).toBoolean ||
ProvidesHoodieConfig.isSchemaEvolutionEnabled(sparkSession)
}
+
+ /**
+ * Checks in parallel whether any of the given file paths are missing from storage.
+ * Uses sparkContext.parallelize to distribute the exists() checks across executors,
+ * avoiding sequential driver-side RPCs that can take minutes on cloud storage for
+ * large file counts.
+ */
+ def checkIfAnyFilesMissing(sqlContext: SQLContext,
+ metaClient: HoodieTableMetaClient,
+ filePaths: java.util.List[StoragePathInfo]): Boolean = {
+ val sc = sqlContext.sparkContext
+ val allFilesToCheck = filePaths.asScala.map(_.getPath.toString)
+ val storageConf = HadoopFSUtils.getStorageConfWithCopy(sc.hadoopConfiguration)
+ val localBasePathStr = metaClient.getBasePath.toString
+ val numPartitions = Math.max(1, Math.min(allFilesToCheck.size, sc.defaultParallelism))
+ val previousJobDescription = sc.getLocalProperty("spark.job.description")
+ try {
+ sc.setJobDescription(s"Checking existence of ${allFilesToCheck.size} files for incremental fallback scan")
+ val missingFileFound = sc.parallelize(allFilesToCheck.toSeq, numPartitions)
+ .map(path => {
+ val storage = HoodieStorageUtils.getStorage(localBasePathStr, storageConf)
Review Comment:
Done in 47c4fd1cb40d. Switched to `mapPartitions`: storage is created once per Spark partition, and each partition stops at the first missing file (`forall`).
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] perf(spark): check file existence in parallel in MOR incremental fallback scan [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20211:
URL: https://github.com/apache/hudi/pull/20211#issuecomment-6023179276
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20211?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 80.36%. Comparing base ([`3b5bb09`](https://app.codecov.io/gh/apache/hudi/commit/3b5bb09b1d365f5e31cf30a8dfdd0a48ec96b2b0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`3649bc9`](https://app.codecov.io/gh/apache/hudi/commit/3649bc9685522d82870ee056f51e157b091cd0f8?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20211 +/- ##
============================================
- Coverage 80.61% 80.36% -0.26%
+ Complexity 35035 34936 -99
============================================
Files 2552 2552
Lines 143281 143277 -4
Branches 17430 17426 -4
============================================
- Hits 115506 115143 -363
- Misses 19867 20206 +339
- Partials 7908 7928 +20
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.11% <ø> (-0.06%)` | :arrow_down: |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.75% <ø> (-0.07%)` | :arrow_down: |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.33% <ø> (-1.65%)` | :arrow_down: |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.98% <ø> (+<0.01%)` | :arrow_up: |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.18% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.78% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `71.08% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.25% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.52% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.06% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20211/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `53.20% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20211/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [flink-integration-tests](https://app.codecov.io/gh/apache/hudi/pull/20211/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `36.40% <ø> (-13.15%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi/pull/20211?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...ain/scala/org/apache/hudi/HoodieBaseRelation.scala](https://app.codecov.io/gh/apache/hudi/pull/20211?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark-common%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fhudi%2FHoodieBaseRelation.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmstY29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvaHVkaS9Ib29kaWVCYXNlUmVsYXRpb24uc2NhbGE=) | `79.06% <ø> (ø)` | |
| [...apache/hudi/MergeOnReadIncrementalRelationV1.scala](https://app.codecov.io/gh/apache/hudi/pull/20211?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark-common%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fhudi%2FMergeOnReadIncrementalRelationV1.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmstY29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvaHVkaS9NZXJnZU9uUmVhZEluY3JlbWVudGFsUmVsYXRpb25WMS5zY2FsYQ==) | `81.48% <ø> (+1.48%)` | :arrow_up: |
| [...apache/hudi/MergeOnReadIncrementalRelationV2.scala](https://app.codecov.io/gh/apache/hudi/pull/20211?src=pr&el=tree&filepath=hudi-spark-datasource%2Fhudi-spark-common%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fhudi%2FMergeOnReadIncrementalRelationV2.scala&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aHVkaS1zcGFyay1kYXRhc291cmNlL2h1ZGktc3BhcmstY29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvaHVkaS9NZXJnZU9uUmVhZEluY3JlbWVudGFsUmVsYXRpb25WMi5zY2FsYQ==) | `88.57% <ø> (-0.22%)` | :arrow_down: |
... and [51 files with indirect coverage changes](https://app.codecov.io/gh/apache/hudi/pull/20211/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org