vinishjail97 commented on code in PR #824:
URL: https://github.com/apache/incubator-xtable/pull/824#discussion_r4223252156
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -306,13 +324,68 @@ 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(
Review Comment:
Thanks @the-other-tim-brown, added a per-sync cache of the previous-commit
metadata keyed by instant time in b4992399, so each instant is read at most
once per sync. The extractor is created per sync, so the cache is bounded by it.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -298,6 +315,7 @@ private AddedAndRemovedFiles getUpdatesToPartition(
List<HoodieBaseFile> baseFiles =
fileGroup.getAllBaseFiles().collect(Collectors.toList());
boolean newBaseFileAdded = false;
+ boolean removeEmitted = false;
for (HoodieBaseFile baseFile : baseFiles) {
if
(baseFile.getCommitTime().equals(instantToConsider.getTimestamp())) {
Review Comment:
Thanks @vaibhavk1992, this was addressed in dbda0b36 when the branch moved
to the Hudi 1.x timeline API; both call sites now use `requestedTime()`.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -306,13 +324,68 @@ 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 timeline.getInstants().stream()
+ .filter(instant -> prevCommitTime.equals(instant.getTimestamp()))
Review Comment:
Same fix in dbda0b36, the lookup now matches on `requestedTime()`.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
@@ -306,13 +324,68 @@ 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 timeline.getInstants().stream()
+ .filter(instant -> prevCommitTime.equals(instant.getTimestamp()))
+ .findFirst()
+ .flatMap(
+ prevInstant -> {
+ try {
+ HoodieCommitMetadata prevMeta =
+ HoodieCommitMetadata.fromBytes(
Review Comment:
Addressed together with Tim's comment in b4992399.
--
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]
