umehrot2 commented on a change in pull request #2893:
URL: https://github.com/apache/hudi/pull/2893#discussion_r654728183
##########
File path:
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/HoodieFileIndex.scala
##########
@@ -151,13 +184,33 @@ case class HoodieFileIndex(
metaClient.reloadActiveTimeline()
val activeInstants =
metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants
fileSystemView = new HoodieTableFileSystemView(metaClient, activeInstants,
allFiles)
- cachedAllInputFiles =
fileSystemView.getLatestBaseFiles.iterator().asScala.toArray
- cachedAllPartitionPaths = partitionFiles.keys.toSeq
- cachedFileSize = cachedAllInputFiles.map(_.getFileLen).sum
+
+ (tableType, queryType) match {
+ case (MERGE_ON_READ, QUERY_TYPE_SNAPSHOT_OPT_VAL) =>
+ // Fetch and store latest base and log files, and their sizes
+ cachedAllInputFiles = partitionFiles.map(p => {
+ val latestSlices =
fileSystemView.getLatestMergedFileSlicesBeforeOrOn(p._1.partitionPath,
activeInstants.lastInstant().get().getTimestamp)
+ val baseAndLogFilesMapping =
latestSlices.iterator().asScala.map(slice => {
+ (slice.getBaseFile.get(),
slice.getLogFiles.sorted(HoodieLogFile.getLogFileComparator).iterator().asScala.toSeq)
+ }).toMap
+ (p._1, baseAndLogFilesMapping)
+ })
+ cachedFileSize =
cachedAllInputFiles.values.flatten.map(baseLogFilesMap => {
+ baseLogFilesMap._1.getFileLen +
baseLogFilesMap._2.map(_.getFileSize).sum
+ }).sum
+ case (_, _) =>
Review comment:
Incremental relations do not use the file index yet. As I understand, it
uses the commit files to get the list. We can explore that separately on how to
integrate it with file index. I will file a jira.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
