vaibhavk1992 commented on code in PR #824:
URL: https://github.com/apache/incubator-xtable/pull/824#discussion_r3702196997
##########
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:
HoodieInstant.getTimestamp() was removed by the Hudi 1.2.0 upgrade (#772,
merged after this
branch diverged). main's getUpdatesToPartition already uses
instantToConsider.requestedTime(),
but this PR's diff still shows instantToConsider.getTimestamp() as
unchanged context, and the
new recoverRemovedFile method also calls instant.getTimestamp(). Needs a
rebase + both call
sites updated to 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:
Caching / re-parsing concern as raised by @the-other-tim-brown
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java:
##########
Review Comment:
Grabs old file via live fsView, same vulnerability as the bug this PR fixes,
I think fix is needed here too.
##########
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 stale API like above. Needs rebase.
--
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]
