[HUDI-2833][Design] Merge small archive files instead of expanding indefinitely.

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java

@@ -249,6 +249,19 @@ public class HoodieCompactionConfig extends HoodieConfig {
           + "record size estimate compute dynamically based on commit metadata. "
           + " This is critical in computing the insert parallelism and bin-packing inserts into small files.");
 
+  public static final ConfigProperty<String> MAX_ARCHIVE_FILES_TO_KEEP_PROP = ConfigProperty
+      .key("hoodie.max.archive.files")
+      .defaultValue("10")
+      .withDocumentation("The numbers of kept archive files under archived.");
+
+  public static final ConfigProperty<String> AUTO_TRIM_ARCHIVE_FILES_DROP = ConfigProperty
+      .key("hoodie.auto.trim.archive.files")
+      .defaultValue("false")
+      .withDocumentation("When enabled, Hoodie will keep the most recent " + MAX_ARCHIVE_FILES_TO_KEEP_PROP.key()
+          + " archive files and delete older one which lose part of archived instants information.");
+
+
+
   /** @deprecated Use {@link #CLEANER_POLICY} and its methods instead */
   @Deprecated
   public static final String CLEANER_POLICY_PROP = CLEANER_POLICY.key();
@@ -541,6 +554,16 @@ public Builder archiveCommitsWith(int minToKeep, int maxToKeep) {
       return this;
     }
 
+    public Builder maxArchiveFilesToKeep(int number) {
+      compactionConfig.setValue(MAX_ARCHIVE_FILES_TO_KEEP_PROP, String.valueOf(number));
+      return this;
+    }
+
+    public Builder withAutoTrimArchiveFiles(boolean enable) {
+      compactionConfig.setValue(AUTO_TRIM_ARCHIVE_FILES_DROP, String.valueOf(enable));
+      return this;
+    }
+
     public Builder compactionSmallFileSize(long smallFileLimitBytes) {
       compactionConfig.setValue(PARQUET_SMALL_FILE_LIMIT, String.valueOf(smallFileLimitBytes));
       return this;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -1073,6 +1073,10 @@ public int getMinCommitsToKeep() {
     return getInt(HoodieCompactionConfig.MIN_COMMITS_TO_KEEP);
   }
 
+  public int getMaxArchiveFilesToKeep() {
+    return getInt(HoodieCompactionConfig.MAX_ARCHIVE_FILES_TO_KEEP_PROP);
+  }
+
   public int getParquetSmallFileLimit() {
     return getInt(HoodieCompactionConfig.PARQUET_SMALL_FILE_LIMIT);
   }
@@ -1101,6 +1105,10 @@ public boolean isAutoClean() {
     return getBoolean(HoodieCompactionConfig.AUTO_CLEAN);
   }
 
+  public boolean getAutoTrimArchiveFilesEnable() {
+    return getBoolean(HoodieCompactionConfig.AUTO_TRIM_ARCHIVE_FILES_DROP);
+  }
+
   public boolean isAutoArchive() {
     return getBoolean(HoodieCompactionConfig.AUTO_ARCHIVE);
   }

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

@@ -26,6 +26,7 @@
 import org.apache.hudi.common.model.HoodieArchivedLogFile;
 import org.apache.hudi.common.model.HoodieAvroPayload;
 import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
+import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.log.HoodieLogFormat;
 import org.apache.hudi.common.table.log.HoodieLogFormat.Writer;
@@ -80,6 +81,7 @@ public class HoodieTimelineArchiveLog<T extends HoodieAvroPayload, I, K, O> {
   private Writer writer;
   private final int maxInstantsToKeep;
   private final int minInstantsToKeep;
+  private final int maxArchiveFilesToKeep;
   private final HoodieTable<T, I, K, O> table;
   private final HoodieTableMetaClient metaClient;
 
@@ -90,6 +92,7 @@ public HoodieTimelineArchiveLog(HoodieWriteConfig config, HoodieTable<T, I, K, O
     this.archiveFilePath = HoodieArchivedTimeline.getArchiveLogPath(metaClient.getArchivePath());
     this.maxInstantsToKeep = config.getMaxCommitsToKeep();
     this.minInstantsToKeep = config.getMinCommitsToKeep();
+    this.maxArchiveFilesToKeep = config.getMaxArchiveFilesToKeep();
   }
 
   private Writer openWriter() {
@@ -130,6 +133,9 @@ public boolean archiveIfRequired(HoodieEngineContext context) throws IOException
         archive(context, instantsToArchive);
         LOG.info("Deleting archived instants " + instantsToArchive);
         success = deleteArchivedInstants(instantsToArchive, context);
+        if (config.getAutoTrimArchiveFilesEnable()) {
+          trimArchiveFilesIfNecessary(context);
+        }
       } else {
         LOG.info("No Instants to archive");
       }
@@ -140,6 +146,48 @@ public boolean archiveIfRequired(HoodieEngineContext context) throws IOException
     }
   }
 
+  private void trimArchiveFilesIfNecessary(HoodieEngineContext context) throws IOException {
+    Stream<HoodieLogFile> allLogFiles = FSUtils.getAllLogFiles(metaClient.getFs(),
+        archiveFilePath.getParent(),
+        archiveFilePath.getName(),
+        HoodieArchivedLogFile.ARCHIVE_EXTENSION,
+        "");
+    List<HoodieLogFile> sortedLogFilesList = allLogFiles.sorted(HoodieLogFile.getReverseLogFileComparator()).collect(Collectors.toList());
+    if (!sortedLogFilesList.isEmpty()) {
+      List<String> skipped = sortedLogFilesList.stream().skip(maxArchiveFilesToKeep).map(HoodieLogFile::getPath).map(Path::toString).collect(Collectors.toList());
+      if (!skipped.isEmpty()) {
+        LOG.info("Deleting archive files :  " + skipped);
+        context.setJobStatus(this.getClass().getSimpleName(), "Delete archive files");
+        Map<String, Boolean> result = deleteFilesParallelize(metaClient, skipped, context, true);
+      }
+    }
+  }
+
+  private Map<String, Boolean> deleteFilesParallelize(HoodieTableMetaClient metaClient, List<String> paths, HoodieEngineContext context, boolean ignoreFailed) {
+
+    return FSUtils.parallelizeFilesProcess(context,
+        metaClient.getFs(),
+        config.getArchiveDeleteParallelism(),
+        pairOfSubPathAndConf -> {
+          Path file = new Path(pairOfSubPathAndConf.getKey());
+          try {
+            FileSystem fs = metaClient.getFs();
+            if (fs.exists(file)) {
+              return fs.delete(file, false);
+            }
+            return true;
+          } catch (IOException e) {
+            if (!ignoreFailed) {
+              throw new HoodieIOException("Failed to delete : " + file, e);
+            } else {
+              LOG.warn("Ignore failed deleting : " + file);
+              return true;
+            }
+          }
+        },
+        paths);
+  }
+
   private Stream<HoodieInstant> getCleanInstantsToArchive() {
     HoodieTimeline cleanAndRollbackTimeline = table.getActiveTimeline()
         .getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.CLEAN_ACTION, HoodieTimeline.ROLLBACK_ACTION)).filterCompletedInstants();
@@ -234,22 +282,7 @@ private boolean deleteArchivedInstants(List<HoodieInstant> archivedInstants, Hoo
     }).map(Path::toString).collect(Collectors.toList());
 
     context.setJobStatus(this.getClass().getSimpleName(), "Delete archived instants");
-    Map<String, Boolean> resultDeleteInstantFiles = FSUtils.parallelizeFilesProcess(context,
-        metaClient.getFs(),
-        config.getArchiveDeleteParallelism(),
-        pairOfSubPathAndConf -> {
-          Path commitFile = new Path(pairOfSubPathAndConf.getKey());
-          try {
-            FileSystem fs = commitFile.getFileSystem(pairOfSubPathAndConf.getValue().get());
-            if (fs.exists(commitFile)) {
-              return fs.delete(commitFile, false);
-            }
-            return true;
-          } catch (IOException e) {
-            throw new HoodieIOException("Failed to delete archived instant " + commitFile, e);
-          }
-        },
-        instantFiles);
+    Map<String, Boolean> resultDeleteInstantFiles = deleteFilesParallelize(metaClient, instantFiles, context, false);
 
     for (Map.Entry<String, Boolean> result : resultDeleteInstantFiles.entrySet()) {
       LOG.info("Archived and deleted instant file " + result.getKey() + " : " + result.getValue());

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

@@ -18,6 +18,8 @@
 
 package org.apache.hudi.io;
 
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hudi.avro.model.HoodieActionInstant;
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
 import org.apache.hudi.avro.model.HoodieCleanerPlan;
@@ -59,6 +61,10 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
@@ -119,18 +125,46 @@ public void clean() throws IOException {
   }
 
   private HoodieWriteConfig initTestTableAndGetWriteConfig(boolean enableMetadata, int minArchivalCommits, int maxArchivalCommits, int maxDeltaCommitsMetadataTable) throws Exception {
-    return initTestTableAndGetWriteConfig(enableMetadata, minArchivalCommits, maxArchivalCommits, maxDeltaCommitsMetadataTable, HoodieTableType.COPY_ON_WRITE);
+    return initTestTableAndGetWriteConfig(enableMetadata, minArchivalCommits, maxArchivalCommits, maxDeltaCommitsMetadataTable, HoodieTableType.COPY_ON_WRITE, false, 10);
   }
 
-  private HoodieWriteConfig initTestTableAndGetWriteConfig(boolean enableMetadata, int minArchivalCommits, int maxArchivalCommits, int maxDeltaCommitsMetadataTable,
+  private HoodieWriteConfig initTestTableAndGetWriteConfig(boolean enableMetadata,
+                                                           int minArchivalCommits,
+                                                           int maxArchivalCommits,
+                                                           int maxDeltaCommitsMetadataTable,
                                                            HoodieTableType tableType) throws Exception {
+    return initTestTableAndGetWriteConfig(enableMetadata, minArchivalCommits, maxArchivalCommits, maxDeltaCommitsMetadataTable, tableType, false, 10);
+  }
+
+  private HoodieWriteConfig initTestTableAndGetWriteConfig(boolean enableMetadata,
+                                                           int minArchivalCommits,
+                                                           int maxArchivalCommits,
+                                                           int maxDeltaCommitsMetadataTable,
+                                                           boolean enableArchiveTrim,
+                                                           int archiveFilesToKeep) throws Exception {
+    return initTestTableAndGetWriteConfig(enableMetadata, minArchivalCommits, maxArchivalCommits, maxDeltaCommitsMetadataTable, HoodieTableType.COPY_ON_WRITE, enableArchiveTrim, archiveFilesToKeep);
+  }
+
+  private HoodieWriteConfig initTestTableAndGetWriteConfig(boolean enableMetadata,
+                                                           int minArchivalCommits,
+                                                           int maxArchivalCommits,
+                                                           int maxDeltaCommitsMetadataTable,
+                                                           HoodieTableType tableType,
+                                                           boolean enableArchiveTrim,
+                                                           int archiveFilesToKeep) throws Exception {
     init(tableType);
     HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder().withPath(basePath)
         .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA).withParallelism(2, 2)
-        .withCompactionConfig(HoodieCompactionConfig.newBuilder().retainCommits(1).archiveCommitsWith(minArchivalCommits, maxArchivalCommits).build())
+        .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+            .retainCommits(1)
+            .withAutoTrimArchiveFiles(enableArchiveTrim)
+            .maxArchiveFilesToKeep(archiveFilesToKeep)
+            .archiveCommitsWith(minArchivalCommits, maxArchivalCommits)
+            .build())
         .withFileSystemViewConfig(FileSystemViewStorageConfig.newBuilder()
             .withRemoteServerPort(timelineServicePort).build())
-        .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(enableMetadata)
+        .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+            .enable(enableMetadata)
             .withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommitsMetadataTable).build())
         .forTable("test-trip-table").build();
     initWriteConfigAndMetatableWriter(writeConfig, enableMetadata);
@@ -183,6 +217,41 @@ public void testArchiveTableWithArchival(boolean enableMetadata) throws Exceptio
     }
   }
 
+  @ParameterizedTest
+  @MethodSource("testArchiveTableWithArchivalTrim")
+  public void testArchiveTableWithArchivalCleanUp(boolean enableMetadata, boolean enableArchiveTrim, int archiveFilesToKeep) throws Exception {
+    HashSet<String> archiveFilesExisted = new HashSet<>();
+    ArrayList<String> currentExistArchiveFiles = new ArrayList<>();
+    HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(enableMetadata, 2, 3, 2, enableArchiveTrim, archiveFilesToKeep);
+    String archivePath = metaClient.getArchivePath();
+    for (int i = 1; i < 10; i++) {
+      testTable.doWriteOperation("0000000" + i, WriteOperationType.UPSERT, i == 1 ? Arrays.asList("p1", "p2") : Collections.emptyList(), Arrays.asList("p1", "p2"), 2);
+      // trigger archival
+      archiveAndGetCommitsList(writeConfig);
+      RemoteIterator<LocatedFileStatus> iter = metaClient.getFs().listFiles(new Path(archivePath), false);
+      ArrayList<String> files = new ArrayList<>();
+      while (iter.hasNext()) {
+        files.add(iter.next().getPath().toString());
+      }
+      archiveFilesExisted.addAll(files);
+      currentExistArchiveFiles = files;
+    }
+
+    assertEquals(archiveFilesToKeep, currentExistArchiveFiles.size());
+
+    if (enableArchiveTrim) {
+      // sort archive files path
+      List<String> sorted = archiveFilesExisted.stream().sorted().collect(Collectors.toList());
+      List<String> archiveFilesDeleted = sorted.subList(0, 3 - archiveFilesToKeep);
+      List<String> archiveFilesKept = sorted.subList(3 - archiveFilesToKeep, sorted.size());
+
+      // assert older archive files are deleted
+      assertFalse(currentExistArchiveFiles.containsAll(archiveFilesDeleted));
+      // assert most recent archive files are preserved
+      assertTrue(currentExistArchiveFiles.containsAll(archiveFilesKept));
+    }
+  }
+
   @ParameterizedTest
   @ValueSource(booleans = {true, false})
   public void testNoArchivalUntilMaxArchiveConfigWithExtraInflightCommits(boolean enableMetadata) throws Exception {
@@ -723,4 +792,14 @@ private HoodieInstant createRollbackMetadata(String rollbackTime, String commitT
     }
     return new HoodieInstant(inflight, "rollback", rollbackTime);
   }
+
+  private static Stream<Arguments> testArchiveTableWithArchivalTrim() {
+    // boolean enableMetadata, boolean enableArchiveTrim, int archiveFilesToKeep
+    return Stream.of(
+        arguments(false, true, 1),
+        arguments(false, false, 3),
+        arguments(true, true, 1),
+        arguments(true, false, 3)
+    );
+  }
 }