This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch rel/1.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.1 by this push:
new c1bec1bb82b [To rel/1.1] Rewrite compaction memory estimator (#10965)
c1bec1bb82b is described below
commit c1bec1bb82bccc8d786e16b77443c04a24115a1c
Author: shuwenwei <[email protected]>
AuthorDate: Tue Aug 29 11:49:07 2023 +0800
[To rel/1.1] Rewrite compaction memory estimator (#10965)
---
.../execute/task/AbstractCompactionTask.java | 2 +
.../execute/task/CrossSpaceCompactionTask.java | 1 -
.../execute/task/InnerSpaceCompactionTask.java | 35 ++-
.../compaction/selector/ICompactionSelector.java | 4 +-
.../estimator/AbstractCompactionEstimator.java | 115 +++++++--
.../estimator/AbstractCrossSpaceEstimator.java | 28 ++-
.../estimator/AbstractInnerSpaceEstimator.java | 14 +-
.../estimator/CompactionEstimateUtils.java | 95 ++++++++
.../selector/estimator/CompactionTaskInfo.java | 101 ++++++++
.../FastCompactionInnerCompactionEstimator.java | 85 +++++++
.../FastCrossSpaceCompactionEstimator.java | 81 +++++++
.../compaction/selector/estimator/FileInfo.java | 48 ++++
.../estimator/InplaceCompactionEstimator.java | 171 --------------
.../ReadChunkInnerCompactionEstimator.java | 73 ++++++
.../ReadPointCrossCompactionEstimator.java | 262 ---------------------
.../impl/RewriteCrossSpaceCompactionSelector.java | 21 +-
.../utils/CrossCompactionTaskResource.java | 6 +-
.../org/apache/iotdb/db/rescon/SystemInfo.java | 3 +
.../utils/CompactionTaskMemCostEstimatorTest.java | 106 +++++++++
.../iotdb/tsfile/read/TsFileSequenceReader.java | 12 +-
20 files changed, 784 insertions(+), 479 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java
index 3246a133866..520a051b13e 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/AbstractCompactionTask.java
@@ -55,6 +55,8 @@ public abstract class AbstractCompactionTask {
protected boolean crossTask;
protected boolean innerSeqTask;
+ protected long memoryCost = 0L;
+
public AbstractCompactionTask(
String storageGroupName,
String dataRegionId,
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CrossSpaceCompactionTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CrossSpaceCompactionTask.java
index 11ec6228e8a..f260a4c7db8 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CrossSpaceCompactionTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/CrossSpaceCompactionTask.java
@@ -64,7 +64,6 @@ public class CrossSpaceCompactionTask extends
AbstractCompactionTask {
protected List<TsFileResource> holdWriteLockList = new ArrayList<>();
protected double selectedSeqFileSize = 0;
protected double selectedUnseqFileSize = 0;
- protected long memoryCost = 0L;
public CrossSpaceCompactionTask(
long timePartition,
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/InnerSpaceCompactionTask.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/InnerSpaceCompactionTask.java
index 2219fc8d08b..eebc5cc7d40 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/InnerSpaceCompactionTask.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/task/InnerSpaceCompactionTask.java
@@ -23,16 +23,21 @@ import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.TsFileMetricManager;
import
org.apache.iotdb.db.engine.compaction.execute.exception.CompactionExceptionHandler;
+import
org.apache.iotdb.db.engine.compaction.execute.exception.CompactionMemoryNotEnoughException;
import
org.apache.iotdb.db.engine.compaction.execute.performer.ICompactionPerformer;
import
org.apache.iotdb.db.engine.compaction.execute.performer.impl.FastCompactionPerformer;
import
org.apache.iotdb.db.engine.compaction.execute.task.subtask.FastCompactionTaskSummary;
import org.apache.iotdb.db.engine.compaction.execute.utils.CompactionUtils;
import
org.apache.iotdb.db.engine.compaction.execute.utils.log.CompactionLogger;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.AbstractInnerSpaceEstimator;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.FastCompactionInnerCompactionEstimator;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.ReadChunkInnerCompactionEstimator;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceList;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
+import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.service.metrics.recorder.CompactionMetricsManager;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
@@ -69,6 +74,8 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
protected long maxModsFileSize;
+ protected AbstractInnerSpaceEstimator innerSpaceEstimator;
+
public InnerSpaceCompactionTask(
long timePartition,
TsFileManager tsFileManager,
@@ -87,6 +94,13 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
this.selectedTsFileResourceList = selectedTsFileResourceList;
this.sequence = sequence;
this.performer = performer;
+ if
(IoTDBDescriptor.getInstance().getConfig().isEnableCompactionMemControl()) {
+ if (this.performer instanceof ReadChunkInnerCompactionEstimator) {
+ innerSpaceEstimator = new ReadChunkInnerCompactionEstimator();
+ } else if (!sequence && this.performer instanceof
FastCompactionInnerCompactionEstimator) {
+ innerSpaceEstimator = new FastCompactionInnerCompactionEstimator();
+ }
+ }
isHoldingReadLock = new boolean[selectedTsFileResourceList.size()];
isHoldingWriteLock = new boolean[selectedTsFileResourceList.size()];
for (int i = 0; i < selectedTsFileResourceList.size(); ++i) {
@@ -317,6 +331,7 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
isSequence());
}
} finally {
+ SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
releaseAllLocksAndResetStatus();
return isSuccess;
}
@@ -458,9 +473,27 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
return false;
}
}
+ if (innerSpaceEstimator != null) {
+ memoryCost =
innerSpaceEstimator.estimateInnerCompactionMemory(selectedTsFileResourceList);
+ }
+ SystemInfo.getInstance().addCompactionMemoryCost(memoryCost, 60);
} catch (Throwable e) {
+ if (e instanceof InterruptedException) {
+ LOGGER.warn("Interrupted when allocating memory for compaction", e);
+ Thread.currentThread().interrupt();
+ } else if (e instanceof CompactionMemoryNotEnoughException) {
+ LOGGER.warn("No enough memory for current compaction task {}", this,
e);
+ }
releaseAllLocksAndResetStatus();
- throw e;
+ return false;
+ } finally {
+ try {
+ if (innerSpaceEstimator != null) {
+ innerSpaceEstimator.close();
+ }
+ } catch (IOException e) {
+ LOGGER.warn("Failed to close InnerSpaceCompactionMemoryEstimator");
+ }
}
return true;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/ICompactionSelector.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/ICompactionSelector.java
index 5f10aeb02f5..f77b868dae5 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/ICompactionSelector.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/ICompactionSelector.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.engine.compaction.selector;
import
org.apache.iotdb.db.engine.compaction.execute.performer.constant.CrossCompactionPerformer;
import org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager;
import
org.apache.iotdb.db.engine.compaction.selector.estimator.AbstractCompactionEstimator;
-import
org.apache.iotdb.db.engine.compaction.selector.estimator.ReadPointCrossCompactionEstimator;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.FastCrossSpaceCompactionEstimator;
import
org.apache.iotdb.db.engine.compaction.selector.utils.CrossCompactionTaskResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -66,7 +66,7 @@ public interface ICompactionSelector {
case READ_POINT:
case FAST:
if (!isInnerSpace) {
- return new ReadPointCrossCompactionEstimator();
+ return new FastCrossSpaceCompactionEstimator();
}
default:
throw new RuntimeException(
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCompactionEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCompactionEstimator.java
index af5a51be07b..47ac39e2049 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCompactionEstimator.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCompactionEstimator.java
@@ -16,55 +16,128 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.engine.compaction.selector.estimator;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.adapter.CompressionRatio;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.engine.storagegroup.timeindex.DeviceTimeIndex;
+import org.apache.iotdb.db.engine.storagegroup.timeindex.FileTimeIndex;
+import org.apache.iotdb.db.engine.storagegroup.timeindex.ITimeIndex;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
/**
* Estimate the memory cost of one compaction task with specific source files
based on its
* corresponding implementation.
*/
-public abstract class AbstractCompactionEstimator {
+public abstract class AbstractCompactionEstimator implements Closeable {
- protected Map<TsFileResource, TsFileSequenceReader> fileReaderCache = new
HashMap<>();
+ protected Map<TsFileResource, FileInfo> fileInfoCache = new HashMap<>();
+ protected Map<TsFileResource, DeviceTimeIndex> deviceTimeIndexCache = new
HashMap<>();
protected IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
protected long compressionRatio = (long)
CompressionRatio.getInstance().getRatio() + 1;
- /**
- * Estimate the memory cost of compacting the unseq file and its
corresponding overlapped seq
- * files in cross space compaction task.
- */
- public abstract long estimateCrossCompactionMemory(
- List<TsFileResource> seqResources, TsFileResource unseqResource) throws
IOException;
+ protected abstract long calculatingMetadataMemoryCost(CompactionTaskInfo
taskInfo);
- /** Estimate the memory cost of compacting the source files in inner space
compaction task. */
- public abstract long estimateInnerCompactionMemory(List<TsFileResource>
resources);
+ protected abstract long calculatingDataMemoryCost(CompactionTaskInfo
taskInfo) throws IOException;
- /** Construct a new or get an existing TsFileSequenceReader of a TsFile. */
- protected TsFileSequenceReader getFileReader(TsFileResource tsFileResource)
throws IOException {
- TsFileSequenceReader reader = fileReaderCache.get(tsFileResource);
- if (reader == null) {
- reader = new TsFileSequenceReader(tsFileResource.getTsFilePath(), true,
false);
- fileReaderCache.put(tsFileResource, reader);
+ protected CompactionTaskInfo
calculatingCompactionTaskInfo(List<TsFileResource> resources)
+ throws IOException {
+ List<FileInfo> fileInfoList = new ArrayList<>();
+ for (TsFileResource resource : resources) {
+ FileInfo fileInfo = getFileInfoFromCache(resource);
+ fileInfoList.add(fileInfo);
}
- return reader;
+ return new CompactionTaskInfo(resources, fileInfoList);
}
- public void close() throws IOException {
- for (TsFileSequenceReader reader : fileReaderCache.values()) {
- reader.close();
+ private FileInfo getFileInfoFromCache(TsFileResource resource) throws
IOException {
+ if (fileInfoCache.containsKey(resource)) {
+ return fileInfoCache.get(resource);
+ }
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(resource.getTsFilePath(), true, false)) {
+ FileInfo fileInfo = CompactionEstimateUtils.calculateFileInfo(reader);
+ fileInfoCache.put(resource, fileInfo);
+ return fileInfo;
+ }
+ }
+
+ protected int
calculatingMaxOverlapFileNumInSubCompactionTask(List<TsFileResource> resources)
+ throws IOException {
+ Set<String> devices = new HashSet<>();
+ List<DeviceTimeIndex> resourceDevices = new ArrayList<>(resources.size());
+ for (TsFileResource resource : resources) {
+ DeviceTimeIndex deviceTimeIndex = getDeviceTimeIndexFromCache(resource);
+ devices.addAll(deviceTimeIndex.getDevices());
+ resourceDevices.add(deviceTimeIndex);
+ }
+ int maxOverlapFileNumInSubCompactionTask = 1;
+ for (String device : devices) {
+ List<DeviceTimeIndex> resourcesContainsCurrentDevice =
+ resourceDevices.stream()
+ .filter(resource -> !resource.definitelyNotContains(device))
+ .sorted(Comparator.comparingLong(resource ->
resource.getStartTime(device)))
+ .collect(Collectors.toList());
+ if (resourcesContainsCurrentDevice.size() <
maxOverlapFileNumInSubCompactionTask) {
+ continue;
+ }
+
+ long maxEndTimeOfCurrentDevice = Long.MIN_VALUE;
+ int overlapFileNumOfCurrentDevice = 0;
+ for (DeviceTimeIndex resource : resourcesContainsCurrentDevice) {
+ long deviceStartTimeInCurrentFile = resource.getStartTime(device);
+ long deviceEndTimeInCurrentFile = resource.getEndTime(device);
+ if (deviceStartTimeInCurrentFile <= maxEndTimeOfCurrentDevice) {
+ // has overlap, update max end time
+ maxEndTimeOfCurrentDevice =
+ Math.max(maxEndTimeOfCurrentDevice, deviceEndTimeInCurrentFile);
+ overlapFileNumOfCurrentDevice++;
+ maxOverlapFileNumInSubCompactionTask =
+ Math.max(maxOverlapFileNumInSubCompactionTask,
overlapFileNumOfCurrentDevice);
+ } else {
+ // reset max end time and overlap file num of current device
+ maxEndTimeOfCurrentDevice = deviceEndTimeInCurrentFile;
+ overlapFileNumOfCurrentDevice = 1;
+ }
+ }
+ // already reach the max value
+ if (maxOverlapFileNumInSubCompactionTask == resources.size()) {
+ return maxOverlapFileNumInSubCompactionTask;
+ }
}
- fileReaderCache.clear();
+ return maxOverlapFileNumInSubCompactionTask;
+ }
+
+ private DeviceTimeIndex getDeviceTimeIndexFromCache(TsFileResource resource)
throws IOException {
+ if (deviceTimeIndexCache.containsKey(resource)) {
+ return deviceTimeIndexCache.get(resource);
+ }
+ ITimeIndex timeIndex = resource.getTimeIndex();
+ if (timeIndex instanceof FileTimeIndex) {
+ timeIndex = resource.buildDeviceTimeIndex();
+ }
+ deviceTimeIndexCache.put(resource, (DeviceTimeIndex) timeIndex);
+ return (DeviceTimeIndex) timeIndex;
+ }
+
+ public void close() throws IOException {
+ deviceTimeIndexCache.clear();
+ fileInfoCache.clear();
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCrossSpaceEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCrossSpaceEstimator.java
index 6b6f4744dbc..e0837a328c6 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCrossSpaceEstimator.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractCrossSpaceEstimator.java
@@ -16,11 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.engine.compaction.selector.estimator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
/**
@@ -28,11 +30,27 @@ import java.util.List;
* its corresponding implementation.
*/
public abstract class AbstractCrossSpaceEstimator extends
AbstractCompactionEstimator {
- public abstract long estimateCrossCompactionMemory(
- List<TsFileResource> seqResources, TsFileResource unseqResource) throws
IOException;
- public long estimateInnerCompactionMemory(List<TsFileResource> resources) {
- throw new RuntimeException(
- "This kind of estimator cannot be used to estimate inner space
compaction task");
+ public long estimateCrossCompactionMemory(
+ List<TsFileResource> seqResources, List<TsFileResource> unseqResources)
throws IOException {
+ if (!config.isEnableCompactionMemControl()) {
+ return 0;
+ }
+ List<TsFileResource> resources = new ArrayList<>(seqResources.size() +
unseqResources.size());
+ resources.addAll(seqResources);
+ resources.addAll(unseqResources);
+ if (!CompactionEstimateUtils.addReadLock(resources)) {
+ return -1L;
+ }
+
+ long cost = 0;
+ try {
+ CompactionTaskInfo taskInfo = calculatingCompactionTaskInfo(resources);
+ cost += calculatingMetadataMemoryCost(taskInfo);
+ cost += calculatingDataMemoryCost(taskInfo);
+ } finally {
+ CompactionEstimateUtils.releaseReadLock(resources);
+ }
+ return cost;
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
index 7010a3cd7c6..e0dc4a7493a 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.engine.compaction.selector.estimator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -28,11 +29,14 @@ import java.util.List;
* its corresponding implementation.
*/
public abstract class AbstractInnerSpaceEstimator extends
AbstractCompactionEstimator {
- public abstract long estimateInnerCompactionMemory(List<TsFileResource>
resources);
- public long estimateCrossCompactionMemory(
- List<TsFileResource> seqResources, TsFileResource unseqResource) throws
IOException {
- throw new RuntimeException(
- "This kind of estimator cannot be used to estimate cross space
compaction task");
+ public long estimateInnerCompactionMemory(List<TsFileResource> resources)
throws IOException {
+ if (!config.isEnableCompactionMemControl()) {
+ return 0;
+ }
+ CompactionTaskInfo taskInfo = calculatingCompactionTaskInfo(resources);
+ long cost = calculatingMetadataMemoryCost(taskInfo);
+ cost += calculatingDataMemoryCost(taskInfo);
+ return cost;
}
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/CompactionEstimateUtils.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/CompactionEstimateUtils.java
new file mode 100644
index 00000000000..e4afb109bcb
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/CompactionEstimateUtils.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.compaction.selector.estimator;
+
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+public class CompactionEstimateUtils {
+
+ /**
+ * Get the details of the tsfile, the returned array contains the following
elements in sequence:
+ *
+ * <p>total chunk num in this tsfile
+ *
+ * <p>max chunk num of one timeseries in this tsfile
+ *
+ * <p>max aligned series num in one device. If there is no aligned series in
this file, then it
+ * turns to be -1.
+ *
+ * <p>max chunk num of one device in this tsfile
+ *
+ * @throws IOException if io errors occurred
+ */
+ public static FileInfo calculateFileInfo(TsFileSequenceReader reader) throws
IOException {
+ int totalChunkNum = 0;
+ int maxChunkNum = 0;
+ int maxAlignedSeriesNumInDevice = -1;
+ int maxDeviceChunkNum = 0;
+ Map<String, List<TimeseriesMetadata>> deviceMetadata =
reader.getAllTimeseriesMetadata(true);
+ for (Map.Entry<String, List<TimeseriesMetadata>> entry :
deviceMetadata.entrySet()) {
+ int deviceChunkNum = 0;
+ List<TimeseriesMetadata> deviceTimeseriesMetadata = entry.getValue();
+ if (deviceTimeseriesMetadata.get(0).getMeasurementId().equals("")) {
+ // aligned device
+ maxAlignedSeriesNumInDevice =
+ Math.max(maxAlignedSeriesNumInDevice,
deviceTimeseriesMetadata.size());
+ }
+ for (TimeseriesMetadata timeseriesMetadata : deviceTimeseriesMetadata) {
+ deviceChunkNum += timeseriesMetadata.getChunkMetadataList().size();
+ totalChunkNum += timeseriesMetadata.getChunkMetadataList().size();
+ maxChunkNum = Math.max(maxChunkNum,
timeseriesMetadata.getChunkMetadataList().size());
+ }
+ maxDeviceChunkNum = Math.max(maxDeviceChunkNum, deviceChunkNum);
+ }
+ long averageChunkMetadataSize =
+ totalChunkNum == 0 ? 0 : reader.getAllMetadataSize() / totalChunkNum;
+ return new FileInfo(
+ totalChunkNum,
+ maxChunkNum,
+ maxAlignedSeriesNumInDevice,
+ maxDeviceChunkNum,
+ averageChunkMetadataSize);
+ }
+
+ public static boolean addReadLock(List<TsFileResource> resources) {
+ for (int i = 0; i < resources.size(); i++) {
+ TsFileResource resource = resources.get(i);
+ resource.readLock();
+ if (resource.isDeleted()) {
+ // release read lock
+ for (int j = 0; j <= i; j++) {
+ resources.get(j).readUnlock();
+ }
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public static void releaseReadLock(List<TsFileResource> resources) {
+ resources.forEach(TsFileResource::readUnlock);
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/CompactionTaskInfo.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/CompactionTaskInfo.java
new file mode 100644
index 00000000000..723505a8918
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/CompactionTaskInfo.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.compaction.selector.estimator;
+
+import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+
+import java.util.List;
+
+public class CompactionTaskInfo {
+ private final List<FileInfo> fileInfoList;
+ private final List<TsFileResource> resources;
+ private int maxConcurrentSeriesNum = 1;
+ private long maxChunkMetadataSize = 0;
+ private int maxChunkMetadataNumInDevice = 0;
+ private int maxChunkMetadataNumInSeries = 0;
+ private long modificationFileSize = 0;
+ private long totalFileSize = 0;
+ private long totalChunkNum = 0;
+ private long totalChunkMetadataSize = 0;
+
+ protected CompactionTaskInfo(List<TsFileResource> resources, List<FileInfo>
fileInfoList) {
+ this.fileInfoList = fileInfoList;
+ this.resources = resources;
+ for (TsFileResource resource : resources) {
+ ModificationFile modificationFile = resource.getModFile();
+ if (modificationFile.exists()) {
+ modificationFileSize += modificationFile.getSize();
+ }
+ this.totalFileSize += resource.getTsFileSize();
+ }
+ for (FileInfo fileInfo : fileInfoList) {
+ maxConcurrentSeriesNum =
+ Math.max(maxConcurrentSeriesNum,
fileInfo.maxAlignedSeriesNumInDevice);
+ maxChunkMetadataNumInSeries =
+ Math.max(maxChunkMetadataNumInSeries, fileInfo.maxSeriesChunkNum);
+ maxChunkMetadataNumInDevice =
+ Math.max(maxChunkMetadataNumInDevice, fileInfo.maxDeviceChunkNum);
+ maxChunkMetadataSize = Math.max(maxChunkMetadataSize,
fileInfo.averageChunkMetadataSize);
+ totalChunkNum += fileInfo.totalChunkNum;
+ totalChunkMetadataSize += fileInfo.totalChunkNum *
fileInfo.averageChunkMetadataSize;
+ }
+ }
+
+ public int getMaxChunkMetadataNumInDevice() {
+ return maxChunkMetadataNumInDevice;
+ }
+
+ public int getMaxChunkMetadataNumInSeries() {
+ return maxChunkMetadataNumInSeries;
+ }
+
+ public long getMaxChunkMetadataSize() {
+ return maxChunkMetadataSize;
+ }
+
+ public List<FileInfo> getFileInfoList() {
+ return fileInfoList;
+ }
+
+ public int getMaxConcurrentSeriesNum() {
+ return maxConcurrentSeriesNum;
+ }
+
+ public long getModificationFileSize() {
+ return modificationFileSize;
+ }
+
+ public long getTotalFileSize() {
+ return totalFileSize;
+ }
+
+ public long getTotalChunkNum() {
+ return totalChunkNum;
+ }
+
+ public List<TsFileResource> getResources() {
+ return resources;
+ }
+
+ public long getTotalChunkMetadataSize() {
+ return totalChunkMetadataSize;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FastCompactionInnerCompactionEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FastCompactionInnerCompactionEstimator.java
new file mode 100644
index 00000000000..a43dd3c2770
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FastCompactionInnerCompactionEstimator.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.compaction.selector.estimator;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.rescon.SystemInfo;
+
+import java.io.IOException;
+
+public class FastCompactionInnerCompactionEstimator extends
AbstractInnerSpaceEstimator {
+
+ @Override
+ public long calculatingMetadataMemoryCost(CompactionTaskInfo taskInfo) {
+ long cost = 0;
+ // add ChunkMetadata size of MultiTsFileDeviceIterator
+ cost +=
+ Math.min(
+ taskInfo.getTotalChunkMetadataSize(),
+ taskInfo.getFileInfoList().size()
+ * taskInfo.getMaxChunkMetadataNumInSeries()
+ * taskInfo.getMaxChunkMetadataSize());
+
+ // add ChunkMetadata size of targetFileWriter
+ long sizeForFileWriter =
+ (long)
+ ((double) SystemInfo.getInstance().getMemorySizeForCompaction()
+ /
IoTDBDescriptor.getInstance().getConfig().getCompactionThreadCount()
+ *
IoTDBDescriptor.getInstance().getConfig().getChunkMetadataSizeProportion());
+ cost += sizeForFileWriter;
+
+ return cost;
+ }
+
+ /**
+ * The data algorithm is: targetChunkWriterSize + mods file size + read
chunk size
+ *
+ * @return estimate data memory cost
+ */
+ @Override
+ public long calculatingDataMemoryCost(CompactionTaskInfo taskInfo) throws
IOException {
+ if (taskInfo.getTotalChunkNum() == 0) {
+ return taskInfo.getModificationFileSize();
+ }
+ long maxConcurrentSeriesNum =
+ Math.max(config.getSubCompactionTaskNum(),
taskInfo.getMaxConcurrentSeriesNum());
+ long averageUncompressedChunkSize =
+ taskInfo.getTotalFileSize() * compressionRatio /
taskInfo.getTotalChunkNum();
+
+ long maxConcurrentSeriesSizeOfTotalFiles =
+ averageUncompressedChunkSize
+ * taskInfo.getFileInfoList().size()
+ * maxConcurrentSeriesNum
+ * taskInfo.getMaxChunkMetadataNumInSeries()
+ / compressionRatio;
+ long maxTargetChunkWriterSize = config.getTargetChunkSize() *
maxConcurrentSeriesNum;
+ long targetChunkWriterSize =
+ Math.min(maxConcurrentSeriesSizeOfTotalFiles,
maxTargetChunkWriterSize);
+
+ long maxConcurrentChunkSizeFromSourceFile =
+ averageUncompressedChunkSize
+ * maxConcurrentSeriesNum
+ *
calculatingMaxOverlapFileNumInSubCompactionTask(taskInfo.getResources());
+
+ return targetChunkWriterSize
+ + maxConcurrentChunkSizeFromSourceFile
+ + taskInfo.getModificationFileSize();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FastCrossSpaceCompactionEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FastCrossSpaceCompactionEstimator.java
new file mode 100644
index 00000000000..4359a2183a1
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FastCrossSpaceCompactionEstimator.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.compaction.selector.estimator;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.rescon.SystemInfo;
+
+import java.io.IOException;
+
+public class FastCrossSpaceCompactionEstimator extends
AbstractCrossSpaceEstimator {
+
+ @Override
+ protected long calculatingMetadataMemoryCost(CompactionTaskInfo taskInfo) {
+ long cost = 0;
+ // add ChunkMetadata size of MultiTsFileDeviceIterator
+ cost +=
+ Math.min(
+ taskInfo.getTotalChunkMetadataSize(),
+ taskInfo.getFileInfoList().size()
+ * taskInfo.getMaxChunkMetadataNumInDevice()
+ * taskInfo.getMaxChunkMetadataSize());
+
+ // add ChunkMetadata size of targetFileWriter
+ long sizeForFileWriter =
+ (long)
+ ((double) SystemInfo.getInstance().getMemorySizeForCompaction()
+ /
IoTDBDescriptor.getInstance().getConfig().getCompactionThreadCount()
+ *
IoTDBDescriptor.getInstance().getConfig().getChunkMetadataSizeProportion());
+ cost += sizeForFileWriter;
+
+ return cost;
+ }
+
+ @Override
+ protected long calculatingDataMemoryCost(CompactionTaskInfo taskInfo) throws
IOException {
+ if (taskInfo.getTotalChunkNum() == 0) {
+ return taskInfo.getModificationFileSize();
+ }
+
+ long maxConcurrentSeriesNum =
+ Math.max(config.getSubCompactionTaskNum(),
taskInfo.getMaxConcurrentSeriesNum());
+ long averageUncompressedChunkSize =
+ taskInfo.getTotalFileSize() * compressionRatio /
taskInfo.getTotalChunkNum();
+
+ long maxConcurrentSeriesSizeOfTotalFiles =
+ averageUncompressedChunkSize
+ * taskInfo.getFileInfoList().size()
+ * maxConcurrentSeriesNum
+ * taskInfo.getMaxChunkMetadataNumInSeries()
+ / compressionRatio;
+ long maxTargetChunkWriterSize = config.getTargetChunkSize() *
maxConcurrentSeriesNum;
+ long targetChunkWriterSize =
+ Math.min(maxConcurrentSeriesSizeOfTotalFiles,
maxTargetChunkWriterSize);
+
+ long maxConcurrentChunkSizeFromSourceFile =
+ averageUncompressedChunkSize
+ * maxConcurrentSeriesNum
+ *
calculatingMaxOverlapFileNumInSubCompactionTask(taskInfo.getResources());
+
+ return targetChunkWriterSize
+ + maxConcurrentChunkSizeFromSourceFile
+ + taskInfo.getModificationFileSize();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FileInfo.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FileInfo.java
new file mode 100644
index 00000000000..ff90a1e21d9
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/FileInfo.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.compaction.selector.estimator;
+
+public class FileInfo {
+ // total chunk num in this tsfile
+ int totalChunkNum = 0;
+ // max chunk num of one timeseries in this tsfile
+ int maxSeriesChunkNum = 0;
+ // max aligned series num in one device. If there is no aligned series in
this file, then it
+ // turns to be -1.
+ int maxAlignedSeriesNumInDevice = -1;
+ // max chunk num of one device in this tsfile
+ @SuppressWarnings("squid:S1068")
+ int maxDeviceChunkNum = 0;
+
+ long averageChunkMetadataSize = 0;
+
+ public FileInfo(
+ int totalChunkNum,
+ int maxSeriesChunkNum,
+ int maxAlignedSeriesNumInDevice,
+ int maxDeviceChunkNum,
+ long averageChunkMetadataSize) {
+ this.totalChunkNum = totalChunkNum;
+ this.maxSeriesChunkNum = maxSeriesChunkNum;
+ this.maxAlignedSeriesNumInDevice = maxAlignedSeriesNumInDevice;
+ this.maxDeviceChunkNum = maxDeviceChunkNum;
+ this.averageChunkMetadataSize = averageChunkMetadataSize;
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/InplaceCompactionEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/InplaceCompactionEstimator.java
deleted file mode 100644
index d8c8e99942a..00000000000
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/InplaceCompactionEstimator.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.db.engine.compaction.selector.estimator;
-
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
-import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
-import org.apache.iotdb.tsfile.read.common.Path;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-public class InplaceCompactionEstimator extends AbstractCrossSpaceEstimator {
- private static final Logger logger =
LoggerFactory.getLogger(InplaceCompactionEstimator.class);
- private static final String LOG_FILE_COST = "Memory cost of file {} is {}";
-
- private boolean tightEstimate;
- private long maxSeqFileCost;
-
- // the number of timeseries being compacted at the same time
- private final int concurrentSeriesNum =
- IoTDBDescriptor.getInstance().getConfig().getSubCompactionTaskNum();
-
- /** Total metadata size of each file. */
- private final Map<TsFileResource, Long> fileMetaSizeMap = new HashMap<>();
-
- /** Maximum memory cost of querying a timeseries in each file. */
- private final Map<TsFileResource, Long> maxSeriesQueryCostMap = new
HashMap<>();
-
- public InplaceCompactionEstimator() {
- this.tightEstimate = false;
- this.maxSeqFileCost = 0;
- }
-
- @Override
- public long estimateCrossCompactionMemory(
- List<TsFileResource> seqResources, TsFileResource unseqResource) throws
IOException {
- if (tightEstimate) {
- return calculateTightMemoryCost(unseqResource, seqResources);
- } else {
- return calculateLooseMemoryCost(unseqResource, seqResources);
- }
- }
-
- private long calculateMemoryCost(
- TsFileResource unseqResource,
- List<TsFileResource> seqResources,
- IFileQueryMemMeasurement unseqMeasurement,
- IFileQueryMemMeasurement seqMeasurement)
- throws IOException {
- long cost = 0;
- Long fileCost = unseqMeasurement.measure(unseqResource);
- cost += fileCost;
-
- for (TsFileResource seqFile : seqResources) {
- fileCost = seqMeasurement.measure(seqFile);
- if (fileCost > maxSeqFileCost) {
- // only one file will be read at the same time, so only the largest
one is recorded here
- cost -= maxSeqFileCost;
- cost += fileCost;
- maxSeqFileCost = fileCost;
- }
- // but writing data into a new file may generate the same amount of
metadata in memory
- cost += calculateMetadataSize(seqFile);
- }
- return cost;
- }
-
- private long calculateLooseMemoryCost(
- TsFileResource unseqResource, List<TsFileResource> seqResources) throws
IOException {
- return calculateMemoryCost(
- unseqResource, seqResources, TsFileResource::getTsFileSize,
this::calculateMetadataSize);
- }
-
- private long calculateTightMemoryCost(
- TsFileResource unseqResource, List<TsFileResource> seqResources) throws
IOException {
- return calculateMemoryCost(
- unseqResource,
- seqResources,
- this::calculateTightUnseqMemoryCost,
- this::calculateTightSeqMemoryCost);
- }
-
- private long calculateMetadataSize(TsFileResource seqFile) throws
IOException {
- Long cost = fileMetaSizeMap.get(seqFile);
- if (cost == null) {
- cost = getFileReader(seqFile).getFileMetadataSize();
- fileMetaSizeMap.put(seqFile, cost);
- logger.debug(LOG_FILE_COST, seqFile, cost);
- }
- return cost;
- }
-
- private long calculateTightFileMemoryCost(
- TsFileResource seqFile, IFileQueryMemMeasurement measurement) throws
IOException {
- Long cost = maxSeriesQueryCostMap.get(seqFile);
- if (cost == null) {
- long[] chunkNums = findTotalAndLargestSeriesChunkNum(seqFile,
getFileReader(seqFile));
- long totalChunkNum = chunkNums[0];
- long maxChunkNum = chunkNums[1];
- cost = measurement.measure(seqFile) * maxChunkNum / totalChunkNum;
- maxSeriesQueryCostMap.put(seqFile, cost);
- logger.debug(LOG_FILE_COST, seqFile, cost);
- }
- return cost;
- }
-
- // this method traverses all ChunkMetadata to find out which series has the
most chunks and uses
- // its proportion to all series to get a maximum estimation
- private long calculateTightSeqMemoryCost(TsFileResource seqFile) throws
IOException {
- long singleSeriesCost = calculateTightFileMemoryCost(seqFile,
this::calculateMetadataSize);
- long multiSeriesCost = concurrentSeriesNum * singleSeriesCost;
- long maxCost = calculateMetadataSize(seqFile);
- return Math.min(multiSeriesCost, maxCost);
- }
-
- // this method traverses all ChunkMetadata to find out which series has the
most chunks and uses
- // its proportion among all series to get a maximum estimation
- private long calculateTightUnseqMemoryCost(TsFileResource unseqFile) throws
IOException {
- long singleSeriesCost = calculateTightFileMemoryCost(unseqFile,
TsFileResource::getTsFileSize);
- long multiSeriesCost = concurrentSeriesNum * singleSeriesCost;
- long maxCost = unseqFile.getTsFileSize();
- return Math.min(multiSeriesCost, maxCost);
- }
-
- // returns totalChunkNum of a file and the max number of chunks of a series
- private long[] findTotalAndLargestSeriesChunkNum(
- TsFileResource tsFileResource, TsFileSequenceReader sequenceReader)
throws IOException {
- long totalChunkNum = 0;
- long maxChunkNum = Long.MIN_VALUE;
- List<Path> paths = sequenceReader.getAllPaths();
-
- for (Path path : paths) {
- List<ChunkMetadata> chunkMetadataList =
sequenceReader.getChunkMetadataList(path, true);
- totalChunkNum += chunkMetadataList.size();
- maxChunkNum = chunkMetadataList.size() > maxChunkNum ?
chunkMetadataList.size() : maxChunkNum;
- }
- logger.debug(
- "In file {}, total chunk num {}, series max chunk num {}",
- tsFileResource,
- totalChunkNum,
- maxChunkNum);
- return new long[] {totalChunkNum, maxChunkNum};
- }
-
- public void setTightEstimate(boolean tightEstimate) {
- this.tightEstimate = tightEstimate;
- }
-}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/ReadChunkInnerCompactionEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/ReadChunkInnerCompactionEstimator.java
new file mode 100644
index 00000000000..9eb64ec2917
--- /dev/null
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/ReadChunkInnerCompactionEstimator.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.compaction.selector.estimator;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.rescon.SystemInfo;
+
+public class ReadChunkInnerCompactionEstimator extends
AbstractInnerSpaceEstimator {
+
+ @Override
+ public long calculatingMetadataMemoryCost(CompactionTaskInfo taskInfo) {
+ long cost = 0;
+ // add ChunkMetadata size of MultiTsFileDeviceIterator
+ cost +=
+ Math.min(
+ taskInfo.getTotalChunkMetadataSize(),
+ taskInfo.getFileInfoList().size()
+ * taskInfo.getMaxChunkMetadataNumInDevice()
+ * taskInfo.getMaxChunkMetadataSize());
+
+ // add ChunkMetadata size of targetFileWriter
+ long sizeForFileWriter =
+ (long)
+ ((double) SystemInfo.getInstance().getMemorySizeForCompaction()
+ /
IoTDBDescriptor.getInstance().getConfig().getCompactionThreadCount()
+ *
IoTDBDescriptor.getInstance().getConfig().getChunkMetadataSizeProportion());
+ cost += sizeForFileWriter;
+
+ return cost;
+ }
+
+ @Override
+ public long calculatingDataMemoryCost(CompactionTaskInfo taskInfo) {
+ if (taskInfo.getTotalChunkNum() == 0) {
+ return taskInfo.getModificationFileSize();
+ }
+ long averageUncompressedChunkSize =
+ taskInfo.getTotalFileSize() * compressionRatio /
taskInfo.getTotalChunkNum();
+
+ long maxConcurrentSeriesSizeOfTotalFiles =
+ averageUncompressedChunkSize
+ * taskInfo.getFileInfoList().size()
+ * taskInfo.getMaxConcurrentSeriesNum()
+ * taskInfo.getMaxChunkMetadataNumInSeries()
+ / compressionRatio;
+ long maxTargetChunkWriterSize =
+ config.getTargetChunkSize() * taskInfo.getMaxConcurrentSeriesNum();
+ long targetChunkWriterSize =
+ Math.min(maxConcurrentSeriesSizeOfTotalFiles,
maxTargetChunkWriterSize);
+
+ long chunkSizeFromSourceFile =
+ averageUncompressedChunkSize * taskInfo.getMaxConcurrentSeriesNum();
+
+ return targetChunkWriterSize + chunkSizeFromSourceFile +
taskInfo.getModificationFileSize();
+ }
+}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/ReadPointCrossCompactionEstimator.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/ReadPointCrossCompactionEstimator.java
deleted file mode 100644
index 9891a48291d..00000000000
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/estimator/ReadPointCrossCompactionEstimator.java
+++ /dev/null
@@ -1,262 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.db.engine.compaction.selector.estimator;
-
-import org.apache.iotdb.commons.conf.IoTDBConstant;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
-import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-public class ReadPointCrossCompactionEstimator extends
AbstractCrossSpaceEstimator {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
-
- // the max cost of reading source seq file among all source seq files of
this cross compaction
- // task
- private long maxCostOfReadingSeqFile;
-
- // the max cost of writing target file
- private long maxCostOfWritingTargetFile;
-
- private int maxConcurrentSeriesNum = 1;
-
- // the number of timeseries being compacted at the same time
- private final int subCompactionTaskNum =
- IoTDBDescriptor.getInstance().getConfig().getSubCompactionTaskNum();
-
- public ReadPointCrossCompactionEstimator() {
- this.maxCostOfReadingSeqFile = 0;
- this.maxCostOfWritingTargetFile = 0;
- }
-
- @Override
- public long estimateCrossCompactionMemory(
- List<TsFileResource> seqResources, TsFileResource unseqResource) throws
IOException {
- if (!addReadLock(seqResources, unseqResource)) {
- // there is file been deleted during selection, return -1
- return -1L;
- }
- try {
- long cost = 0;
- cost += calculateReadingUnseqFile(unseqResource);
- cost += calculateReadingSeqFiles(seqResources);
- cost += calculatingWritingTargetFiles(seqResources, unseqResource);
- return cost;
- } finally {
- releaseReadLock(seqResources, unseqResource);
- }
- }
-
- /** Add read lock. Return false if any of the file were deleted. */
- private boolean addReadLock(List<TsFileResource> seqResources,
TsFileResource unseqResource) {
- List<TsFileResource> allResources = new ArrayList<>(seqResources);
- allResources.add(unseqResource);
- for (int i = 0; i < allResources.size(); i++) {
- TsFileResource resource = allResources.get(i);
- resource.readLock();
- if (resource.isDeleted()) {
- // release read lock
- for (int j = 0; j <= i; j++) {
- allResources.get(j).readUnlock();
- }
- return false;
- }
- }
- return true;
- }
-
- private void releaseReadLock(List<TsFileResource> seqResources,
TsFileResource unseqResource) {
- seqResources.forEach(TsFileResource::readUnlock);
- unseqResource.readUnlock();
- }
-
- /**
- * Calculate memory cost of reading source unseq files in the cross space
compaction. Double the
- * total size of the timeseries to be compacted at the same time in all
unseq files.
- */
- private long calculateReadingUnseqFile(TsFileResource unseqResource) throws
IOException {
- TsFileSequenceReader reader = getFileReader(unseqResource);
- FileInfo fileInfo = getSeriesAndDeviceChunkNum(reader);
- // it is max aligned series num of one device when tsfile contains aligned
series,
- // else is sub compaction task num.
- int concurrentSeriesNum =
- fileInfo.maxAlignedSeriesNumInDevice == -1
- ? subCompactionTaskNum
- : fileInfo.maxAlignedSeriesNumInDevice;
- maxConcurrentSeriesNum = Math.max(maxConcurrentSeriesNum,
concurrentSeriesNum);
- if (fileInfo.totalChunkNum == 0) { // If totalChunkNum ==0, i.e. this
unSeq tsFile has no chunk.
- logger.warn(
- "calculateReadingUnseqFile(), find 1 empty unSeq tsFile: {}.",
- unseqResource.getTsFilePath());
- return 0;
- }
- // it means the max size of a timeseries in this file when reading all of
its chunk into memory.
-
- long resourceFileSize =
- compressionRatio
- * concurrentSeriesNum
- * (unseqResource.getTsFileSize() * fileInfo.maxSeriesChunkNum /
fileInfo.totalChunkNum);
-
- // add mod file size
- long modFileSize = unseqResource.getModFile().getSize();
- return resourceFileSize + modFileSize;
- }
-
- /**
- * Calculate memory cost of reading source seq files in the cross space
compaction. Select the
- * maximun size of the timeseries to be compacted at the same time in one
seq file, because only
- * one seq file will be queried at the same time.
- */
- private long calculateReadingSeqFiles(List<TsFileResource> seqResources)
throws IOException {
- long cost = 0;
- for (TsFileResource seqResource : seqResources) {
- TsFileSequenceReader reader = getFileReader(seqResource);
- FileInfo fileInfo = getSeriesAndDeviceChunkNum(reader);
- // it is max aligned series num of one device when tsfile contains
aligned series,
- // else is sub compaction task num.
- int concurrentSeriesNum =
- fileInfo.maxAlignedSeriesNumInDevice == -1
- ? subCompactionTaskNum
- : fileInfo.maxAlignedSeriesNumInDevice;
- maxConcurrentSeriesNum = Math.max(maxConcurrentSeriesNum,
concurrentSeriesNum);
- long seqFileCost;
- if (fileInfo.totalChunkNum == 0) { // If totalChunkNum ==0, i.e. this
seq tsFile has no chunk.
- logger.warn(
- "calculateReadingSeqFiles(), find 1 empty seq tsFile: {}.",
- seqResource.getTsFilePath());
- seqFileCost = 0;
- } else {
- // We need to multiply the compression ratio here.
- seqFileCost =
- compressionRatio
- * seqResource.getTsFileSize()
- * concurrentSeriesNum
- / fileInfo.totalChunkNum;
- }
-
- if (seqFileCost > maxCostOfReadingSeqFile) {
- // Only one seq file will be read at the same time.
- // not only reading chunk into chunk cache, but also need to
deserialize data point into
- // merge reader. We have to add the cost in merge reader here and the
cost of chunk cache is
- // unnecessary.
- cost -= maxCostOfReadingSeqFile;
- cost += seqFileCost;
- maxCostOfReadingSeqFile = seqFileCost;
- }
-
- // add mod file size
- cost += seqResource.getModFile().getSize();
- }
- return cost;
- }
-
- /**
- * Calculate memory cost of writing target files in the cross space
compaction. Including metadata
- * size of all source files and size of concurrent target chunks.
- */
- private long calculatingWritingTargetFiles(
- List<TsFileResource> seqResources, TsFileResource unseqResource) throws
IOException {
- long cost = 0;
- for (TsFileResource seqResource : seqResources) {
- TsFileSequenceReader reader = getFileReader(seqResource);
- // add seq file metadata size
- cost += reader.getFileMetadataSize();
- }
- // add unseq file metadata size
- cost += getFileReader(unseqResource).getFileMetadataSize();
-
- // concurrent series chunk size
- long writingTargetCost = maxConcurrentSeriesNum *
config.getTargetChunkSize();
- if (writingTargetCost > maxCostOfWritingTargetFile) {
- cost -= maxCostOfWritingTargetFile;
- cost += writingTargetCost;
- maxCostOfWritingTargetFile = writingTargetCost;
- }
-
- return cost;
- }
-
- /**
- * Get the details of the tsfile, the returned array contains the following
elements in sequence:
- *
- * <p>total chunk num in this tsfile
- *
- * <p>max chunk num of one timeseries in this tsfile
- *
- * <p>max aligned series num in one device. If there is no aligned series in
this file, then it
- * turns to be -1.
- *
- * <p>max chunk num of one device in this tsfile
- */
- private FileInfo getSeriesAndDeviceChunkNum(TsFileSequenceReader reader)
throws IOException {
- int totalChunkNum = 0;
- int maxChunkNum = 0;
- int maxAlignedSeriesNumInDevice = -1;
- int maxDeviceChunkNum = 0;
- Map<String, List<TimeseriesMetadata>> deviceMetadata =
reader.getAllTimeseriesMetadata(true);
- for (Map.Entry<String, List<TimeseriesMetadata>> entry :
deviceMetadata.entrySet()) {
- int deviceChunkNum = 0;
- List<TimeseriesMetadata> deviceTimeseriesMetadata = entry.getValue();
- if (deviceTimeseriesMetadata.get(0).getMeasurementId().equals("")) {
- // aligned device
- maxAlignedSeriesNumInDevice =
- Math.max(maxAlignedSeriesNumInDevice,
deviceTimeseriesMetadata.size());
- }
- for (TimeseriesMetadata timeseriesMetadata : deviceTimeseriesMetadata) {
- deviceChunkNum += timeseriesMetadata.getChunkMetadataList().size();
- totalChunkNum += timeseriesMetadata.getChunkMetadataList().size();
- maxChunkNum = Math.max(maxChunkNum,
timeseriesMetadata.getChunkMetadataList().size());
- }
- maxDeviceChunkNum = Math.max(maxDeviceChunkNum, deviceChunkNum);
- }
- return new FileInfo(totalChunkNum, maxChunkNum,
maxAlignedSeriesNumInDevice, maxDeviceChunkNum);
- }
-
- private class FileInfo {
- // total chunk num in this tsfile
- public int totalChunkNum = 0;
- // max chunk num of one timeseries in this tsfile
- public int maxSeriesChunkNum = 0;
- // max aligned series num in one device. If there is no aligned series in
this file, then it
- // turns to be -1.
- public int maxAlignedSeriesNumInDevice = -1;
- // max chunk num of one device in this tsfile
- public int maxDeviceChunkNum = 0;
-
- public FileInfo(
- int totalChunkNum,
- int maxSeriesChunkNum,
- int maxAlignedSeriesNumInDevice,
- int maxDeviceChunkNum) {
- this.totalChunkNum = totalChunkNum;
- this.maxSeriesChunkNum = maxSeriesChunkNum;
- this.maxAlignedSeriesNumInDevice = maxAlignedSeriesNumInDevice;
- this.maxDeviceChunkNum = maxDeviceChunkNum;
- }
- }
-}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/RewriteCrossSpaceCompactionSelector.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/RewriteCrossSpaceCompactionSelector.java
index 596b663e161..4c7acaf1269 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/RewriteCrossSpaceCompactionSelector.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/RewriteCrossSpaceCompactionSelector.java
@@ -24,7 +24,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager;
import org.apache.iotdb.db.engine.compaction.selector.ICompactionSelector;
import org.apache.iotdb.db.engine.compaction.selector.ICrossSpaceSelector;
-import
org.apache.iotdb.db.engine.compaction.selector.estimator.AbstractCompactionEstimator;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.AbstractCrossSpaceEstimator;
import
org.apache.iotdb.db.engine.compaction.selector.utils.CrossCompactionTaskResource;
import
org.apache.iotdb.db.engine.compaction.selector.utils.CrossSpaceCompactionCandidate;
import
org.apache.iotdb.db.engine.compaction.selector.utils.CrossSpaceCompactionCandidate.CrossCompactionTaskResourceSplit;
@@ -39,6 +39,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -59,7 +60,7 @@ public class RewriteCrossSpaceCompactionSelector implements
ICrossSpaceSelector
private final int maxCrossCompactionFileNum;
private final long maxCrossCompactionFileSize;
- private AbstractCompactionEstimator compactionEstimator;
+ private final AbstractCrossSpaceEstimator compactionEstimator;
public RewriteCrossSpaceCompactionSelector(
String logicalStorageGroupName,
@@ -81,8 +82,9 @@ public class RewriteCrossSpaceCompactionSelector implements
ICrossSpaceSelector
IoTDBDescriptor.getInstance().getConfig().getMaxCrossCompactionCandidateFileSize();
this.compactionEstimator =
- ICompactionSelector.getCompactionEstimator(
-
IoTDBDescriptor.getInstance().getConfig().getCrossCompactionPerformer(), false);
+ (AbstractCrossSpaceEstimator)
+ ICompactionSelector.getCompactionEstimator(
+
IoTDBDescriptor.getInstance().getConfig().getCrossCompactionPerformer(), false);
}
/**
@@ -168,8 +170,15 @@ public class RewriteCrossSpaceCompactionSelector
implements ICrossSpaceSelector
}
}
+ List<TsFileResource> newSelectedSeqResources = new
ArrayList<>(taskResource.getSeqFiles());
+ newSelectedSeqResources.addAll(targetSeqFiles);
+ List<TsFileResource> newSelectedUnseqResources =
+ new ArrayList<>(taskResource.getUnseqFiles());
+ newSelectedUnseqResources.add(unseqFile);
+
long memoryCost =
- compactionEstimator.estimateCrossCompactionMemory(targetSeqFiles,
unseqFile);
+ compactionEstimator.estimateCrossCompactionMemory(
+ newSelectedSeqResources, newSelectedUnseqResources);
if (!canAddToTaskResource(taskResource, unseqFile, targetSeqFiles,
memoryCost)) {
break;
}
@@ -247,7 +256,7 @@ public class RewriteCrossSpaceCompactionSelector implements
ICrossSpaceSelector
if (taskResource.getTotalFileNums() + 1 + seqFiles.size() <=
maxCrossCompactionFileNum
&& taskResource.getTotalFileSize() + totalFileSize <=
maxCrossCompactionFileSize
- && taskResource.getTotalMemoryCost() + memoryCost < memoryBudget) {
+ && memoryCost < memoryBudget) {
return true;
}
return false;
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossCompactionTaskResource.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossCompactionTaskResource.java
index 1c9ae6f3d7d..3c46f790910 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossCompactionTaskResource.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossCompactionTaskResource.java
@@ -69,7 +69,7 @@ public class CrossCompactionTaskResource {
TsFileResource unseqFile, List<TsFileResource> seqFiles, long
memoryCost) {
addUnseqFile(unseqFile);
addTargetSeqFiles(seqFiles);
- increaseMemoryCost(memoryCost);
+ updateMemoryCost(memoryCost);
}
private void addUnseqFile(TsFileResource file) {
@@ -88,8 +88,8 @@ public class CrossCompactionTaskResource {
countStatistic(file);
}
- private void increaseMemoryCost(long newMemoryCost) {
- this.totalMemoryCost += newMemoryCost;
+ private void updateMemoryCost(long newMemoryCost) {
+ this.totalMemoryCost = Math.max(totalMemoryCost, newMemoryCost);
}
private void countStatistic(TsFileResource file) {
diff --git a/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java
b/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java
index c77825ddd61..7bb52307adb 100644
--- a/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java
+++ b/server/src/main/java/org/apache/iotdb/db/rescon/SystemInfo.java
@@ -212,6 +212,9 @@ public class SystemInfo {
}
public synchronized void resetCompactionMemoryCost(long
compactionMemoryCost) {
+ if (!config.isEnableCompactionMemControl()) {
+ return;
+ }
this.compactionMemoryCost.addAndGet(-compactionMemoryCost);
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionTaskMemCostEstimatorTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionTaskMemCostEstimatorTest.java
new file mode 100644
index 00000000000..e0f6864371d
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionTaskMemCostEstimatorTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.engine.compaction.utils;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.db.engine.compaction.AbstractCompactionTest;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.FastCompactionInnerCompactionEstimator;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.FastCrossSpaceCompactionEstimator;
+import
org.apache.iotdb.db.engine.compaction.selector.estimator.ReadChunkInnerCompactionEstimator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+public class CompactionTaskMemCostEstimatorTest extends AbstractCompactionTest
{
+
+ @Before
+ public void setUp()
+ throws IOException, WriteProcessException, MetadataException,
InterruptedException {
+ super.setUp();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ }
+
+ @Test
+ public void testEstimateReadChunkInnerSpaceCompactionTaskMemCost()
+ throws IOException, MetadataException, WriteProcessException {
+ createFiles(3, 10, 5, 100000, 0, 0, 50, 50, true, true);
+ tsFileManager.addAll(seqResources, true);
+ List<TsFileResource> tsFileList = tsFileManager.getTsFileList(true);
+ System.out.println(tsFileList.get(0).getTsFile().getAbsolutePath());
+ long cost = new
ReadChunkInnerCompactionEstimator().estimateInnerCompactionMemory(tsFileList);
+ Assert.assertTrue(cost > 0);
+ }
+
+ @Test
+ public void testEstimateReadChunkInnerSpaceCompactionTaskMemCost2()
+ throws IOException, MetadataException, WriteProcessException {
+ createFiles(3, 10, 5, 100, 0, 0, 50, 50, false, true);
+ tsFileManager.addAll(seqResources, true);
+ List<TsFileResource> tsFileList = tsFileManager.getTsFileList(true);
+ long cost = new
ReadChunkInnerCompactionEstimator().estimateInnerCompactionMemory(tsFileList);
+ Assert.assertTrue(cost > 0);
+ }
+
+ @Test
+ public void testEstimateFastCompactionInnerSpaceCompactionTaskMemCost()
+ throws IOException, MetadataException, WriteProcessException {
+ createFiles(5, 10, 5, 10000, 0, 0, 50, 50, true, false);
+ createFiles(10, 4, 5, 10000, 1000, 0, 30, 90, true, false);
+
+ tsFileManager.addAll(unseqResources, false);
+ long cost =
+ new
FastCompactionInnerCompactionEstimator().estimateInnerCompactionMemory(unseqResources);
+ Assert.assertTrue(cost > 0);
+ }
+
+ @Test
+ public void testEstimateFastCompactionInnerSpaceCompactionTaskMemCost2()
+ throws IOException, MetadataException, WriteProcessException {
+ createFiles(3, 10, 5, 100, 0, 0, 50, 50, false, true);
+ tsFileManager.addAll(seqResources, true);
+ List<TsFileResource> tsFileList = tsFileManager.getTsFileList(true);
+ long cost =
+ new
FastCompactionInnerCompactionEstimator().estimateInnerCompactionMemory(tsFileList);
+ Assert.assertTrue(cost > 0);
+ }
+
+ @Test
+ public void testEstimateFastCompactionCrossSpaceCompactionTaskMemCost1()
+ throws IOException, MetadataException, WriteProcessException {
+ createFiles(3, 10, 5, 100, 0, 0, 50, 50, false, true);
+ createFiles(4, 10, 5, 400, 0, 0, 30, 50, false, false);
+ long cost =
+ new FastCrossSpaceCompactionEstimator()
+ .estimateCrossCompactionMemory(seqResources, unseqResources);
+ Assert.assertTrue(cost > 0);
+ }
+}
diff --git
a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
index 6c29e7c4c44..1d896f53ca3 100644
---
a/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
+++
b/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
@@ -223,12 +223,20 @@ public class TsFileSequenceReader implements
AutoCloseable {
return fileMetadataSize;
}
+ /** Return the tsfile meta data size of this tsfile. */
+ public long getFileMetadataSize() throws IOException {
+ return tsFileInput.size() - getFileMetadataPos();
+ }
+
/**
* Return the whole meta data size of this tsfile, including ChunkMetadata,
TimeseriesMetadata and
* etc.
*/
- public long getFileMetadataSize() throws IOException {
- return tsFileInput.size() - getFileMetadataPos();
+ public long getAllMetadataSize() throws IOException {
+ if (tsFileMetaData == null) {
+ readFileMetadata();
+ }
+ return tsFileInput.size() - tsFileMetaData.getMetaOffset();
}
/** this function does not modify the position of the file reader. */
