[HUDI-1740] Fix insert-overwrite API archival (#2784)

- fix problem of archiving replace commits
- Fix problem when getting empty replacecommit.requested
- Improved the logic of handling empty and non-empty requested/inflight commit files. Added unit tests to cover both empty and non-empty inflight files cases and cleaned up some unused test util methods

Co-authored-by: yorkzero831 <yorkzero8312@gmail.com>
Co-authored-by: zheren.yu <zheren.yu@paypay-corp.co.jp>

hudi-cli/src/test/java/org/apache/hudi/cli/commands/TestCommitsCommand.java

@@ -24,7 +24,7 @@ import org.apache.hudi.cli.HoodieTableHeaderFields;
 import org.apache.hudi.cli.TableHeader;
 import org.apache.hudi.cli.testutils.AbstractShellIntegrationTest;
 import org.apache.hudi.cli.testutils.HoodieTestCommitMetadataGenerator;
-import org.apache.hudi.cli.testutils.HoodieTestReplaceCommitMetadatGenerator;
+import org.apache.hudi.cli.testutils.HoodieTestReplaceCommitMetadataGenerator;
 import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -122,7 +122,7 @@ public class TestCommitsCommand extends AbstractShellIntegrationTest {
     for (Map.Entry<HoodieInstant, Integer[]> entry : replaceCommitData.entrySet()) {
       String key = entry.getKey().getTimestamp();
       Integer[] value = entry.getValue();
-      HoodieTestReplaceCommitMetadatGenerator.createReplaceCommitFileWithMetadata(tablePath, key,
+      HoodieTestReplaceCommitMetadataGenerator.createReplaceCommitFileWithMetadata(tablePath, key,
               Option.of(value[0]), Option.of(value[1]), metaClient);
     }
 

hudi-cli/src/test/java/org/apache/hudi/cli/testutils/HoodieTestReplaceCommitMetadataGenerator.java

@@ -35,7 +35,7 @@ import java.util.UUID;
 import static org.apache.hudi.common.testutils.FileCreateUtils.baseFileName;
 import static org.apache.hudi.common.util.CollectionUtils.createImmutableList;
 
-public class HoodieTestReplaceCommitMetadatGenerator extends HoodieTestCommitMetadataGenerator {
+public class HoodieTestReplaceCommitMetadataGenerator extends HoodieTestCommitMetadataGenerator {
   public static void createReplaceCommitFileWithMetadata(String basePath, String commitTime, Option<Integer> writes, Option<Integer> updates,
                                                          HoodieTableMetaClient metaclient) throws Exception {
 
@@ -43,7 +43,7 @@ public class HoodieTestReplaceCommitMetadatGenerator extends HoodieTestCommitMet
         UUID.randomUUID().toString(), writes, updates);
     HoodieRequestedReplaceMetadata requestedReplaceMetadata = getHoodieRequestedReplaceMetadata();
 
-    HoodieTestTable.of(metaclient).addReplaceCommit(commitTime, requestedReplaceMetadata, replaceMetadata);
+    HoodieTestTable.of(metaclient).addReplaceCommit(commitTime, Option.ofNullable(requestedReplaceMetadata), Option.empty(), replaceMetadata);
   }
 
   private static HoodieRequestedReplaceMetadata getHoodieRequestedReplaceMetadata() {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/ReplaceArchivalHelper.java

@@ -68,6 +68,11 @@ public class ReplaceArchivalHelper implements Serializable {
   public static boolean deleteReplacedFileGroups(HoodieEngineContext context, HoodieTableMetaClient metaClient,
                                                  TableFileSystemView fileSystemView,
                                                  HoodieInstant instant, List<String> replacedPartitions) {
+    // There is no file id to be replaced in the very first replace commit file for insert overwrite operation
+    if (replacedPartitions.isEmpty()) {
+      LOG.warn("Found no partition files to replace");
+      return true;
+    }
     context.setJobStatus(ReplaceArchivalHelper.class.getSimpleName(), "Delete replaced file groups");
     List<Boolean> f = context.map(replacedPartitions, partition -> {
       Stream<FileSlice> fileSlices =  fileSystemView.getReplacedFileGroupsBeforeOrOn(instant.getTimestamp(), partition)

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/MetadataConversionUtils.java

@@ -21,6 +21,7 @@ package org.apache.hudi.client.utils;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.io.IOException;
+
 import org.apache.hudi.avro.model.HoodieArchivedMetaEntry;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
 import org.apache.hudi.avro.model.HoodieRequestedReplaceMetadata;
@@ -37,8 +38,8 @@ import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
 import org.apache.hudi.common.util.CleanerUtils;
-import org.apache.hudi.common.util.ClusteringUtils;
 import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.Option;
 
 /**
  * Helper class to convert between different action related payloads and {@link HoodieArchivedMetaEntry}.
@@ -72,10 +73,21 @@ public class MetadataConversionUtils {
           HoodieReplaceCommitMetadata replaceCommitMetadata = HoodieReplaceCommitMetadata
               .fromBytes(metaClient.getActiveTimeline().getInstantDetails(hoodieInstant).get(), HoodieReplaceCommitMetadata.class);
           archivedMetaWrapper.setHoodieReplaceCommitMetadata(ReplaceArchivalHelper.convertReplaceCommitMetadata(replaceCommitMetadata));
+        } else if (hoodieInstant.isInflight()) {
+          // inflight replacecommit files have the same meta data body as HoodieCommitMetadata
+          // so we could re-use it without further creating an inflight extension.
+          // Or inflight replacecommit files are empty under clustering circumstance
+          Option<HoodieCommitMetadata> inflightCommitMetadata = getInflightReplaceMetadata(metaClient, hoodieInstant);
+          if (inflightCommitMetadata.isPresent()) {
+            archivedMetaWrapper.setHoodieInflightReplaceMetadata(convertCommitMetadata(inflightCommitMetadata.get()));
+          }
         } else {
-          HoodieRequestedReplaceMetadata requestedReplaceMetadata =
+          // we may have cases with empty HoodieRequestedReplaceMetadata e.g. insert_overwrite_table or insert_overwrite
-              ClusteringUtils.getRequestedReplaceMetadata(metaClient, hoodieInstant).get();
+          // without clustering. However, we should revisit the requested commit file standardization
-          archivedMetaWrapper.setHoodieRequestedReplaceMetadata(requestedReplaceMetadata);
+          Option<HoodieRequestedReplaceMetadata> requestedReplaceMetadata = getRequestedReplaceMetadata(metaClient, hoodieInstant);
+          if (requestedReplaceMetadata.isPresent()) {
+            archivedMetaWrapper.setHoodieRequestedReplaceMetadata(requestedReplaceMetadata.get());
+          }
         }
         archivedMetaWrapper.setActionType(ActionType.replacecommit.name());
         break;
@@ -107,14 +119,25 @@ public class MetadataConversionUtils {
     return archivedMetaWrapper;
   }
 
-  public static HoodieArchivedMetaEntry createMetaWrapper(HoodieInstant hoodieInstant,
+  public static Option<HoodieCommitMetadata> getInflightReplaceMetadata(HoodieTableMetaClient metaClient, HoodieInstant instant) throws IOException {
-                                                          HoodieCommitMetadata hoodieCommitMetadata) {
+    Option<byte[]> inflightContent = metaClient.getActiveTimeline().getInstantDetails(instant);
-    HoodieArchivedMetaEntry archivedMetaWrapper = new HoodieArchivedMetaEntry();
+    if (!inflightContent.isPresent() || inflightContent.get().length == 0) {
-    archivedMetaWrapper.setCommitTime(hoodieInstant.getTimestamp());
+      // inflight files can be empty in some certain cases, e.g. when users opt in clustering
-    archivedMetaWrapper.setActionState(hoodieInstant.getState().name());
+      return Option.empty();
-    archivedMetaWrapper.setHoodieCommitMetadata(convertCommitMetadata(hoodieCommitMetadata));
+    }
-    archivedMetaWrapper.setActionType(ActionType.commit.name());
+    return Option.of(HoodieCommitMetadata.fromBytes(inflightContent.get(), HoodieCommitMetadata.class));
-    return archivedMetaWrapper;
+  }
+
+  public static Option<HoodieRequestedReplaceMetadata> getRequestedReplaceMetadata(HoodieTableMetaClient metaClient, HoodieInstant instant) throws IOException {
+    Option<byte[]> requestedContent = metaClient.getActiveTimeline().getInstantDetails(instant);
+    if (!requestedContent.isPresent() || requestedContent.get().length == 0) {
+      // requested commit files can be empty in some certain cases, e.g. insert_overwrite or insert_overwrite_table.
+      // However, it appears requested files are supposed to contain meta data and we should revisit the standardization
+      // of requested commit files
+      // TODO revisit requested commit file standardization https://issues.apache.org/jira/browse/HUDI-1739
+      return Option.empty();
+    }
+    return Option.of(TimelineMetadataUtils.deserializeRequestedReplaceMetadata(requestedContent.get()));
   }
 
   public static org.apache.hudi.avro.model.HoodieCommitMetadata convertCommitMetadata(

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java

@@ -296,7 +296,6 @@ public class HoodieTimelineArchiveLog<T extends HoodieAvroPayload, I, K, O> {
 
   public void archive(HoodieEngineContext context, List<HoodieInstant> instants) throws HoodieCommitException {
     try {
-      HoodieTimeline commitTimeline = metaClient.getActiveTimeline().getAllCommitsTimeline().filterCompletedInstants();
       Schema wrapperSchema = HoodieArchivedMetaEntry.getClassSchema();
       LOG.info("Wrapper schema " + wrapperSchema.toString());
       List<IndexedRecord> records = new ArrayList<>();
@@ -308,7 +307,7 @@ public class HoodieTimelineArchiveLog<T extends HoodieAvroPayload, I, K, O> {
         }
         try {
           deleteAnyLeftOverMarkerFiles(context, hoodieInstant);
-          records.add(convertToAvroRecord(commitTimeline, hoodieInstant));
+          records.add(convertToAvroRecord(hoodieInstant));
           if (records.size() >= this.config.getCommitArchivalBatchSize()) {
             writeToFile(wrapperSchema, records);
           }
@@ -365,8 +364,8 @@ public class HoodieTimelineArchiveLog<T extends HoodieAvroPayload, I, K, O> {
     }
   }
 
-  private IndexedRecord convertToAvroRecord(HoodieTimeline commitTimeline, HoodieInstant hoodieInstant)
+  private IndexedRecord convertToAvroRecord(HoodieInstant hoodieInstant)
-          throws IOException {
+      throws IOException {
     return MetadataConversionUtils.createMetaWrapper(hoodieInstant, metaClient);
   }
 }

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java

@@ -383,7 +383,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
     requestedReplaceMetadata.setClusteringPlan(clusteringPlan);
     requestedReplaceMetadata.setVersion(TimelineLayoutVersion.CURR_VERSION);
     HoodieTestTable.of(metaClient)
-        .addRequestedReplace(instantTime, requestedReplaceMetadata)
+        .addRequestedReplace(instantTime, Option.of(requestedReplaceMetadata))
         .withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
   }
 
@@ -413,7 +413,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
     requestedReplaceMetadata.setClusteringPlan(clusteringPlan);
     requestedReplaceMetadata.setVersion(TimelineLayoutVersion.CURR_VERSION);
     HoodieTestTable.of(metaClient)
-        .addReplaceCommit(instantTime, requestedReplaceMetadata, replaceMetadata)
+        .addReplaceCommit(instantTime, Option.of(requestedReplaceMetadata), Option.empty(), replaceMetadata)
         .withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
   }
 

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/utils/TestMetadataConversionUtils.java

@@ -20,6 +20,7 @@ package org.apache.hudi.utils;
 
 import static org.apache.hudi.common.util.CleanerUtils.convertCleanMetadata;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -80,13 +81,43 @@ public class TestMetadataConversionUtils extends HoodieCommonTestHarness {
   @Test
   public void testCompletedReplace() throws Exception {
     String newCommitTime = HoodieTestTable.makeNewCommitTime();
-    createReplace(newCommitTime, WriteOperationType.INSERT_OVERWRITE);
+    createReplace(newCommitTime, WriteOperationType.INSERT_OVERWRITE, true);
     HoodieArchivedMetaEntry metaEntry = MetadataConversionUtils.createMetaWrapper(
         new HoodieInstant(State.COMPLETED, HoodieTimeline.REPLACE_COMMIT_ACTION, newCommitTime), metaClient);
     assertEquals(metaEntry.getActionState(), State.COMPLETED.toString());
     assertEquals(metaEntry.getHoodieReplaceCommitMetadata().getOperationType(), WriteOperationType.INSERT_OVERWRITE.toString());
   }
 
+  @Test
+  public void testEmptyRequestedReplace() throws Exception {
+    String newCommitTime = HoodieTestTable.makeNewCommitTime();
+    createReplace(newCommitTime, WriteOperationType.INSERT_OVERWRITE_TABLE, false);
+    HoodieArchivedMetaEntry metaEntry = MetadataConversionUtils.createMetaWrapper(
+            new HoodieInstant(State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, newCommitTime), metaClient);
+    assertEquals(metaEntry.getActionState(), State.REQUESTED.toString());
+    assertNull(metaEntry.getHoodieRequestedReplaceMetadata());
+  }
+
+  @Test
+  public void testEmptyInflightReplace() throws Exception {
+    String newCommitTime = HoodieTestTable.makeNewCommitTime();
+    createReplace(newCommitTime, WriteOperationType.INSERT_OVERWRITE_TABLE, true);
+    HoodieArchivedMetaEntry metaEntry = MetadataConversionUtils.createMetaWrapper(
+            new HoodieInstant(State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, newCommitTime), metaClient);
+    assertEquals(metaEntry.getActionState(), State.INFLIGHT.toString());
+    assertNull(metaEntry.getHoodieInflightReplaceMetadata());
+  }
+
+  @Test
+  public void testNonEmptyInflightReplace() throws Exception {
+    String newCommitTime = HoodieTestTable.makeNewCommitTime();
+    createReplace(newCommitTime, WriteOperationType.INSERT_OVERWRITE_TABLE, false);
+    HoodieArchivedMetaEntry metaEntry = MetadataConversionUtils.createMetaWrapper(
+        new HoodieInstant(State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, newCommitTime), metaClient);
+    assertEquals(metaEntry.getActionState(), State.INFLIGHT.toString());
+    assertEquals(metaEntry.getHoodieInflightReplaceMetadata().getOperationType(), WriteOperationType.INSERT_OVERWRITE_TABLE.name());
+  }
+
   @Test
   public void testCompletedCommitOrDeltaCommit() throws Exception {
     String newCommitTime = HoodieTestTable.makeNewCommitTime();
@@ -169,7 +200,8 @@ public class TestMetadataConversionUtils extends HoodieCommonTestHarness {
         .withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
   }
 
-  private void createReplace(String instantTime, WriteOperationType writeOperationType) throws Exception {
+  private void createReplace(String instantTime, WriteOperationType writeOperationType, Boolean isClustering)
+          throws Exception {
     String fileId1 = "file-1";
     String fileId2 = "file-2";
 
@@ -182,18 +214,29 @@ public class TestMetadataConversionUtils extends HoodieCommonTestHarness {
     writeStat.setFileId("file-1");
     replaceMetadata.addWriteStat(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, writeStat);
     replaceMetadata.setOperationType(writeOperationType);
-    // create replace instant to mark fileId1 as deleted
+    // some cases requestedReplaceMetadata will be null
-    HoodieRequestedReplaceMetadata requestedReplaceMetadata = new HoodieRequestedReplaceMetadata();
+    // e.g. insert_overwrite_table or insert_overwrite without clustering
-    requestedReplaceMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE.name());
+    HoodieRequestedReplaceMetadata requestedReplaceMetadata = null;
-    HoodieClusteringPlan clusteringPlan = new HoodieClusteringPlan();
+    HoodieCommitMetadata inflightReplaceMetadata = null;
-    HoodieClusteringGroup clusteringGroup = new HoodieClusteringGroup();
+    if (isClustering) {
-    HoodieSliceInfo sliceInfo = new HoodieSliceInfo();
+      requestedReplaceMetadata = new HoodieRequestedReplaceMetadata();
-    clusteringGroup.setSlices(Arrays.asList(sliceInfo));
+      requestedReplaceMetadata.setOperationType(writeOperationType.name());
-    clusteringPlan.setInputGroups(Arrays.asList(clusteringGroup));
+      HoodieClusteringPlan clusteringPlan = new HoodieClusteringPlan();
-    requestedReplaceMetadata.setClusteringPlan(clusteringPlan);
+      HoodieClusteringGroup clusteringGroup = new HoodieClusteringGroup();
-    requestedReplaceMetadata.setVersion(TimelineLayoutVersion.CURR_VERSION);
+      HoodieSliceInfo sliceInfo = new HoodieSliceInfo();
+      clusteringGroup.setSlices(Arrays.asList(sliceInfo));
+      clusteringPlan.setInputGroups(Arrays.asList(clusteringGroup));
+      requestedReplaceMetadata.setClusteringPlan(clusteringPlan);
+      requestedReplaceMetadata.setVersion(TimelineLayoutVersion.CURR_VERSION);
+    } else {
+      // inflightReplaceMetadata will be null in clustering but not null
+      // in insert_overwrite or insert_overwrite_table
+      inflightReplaceMetadata = new HoodieCommitMetadata();
+      inflightReplaceMetadata.setOperationType(writeOperationType);
+      inflightReplaceMetadata.setCompacted(false);
+    }
     HoodieTestTable.of(metaClient)
-        .addReplaceCommit(instantTime, requestedReplaceMetadata, replaceMetadata)
+        .addReplaceCommit(instantTime, Option.ofNullable(requestedReplaceMetadata), Option.ofNullable(inflightReplaceMetadata), replaceMetadata)
         .withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
   }
 

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiveLog.java

@@ -241,6 +241,10 @@ public class TestHoodieTimelineArchiveLog extends HoodieClientTestHarness {
         .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 3).build())
         .build();
 
+    // when using insert_overwrite or insert_overwrite_table
+    // first commit may without replaceFileIds
+    createReplaceMetadataWithoutReplaceFileId("000");
+
     int numCommits = 4;
     int commitInstant = 100;
     for (int i = 0; i < numCommits; i++) {
@@ -251,7 +255,7 @@ public class TestHoodieTimelineArchiveLog extends HoodieClientTestHarness {
     metaClient = HoodieTableMetaClient.reload(metaClient);
     HoodieTimeline timeline = metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
     HoodieTable table = HoodieSparkTable.create(cfg, context, metaClient);
-    assertEquals(4, timeline.countInstants(), "Loaded 4 commits and the count should match");
+    assertEquals(5, timeline.countInstants(), "Loaded 5 commits and the count should match");
     HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
     boolean result = archiveLog.archiveIfRequired(context);
     assertTrue(result);
@@ -513,13 +517,7 @@ public class TestHoodieTimelineArchiveLog extends HoodieClientTestHarness {
     HoodieCommitMetadata hoodieCommitMetadata = new HoodieCommitMetadata();
     hoodieCommitMetadata.setOperationType(WriteOperationType.INSERT);
 
-    HoodieWriteConfig cfg = HoodieWriteConfig.newBuilder().withPath(basePath)
-        .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2).forTable("test-commitMetadata-converter")
-        .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(2, 5).build())
-        .build();
     metaClient = HoodieTableMetaClient.reload(metaClient);
-    HoodieTable table = HoodieSparkTable.create(cfg, context, metaClient);
-    HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(cfg, table);
 
     org.apache.hudi.avro.model.HoodieCommitMetadata expectedCommitMetadata = MetadataConversionUtils
         .convertCommitMetadata(hoodieCommitMetadata);
@@ -681,6 +679,22 @@ public class TestHoodieTimelineArchiveLog extends HoodieClientTestHarness {
     assertEquals(notArchivedInstants, Arrays.asList(notArchivedInstant1, notArchivedInstant2, notArchivedInstant3), "");
   }
 
+  private void createReplaceMetadataWithoutReplaceFileId(String instantTime) throws Exception {
+
+    // create replace instant without a previous replace commit
+    HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
+        .setOperationType(WriteOperationType.INSERT_OVERWRITE_TABLE.toString())
+        .setVersion(1)
+        .setExtraMetadata(Collections.emptyMap())
+        .build();
+    HoodieReplaceCommitMetadata completeReplaceMetadata = new HoodieReplaceCommitMetadata();
+    HoodieCommitMetadata inflightReplaceMetadata = new HoodieCommitMetadata();
+    completeReplaceMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE_TABLE);
+    inflightReplaceMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE_TABLE);
+    HoodieTestTable.of(metaClient)
+        .addReplaceCommit(instantTime, Option.of(requestedReplaceMetadata), Option.of(inflightReplaceMetadata), completeReplaceMetadata);
+  }
+
   private void createReplaceMetadata(String instantTime) throws Exception {
     String fileId1 = "file-" + instantTime + "-1";
     String fileId2 = "file-" + instantTime + "-2";
@@ -691,11 +705,13 @@ public class TestHoodieTimelineArchiveLog extends HoodieClientTestHarness {
         .setVersion(1)
         .setExtraMetadata(Collections.emptyMap())
         .build();
-    HoodieReplaceCommitMetadata replaceMetadata = new HoodieReplaceCommitMetadata();
+    HoodieReplaceCommitMetadata completeReplaceMetadata = new HoodieReplaceCommitMetadata();
-    replaceMetadata.addReplaceFileId(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1);
+    HoodieCommitMetadata inflightReplaceMetadata = new HoodieCommitMetadata();
-    replaceMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE);
+    completeReplaceMetadata.addReplaceFileId(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1);
+    completeReplaceMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE);
+    inflightReplaceMetadata.setOperationType(WriteOperationType.INSERT_OVERWRITE);
     HoodieTestTable.of(metaClient)
-        .addReplaceCommit(instantTime, requestedReplaceMetadata, replaceMetadata)
+        .addReplaceCommit(instantTime, Option.of(requestedReplaceMetadata), Option.of(inflightReplaceMetadata), completeReplaceMetadata)
         .withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
   }
 

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestCleaner.java

@@ -843,10 +843,11 @@ public class TestCleaner extends HoodieClientTestBase {
     assertTrue(testTable.baseFileExists(p1, "00000000000001", file1P1C0));
 
     // make next replacecommit, with 1 clustering operation. logically delete p0. No change to p1
+    // notice that clustering generates empty inflight commit files
     Map<String, String> partitionAndFileId002 = testTable.forReplaceCommit("00000000000002").getFileIdsWithBaseFilesInPartitions(p0);
     String file2P0C1 = partitionAndFileId002.get(p0);
     Pair<HoodieRequestedReplaceMetadata, HoodieReplaceCommitMetadata> replaceMetadata = generateReplaceCommitMetadata(p0, file1P0C0, file2P0C1);
-    testTable.addReplaceCommit("00000000000002", replaceMetadata.getKey(), replaceMetadata.getValue());
+    testTable.addReplaceCommit("00000000000002", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
 
     // run cleaner
     List<HoodieCleanStat> hoodieCleanStatsTwo = runCleaner(config);
@@ -856,10 +857,11 @@ public class TestCleaner extends HoodieClientTestBase {
     assertTrue(testTable.baseFileExists(p1, "00000000000001", file1P1C0));
 
     // make next replacecommit, with 1 clustering operation. Replace data in p1. No change to p0
+    // notice that clustering generates empty inflight commit files
     Map<String, String> partitionAndFileId003 = testTable.forReplaceCommit("00000000000003").getFileIdsWithBaseFilesInPartitions(p1);
     String file3P1C2 = partitionAndFileId003.get(p1);
     replaceMetadata = generateReplaceCommitMetadata(p1, file1P1C0, file3P1C2);
-    testTable.addReplaceCommit("00000000000003", replaceMetadata.getKey(), replaceMetadata.getValue());
+    testTable.addReplaceCommit("00000000000003", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
 
     // run cleaner
     List<HoodieCleanStat> hoodieCleanStatsThree = runCleaner(config);
@@ -870,10 +872,11 @@ public class TestCleaner extends HoodieClientTestBase {
     assertTrue(testTable.baseFileExists(p1, "00000000000001", file1P1C0));
 
     // make next replacecommit, with 1 clustering operation. Replace data in p0 again
+    // notice that clustering generates empty inflight commit files
     Map<String, String> partitionAndFileId004 = testTable.forReplaceCommit("00000000000004").getFileIdsWithBaseFilesInPartitions(p0);
     String file4P0C3 = partitionAndFileId004.get(p0);
     replaceMetadata = generateReplaceCommitMetadata(p0, file2P0C1, file4P0C3);
-    testTable.addReplaceCommit("00000000000004", replaceMetadata.getKey(), replaceMetadata.getValue());
+    testTable.addReplaceCommit("00000000000004", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
 
     // run cleaner
     List<HoodieCleanStat> hoodieCleanStatsFour = runCleaner(config);
@@ -885,10 +888,11 @@ public class TestCleaner extends HoodieClientTestBase {
     assertTrue(testTable.baseFileExists(p1, "00000000000001", file1P1C0));
 
     // make next replacecommit, with 1 clustering operation. Replace all data in p1. no new files created
+    // notice that clustering generates empty inflight commit files
     Map<String, String> partitionAndFileId005 = testTable.forReplaceCommit("00000000000005").getFileIdsWithBaseFilesInPartitions(p1);
     String file4P1C4 = partitionAndFileId005.get(p1);
     replaceMetadata = generateReplaceCommitMetadata(p0, file3P1C2, file4P1C4);
-    testTable.addReplaceCommit("00000000000005", replaceMetadata.getKey(), replaceMetadata.getValue());
+    testTable.addReplaceCommit("00000000000005", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
     
     List<HoodieCleanStat> hoodieCleanStatsFive = runCleaner(config, 2);
     assertTrue(testTable.baseFileExists(p0, "00000000000004", file4P0C3));

hudi-common/src/main/avro/HoodieArchivedMetaEntry.avsc

@@ -112,6 +112,14 @@
             "HoodieRequestedReplaceMetadata"
          ],
          "default": null
+      },
+      {
+         "name":"HoodieInflightReplaceMetadata",
+         "type":[
+            "null",
+            "HoodieCommitMetadata"
+         ],
+         "default": null
       }
    ]
 }

hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java

@@ -34,6 +34,7 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
 import org.apache.hudi.common.table.view.TableFileSystemView;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.exception.HoodieException;
 
 import org.apache.hadoop.fs.FileSystem;
@@ -158,12 +159,20 @@ public class FileCreateUtils {
     createMetaFile(basePath, instantTime, HoodieTimeline.REPLACE_COMMIT_EXTENSION, metadata.toJsonString().getBytes(StandardCharsets.UTF_8));
   }
 
-  public static void createRequestedReplaceCommit(String basePath, String instantTime, HoodieRequestedReplaceMetadata requestedReplaceMetadata) throws IOException {
+  public static void createRequestedReplaceCommit(String basePath, String instantTime, Option<HoodieRequestedReplaceMetadata> requestedReplaceMetadata) throws IOException {
-    createMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_REPLACE_COMMIT_EXTENSION, serializeRequestedReplaceMetadata(requestedReplaceMetadata).get());
+    if (requestedReplaceMetadata.isPresent()) {
+      createMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_REPLACE_COMMIT_EXTENSION, serializeRequestedReplaceMetadata(requestedReplaceMetadata.get()).get());
+    } else {
+      createMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_REPLACE_COMMIT_EXTENSION);
+    }
   }
 
-  public static void createInflightReplaceCommit(String basePath, String instantTime) throws IOException {
+  public static void createInflightReplaceCommit(String basePath, String instantTime, Option<HoodieCommitMetadata> inflightReplaceMetadata) throws IOException {
-    createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_REPLACE_COMMIT_EXTENSION);
+    if (inflightReplaceMetadata.isPresent()) {
+      createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_REPLACE_COMMIT_EXTENSION, inflightReplaceMetadata.get().toJsonString().getBytes(StandardCharsets.UTF_8));
+    } else {
+      createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_REPLACE_COMMIT_EXTENSION);
+    }
   }
 
   public static void createCleanFile(String basePath, String instantTime, HoodieCleanMetadata metadata) throws IOException {
@@ -199,10 +208,6 @@ public class FileCreateUtils {
     createAuxiliaryMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_COMPACTION_EXTENSION);
   }
 
-  public static void createInflightCompaction(String basePath, String instantTime) throws IOException {
-    createAuxiliaryMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_COMPACTION_EXTENSION);
-  }
-
   public static void createPartitionMetaFile(String basePath, String partitionPath) throws IOException {
     Path parentPath = Paths.get(basePath, partitionPath);
     Files.createDirectories(parentPath);

hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java

@@ -65,7 +65,6 @@ import static org.apache.hudi.common.testutils.FileCreateUtils.createCommit;
 import static org.apache.hudi.common.testutils.FileCreateUtils.createDeltaCommit;
 import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCleanFile;
 import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCommit;
-import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCompaction;
 import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightDeltaCommit;
 import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightReplaceCommit;
 import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightRollbackFile;
@@ -128,13 +127,6 @@ public class HoodieTestTable {
     return this;
   }
 
-  public HoodieTestTable addRequestedDeltaCommit(String instantTime) throws Exception {
-    createRequestedDeltaCommit(basePath, instantTime);
-    currentInstantTime = instantTime;
-    metaClient = HoodieTableMetaClient.reload(metaClient);
-    return this;
-  }
-
   public HoodieTestTable addInflightCommit(String instantTime) throws Exception {
     createRequestedCommit(basePath, instantTime);
     createInflightCommit(basePath, instantTime);
@@ -143,14 +135,6 @@ public class HoodieTestTable {
     return this;
   }
 
-  public HoodieTestTable addInflightDeltaCommit(String instantTime) throws Exception {
-    createRequestedDeltaCommit(basePath, instantTime);
-    createInflightDeltaCommit(basePath, instantTime);
-    currentInstantTime = instantTime;
-    metaClient = HoodieTableMetaClient.reload(metaClient);
-    return this;
-  }
-
   public HoodieTestTable addCommit(String instantTime) throws Exception {
     createRequestedCommit(basePath, instantTime);
     createInflightCommit(basePath, instantTime);
@@ -178,16 +162,20 @@ public class HoodieTestTable {
     return this;
   }
 
-  public HoodieTestTable addReplaceCommit(String instantTime, HoodieRequestedReplaceMetadata requestedReplaceMetadata, HoodieReplaceCommitMetadata metadata) throws Exception {
+  public HoodieTestTable addReplaceCommit(
+      String instantTime,
+      Option<HoodieRequestedReplaceMetadata> requestedReplaceMetadata,
+      Option<HoodieCommitMetadata> inflightReplaceMetadata,
+      HoodieReplaceCommitMetadata completeReplaceMetadata) throws Exception {
     createRequestedReplaceCommit(basePath, instantTime, requestedReplaceMetadata);
-    createInflightReplaceCommit(basePath, instantTime);
+    createInflightReplaceCommit(basePath, instantTime, inflightReplaceMetadata);
-    createReplaceCommit(basePath, instantTime, metadata);
+    createReplaceCommit(basePath, instantTime, completeReplaceMetadata);
     currentInstantTime = instantTime;
     metaClient = HoodieTableMetaClient.reload(metaClient);
     return this;
   }
 
-  public HoodieTestTable addRequestedReplace(String instantTime, HoodieRequestedReplaceMetadata requestedReplaceMetadata) throws Exception {
+  public HoodieTestTable addRequestedReplace(String instantTime, Option<HoodieRequestedReplaceMetadata> requestedReplaceMetadata) throws Exception {
     createRequestedReplaceCommit(basePath, instantTime, requestedReplaceMetadata);
     currentInstantTime = instantTime;
     metaClient = HoodieTableMetaClient.reload(metaClient);
@@ -247,14 +235,6 @@ public class HoodieTestTable {
     return addRequestedCompaction(instantTime, plan);
   }
 
-  public HoodieTestTable addCompaction(String instantTime) throws IOException {
-    createRequestedCompaction(basePath, instantTime);
-    createInflightCompaction(basePath, instantTime);
-    currentInstantTime = instantTime;
-    metaClient = HoodieTableMetaClient.reload(metaClient);
-    return this;
-  }
-
   public HoodieTestTable forCommit(String instantTime) {
     currentInstantTime = instantTime;
     return this;
@@ -270,11 +250,6 @@ public class HoodieTestTable {
     return this;
   }
 
-  public HoodieTestTable forCompaction(String instantTime) {
-    currentInstantTime = instantTime;
-    return this;
-  }
-
   public HoodieTestTable withPartitionMetaFiles(String... partitionPaths) throws IOException {
     for (String partitionPath : partitionPaths) {
       FileCreateUtils.createPartitionMetaFile(basePath, partitionPath);
@@ -282,10 +257,6 @@ public class HoodieTestTable {
     return this;
   }
 
-  public HoodieTestTable withMarkerFile(String partitionPath, IOType ioType) throws IOException {
-    return withMarkerFile(partitionPath, UUID.randomUUID().toString(), ioType);
-  }
-
   public HoodieTestTable withMarkerFile(String partitionPath, String fileId, IOType ioType) throws IOException {
     createMarkerFile(basePath, partitionPath, currentInstantTime, fileId, ioType);
     return this;
@@ -357,10 +328,6 @@ public class HoodieTestTable {
     return this;
   }
 
-  public boolean inflightCommitsExist(String... instantTime) {
-    return Arrays.stream(instantTime).allMatch(this::inflightCommitExists);
-  }
-
   public boolean inflightCommitExists(String instantTime) {
     try {
       return fs.exists(getInflightCommitFilePath(instantTime));
@@ -369,10 +336,6 @@ public class HoodieTestTable {
     }
   }
 
-  public boolean commitsExist(String... instantTime) {
-    return Arrays.stream(instantTime).allMatch(this::commitExists);
-  }
-
   public boolean commitExists(String instantTime) {
     try {
       return fs.exists(getCommitFilePath(instantTime));
@@ -389,10 +352,6 @@ public class HoodieTestTable {
     });
   }
 
-  public boolean baseFilesExist(String partition, String instantTime, String... fileIds) {
-    return Arrays.stream(fileIds).allMatch(f -> baseFileExists(partition, instantTime, f));
-  }
-
   public boolean baseFileExists(String partition, String instantTime, String fileId) {
     try {
       return fs.exists(new Path(Paths.get(basePath, partition, baseFileName(instantTime, fileId)).toString()));