This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new ec2831ec932d perf(spark): check file existence in parallel in MOR
incremental fallback scan (#20211)
ec2831ec932d is described below
commit ec2831ec932d3da193b15fee0164724913d33da5
Author: Prashant Wason <[email protected]>
AuthorDate: Tue Oct 6 16:50:07 2026 -0700
perf(spark): check file existence in parallel in MOR incremental fallback
scan (#20211)
---
.../scala/org/apache/hudi/HoodieBaseRelation.scala | 29 ++++++++++++++++++++++
.../org/apache/hudi/IncrementalRelationV1.scala | 15 +++--------
.../org/apache/hudi/IncrementalRelationV2.scala | 15 +++--------
.../hudi/MergeOnReadIncrementalRelationV1.scala | 14 +++++++++--
.../hudi/MergeOnReadIncrementalRelationV2.scala | 14 +++++++++--
5 files changed, 59 insertions(+), 28 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
index 9d804622e6fd..e7d389d666c3 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
@@ -796,4 +796,33 @@ 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: Iterable[String]): Boolean = {
+ val sc = sqlContext.sparkContext
+ val allFilesToCheck = filePaths.toSeq
+ // Use the meta client's storage conf so that session-scoped fs.* settings
reach the executors
+ val storageConf = metaClient.getStorageConf
+ 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 query")
+ // Instantiate the storage once per Spark partition and stop at the
first missing file
+ sc.parallelize(allFilesToCheck, numPartitions)
+ .mapPartitions(paths => {
+ val storage = HoodieStorageUtils.getStorage(localBasePathStr,
storageConf)
+ Iterator.single(paths.forall(path => storage.exists(new
StoragePath(path))))
+ }).collect().exists(allExist => !allExist)
+ } finally {
+ sc.setJobDescription(previousJobDescription)
+ }
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV1.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV1.scala
index 52b80cd4ac3e..f2e4511e2783 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV1.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV1.scala
@@ -32,11 +32,8 @@ import org.apache.hudi.common.table.timeline.{HoodieInstant,
HoodieTimeline}
import
org.apache.hudi.common.table.timeline.TimelineUtils.{handleHollowCommitIfNeeded,
HollowCommitHandling}
import
org.apache.hudi.common.table.timeline.TimelineUtils.HollowCommitHandling.USE_TRANSITION_TIME
import org.apache.hudi.common.util.{HoodieTimer, InternalSchemaCache}
-import org.apache.hudi.common.util.HoodieStorageUtils
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.hudi.exception.{HoodieException,
HoodieIncrementalPathNotFoundException}
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
-import org.apache.hudi.storage.StoragePath
import org.apache.hudi.table.HoodieSparkTable
import org.apache.hudi.util.IncrementalRelationUtil
@@ -255,18 +252,12 @@ class IncrementalRelationV1(val sqlContext: SQLContext,
if (fallbackToFullTableScan) {
val timer = HoodieTimer.start
- val allFilesToCheck = filteredMetaBootstrapFullPaths ++
filteredRegularFullPaths
- val storageConf =
HadoopFSUtils.getStorageConfWithCopy(sqlContext.sparkContext.hadoopConfiguration)
- val localBasePathStr = basePath.toString
- val firstNotFoundPath =
sqlContext.sparkContext.parallelize(allFilesToCheck.toSeq, allFilesToCheck.size)
- .map(path => {
- val storage = HoodieStorageUtils.getStorage(localBasePathStr,
storageConf)
- storage.exists(new StoragePath(path))
- }).collect().find(v => !v)
+ val missingFileFound = HoodieBaseRelation.checkIfAnyFilesMissing(
+ sqlContext, metaClient, filteredMetaBootstrapFullPaths ++
filteredRegularFullPaths)
val timeTaken = timer.endTimer()
log.info("Checking if paths exists took " + timeTaken + "ms")
- if (firstNotFoundPath.isDefined) {
+ if (missingFileFound) {
doFullTableScan = true
log.info("Falling back to full table scan as some files cannot
be found.")
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV2.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV2.scala
index 49c94c0f8532..48e5ea81094e 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV2.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/IncrementalRelationV2.scala
@@ -30,10 +30,7 @@ import
org.apache.hudi.common.table.log.InstantRange.RangeType
import org.apache.hudi.common.table.read.IncrementalQueryAnalyzer
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.hudi.common.util.{HoodieTimer, InternalSchemaCache}
-import org.apache.hudi.common.util.HoodieStorageUtils
import org.apache.hudi.exception.{HoodieException,
HoodieIncrementalPathNotFoundException}
-import org.apache.hudi.hadoop.fs.HadoopFSUtils
-import org.apache.hudi.storage.StoragePath
import org.apache.hudi.util.IncrementalRelationUtil
import org.apache.avro.Schema
@@ -242,18 +239,12 @@ class IncrementalRelationV2(val sqlContext: SQLContext,
if (fallbackToFullTableScan) {
val timer = HoodieTimer.start
- val allFilesToCheck = filteredMetaBootstrapFullPaths ++
filteredRegularFullPaths
- val storageConf =
HadoopFSUtils.getStorageConfWithCopy(sqlContext.sparkContext.hadoopConfiguration)
- val localBasePathStr = basePath.toString
- val firstNotFoundPath =
sqlContext.sparkContext.parallelize(allFilesToCheck.toSeq, allFilesToCheck.size)
- .map(path => {
- val storage = HoodieStorageUtils.getStorage(localBasePathStr,
storageConf)
- storage.exists(new StoragePath(path))
- }).collect().find(v => !v)
+ val missingFileFound = HoodieBaseRelation.checkIfAnyFilesMissing(
+ sqlContext, metaClient, filteredMetaBootstrapFullPaths ++
filteredRegularFullPaths)
val timeTaken = timer.endTimer()
log.info("Checking if paths exists took " + timeTaken + "ms")
- if (firstNotFoundPath.isDefined) {
+ if (missingFileFound) {
doFullTableScan = true
log.info("Falling back to full table scan as some files cannot
be found.")
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
index 46444a719a38..0cc4be8df532 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
@@ -198,8 +198,18 @@ trait HoodieIncrementalRelationV1Trait extends
HoodieBaseRelation {
// 3. there are files in metadata be deleted
protected lazy val fullTableScan: Boolean = {
val fallbackToFullTableScan =
optParams.getOrElse(DataSourceReadOptions.INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN.key,
"false").toBoolean
- fallbackToFullTableScan && (startInstantArchived || endInstantArchived
- || affectedFilesInCommits.asScala.exists(fileStatus =>
!metaClient.getStorage.exists(fileStatus.getPath)))
+ if (!fallbackToFullTableScan) {
+ false
+ } else if (startInstantArchived || endInstantArchived) {
+ true
+ } else {
+ val missingFileFound = HoodieBaseRelation.checkIfAnyFilesMissing(
+ sqlContext, metaClient,
affectedFilesInCommits.asScala.map(_.getPath.toString))
+ if (missingFileFound) {
+ logInfo("Falling back to full table scan as some files cannot be
found.")
+ }
+ missingFileFound
+ }
}
protected lazy val includedCommits: immutable.Seq[HoodieInstant] = {
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
index 169ea5936c44..fb4ef6c38342 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
@@ -199,8 +199,18 @@ trait HoodieIncrementalRelationV2Trait extends
HoodieBaseRelation {
val fallbackToFullTableScan =
optParams.getOrElse(DataSourceReadOptions.INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN.key,
DataSourceReadOptions.INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN.defaultValue).toBoolean
- fallbackToFullTableScan && (startInstantArchived
- || affectedFilesInCommits.asScala.exists(fileStatus =>
!metaClient.getStorage.exists(fileStatus.getPath)))
+ if (!fallbackToFullTableScan) {
+ false
+ } else if (startInstantArchived) {
+ true
+ } else {
+ val missingFileFound = HoodieBaseRelation.checkIfAnyFilesMissing(
+ sqlContext, metaClient,
affectedFilesInCommits.asScala.map(_.getPath.toString))
+ if (missingFileFound) {
+ logInfo("Falling back to full table scan as some files cannot be
found.")
+ }
+ missingFileFound
+ }
}
protected val rangeType: RangeType
