You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by vi...@apache.org on 2026/04/11 23:56:57 UTC
(incubator-xtable) branch main updated: [bug] Fix batch INSERT_OVERWRITE replacecommits dropping adds in HudiDataFileExtractor (#816)
This is an automated email from the ASF dual-hosted git repository.
vinish 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 634bcb6b [bug] Fix batch INSERT_OVERWRITE replacecommits dropping adds in HudiDataFileExtractor (#816)
634bcb6b is described below
commit 634bcb6bded6ec7b01fde5af0ecab3f11278e2b5
Author: Vinish Reddy <vi...@gmail.com>
AuthorDate: Sun Apr 12 05:26:53 2026 +0530
[bug] Fix batch INSERT_OVERWRITE replacecommits dropping adds in HudiDataFileExtractor (#816)
---
.../apache/xtable/hudi/HudiDataFileExtractor.java | 4 +-
.../java/org/apache/xtable/TestSparkHudiTable.java | 11 +++++
.../apache/xtable/hudi/ITHudiConversionSource.java | 48 ++++++++++++++++++++++
3 files changed, 60 insertions(+), 3 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 5e17b389..f3b73ec2 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
@@ -326,9 +326,7 @@ public class HudiDataFileExtractor implements AutoCloseable {
partitionValuesExtractor.extractPartitionValues(partitioningFields, partitionPath);
Stream<HoodieFileGroup> fileGroups =
Stream.concat(
- fsView.getAllFileGroups(partitionPath),
- fsView.getReplacedFileGroupsBeforeOrOn(
- instantToConsider.getTimestamp(), partitionPath));
+ fsView.getAllFileGroups(partitionPath), fsView.getAllReplacedFileGroups(partitionPath));
fileGroups.forEach(
fileGroup -> {
List<HoodieBaseFile> baseFiles = fileGroup.getAllBaseFiles().collect(Collectors.toList());
diff --git a/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java b/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java
index 1aaf61f9..a81bed20 100644
--- a/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java
+++ b/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java
@@ -220,6 +220,17 @@ public class TestSparkHudiTable extends TestAbstractHudiTable {
assertNoWriteErrors(result);
}
+ public void insertOverwrite(
+ List<HoodieRecord<HoodieAvroPayload>> records, HoodieTableType tableType) {
+ String actionType =
+ CommitUtils.getCommitActionType(WriteOperationType.INSERT_OVERWRITE, tableType);
+ String instant = getStartCommitOfActionType(actionType);
+ JavaRDD<HoodieRecord<HoodieAvroPayload>> writeRecords = jsc.parallelize(records, 1);
+ HoodieWriteResult writeResult = writeClient.insertOverwrite(writeRecords, instant);
+ List<WriteStatus> result = writeResult.getWriteStatuses().collect();
+ assertNoWriteErrors(result);
+ }
+
public void cluster() {
String instant = writeClient.scheduleClustering(Option.empty()).get();
writeClient.cluster(instant, true);
diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java
index 99d932f7..bc0b99c9 100644
--- a/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java
+++ b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java
@@ -419,6 +419,54 @@ public class ITHudiConversionSource {
}
}
+ @ParameterizedTest
+ @MethodSource("testsForAllTableTypes")
+ public void testMultipleInsertOverwriteOnSamePartitions(HoodieTableType tableType) {
+ String tableName = "test_table_" + UUID.randomUUID();
+ try (TestSparkHudiTable table =
+ TestSparkHudiTable.forStandardSchema(tableName, tempDir, jsc, "level:SIMPLE", tableType)) {
+ List<List<String>> allBaseFilePaths = new ArrayList<>();
+ List<TableChange> allTableChanges = new ArrayList<>();
+
+ // Initial insert into partition "INFO"
+ String commitInstant1 = table.startCommit();
+ List<HoodieRecord<HoodieAvroPayload>> insertsForCommit1 = table.generateRecords(50, "INFO");
+ table.insertRecordsWithCommitAlreadyStarted(insertsForCommit1, commitInstant1, true);
+ allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
+
+ // INSERT_OVERWRITE on "INFO" partition (replacecommit A — new file groups replace initial)
+ List<HoodieRecord<HoodieAvroPayload>> overwriteRecords1 = table.generateRecords(30, "INFO");
+ table.insertOverwrite(overwriteRecords1, tableType);
+ allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
+
+ // INSERT_OVERWRITE on "INFO" partition again (replacecommit B — new file groups replace A's)
+ List<HoodieRecord<HoodieAvroPayload>> overwriteRecords2 = table.generateRecords(20, "INFO");
+ table.insertOverwrite(overwriteRecords2, tableType);
+ allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
+
+ HudiConversionSource hudiClient =
+ getHudiSourceClient(CONFIGURATION, table.getBasePath(), "level:VALUE");
+ // Get the current snapshot
+ InternalSnapshot internalSnapshot = hudiClient.getCurrentSnapshot();
+ ValidationTestHelper.validateSnapshot(
+ internalSnapshot, allBaseFilePaths.get(allBaseFilePaths.size() - 1));
+ // Get changes in Incremental format since the initial insert
+ InstantsForIncrementalSync instantsForIncrementalSync =
+ InstantsForIncrementalSync.builder()
+ .lastSyncInstant(HudiInstantUtils.parseFromInstantTime(commitInstant1))
+ .build();
+ CommitsBacklog<HoodieInstant> instantCommitsBacklog =
+ hudiClient.getCommitsBacklog(instantsForIncrementalSync);
+ for (HoodieInstant instant : instantCommitsBacklog.getCommitsToProcess()) {
+ TableChange tableChange = hudiClient.getTableChangeForCommit(instant);
+ allTableChanges.add(tableChange);
+ }
+ // Without the fix, replacecommit A would have 0 adds because the FileSystemView
+ // built from the full timeline marks A's file groups as replaced by B.
+ ValidationTestHelper.validateTableChanges(allBaseFilePaths, allTableChanges);
+ }
+ }
+
@ParameterizedTest
@MethodSource("testsForAllTableTypes")
public void testsForDeleteAllRecordsInPartition(HoodieTableType tableType) {