[core] Rename CompactFutureManager.isCompaction to compactNotCompleted

Author: JingsongLi

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2539,7 +2539,7 @@ public boolean prepareCommitWaitCompaction() {
         return options.get(LOOKUP_WAIT);
     }
 
-    public boolean statefulLookup() {
+    public boolean laziedLookup() {
         return needLookup()
                 && (!options.get(LOOKUP_WAIT) || LookupCompactMode.GENTLE.equals(lookupCompact()));
     }

paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java

@@ -210,9 +210,9 @@ public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
     }
 
     @Override
-    public boolean isCompacting() {
+    public boolean compactNotCompleted() {
         compactManager.triggerCompaction(false);
-        return compactManager.isCompacting();
+        return compactManager.compactNotCompleted();
     }
 
     @VisibleForTesting

paimon-core/src/main/java/org/apache/paimon/compact/CompactFutureManager.java

@@ -40,7 +40,7 @@ public void cancelCompaction() {
     }
 
     @Override
-    public boolean isCompacting() {
+    public boolean compactNotCompleted() {
         return taskFuture != null;
     }
 

paimon-core/src/main/java/org/apache/paimon/compact/CompactManager.java

@@ -56,5 +56,5 @@ Optional<CompactResult> getCompactionResult(boolean blocking)
      * Check if a compaction is in progress, or if a compaction result remains to be fetched, or if
      * a compaction should be triggered later.
      */
-    boolean isCompacting();
+    boolean compactNotCompleted();
 }

paimon-core/src/main/java/org/apache/paimon/compact/NoopCompactManager.java

@@ -71,7 +71,7 @@ public Optional<CompactResult> getCompactionResult(boolean blocking)
     public void cancelCompaction() {}
 
     @Override
-    public boolean isCompacting() {
+    public boolean compactNotCompleted() {
         return false;
     }
 

paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java

@@ -278,9 +278,9 @@ public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
     }
 
     @Override
-    public boolean isCompacting() {
+    public boolean compactNotCompleted() {
         compactManager.triggerCompaction(false);
-        return compactManager.isCompacting();
+        return compactManager.compactNotCompleted();
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java

@@ -244,8 +244,11 @@ public Optional<CompactResult> getCompactionResult(boolean blocking)
     }
 
     @Override
-    public boolean isCompacting() {
-        return super.isCompacting() || (needLookup && !levels().level0().isEmpty());
+    public boolean compactNotCompleted() {
+        // If it is a lookup compaction, we should ensure that all level 0 files are consumed, so
+        // here we need to make the outside think that we still need to do unfinished compact
+        // working
+        return super.compactNotCompleted() || (needLookup && !levels().level0().isEmpty());
     }
 
     private void reportMetrics() {

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/UniversalCompaction.java

@@ -78,7 +78,8 @@ public UniversalCompaction(
         this.opCompactionInterval =
                 opCompactionInterval == null ? null : opCompactionInterval.toMillis();
         this.maxLookupCompactInterval = maxLookupCompactInterval;
-        this.lookupCompactTriggerCount = new AtomicInteger(0);
+        this.lookupCompactTriggerCount =
+                maxLookupCompactInterval == null ? null : new AtomicInteger(0);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java

@@ -294,7 +294,7 @@ Function<WriterContainer<T>, Boolean> createConflictAwareWriterCleanChecker(
         // Condition 3: The writer has no postponed compaction like gentle lookup compaction.
         return writerContainer ->
                 writerContainer.lastModifiedCommitIdentifier < latestCommittedIdentifier
-                        && !writerContainer.writer.isCompacting();
+                        && !writerContainer.writer.compactNotCompleted();
     }
 
     protected static <T>

paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java

@@ -84,7 +84,7 @@ public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
     }
 
     @Override
-    public boolean isCompacting() {
+    public boolean compactNotCompleted() {
         return false;
     }