[HUDI-5032] Add archive to cli

hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java

@@ -28,7 +28,12 @@
 import org.apache.hudi.cli.HoodieCLI;
 import org.apache.hudi.cli.HoodiePrintHelper;
 import org.apache.hudi.cli.TableHeader;
+import org.apache.hudi.client.HoodieTimelineArchiver;
+import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieAvroPayload;
 import org.apache.hudi.common.model.HoodieLogFile;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.log.HoodieLogFormat;
@@ -37,6 +42,16 @@
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.ClosableIterator;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCleanConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.table.HoodieSparkTable;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
 import org.springframework.shell.standard.ShellComponent;
 import org.springframework.shell.standard.ShellMethod;
 import org.springframework.shell.standard.ShellOption;
@@ -52,6 +67,35 @@
  */
 @ShellComponent
 public class ArchivedCommitsCommand {
+  private static final Logger LOG = LogManager.getLogger(ArchivedCommitsCommand.class);
+  private JavaSparkContext jsc;
+  @ShellMethod(key = "trigger archival", value = "trigger archival")
+  public void triggerArchival(
+      @ShellOption(value = {"--minCommits"},
+        help = "Minimum number of instants to retain in the active timeline. See hoodie.keep.min.commits",
+        defaultValue = "20") int minCommits,
+      @ShellOption(value = {"--maxCommits"},
+          help = "Maximum number of instants to retain in the active timeline. See hoodie.keep.max.commits",
+          defaultValue = "30") int maxCommits,
+      @ShellOption(value = {"--commitsRetained"}, help = "Number of commits to retain, without cleaning",
+          defaultValue = "10") int retained) {
+
+    initJavaSparkContext();
+    HoodieWriteConfig config = HoodieWriteConfig.newBuilder().withPath(HoodieCLI.basePath)
+        .withArchivalConfig(HoodieArchivalConfig.newBuilder().archiveCommitsWith(minCommits,maxCommits).build())
+        .withCleanConfig(HoodieCleanConfig.newBuilder().retainCommits(retained).build())
+        .withEmbeddedTimelineServerEnabled(false)
+        .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
+        .build();
+    HoodieEngineContext context = new HoodieSparkEngineContext(jsc);
+    HoodieSparkTable<HoodieAvroPayload> table = HoodieSparkTable.create(config, context);
+    try {
+      HoodieTimelineArchiver archiver = new HoodieTimelineArchiver(config, table);
+      archiver.archiveIfRequired(context,true);
+    } catch (IOException ioe) {
+      LOG.error("Failed to archive with IOException: " + ioe);
+    }
+  }
 
   @ShellMethod(key = "show archived commit stats", value = "Read commits from archived files and show details")
   public String showArchivedCommits(
@@ -206,4 +250,22 @@ private Comparable[] readCommit(GenericRecord record, boolean skipMetadata) {
       return new Comparable[] {};
     }
   }
+
+  private void initJavaSparkContext() {
+    if (jsc == null) {
+      SparkConf sparkConf = new SparkConf();
+      sparkConf.set("spark.app.name", getClass().getName());
+      sparkConf.set("spark.master", "local[*]");
+      sparkConf.set("spark.default.parallelism", "4");
+      sparkConf.set("spark.sql.shuffle.partitions", "4");
+      sparkConf.set("spark.driver.maxResultSize", "2g");
+      sparkConf.set("spark.hadoop.mapred.output.compress", "true");
+      sparkConf.set("spark.hadoop.mapred.output.compression.codec", "true");
+      sparkConf.set("spark.hadoop.mapred.output.compression.codec", "org.apache.hadoop.io.compress.GzipCodec");
+      sparkConf.set("spark.hadoop.mapred.output.compression.type", "BLOCK");
+      sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
+      SparkSession spark = SparkSession.builder().config(sparkConf).getOrCreate();
+      jsc = new JavaSparkContext(spark.sparkContext());
+    }
+  }
 }

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

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.cli.commands;
+
+import org.apache.hudi.cli.HoodieCLI;
+import org.apache.hudi.cli.functional.CLIFunctionalTestHarness;
+import org.apache.hudi.cli.testutils.HoodieTestCommitMetadataGenerator;
+import org.apache.hudi.cli.testutils.ShellEvaluationResultUtil;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.shell.Shell;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@Tag("functional")
+@SpringBootTest(properties = {"spring.shell.interactive.enabled=false", "spring.shell.command.script.enabled=false"})
+public class TestArchiveCommand extends CLIFunctionalTestHarness {
+
+  @Autowired
+  private Shell shell;
+
+  @Test
+  public void testArchiving() throws Exception {
+    HoodieCLI.conf = hadoopConf();
+
+    // Create table and connect
+    String tableName = tableName();
+    String tablePath = tablePath(tableName);
+
+    new TableCommand().createTable(
+        tablePath, tableName,
+        "COPY_ON_WRITE", "", 1, "org.apache.hudi.common.model.HoodieAvroPayload");
+
+    HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
+
+    // Create six commits
+    for (int i = 100; i < 106; i++) {
+      String timestamp = String.valueOf(i);
+      HoodieTestCommitMetadataGenerator.createCommitFileWithMetadata(tablePath,timestamp, hadoopConf());
+    }
+
+    Object cmdResult = shell.evaluate(() -> "trigger archival --minCommits 2 --maxCommits 3 --commitsRetained 1");
+    assertTrue(ShellEvaluationResultUtil.isSuccess(cmdResult));
+    metaClient = HoodieTableMetaClient.reload(metaClient);
+
+    //get instants in the active timeline only returns the latest state of the commit
+    //therefore we expect 2 instants because minCommits is 2
+    assertEquals(2, metaClient.getActiveTimeline().getInstants().count());
+
+    //get instants in the archived timeline returns all instants in the commit
+    //therefore we expect 12 instants because 6 commits - 2 commits in active timeline = 4 in archived
+    //since each commit is completed, there are 3 instances per commit (requested, inflight, completed)
+    //and 3 instances per commit * 4 commits = 12 instances
+    assertEquals(12, metaClient.getArchivedTimeline().getInstants().count());
+  }
+
+}
+