This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new b3a07cc8 fix(hudi): Emit REMOVE in HudiDataFileExtractor when CLEAN
deletes the previous file version before incremental sync (#824)
b3a07cc8 is described below
commit b3a07cc85d491e1f383667d373427e7082f0baa4
Author: Vinish Reddy <[email protected]>
AuthorDate: Sun Oct 11 11:15:30 2026 +0530
fix(hudi): Emit REMOVE in HudiDataFileExtractor when CLEAN deletes the
previous file version before incremental sync (#824)
* fix(hudi): emit REMOVE when CLEAN deletes previous file version before
incremental sync
When Hudi CLEAN physically removes an old base file from storage before
XTable
processes the corresponding UPSERT commit incrementally, getAllBaseFiles()
only
returns the new file. The else-if branch in getUpdatesToPartition that
emits a
REMOVE for the previous version never fires, leaving a stale ADD in the
downstream
Delta log that causes FILE_NOT_FOUND errors on read.
Fix: after the base-file loop, if an ADD was emitted but no REMOVE was
emitted
and HoodieWriteStat.prevCommit indicates a prior version exists, recover
the old
file path by reading the previous commit's metadata from the visible Hudi
timeline
and emit a REMOVE. Gracefully degrades (logs warning, skips) if the previous
commit has been archived.
* Adapt the CLEAN REMOVE recovery to the Hudi 1.x timeline API
* Cache previous-commit metadata per sync when recovering cleaned files
* Add IT for CLEAN deleting replaced file groups of a pending clustering
commit
Async clustering that completes after a sync and is cleaned before the next
sync must not be processed incrementally. The controller falls back to
snapshot sync because the pending replace commit is older than the clean's
earliest commit to retain.
* Simplify readCommitMetadata with Optional.flatMap
---
.../apache/xtable/hudi/HudiDataFileExtractor.java | 91 ++++++++++-
.../org/apache/xtable/ITConversionController.java | 94 ++++++++++++
.../java/org/apache/xtable/TestJavaHudiTable.java | 9 +-
.../xtable/hudi/TestHudiDataFileExtractor.java | 170 +++++++++++++++++++++
4 files changed, 357 insertions(+), 7 deletions(-)
diff --git
a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java
b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java
index 2a45ae79..c94e3dbc 100644
---
a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java
+++
b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java
@@ -23,14 +23,18 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import lombok.Builder;
import lombok.Value;
+import lombok.extern.log4j.Log4j2;
import org.apache.hadoop.fs.Path;
@@ -71,6 +75,7 @@ import org.apache.xtable.model.storage.InternalFilesDiff;
import org.apache.xtable.model.storage.PartitionFileGroup;
/** Extracts all the files for Hudi table represented by {@link
InternalTable}. */
+@Log4j2
public class HudiDataFileExtractor implements AutoCloseable {
private final HoodieTableMetadata tableMetadata;
private final HoodieTableMetaClient metaClient;
@@ -80,6 +85,10 @@ public class HudiDataFileExtractor implements AutoCloseable {
private final HoodieMetadataConfig metadataConfig;
private final FileSystemViewManager fileSystemViewManager;
private final Path basePath;
+ // Previous-commit metadata read to recover cleaned files, keyed by instant
time. The extractor
+ // lives for a single sync, so each instant is read at most once per sync.
+ private final Map<String, Optional<HoodieCommitMetadata>>
prevCommitMetadataCache =
+ new HashMap<>();
public HudiDataFileExtractor(
HoodieTableMetaClient metaClient,
@@ -165,16 +174,13 @@ public class HudiDataFileExtractor implements
AutoCloseable {
.getPartitionToWriteStats()
.forEach(
(partitionPath, writeStats) -> {
- Set<String> affectedFileIds =
- writeStats.stream()
- .map(HoodieWriteStat::getFileId)
- .collect(Collectors.toSet());
AddedAndRemovedFiles addedAndRemovedFiles =
getUpdatesToPartition(
fsView,
+ timeline,
instantToConsider,
partitionPath,
- affectedFileIds,
+ writeStats,
partitioningFields);
addedFiles.addAll(addedAndRemovedFiles.getAdded());
removedFiles.addAll(addedAndRemovedFiles.getRemoved());
@@ -282,10 +288,26 @@ public class HudiDataFileExtractor implements
AutoCloseable {
private AddedAndRemovedFiles getUpdatesToPartition(
TableFileSystemView fsView,
+ HoodieTimeline timeline,
HoodieInstant instantToConsider,
String partitionPath,
- Set<String> affectedFileIds,
+ List<HoodieWriteStat> writeStats,
List<InternalPartitionField> partitioningFields) {
+ Set<String> affectedFileIds =
+
writeStats.stream().map(HoodieWriteStat::getFileId).collect(Collectors.toSet());
+ // Build a map from fileId → prevCommit for file groups that are UPSERTs
(not brand-new).
+ // Used below to recover the old file path when CLEAN has already deleted
it from the FS.
+ Map<String, String> fileIdToPrevCommit =
+ writeStats.stream()
+ .filter(
+ ws ->
+ ws.getPrevCommit() != null
+ &&
!HoodieTimeline.INIT_INSTANT_TS.equals(ws.getPrevCommit()))
+ .collect(
+ Collectors.toMap(
+ HoodieWriteStat::getFileId,
+ HoodieWriteStat::getPrevCommit,
+ (existing, replacement) -> existing));
List<InternalDataFile> filesToAdd = new
ArrayList<>(affectedFileIds.size());
List<InternalDataFile> filesToRemove = new
ArrayList<>(affectedFileIds.size());
List<PartitionValue> partitionValues =
@@ -300,6 +322,7 @@ public class HudiDataFileExtractor implements AutoCloseable
{
List<HoodieBaseFile> baseFiles =
fileGroup.getAllBaseFiles().collect(Collectors.toList());
boolean newBaseFileAdded = false;
+ boolean removeEmitted = false;
for (HoodieBaseFile baseFile : baseFiles) {
if
(baseFile.getCommitTime().equals(instantToConsider.requestedTime())) {
newBaseFileAdded = true;
@@ -308,13 +331,69 @@ public class HudiDataFileExtractor implements
AutoCloseable {
// 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) {
+ return timeline.getInstants().stream()
+ .filter(hoodieInstant ->
commitTime.equals(hoodieInstant.requestedTime()))
+ .findFirst()
+ .flatMap(
+ instant -> {
+ try {
+ return Optional.of(timeline.readCommitMetadata(instant));
+ } catch (IOException e) {
+ log.warn(
+ "Unable to read previous commit {} to recover removed
files", commitTime, e);
+ return Optional.empty();
+ }
+ });
+ }
+
private AddedAndRemovedFiles getUpdatesToPartitionForReplaceCommit(
TableFileSystemView fsView,
HoodieInstant instantToConsider,
diff --git
a/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
index ee6c5c98..d282cd0a 100644
--- a/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
+++ b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java
@@ -729,6 +729,100 @@ public class ITConversionController {
}
}
+ /**
+ * Verifies that incremental sync emits a REMOVE when Hudi CLEAN deletes the
previous base file
+ * version from storage before XTable processes the UPSERT commit.
+ */
+ @Test
+ public void testIncrementalSyncEmitsRemoveWhenHudiCleanRunsBeforeSync() {
+ String tableName = getTableName();
+ ConversionSourceProvider<?> conversionSourceProvider =
getConversionSourceProvider(HUDI);
+ try (TestJavaHudiTable table =
+ TestJavaHudiTable.forStandardSchema(
+ tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
+ ConversionConfig conversionConfig =
+ getTableSyncConfig(
+ HUDI, SyncMode.INCREMENTAL, tableName, table,
ImmutableList.of(DELTA), null, null);
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
+
+ // Commit A: insert 100 records; sync so Delta has ADD actions for
commitA files.
+ List<HoodieRecord<HoodieAvroPayload>> insertedRecords =
table.insertRecords(100, true);
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ checkDatasetEquivalence(HUDI, table, Collections.singletonList(DELTA),
100);
+
+ // Commit B: upsert 50 records, creating new file versions for those
file groups.
+ // XTable does NOT sync here — simulating the lag between Hudi writes
and XTable runs.
+ table.upsertRecords(insertedRecords.subList(0, 50), true);
+
+ // Commit C: insert 20 brand-new records. A third commit is required
before CLEAN so
+ // that Hudi's clean plan is non-empty (retainCommits=1 needs N+1
commits to purge N).
+ table.insertRecords(20, true);
+
+ // CLEAN physically deletes commitA's file versions from disk — this is
the race condition.
+ // After this point getAllBaseFiles() for the upserted file groups
returns only commitB files.
+ table.clean();
+
+ // Incremental sync picks up commits B and C. For the upserted file
groups in commit B, the
+ // fix recovers the commitA file path from prevCommit metadata and emits
a REMOVE, preventing
+ // Delta from retaining a stale ADD that would cause
FileNotFoundException on read.
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+
+ // 100 original + 20 new inserts = 120 records (50 upserted records are
updated, not added).
+ // Reading via Spark would throw FileNotFoundException on stale commitA
ADDs without the fix.
+ checkDatasetEquivalence(HUDI, table, Collections.singletonList(DELTA),
120);
+ }
+ }
+
+ @Test
+ public void
testSyncFallsBackToSnapshotWhenCleanDeletesReplacedFileGroupsOfPendingCommit() {
+ String tableName = getTableName();
+ ConversionSourceProvider<?> conversionSourceProvider =
getConversionSourceProvider(HUDI);
+ try (TestJavaHudiTable table =
+ TestJavaHudiTable.forStandardSchema(
+ tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE)) {
+ ConversionConfig conversionConfig =
+ getTableSyncConfig(
+ HUDI,
+ SyncMode.INCREMENTAL,
+ tableName,
+ table,
+ ImmutableList.of(DELTA, ICEBERG),
+ null,
+ null);
+ ConversionController conversionController =
+ new ConversionController(jsc.hadoopConfiguration());
+
+ // Commit A: insert 100 records; sync so the targets reference commitA
files.
+ table.insertRecords(100, true);
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ checkDatasetEquivalence(HUDI, table, ImmutableList.of(DELTA, ICEBERG),
100);
+
+ // Replace commit B: schedule async clustering over the commitA file
groups, but do not run
+ // it yet.
+ String clusteringInstant = table.onlyScheduleClustering();
+
+ // Commit C: insert 20 records and sync. XTable syncs C and records B as
a pending commit.
+ table.insertRecords(20, true);
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ checkDatasetEquivalence(HUDI, table, ImmutableList.of(DELTA, ICEBERG),
120);
+
+ // B completes after the sync and marks the commitA file groups as
replaced.
+ table.completeScheduledClustering(clusteringInstant);
+
+ // CLEAN retains C, so the earliest commit to retain equals the last
synced commit and the
+ // next sync stays incremental. B is older than C, so CLEAN physically
deletes the replaced
+ // commitA file groups before XTable processes B.
+ table.clean();
+
+ // B is still a pending commit for the targets and is older than the
earliest commit to
+ // retain, so the controller falls back to snapshot sync instead of
processing B
+ // incrementally against a file system view that no longer has the
commitA file groups.
+ conversionController.sync(conversionConfig, conversionSourceProvider);
+ checkDatasetEquivalence(HUDI, table, ImmutableList.of(DELTA, ICEBERG),
120);
+ }
+ }
+
@ParameterizedTest
@EnumSource(value = SyncMode.class)
public void testSyncWithSingleFormat(SyncMode syncMode) {
diff --git a/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java
b/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java
index 8e8d05c0..99b362c4 100644
--- a/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java
+++ b/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java
@@ -292,7 +292,14 @@ public class TestJavaHudiTable extends
TestAbstractHudiTable {
}
public void cluster() {
- String instant = writeClient.scheduleClustering(Option.empty()).get();
+ completeScheduledClustering(onlyScheduleClustering());
+ }
+
+ public String onlyScheduleClustering() {
+ return writeClient.scheduleClustering(Option.empty()).get();
+ }
+
+ public void completeScheduledClustering(String instant) {
writeClient.cluster(instant, true);
// Reinitializing as clustering disables auto commit and we want to enable
it back.
writeClient = initJavaWriteClient(schema, typedProperties, null);
diff --git
a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiDataFileExtractor.java
b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiDataFileExtractor.java
new file mode 100644
index 00000000..883bc0b4
--- /dev/null
+++
b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiDataFileExtractor.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hudi;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.Optional;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.Test;
+
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.versioning.v2.InstantComparatorV2;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import org.apache.xtable.model.storage.InternalDataFile;
+
+class TestHudiDataFileExtractor {
+
+ @Test
+ void recoverRemovedFile_returnsFileWhenPreviousCommitInTimeline() throws
Exception {
+ HudiDataFileExtractor extractor =
buildExtractorWithBasePath("file:///tmp/test-table");
+ String partition = "year=2023";
+ String fileId = "fg-1";
+ String prevCommitTime = "20230101000000000";
+ String oldPath = partition + "/" + fileId + "_0-0-0_" + prevCommitTime +
".parquet";
+
+ HoodieInstant prevInstant =
+ new HoodieInstant(
+ HoodieInstant.State.COMPLETED,
+ HoodieTimeline.COMMIT_ACTION,
+ prevCommitTime,
+ InstantComparatorV2.REQUESTED_TIME_BASED_COMPARATOR);
+ HoodieTimeline timeline = mock(HoodieTimeline.class);
+
when(timeline.getInstants()).thenReturn(Collections.singletonList(prevInstant));
+ when(timeline.readCommitMetadata(prevInstant))
+ .thenReturn(singleStatCommit(partition, fileId, oldPath));
+
+ Optional<InternalDataFile> result =
+ extractor.recoverRemovedFile(
+ timeline, partition, fileId, prevCommitTime,
Collections.emptyList());
+
+ assertTrue(result.isPresent());
+ assertEquals(oldPath, result.get().getPhysicalPath());
+ }
+
+ @Test
+ void recoverRemovedFile_returnsEmptyWhenPreviousCommitArchived() {
+ HudiDataFileExtractor extractor =
buildExtractorWithBasePath("file:///tmp/test-table");
+
+ HoodieTimeline timeline = mock(HoodieTimeline.class);
+ when(timeline.getInstants()).thenReturn(Collections.emptyList());
+
+ Optional<InternalDataFile> result =
+ extractor.recoverRemovedFile(
+ timeline, "year=2023", "fg-1", "20230101000000000",
Collections.emptyList());
+
+ assertFalse(result.isPresent());
+ }
+
+ @Test
+ void recoverRemovedFile_returnsEmptyWhenFileIdNotInPrevCommit() throws
Exception {
+ HudiDataFileExtractor extractor =
buildExtractorWithBasePath("file:///tmp/test-table");
+ String partition = "year=2023";
+ String prevCommitTime = "20230101000000000";
+
+ HoodieInstant prevInstant =
+ new HoodieInstant(
+ HoodieInstant.State.COMPLETED,
+ HoodieTimeline.COMMIT_ACTION,
+ prevCommitTime,
+ InstantComparatorV2.REQUESTED_TIME_BASED_COMPARATOR);
+ HoodieTimeline timeline = mock(HoodieTimeline.class);
+
when(timeline.getInstants()).thenReturn(Collections.singletonList(prevInstant));
+ when(timeline.readCommitMetadata(prevInstant))
+ .thenReturn(
+ singleStatCommit(partition, "different-fg", partition +
"/different-fg.parquet"));
+
+ Optional<InternalDataFile> result =
+ extractor.recoverRemovedFile(
+ timeline, partition, "fg-1", prevCommitTime,
Collections.emptyList());
+
+ assertFalse(result.isPresent());
+ }
+
+ @Test
+ void recoverRemovedFile_readsPreviousCommitMetadataOncePerInstant() throws
Exception {
+ HudiDataFileExtractor extractor =
buildExtractorWithBasePath("file:///tmp/test-table");
+ String partition = "year=2023";
+ String prevCommitTime = "20230101000000000";
+ String oldPath1 = partition + "/fg-1_0-0-0_" + prevCommitTime + ".parquet";
+ String oldPath2 = partition + "/fg-2_0-0-0_" + prevCommitTime + ".parquet";
+
+ HoodieInstant prevInstant =
+ new HoodieInstant(
+ HoodieInstant.State.COMPLETED,
+ HoodieTimeline.COMMIT_ACTION,
+ prevCommitTime,
+ InstantComparatorV2.REQUESTED_TIME_BASED_COMPARATOR);
+ HoodieCommitMetadata prevMetadata = singleStatCommit(partition, "fg-1",
oldPath1);
+ HoodieWriteStat secondStat = new HoodieWriteStat();
+ secondStat.setFileId("fg-2");
+ secondStat.setPath(oldPath2);
+ prevMetadata.addWriteStat(partition, secondStat);
+ HoodieTimeline timeline = mock(HoodieTimeline.class);
+
when(timeline.getInstants()).thenReturn(Collections.singletonList(prevInstant));
+ when(timeline.readCommitMetadata(prevInstant)).thenReturn(prevMetadata);
+
+ Optional<InternalDataFile> result1 =
+ extractor.recoverRemovedFile(
+ timeline, partition, "fg-1", prevCommitTime,
Collections.emptyList());
+ Optional<InternalDataFile> result2 =
+ extractor.recoverRemovedFile(
+ timeline, partition, "fg-2", prevCommitTime,
Collections.emptyList());
+
+ assertEquals(oldPath1, result1.get().getPhysicalPath());
+ assertEquals(oldPath2, result2.get().getPhysicalPath());
+ verify(timeline, times(1)).readCommitMetadata(prevInstant);
+ }
+
+ private static HoodieCommitMetadata singleStatCommit(
+ String partition, String fileId, String path) {
+ HoodieWriteStat stat = new HoodieWriteStat();
+ stat.setFileId(fileId);
+ stat.setPath(path);
+ HoodieCommitMetadata metadata = new HoodieCommitMetadata();
+ metadata.addWriteStat(partition, stat);
+ return metadata;
+ }
+
+ private static HudiDataFileExtractor buildExtractorWithBasePath(String
basePathStr) {
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ doReturn(new HadoopStorageConfiguration(new
Configuration())).when(metaClient).getStorageConf();
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ when(tableConfig.isMetadataTableAvailable()).thenReturn(false);
+ when(metaClient.getBasePath()).thenReturn(new StoragePath(basePathStr));
+ return new HudiDataFileExtractor(metaClient, null, null);
+ }
+}
