codope commented on code in PR #12091:
URL: https://github.com/apache/hudi/pull/12091#discussion_r1799771414
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1088,28 +1102,12 @@ private void
updateFunctionalIndexIfPresent(HoodieCommitMetadata commitMetadata,
*/
private HoodieData<HoodieRecord>
getFunctionalIndexUpdates(HoodieCommitMetadata commitMetadata, String
indexPartition, String instantTime) throws Exception {
HoodieIndexDefinition indexDefinition =
getFunctionalIndexDefinition(indexPartition);
- List<Pair<String, FileSlice>> partitionFileSlicePairs = new ArrayList<>();
- HoodieTableFileSystemView fsv =
HoodieTableMetadataUtil.getFileSystemView(dataMetaClient);
- commitMetadata.getPartitionToWriteStats().forEach((dataPartition,
writeStats) -> {
- // collect list of FileIDs touched as part of this commit.
- Set<String> fileIds = writeStats.stream().map(writeStat ->
writeStat.getFileId()).collect(Collectors.toSet());
- List<FileSlice> fileSlices =
getPartitionLatestFileSlicesIncludingInflight(dataMetaClient, Option.of(fsv),
dataPartition)
- .stream().filter(fileSlice ->
fileIds.contains(fileSlice.getFileId())).collect(Collectors.toList());
- // process only the fileSlices touched in this commit meta
- // data.
- fileSlices.forEach(fileSlice -> {
- // Filter log files for the instant time and add to this partition
fileSlice pairs
- List<HoodieLogFile> logFilesForInstant = fileSlice.getLogFiles()
- .filter(logFile ->
logFile.getDeltaCommitTime().equals(instantTime))
- .collect(Collectors.toList());
- Option<HoodieBaseFile> baseFileOpt =
fileSlice.getBaseInstantTime().equals(instantTime) ? fileSlice.getBaseFile() :
Option.empty();
- partitionFileSlicePairs.add(Pair.of(dataPartition, new FileSlice(
- fileSlice.getFileGroupId(), fileSlice.getBaseInstantTime(),
baseFileOpt.orElse(null), logFilesForInstant)));
- });
- });
- int parallelism = Math.min(partitionFileSlicePairs.size(),
dataWriteConfig.getMetadataConfig().getFunctionalIndexParallelism());
+ List<Pair<String, Pair<String, Long>>> partitionFilePathPairs = new
ArrayList<>();
+ commitMetadata.getPartitionToWriteStats().forEach((dataPartition,
writeStats) -> writeStats.forEach(writeStat -> partitionFilePathPairs.add(
+ Pair.of(writeStat.getPartitionPath(), Pair.of(new
StoragePath(dataMetaClient.getBasePath(), writeStat.getPath()).toString(),
writeStat.getFileSizeInBytes())))));
Review Comment:
needed to combine with base path to form the full path (that's what readers
expect)
--
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]
