Optimize ttl deletion in fast compaction performer (#16367) (#16383)

* optimize ttl deletion in fast compaction performer

* fix ut

* check isEmpty

Author: shuwenwei

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java

@@ -142,21 +142,16 @@ public void perform() throws Exception {
         // checked above
         //noinspection OptionalGetWithoutIsPresent
         sortedSourceFiles.sort(Comparator.comparingLong(x -> x.getStartTime(device).get()));
+        Deletion ttlDeletion = null;
         if (ttl != Long.MAX_VALUE) {
-          Deletion ttlDeletion =
+          ttlDeletion =
               new Deletion(
                   new PartialPath(device, IoTDBConstant.ONE_LEVEL_PATH_WILDCARD),
                   Long.MAX_VALUE,
                   Long.MIN_VALUE,
                   deviceIterator.getTimeLowerBoundForCurrentDevice());
-          for (TsFileResource sourceFile : sortedSourceFiles) {
-            modificationCache
-                .computeIfAbsent(
-                    sourceFile.getTsFile().getName(),
-                    k -> PatternTreeMapFactory.getModsPatternTreeMap())
-                .append(ttlDeletion.getPath(), ttlDeletion);
-          }
         }
+        compactionWriter.setTTLDeletion(ttlDeletion);
 
         if (sortedSourceFiles.isEmpty()) {
           // device is out of dated in all source files

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java

@@ -30,6 +30,7 @@
 import org.apache.iotdb.db.service.metrics.FileMetrics;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.constant.CompactionTaskType;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager;
+import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion;
 import org.apache.iotdb.db.storageengine.dataregion.modification.Modification;
 import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
@@ -48,6 +49,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -376,13 +378,23 @@ public static void deleteTsFileResourceWithoutLock(TsFileResource resource) {
   public static List<Modification> getMatchedModifications(
       PatternTreeMap<Modification, PatternTreeMapFactory.ModsSerializer> patternTreeMap,
       IDeviceID deviceID,
-      String measurement)
+      String measurement,
+      Deletion ttlDeletion)
       throws IllegalPathException {
-    if (patternTreeMap == null) {
-      return Collections.emptyList();
+    if ((patternTreeMap == null) || patternTreeMap.isEmpty()) {
+      return ttlDeletion == null ? Collections.emptyList() : Collections.singletonList(ttlDeletion);
     }
     PartialPath path = CompactionPathUtils.getPath(deviceID, measurement);
-    return ModificationFile.sortAndMerge(patternTreeMap.getOverlapped(path));
+    List<Modification> modifications = patternTreeMap.getOverlapped(path);
+    if (ttlDeletion != null) {
+      if (!(modifications instanceof ArrayList)) {
+        List<Modification> newModEntries = new ArrayList<>(modifications.size() + 1);
+        newModEntries.addAll(modifications);
+        modifications = newModEntries;
+      }
+      modifications.add(ttlDeletion);
+    }
+    return ModificationFile.sortAndMerge(modifications);
   }
 
   public static boolean isDiskHasSpace() {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/MultiTsFileDeviceIterator.java

@@ -451,10 +451,7 @@ private void applyModificationForAlignedChunkMetadataList(
       }
       List<Modification> modificationList =
           CompactionUtils.getMatchedModifications(
-              modifications, device, valueChunkMetadata.getMeasurementUid());
-      if (ttlDeletion != null) {
-        modificationList.add(ttlDeletion);
-      }
+              modifications, device, valueChunkMetadata.getMeasurementUid(), ttlDeletion);
       modificationForCurDevice.add(
           modificationList.isEmpty() ? Collections.emptyList() : modificationList);
     }
@@ -666,11 +663,7 @@ public String nextSeries() throws IllegalPathException {
           // collect the modifications for current series
           List<Modification> modificationForCurrentSeries =
               CompactionUtils.getMatchedModifications(
-                  modificationsInThisResource, device, currentCompactingSeries);
-          // add ttl deletion for current series
-          if (ttlDeletion != null) {
-            modificationForCurrentSeries.add(ttlDeletion);
-          }
+                  modificationsInThisResource, device, currentCompactingSeries, ttlDeletion);
 
           // if there are modifications of current series, apply them to the chunk metadata
           if (!modificationForCurrentSeries.isEmpty()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/executor/fast/SeriesCompactionExecutor.java

@@ -478,7 +478,11 @@ protected List<Modification> getModificationsFromCache(
       throws IllegalPathException {
     PatternTreeMap<Modification, PatternTreeMapFactory.ModsSerializer> allModifications =
         modificationCacheMap.get(tsFileResource.getTsFile().getName());
-    return CompactionUtils.getMatchedModifications(allModifications, deviceId, measurement);
+    return CompactionUtils.getMatchedModifications(
+        allModifications,
+        deviceId,
+        measurement,
+        compactionWriter.getTTLLowerBoundForCurrentDevice());
   }
 
   @SuppressWarnings("squid:S3776")

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/AbstractCompactionWriter.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.executor.fast.element.ChunkMetadataElement;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.writer.flushcontroller.AbstractCompactionFlushController;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.io.CompactionTsFileWriter;
+import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion;
 
 import org.apache.tsfile.exception.write.PageException;
 import org.apache.tsfile.file.header.PageHeader;
@@ -99,10 +100,20 @@ public abstract class AbstractCompactionWriter implements AutoCloseable {
 
   protected String[] measurementId = new String[subTaskNum];
 
+  protected Deletion ttlDeletionForCurrentDevice;
+
   public abstract void startChunkGroup(IDeviceID deviceId, boolean isAlign) throws IOException;
 
   public abstract void endChunkGroup() throws IOException;
 
+  public void setTTLDeletion(Deletion ttlDeletion) {
+    this.ttlDeletionForCurrentDevice = ttlDeletion;
+  }
+
+  public Deletion getTTLLowerBoundForCurrentDevice() {
+    return ttlDeletionForCurrentDevice;
+  }
+
   public void startMeasurement(String measurement, IChunkWriter chunkWriter, int subTaskId) {
     lastCheckIndex = 0;
     lastTimeSet[subTaskId] = false;

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

@@ -246,6 +246,10 @@ public FileInfo.RoughFileInfo getRoughFileInfo(TsFileResource resource) {
     return roughFileInfo;
   }
 
+  public boolean supportsRoughEstimation() {
+    return true;
+  }
+
   public static void removeFileInfoFromGlobalFileInfoCache(TsFileResource resource) {
     if (resource == null || resource.getTsFile() == null) {
       return;

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

@@ -89,4 +89,9 @@ public long roughEstimateInnerCompactionMemory(
       CompactionScheduleContext context, List<TsFileResource> resources) throws IOException {
     throw new RuntimeException("unimplemented");
   }
+
+  @Override
+  public boolean supportsRoughEstimation() {
+    return false;
+  }
 }

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

@@ -201,7 +201,7 @@ private void resetMemoryEstimationFields() {
       performer =
           sequence ? context.getSeqCompactionPerformer() : context.getUnseqCompactionPerformer();
       estimator = performer.getInnerSpaceEstimator().orElse(null);
-      if (estimator == null) {
+      if (estimator == null || !estimator.supportsRoughEstimation()) {
         estimateCompactionTaskMemoryDuringSelection = false;
       }
       memoryCost = 0;

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

@@ -288,7 +288,7 @@ private boolean isDeviceDeletedByMods(
       throws IllegalPathException {
     List<Modification> deviceModifications =
         CompactionUtils.getMatchedModifications(
-            modifications, device, AlignedPath.VECTOR_PLACEHOLDER);
+            modifications, device, AlignedPath.VECTOR_PLACEHOLDER, null);
     for (Modification modification : deviceModifications) {
       if (((Deletion) modification).getTimeRange().contains(startTime, endTime)) {
         return true;

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/path/PatternTreeMap.java

@@ -61,6 +61,10 @@ public PatternTreeMap(
     this.serializer = serializer;
   }
 
+  public boolean isEmpty() {
+    return root.isLeaf() && root.getValues().isEmpty();
+  }
+
   /**
    * Append key and value to PatternTreeMap.
    *