[HUDI-5241] Optimize HoodieDefaultTimeline API

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

@@ -65,7 +65,7 @@ private String printCommits(HoodieDefaultTimeline timeline,
     final List<Comparable[]> rows = new ArrayList<>();
 
     final List<HoodieInstant> commits = timeline.getCommitsTimeline().filterCompletedInstants()
-        .getInstants().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
+        .getInstantsAsStream().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
 
     for (final HoodieInstant commit : commits) {
       if (timeline.getInstantDetails(commit).isPresent()) {
@@ -103,7 +103,7 @@ private String printCommitsWithMetadata(HoodieDefaultTimeline timeline,
     final List<Comparable[]> rows = new ArrayList<>();
 
     final List<HoodieInstant> commits = timeline.getCommitsTimeline().filterCompletedInstants()
-        .getInstants().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
+        .getInstantsAsStream().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
 
     for (final HoodieInstant commit : commits) {
       if (timeline.getInstantDetails(commit).isPresent()) {
@@ -372,20 +372,20 @@ public String compareCommits(@ShellOption(value = {"--path"}, help = "Path of th
     HoodieTimeline targetTimeline = target.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
     HoodieTimeline sourceTimeline = source.getActiveTimeline().getCommitsTimeline().filterCompletedInstants();
     String targetLatestCommit =
-        targetTimeline.getInstants().iterator().hasNext() ? targetTimeline.lastInstant().get().getTimestamp() : "0";
+        targetTimeline.getInstantsAsStream().iterator().hasNext() ? targetTimeline.lastInstant().get().getTimestamp() : "0";
     String sourceLatestCommit =
-        sourceTimeline.getInstants().iterator().hasNext() ? sourceTimeline.lastInstant().get().getTimestamp() : "0";
+        sourceTimeline.getInstantsAsStream().iterator().hasNext() ? sourceTimeline.lastInstant().get().getTimestamp() : "0";
 
     if (sourceLatestCommit != null
         && HoodieTimeline.compareTimestamps(targetLatestCommit, HoodieTimeline.GREATER_THAN, sourceLatestCommit)) {
       // source is behind the target
       List<String> commitsToCatchup = targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE)
-          .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+          .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
       return "Source " + source.getTableConfig().getTableName() + " is behind by " + commitsToCatchup.size()
           + " commits. Commits to catch up - " + commitsToCatchup;
     } else {
       List<String> commitsToCatchup = sourceTimeline.findInstantsAfter(targetLatestCommit, Integer.MAX_VALUE)
-          .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+          .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
       return "Source " + source.getTableConfig().getTableName() + " is ahead by " + commitsToCatchup.size()
           + " commits. Commits to catch up - " + commitsToCatchup;
     }

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

@@ -317,7 +317,7 @@ private static String printAllCompactions(HoodieDefaultTimeline timeline,
         .collect(Collectors.toList());
 
     Set<String> committedInstants = timeline.getCommitTimeline().filterCompletedInstants()
-        .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
+        .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
 
     List<Comparable[]> rows = new ArrayList<>();
     for (Pair<HoodieInstant, HoodieCompactionPlan> compactionPlan : compactionPlans) {

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

@@ -137,7 +137,7 @@ private String printDiffWithMetadata(HoodieDefaultTimeline timeline, Integer lim
                                        BiFunction<HoodieWriteStat, String, Boolean> diffEntityChecker) throws IOException {
     List<Comparable[]> rows = new ArrayList<>();
     List<HoodieInstant> commits = timeline.getCommitsTimeline().filterCompletedInstants()
-        .getInstants().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
+        .getInstantsAsStream().sorted(HoodieInstant.COMPARATOR.reversed()).collect(Collectors.toList());
 
     for (final HoodieInstant commit : commits) {
       Option<byte[]> instantDetails = timeline.getInstantDetails(commit);

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

@@ -87,7 +87,7 @@ public String exportInstants(
     // The non archived instants can be listed from the Timeline.
     HoodieTimeline timeline = HoodieCLI.getTableMetaClient().getActiveTimeline().filterCompletedInstants()
         .filter(i -> actionSet.contains(i.getAction()));
-    List<HoodieInstant> nonArchivedInstants = timeline.getInstants().collect(Collectors.toList());
+    List<HoodieInstant> nonArchivedInstants = timeline.getInstants();
 
     // Archived instants are in the commit archive files
     FileStatus[] statuses = FSUtils.getFs(basePath, HoodieCLI.conf).globStatus(archivePath);

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

@@ -256,7 +256,7 @@ private HoodieTableFileSystemView buildFileSystemView(String globRegex, String m
       timeline = timeline.filterCompletedInstants();
     }
 
-    instantsStream = timeline.getInstants();
+    instantsStream = timeline.getInstantsAsStream();
 
     if (!maxInstant.isEmpty()) {
       final BiPredicate<String, String> predicate;

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

@@ -90,7 +90,7 @@ public String validateSync(
   private String getString(HoodieTableMetaClient target, HoodieTimeline targetTimeline, HoodieTableMetaClient source, long sourceCount, long targetCount, String sourceLatestCommit)
       throws IOException {
     List<HoodieInstant> commitsToCatchup = targetTimeline.findInstantsAfter(sourceLatestCommit, Integer.MAX_VALUE)
-        .getInstants().collect(Collectors.toList());
+        .getInstants();
     if (commitsToCatchup.isEmpty()) {
       return "Count difference now is (count(" + target.getTableConfig().getTableName() + ") - count("
           + source.getTableConfig().getTableName() + ") == " + (targetCount - sourceCount);

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

@@ -211,7 +211,7 @@ public void removeCorruptedPendingCleanAction() {
   public void showFailedCommits() {
     HoodieTableMetaClient metaClient = HoodieCLI.getTableMetaClient();
     HoodieActiveTimeline activeTimeline =  metaClient.getActiveTimeline();
-    activeTimeline.filterCompletedInstants().getInstants().filter(activeTimeline::isEmpty).forEach(hoodieInstant -> LOG.warn("Empty Commit: " + hoodieInstant.toString()));
+    activeTimeline.filterCompletedInstants().getInstantsAsStream().filter(activeTimeline::isEmpty).forEach(hoodieInstant -> LOG.warn("Empty Commit: " + hoodieInstant.toString()));
   }
 
   @ShellMethod(key = "repair migrate-partition-meta", value = "Migrate all partition meta file currently stored in text format "

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

@@ -106,7 +106,7 @@ public String rollbackToSavepoint(
     }
     HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
     HoodieTimeline timeline = activeTimeline.getCommitsTimeline().filterCompletedInstants();
-    List<HoodieInstant> instants = timeline.getInstants().filter(instant -> instant.getTimestamp().equals(instantTime)).collect(Collectors.toList());
+    List<HoodieInstant> instants = timeline.getInstantsAsStream().filter(instant -> instant.getTimestamp().equals(instantTime)).collect(Collectors.toList());
 
     if (instants.isEmpty()) {
       return "Commit " + instantTime + " not found in Commits " + timeline;

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

@@ -45,7 +45,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 
 /**
  * CLI command to displays stats options.
@@ -73,7 +72,7 @@ public String writeAmplificationStats(
 
     List<Comparable[]> rows = new ArrayList<>();
     DecimalFormat df = new DecimalFormat("#.00");
-    for (HoodieInstant instantTime : timeline.getInstants().collect(Collectors.toList())) {
+    for (HoodieInstant instantTime : timeline.getInstants()) {
       String waf = "0";
       HoodieCommitMetadata commit = HoodieCommitMetadata.fromBytes(activeTimeline.getInstantDetails(instantTime).get(),
           HoodieCommitMetadata.class);

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

@@ -208,7 +208,7 @@ private String printTimelineInfo(
       Integer limit, String sortByField, boolean descending, boolean headerOnly, boolean withRowNo,
       boolean showTimeSeconds, boolean showRollbackInfo) {
     Map<String, List<String>> rollbackInfo = getRolledBackInstantInfo(timeline);
-    final List<Comparable[]> rows = timeline.getInstants().map(instant -> {
+    final List<Comparable[]> rows = timeline.getInstantsAsStream().map(instant -> {
       int numColumns = showRollbackInfo ? 7 : 6;
       Comparable[] row = new Comparable[numColumns];
       String instantTimestamp = instant.getTimestamp();
@@ -343,8 +343,7 @@ private Map<String, List<String>> getRolledBackInstantInfo(HoodieTimeline timeli
     // Instant rolled back or to roll back -> rollback instants
     Map<String, List<String>> rollbackInfoMap = new HashMap<>();
     List<HoodieInstant> rollbackInstants = timeline.filter(instant ->
-            HoodieTimeline.ROLLBACK_ACTION.equalsIgnoreCase(instant.getAction()))
-        .getInstants().collect(Collectors.toList());
+            HoodieTimeline.ROLLBACK_ACTION.equalsIgnoreCase(instant.getAction())).getInstants();
     rollbackInstants.forEach(rollbackInstant -> {
       try {
         if (rollbackInstant.isInflight()) {

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

@@ -66,13 +66,13 @@ public void testArchiving() throws Exception {
 
     //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());
+    assertEquals(2, metaClient.getActiveTimeline().countInstants());
 
     //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());
+    assertEquals(12, metaClient.getArchivedTimeline().countInstants());
   }
 
 }

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

@@ -127,13 +127,13 @@ public void testShowCleans() throws Exception {
 
     // First, run clean
     SparkMain.clean(jsc(), HoodieCLI.basePath, propsFilePath.getPath(), new ArrayList<>());
-    assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().count(),
+    assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().countInstants(),
         "Loaded 1 clean and the count should match");
 
     Object result = shell.evaluate(() -> "cleans show");
     assertTrue(ShellEvaluationResultUtil.isSuccess(result));
 
-    HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().findFirst().orElse(null);
+    HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstantsAsStream().findFirst().orElse(null);
     assertNotNull(clean);
 
     TableHeader header =
@@ -163,10 +163,10 @@ public void testShowCleanPartitions() {
 
     // First, run clean with two partition
     SparkMain.clean(jsc(), HoodieCLI.basePath, propsFilePath.toString(), new ArrayList<>());
-    assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().count(),
+    assertEquals(1, metaClient.getActiveTimeline().reload().getCleanerTimeline().countInstants(),
         "Loaded 1 clean and the count should match");
 
-    HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstants().findFirst().get();
+    HoodieInstant clean = metaClient.getActiveTimeline().reload().getCleanerTimeline().getInstantsAsStream().findFirst().get();
 
     Object result = shell.evaluate(() -> "clean showpartitions --clean " + clean.getTimestamp());
     assertTrue(ShellEvaluationResultUtil.isSuccess(result));

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

@@ -521,7 +521,7 @@ public void testCompareCommits(HoodieTableType tableType) throws Exception {
 
     // the latest instant of test_table2 is 101
     List<String> commitsToCatchup = metaClient.getActiveTimeline().findInstantsAfter("101", Integer.MAX_VALUE)
-        .getInstants().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
+        .getInstantsAsStream().map(HoodieInstant::getTimestamp).collect(Collectors.toList());
     String expected = String.format("Source %s is ahead by %d commits. Commits to catch up - %s",
         tableName1, commitsToCatchup.size(), commitsToCatchup);
     assertEquals(expected, result.toString());

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

@@ -255,14 +255,14 @@ public void testRemoveCorruptedPendingCleanAction() throws IOException {
     // reload meta client
     metaClient = HoodieTableMetaClient.reload(metaClient);
     // first, there are four instants
-    assertEquals(4, metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants().count());
+    assertEquals(4, metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
 
     Object result = shell.evaluate(() -> "repair corrupted clean files");
     assertTrue(ShellEvaluationResultUtil.isSuccess(result));
 
     // reload meta client
     metaClient = HoodieTableMetaClient.reload(metaClient);
-    assertEquals(0, metaClient.getActiveTimeline().filterInflightsAndRequested().getInstants().count());
+    assertEquals(0, metaClient.getActiveTimeline().filterInflightsAndRequested().countInstants());
   }
 
   /**
@@ -283,7 +283,7 @@ public void testShowFailedCommits() {
       HoodieTestCommitMetadataGenerator.createCommitFile(tablePath, timestamp, conf);
     }
 
-    metaClient.getActiveTimeline().getInstants().filter(hoodieInstant -> Integer.parseInt(hoodieInstant.getTimestamp()) % 4 == 0).forEach(hoodieInstant -> {
+    metaClient.getActiveTimeline().getInstantsAsStream().filter(hoodieInstant -> Integer.parseInt(hoodieInstant.getTimestamp()) % 4 == 0).forEach(hoodieInstant -> {
       metaClient.getActiveTimeline().deleteInstantFileIfExists(hoodieInstant);
       metaClient.getActiveTimeline().createNewInstant(hoodieInstant);
     });

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

@@ -128,7 +128,7 @@ public void testShowRollbacks() {
 
     // get rollback instants
     HoodieActiveTimeline activeTimeline = new RollbacksCommand.RollbackTimeline(HoodieCLI.getTableMetaClient());
-    Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstants();
+    Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstantsAsStream();
 
     List<Comparable[]> rows = new ArrayList<>();
     rollback.sorted().forEach(instant -> {
@@ -169,7 +169,7 @@ public void testShowRollbacks() {
   public void testShowRollback() throws IOException {
     // get instant
     HoodieActiveTimeline activeTimeline = new RollbacksCommand.RollbackTimeline(HoodieCLI.getTableMetaClient());
-    Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstants();
+    Stream<HoodieInstant> rollback = activeTimeline.getRollbackTimeline().filterCompletedInstants().getInstantsAsStream();
     HoodieInstant instant = rollback.findFirst().orElse(null);
     assertNotNull(instant, "The instant can not be null.");
 

hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java

@@ -128,12 +128,12 @@ public void testClustering() throws IOException {
 
     // assert clustering complete
     assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
-        .filterCompletedInstants().getInstants()
+        .filterCompletedInstants().getInstantsAsStream()
         .map(HoodieInstant::getTimestamp).collect(Collectors.toList()).contains(instance),
         "Pending clustering must be completed");
 
     assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
-            .getCompletedReplaceTimeline().getInstants()
+            .getCompletedReplaceTimeline().getInstantsAsStream()
             .map(HoodieInstant::getTimestamp).collect(Collectors.toList()).contains(instance),
         "Pending clustering must be completed");
   }
@@ -156,7 +156,7 @@ public void testClusteringScheduleAndExecute() throws IOException {
 
     // assert clustering complete
     assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
-            .getCompletedReplaceTimeline().getInstants()
+            .getCompletedReplaceTimeline().getInstantsAsStream()
             .map(HoodieInstant::getTimestamp).count() > 0,
         "Completed clustering couldn't be 0");
   }

hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestCompactionCommand.java

@@ -136,7 +136,7 @@ public void testCompact() throws IOException {
 
     // assert compaction complete
     assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
-        .filterCompletedInstants().getInstants()
+        .filterCompletedInstants().getInstantsAsStream()
         .map(HoodieInstant::getTimestamp).collect(Collectors.toList()).contains(instance),
         "Pending compaction must be completed");
   }
@@ -164,7 +164,7 @@ public void testCompactScheduleAndExecute() throws IOException {
 
     // assert compaction complete
     assertTrue(HoodieCLI.getTableMetaClient().getActiveTimeline().reload()
-            .filterCompletedInstants().getInstants()
+            .filterCompletedInstants().getInstantsAsStream()
             .map(HoodieInstant::getTimestamp).count() > 0,
         "Completed compaction couldn't be 0");
   }