Modify the strategy of compaction schedule

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

@@ -2427,6 +2427,9 @@ public int executeCompaction() throws InterruptedException {
         // could
         // be evicted due to the low priority of the task
         for (long timePartition : timePartitions) {
+          if (CompactionTaskManager.getInstance().isWaitingQueueFull()) {
+            break;
+          }
           CompactionScheduler.sharedLockCompactionSelection();
           try {
             trySubmitCount +=

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionScheduler.java

@@ -186,6 +186,9 @@ private static int addTaskToWaitingQueue(List<? extends AbstractCompactionTask>
 
   private static boolean canAddTaskToWaitingQueue(AbstractCompactionTask task)
       throws InterruptedException {
+    if (CompactionTaskManager.getInstance().isWaitingQueueFull()) {
+      return false;
+    }
     if (Thread.interrupted()) {
       throw new InterruptedException();
     }
@@ -244,6 +247,9 @@ private static int tryToSubmitCrossSpaceCompactionTask(
     if (!config.isEnableCrossSpaceCompaction()) {
       return 0;
     }
+    if (CompactionTaskManager.getInstance().isWaitingQueueFull()) {
+      return 0;
+    }
     String logicalStorageGroupName = tsFileManager.getStorageGroupName();
     String dataRegionId = tsFileManager.getDataRegionId();
     ICrossSpaceSelector crossSpaceCompactionSelector =

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskManager.java

@@ -101,6 +101,8 @@ public class CompactionTaskManager implements IService {
 
   private volatile boolean init = false;
 
+  private final AtomicInteger zeroLevelInnerCompactionTaskNum = new AtomicInteger(0);
+
   public static CompactionTaskManager getInstance() {
     return INSTANCE;
   }
@@ -120,6 +122,15 @@ public synchronized void start() {
       candidateCompactionTaskQueue.regsitPollLastHook(
           AbstractCompactionTask::resetCompactionCandidateStatusForAllSourceFiles);
       candidateCompactionTaskQueue.regsitPollLastHook(AbstractCompactionTask::handleTaskCleanup);
+      candidateCompactionTaskQueue.regsitPollLastHook(
+          task -> {
+            if (task instanceof InnerSpaceCompactionTask
+                && ((InnerSpaceCompactionTask) task).getSumOfCompactionCount() == 0) {
+              CompactionTaskManager.getInstance()
+                  .getZeroLevelInnerCompactionTaskNum()
+                  .decrementAndGet();
+            }
+          });
       init = true;
     }
     logger.info("Compaction task manager started.");
@@ -233,8 +244,17 @@ public ServiceType getID() {
     return ServiceType.COMPACTION_SERVICE;
   }
 
+  public AtomicInteger getZeroLevelInnerCompactionTaskNum() {
+    return this.zeroLevelInnerCompactionTaskNum;
+  }
+
   public boolean isWaitingQueueFull() {
-    return candidateCompactionTaskQueue.size() == candidateCompactionTaskQueue.getMaxSize();
+    if (config.getCandidateCompactionTaskQueueSize()
+        > 2 * config.getCompactionScheduleThreadNum()) {
+      return candidateCompactionTaskQueue.size()
+          >= (candidateCompactionTaskQueue.getMaxSize() - config.getCompactionScheduleThreadNum());
+    }
+    return candidateCompactionTaskQueue.size() >= candidateCompactionTaskQueue.getMaxSize();
   }
 
   /**
@@ -250,6 +270,10 @@ public synchronized boolean addTaskToWaitingQueue(AbstractCompactionTask compact
         && !candidateCompactionTaskQueue.contains(compactionTask)
         && !isTaskRunning(compactionTask)
         && compactionTask.setSourceFilesToCompactionCandidate()) {
+      if (compactionTask instanceof InnerSpaceCompactionTask
+          && ((InnerSpaceCompactionTask) compactionTask).getSumOfCompactionCount() == 0) {
+        zeroLevelInnerCompactionTaskNum.incrementAndGet();
+      }
       candidateCompactionTaskQueue.put(compactionTask);
       return true;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/CompactionTaskQueue.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.storageengine.dataregion.compaction.schedule;
 
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.AbstractCompactionTask;
+import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
 import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
 import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
 
@@ -51,6 +52,10 @@ public AbstractCompactionTask take() throws InterruptedException {
         Thread.sleep(TimeUnit.SECONDS.toMillis(1));
         continue;
       }
+      if (task instanceof InnerSpaceCompactionTask
+          && ((InnerSpaceCompactionTask) task).getSumOfCompactionCount() == 0) {
+        CompactionTaskManager.getInstance().getZeroLevelInnerCompactionTaskNum().decrementAndGet();
+      }
       return task;
     }
   }
@@ -75,6 +80,10 @@ private boolean prepareTask(AbstractCompactionTask task) throws InterruptedExcep
   }
 
   private void dropCompactionTask(AbstractCompactionTask task) {
+    if (task instanceof InnerSpaceCompactionTask
+        && ((InnerSpaceCompactionTask) task).getSumOfCompactionCount() == 0) {
+      CompactionTaskManager.getInstance().getZeroLevelInnerCompactionTaskNum().decrementAndGet();
+    }
     task.resetCompactionCandidateStatusForAllSourceFiles();
     task.handleTaskCleanup();
     task.releaseOccupiedResources();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SizeTieredCompactionSelector.java

@@ -210,6 +210,10 @@ private List<InnerSpaceCompactionTask> selectTaskBaseOnLevel() throws IOExceptio
       if (!selectedResourceList.isEmpty()) {
         return createCompactionTasks(selectedResourceList, CompactionTaskPriorityType.NORMAL);
       }
+      // If any zero level task exist, skip the selection of high level compaction task
+      if (CompactionTaskManager.getInstance().getZeroLevelInnerCompactionTaskNum().get() > 0) {
+        return Collections.emptyList();
+      }
     }
     return Collections.emptyList();
   }