From 2ebd741985b38a417774e10f3b4a6b0c2622958b Mon Sep 17 00:00:00 2001
From: XuQianJin-Stars <forwardxu@apache.com>
Date: Sun, 18 Sep 2022 17:57:13 +0800
Subject: [PATCH] add filterMorCompactionInstants for append

---
 .../main/java/org/apache/hudi/io/HoodieAppendHandle.java   | 2 +-
 .../hudi/common/table/timeline/HoodieDefaultTimeline.java  | 7 ++++---
 .../apache/hudi/common/table/timeline/HoodieTimeline.java  | 2 +-
 3 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
index 0e5aca3f7fec..fd0bfd147334 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java
@@ -149,7 +149,7 @@ private void init(HoodieRecord record) {
       List<String> logFiles = new ArrayList<>();
 
       Option<HoodieInstant> maxCompleteInstant = hoodieTable.getMetaClient().getActiveTimeline().getWriteTimeline()
-          .filterCompletedCompactionInstants().lastInstant();
+          .filterMorCompactionInstants().lastInstant();
       if (maxCompleteInstant.isPresent()) {
         if (fileSlice.isPresent()) {
           baseInstantTime = fileSlice.get().getBaseInstantTime();
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
index 560407c1a806..49511243ff82 100644
--- 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
@@ -109,10 +109,11 @@ public HoodieTimeline filterCompletedAndCompactionInstants() {
   }
 
   @Override
-  public HoodieTimeline filterCompletedCompactionInstants() {
+  public HoodieTimeline filterMorCompactionInstants() {
     // Note: use for mor.
-    return new HoodieDefaultTimeline(instants.stream().filter(s -> s.isCompleted()
-        && s.getAction().equals(HoodieTimeline.COMMIT_ACTION)), details);
+    return new HoodieDefaultTimeline(instants.stream().filter(s ->
+        ((s.isRequested() || s.isInflight()) && s.getAction().equals(HoodieTimeline.COMPACTION_ACTION))
+            || (s.isCompleted() && s.getAction().equals(HoodieTimeline.COMMIT_ACTION))), details);
   }
 
   @Override
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
index 58645a7c847a..a5695842caad 100644
--- 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
@@ -146,7 +146,7 @@ public interface HoodieTimeline extends Serializable {
    *
    * @return New instance of HoodieTimeline with just completed compaction instants
    */
-  HoodieTimeline filterCompletedCompactionInstants();
+  HoodieTimeline filterMorCompactionInstants();
 
   /**
    * Timeline to just include commits (commit/deltacommit), compaction and replace actions.