Fixing the instant listing using completion time

apache · Sep 20, 2024 · 7046d40 · 7046d40
1 parent ad4952b
commit 7046d40
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 6 deletions.
diff --git a/...i-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java b/...i-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java
@@ -51,7 +51,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -288,9 +287,9 @@ public HoodieCleanMetadata execute() {
   }
 
   private HoodieCleanMetadata getMetadataWithMaxCompletionTime(List<HoodieCleanMetadata> cleanMetadata) {
-    Optional<HoodieCleanMetadata> r = cleanMetadata.stream().reduce((e1, e2) ->
-        e1.getLastCompletedCommitTimestamp().compareTo(e2.getLastCompletedCommitTimestamp()) > 0 ? e1 : e2);
-    return r.orElse(null);
+    return cleanMetadata.stream().reduce((e1, e2) ->
+            e1.getLastCompletedCommitTimestamp().compareTo(e2.getLastCompletedCommitTimestamp()) > 0 ? e1 : e2)
+        .orElse(null);
   }
 
   private void checkIfOtherWriterCommitted(HoodieInstant hoodieInstant, HoodieIOException e) {

diff --git a/...ient-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanActionExecutor.java b/...ient-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanActionExecutor.java
@@ -49,8 +49,8 @@
 import java.util.stream.Collectors;
 
 import static org.apache.hudi.client.utils.MetadataTableUtils.shouldUseBatchLookup;
-import static org.apache.hudi.common.util.MapUtils.nonEmpty;
 import static org.apache.hudi.common.util.CleanerUtils.SAVEPOINTED_TIMESTAMPS;
+import static org.apache.hudi.common.util.MapUtils.nonEmpty;
 
 public class CleanPlanActionExecutor<T, I, K, O> extends BaseActionExecutor<T, I, K, O, Option<HoodieCleanerPlan>> {
 
@@ -76,7 +76,7 @@ private int getCommitsSinceLastCleaning() {
         HoodieCleanMetadata cleanMetadata = TimelineMetadataUtils
             .deserializeHoodieCleanMetadata(table.getActiveTimeline().getInstantDetails(lastCleanInstant.get()).get());
         String lastCompletedCommitTimestamp = cleanMetadata.getLastCompletedCommitTimestamp();
-        numCommits = commitTimeline.findInstantsAfter(lastCompletedCommitTimestamp).countInstants();
+        numCommits = commitTimeline.findInstantsAfterCompletionTime(lastCompletedCommitTimestamp).countInstants();
       } catch (IOException e) {
         throw new HoodieIOException("Parsing of last clean instant " + lastCleanInstant.get() + " failed", e);
       }

diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
@@ -282,6 +282,12 @@ public HoodieTimeline findInstantsAfter(String instantTime) {
         .filter(s -> compareTimestamps(s.getTimestamp(), GREATER_THAN, instantTime)), details);
   }
 
+  @Override
+  public HoodieTimeline findInstantsAfterCompletionTime(String instantTime) {
+    return new HoodieDefaultTimeline(getInstantsAsStream()
+        .filter(s -> compareTimestamps(s.getCompletionTime(), GREATER_THAN, instantTime)), details);
+  }
+
   @Override
   public HoodieDefaultTimeline findInstantsAfterOrEquals(String commitTime, int numCommits) {
     return new HoodieDefaultTimeline(getInstantsAsStream()

diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
@@ -291,6 +291,8 @@ public interface HoodieTimeline extends Serializable {
    */
   HoodieTimeline findInstantsAfter(String instantTime);
 
+  HoodieTimeline findInstantsAfterCompletionTime(String instantTime);
+
   /**
    * Create a new Timeline with all instants before specified time.
    */