vinishjail97 commented on code in PR #824:
URL: https://github.com/apache/incubator-xtable/pull/824#discussion_r4240334328
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -308,13 +331,69 @@ private AddedAndRemovedFiles getUpdatesToPartition(
// if a new base file was added, then the previous base file
for the group needs
// to be removed
filesToRemove.add(buildFileWithoutStats(partitionValues,
baseFile));
+ removeEmitted = true;
break;
}
}
+ // Recover old file path from timeline when CLEAN deleted it
before this sync.
+ if (newBaseFileAdded && !removeEmitted) {
+ String fileId = fileGroup.getFileGroupId().getFileId();
+ String prevCommit = fileIdToPrevCommit.get(fileId);
+ if (prevCommit != null) {
+ recoverRemovedFile(timeline, partitionPath, fileId,
prevCommit, partitionValues)
+ .ifPresent(filesToRemove::add);
+ }
+ }
});
return
AddedAndRemovedFiles.builder().added(filesToAdd).removed(filesToRemove).build();
}
+ /**
+ * Looks up the previous base file path from the timeline when CLEAN has
already deleted it from
+ * storage, so the downstream format can emit a proper REMOVE for the stale
file.
+ *
+ * @return the old {@link InternalDataFile}, or empty if the previous commit
is no longer in the
+ * visible timeline (e.g. archived before this sync ran)
+ */
+ Optional<InternalDataFile> recoverRemovedFile(
+ HoodieTimeline timeline,
+ String partitionPath,
+ String fileId,
+ String prevCommitTime,
+ List<PartitionValue> partitionValues) {
+ return prevCommitMetadataCache
+ .computeIfAbsent(prevCommitTime, commitTime ->
readCommitMetadata(timeline, commitTime))
+ .flatMap(
+ prevMeta ->
+ prevMeta
+ .getPartitionToWriteStats()
+ .getOrDefault(partitionPath, Collections.emptyList())
+ .stream()
+ .filter(ws -> fileId.equals(ws.getFileId()))
+ .findFirst()
+ .map(
+ ws ->
+ buildFileWithoutStats(
+ partitionValues, new
HoodieBaseFile(ws.getPath()))));
+ }
+
+ private Optional<HoodieCommitMetadata> readCommitMetadata(
+ HoodieTimeline timeline, String commitTime) {
+ Optional<HoodieInstant> instant =
+ timeline.getInstants().stream()
+ .filter(hoodieInstant ->
commitTime.equals(hoodieInstant.requestedTime()))
+ .findFirst();
+ if (!instant.isPresent()) {
+ return Optional.empty();
+ }
+ try {
+ return Optional.of(timeline.readCommitMetadata(instant.get()));
Review Comment:
Thanks @the-other-tim-brown, simplified in 487557bc. `readCommitMetadata`
throws a checked `IOException`, so a plain `map` does not compile; I used
`flatMap` with the catch inside the lambda.
--
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]
