feat: Introduce the configuration for forcibly triggering timeline compaction

1. Introduce the configuration for forcibly triggering timeline compaction

Signed-off-by: TheR1sing3un <[email protected]>

feat: enable timeline compaction by default

1. enable timeline compaction by default

Signed-off-by: TheR1sing3un <[email protected]>

rerun

Author: TheR1sing3un

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/timeline/versioning/v2/TimelineArchiverV2.java

@@ -115,8 +115,6 @@ public int archiveIfRequired(HoodieEngineContext context, boolean acquireLock) t
         this.timelineWriter.write(instantsToArchive, Option.of(action -> deleteAnyLeftOverMarkers(context, action)), Option.of(exceptionHandler));
         log.debug("Deleting archived instants");
         deleteArchivedActions(instantsToArchive, context);
-        // triggers compaction and cleaning only after archiving action
-        this.timelineWriter.compactAndClean(context);
         Supplier<List<HoodieInstant>> archivedInstants = () -> instantsToArchive.stream()
             .flatMap(action -> Stream.concat(action.getCompletedInstants().stream(), action.getPendingInstants().stream()))
             .collect(Collectors.toList());
@@ -125,6 +123,10 @@ public int archiveIfRequired(HoodieEngineContext context, boolean acquireLock) t
       } else {
         log.info("No Instants to archive");
       }
+      // run compact and clean if needed even no instants were archived
+      if (!instantsToArchive.isEmpty() || config.isTimelineCompactionForced()) {
+        this.timelineWriter.compactAndClean(context);
+      }
       return instantsToArchive.size();
     } finally {
       if (acquireLock) {

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

@@ -102,6 +102,14 @@ public class HoodieArchivalConfig extends HoodieConfig {
       .withDocumentation("If enabled, archival will proceed beyond savepoint, skipping savepoint commits."
           + " If disabled, archival will stop at the earliest savepoint commit.");
 
+  public static final ConfigProperty<Boolean> TIMELINE_COMPACTION_FORCED = ConfigProperty
+      .key("hoodie.timeline.compaction.forced")
+      .defaultValue(true)
+      .markAdvanced()
+      .withDocumentation("If enabled, timeline compaction will be forced to run during archival of timeline."
+          + " This helps in reducing the number of files in the archived timeline, at the cost of"
+          + " additional compaction time during archival.");
+
   public static final ConfigProperty<Long> TIMELINE_COMPACTION_TARGET_FILE_MAX_BYTES = ConfigProperty
         .key("hoodie.timeline.compaction.target.file.max.bytes")
         .defaultValue(1000L * 1024 * 1024)
@@ -198,6 +206,11 @@ public Builder withArchiveBeyondSavepoint(boolean archiveBeyondSavepoint) {
       return this;
     }
 
+    public Builder withTimelineCompactionForced(boolean timelineCompactionForced) {
+      archivalConfig.setValue(TIMELINE_COMPACTION_FORCED, String.valueOf(timelineCompactionForced));
+      return this;
+    }
+
     public HoodieArchivalConfig build() {
       archivalConfig.setDefaults(HoodieArchivalConfig.class.getName());
       return archivalConfig;

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

@@ -1916,6 +1916,10 @@ public int getCommitArchivalBatchSize() {
     return getInt(HoodieArchivalConfig.COMMITS_ARCHIVAL_BATCH_SIZE);
   }
 
+  public boolean isTimelineCompactionForced() {
+    return getBoolean(HoodieArchivalConfig.TIMELINE_COMPACTION_FORCED);
+  }
+
   public Boolean shouldCleanBootstrapBaseFile() {
     return getBoolean(HoodieCleanConfig.CLEANER_BOOTSTRAP_BASE_FILE_ENABLE);
   }

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

@@ -725,6 +725,49 @@ public void testCompactionCleaning() throws Exception {
     assertEquals(Arrays.asList(7, 8, 9), LSMTimeline.allSnapshotVersions(metaClient, metaClient.getArchivePath()).stream().sorted().collect(Collectors.toList()));
   }
 
+  @Test
+  public void testCompactionForced() throws Exception {
+    HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(true, 4, 5, 2, 3);
+
+    // do ingestion and trigger archive actions here.
+    for (int i = 1; i < 21; i++) {
+      testTable.doWriteOperation(
+          WriteClientTestUtils.createNewInstantTime(), WriteOperationType.UPSERT, i == 1 ? Arrays.asList("p1", "p2") : Collections.emptyList(), Arrays.asList("p1", "p2"), 2);
+      archiveAndGetCommitsList(writeConfig);
+    }
+
+    // loading archived timeline and active timeline success
+    HoodieActiveTimeline rawActiveTimeline = TIMELINE_FACTORY.createActiveTimeline(metaClient, false);
+    HoodieArchivedTimeline archivedTimeLine = metaClient.getArchivedTimeline();
+    assertEquals(4 * 3 + 16, rawActiveTimeline.countInstants() + archivedTimeLine.countInstants());
+
+    assertEquals(10, LSMTimeline.latestSnapshotVersion(metaClient, metaClient.getArchivePath()));
+    assertEquals(Arrays.asList(8, 9, 10), LSMTimeline.allSnapshotVersions(metaClient, metaClient.getArchivePath()).stream().sorted().collect(Collectors.toList()));
+
+    HoodieLSMTimelineManifest latestSnapshotManifest = LSMTimeline.latestSnapshotManifest(metaClient, metaClient.getArchivePath());
+    Map<Integer, List<Pair<Integer, HoodieLSMTimelineManifest.LSMFileEntry>>> layeredFiles =
+        latestSnapshotManifest.getFiles().stream().map(file -> Pair.of(LSMTimeline.getFileLayer(file.getFileName()), file)).collect(Collectors.groupingBy(Pair::getKey));
+    assertEquals(2, layeredFiles.get(0).size());
+    assertEquals(2, layeredFiles.get(1).size());
+
+    // run archive again without new commits and force-compaction, there will be no new snapshot version created
+    writeConfig.setValue(HoodieArchivalConfig.TIMELINE_COMPACTION_FORCED.key(), "false");
+    writeConfig.setValue(HoodieArchivalConfig.TIMELINE_COMPACTION_BATCH_SIZE.key(), "2");
+    archiveAndGetCommitsList(writeConfig, true);
+    assertEquals(10, LSMTimeline.latestSnapshotVersion(metaClient, metaClient.getArchivePath()));
+
+    // run archive again without new commits but with force-compaction and there will be new snapshot version created
+    writeConfig.setValue(HoodieArchivalConfig.TIMELINE_COMPACTION_FORCED.key(), "true");
+    archiveAndGetCommitsList(writeConfig, true);
+    assertEquals(12, LSMTimeline.latestSnapshotVersion(metaClient, metaClient.getArchivePath()));
+    latestSnapshotManifest = LSMTimeline.latestSnapshotManifest(metaClient, metaClient.getArchivePath());
+    layeredFiles = latestSnapshotManifest.getFiles().stream().map(file -> Pair.of(LSMTimeline.getFileLayer(file.getFileName()), file)).collect(Collectors.groupingBy(Pair::getKey));
+    assertFalse(layeredFiles.containsKey(0));
+    assertEquals(1, layeredFiles.get(1).size());
+    assertEquals(1, layeredFiles.get(2).size());
+
+  }
+
   @Test
   public void testCompactionWithLargeL0File() throws Exception {
     HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(true, 4, 5, 2, 3);