Fix estimating compaction memory without acquire read lock (apache#11972

)
Caideyipi · Jan 26, 2024 · 6c15634 · 6c15634
1 parent 40fc15b
commit 6c15634
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 3 deletions.
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRepairDataIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRepairDataIT.java
@@ -44,6 +44,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
 import static org.junit.Assert.assertNotNull;
@@ -81,6 +82,7 @@ public void testRepairData() {
         if (sorted) {
           return;
         }
+        Thread.sleep(TimeUnit.SECONDS.toMillis(1));
       }
       Assert.fail();
     } catch (Exception e) {

diff --git a/...b/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java b/...b/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java
@@ -23,6 +23,7 @@
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.storageengine.dataregion.flush.CompressionRatio;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.DeviceTimeIndex;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex;
@@ -63,6 +64,15 @@ public abstract class AbstractCompactionEstimator {
 
   protected abstract long calculatingDataMemoryCost(CompactionTaskInfo taskInfo) throws IOException;
 
+  protected boolean isAllSourceFileExist(List<TsFileResource> resources) {
+    for (TsFileResource resource : resources) {
+      if (resource.getStatus() == TsFileResourceStatus.DELETED) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   protected CompactionTaskInfo calculatingCompactionTaskInfo(List<TsFileResource> resources)
       throws IOException {
     List<FileInfo> fileInfoList = new ArrayList<>();

diff --git a/...b/storageengine/dataregion/compaction/selector/estimator/AbstractCrossSpaceEstimator.java b/...b/storageengine/dataregion/compaction/selector/estimator/AbstractCrossSpaceEstimator.java
@@ -45,6 +45,9 @@ public long estimateCrossCompactionMemory(
 
     long cost = 0;
     try {
+      if (!isAllSourceFileExist(resources)) {
+        return -1L;
+      }
       CompactionTaskInfo taskInfo = calculatingCompactionTaskInfo(resources);
       cost += calculatingMetadataMemoryCost(taskInfo);
       cost += calculatingDataMemoryCost(taskInfo);

diff --git a/...b/storageengine/dataregion/compaction/selector/estimator/AbstractInnerSpaceEstimator.java b/...b/storageengine/dataregion/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
@@ -34,9 +34,22 @@ public long estimateInnerCompactionMemory(List<TsFileResource> resources) throws
     if (!config.isEnableCompactionMemControl()) {
       return 0;
     }
-    CompactionTaskInfo taskInfo = calculatingCompactionTaskInfo(resources);
-    long cost = calculatingMetadataMemoryCost(taskInfo);
-    cost += calculatingDataMemoryCost(taskInfo);
+
+    if (!CompactionEstimateUtils.addReadLock(resources)) {
+      return -1L;
+    }
+    long cost = 0;
+    try {
+      if (!isAllSourceFileExist(resources)) {
+        return -1L;
+      }
+
+      CompactionTaskInfo taskInfo = calculatingCompactionTaskInfo(resources);
+      cost = calculatingMetadataMemoryCost(taskInfo);
+      cost += calculatingDataMemoryCost(taskInfo);
+    } finally {
+      CompactionEstimateUtils.releaseReadLock(resources);
+    }
     return cost;
   }
 }