[bug] Fix batch INSERT_OVERWRITE replacecommits dropping adds in HudiDataFileExtractor (#816)

Author: vinishjail97

xtable-core/src/main/java/org/apache/xtable/hudi/HudiDataFileExtractor.java

@@ -326,9 +326,7 @@ private AddedAndRemovedFiles getUpdatesToPartitionForReplaceCommit(
         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());

xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java

@@ -220,6 +220,17 @@ public void deletePartition(String partition, HoodieTableType tableType) {
     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);

xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java

@@ -419,6 +419,54 @@ public void testsForDropPartition(HoodieTableType tableType) {
     }
   }
 
+  @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) {