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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
