This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.2 by this push:
new e724c8c8453 [To rel/1.2] add memory estimator on inner space
compaction (#10691)
e724c8c8453 is described below
commit e724c8c84539f56064a0e6f40b290b4340934c9f
Author: shuwenwei <[email protected]>
AuthorDate: Thu Jul 27 16:12:58 2023 +0800
[To rel/1.2] add memory estimator on inner space compaction (#10691)
---
.../execute/task/AbstractCompactionTask.java | 2 +
.../execute/task/CrossSpaceCompactionTask.java | 1 -
.../execute/task/InnerSpaceCompactionTask.java | 37 ++++++++-
.../estimator/AbstractCompactionEstimator.java | 3 +-
.../estimator/AbstractInnerSpaceEstimator.java | 74 ++++++++++++++++-
.../estimator/CompactionEstimateUtils.java | 96 ++++++++++++++++++++++
.../FastCompactionInnerCompactionEstimator.java | 52 ++++++++++++
.../compaction/selector/estimator/FileInfo.java | 48 +++++++++++
.../ReadChunkInnerCompactionEstimator.java | 59 +++++++++++++
.../ReadPointCrossCompactionEstimator.java | 81 +-----------------
.../compaction/CompactionSchedulerTest.java | 1 +
.../utils/CompactionTaskMemCostEstimatorTest.java | 94 +++++++++++++++++++++
.../iotdb/tsfile/read/TsFileSequenceReader.java | 12 ++-
13 files changed, 476 insertions(+), 84 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java
index 4502f678b5a..51c9de96889 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/AbstractCompactionTask.java
@@ -57,6 +57,8 @@ public abstract class AbstractCompactionTask {
protected boolean crossTask;
protected boolean innerSeqTask;
+ protected long memoryCost = 0L;
+
protected AbstractCompactionTask(
String storageGroupName,
String dataRegionId,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
index a8a74e443da..cfc42efda61 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
@@ -61,7 +61,6 @@ public class CrossSpaceCompactionTask extends
AbstractCompactionTask {
protected List<TsFileResource> holdWriteLockList = new ArrayList<>();
protected double selectedSeqFileSize = 0;
protected double selectedUnseqFileSize = 0;
- protected long memoryCost = 0L;
@SuppressWarnings("squid:S107")
public CrossSpaceCompactionTask(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
index 6a843710332..17d70879032 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
@@ -20,21 +20,29 @@
package org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task;
import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.service.metrics.CompactionMetrics;
import org.apache.iotdb.db.service.metrics.FileMetrics;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionExceptionHandler;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionFileCountExceededException;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionMemoryNotEnoughException;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionValidationFailedException;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.subtask.FastCompactionTaskSummary;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.validator.CompactionValidator;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator.AbstractInnerSpaceEstimator;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator.FastCompactionInnerCompactionEstimator;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator.ReadChunkInnerCompactionEstimator;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceList;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.generator.TsFileNameGenerator;
+import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
@@ -70,6 +78,8 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
protected long maxModsFileSize;
+ protected AbstractInnerSpaceEstimator innerSpaceEstimator;
+
public InnerSpaceCompactionTask(
long timePartition,
TsFileManager tsFileManager,
@@ -88,6 +98,13 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
this.selectedTsFileResourceList = selectedTsFileResourceList;
this.sequence = sequence;
this.performer = performer;
+ if
(IoTDBDescriptor.getInstance().getConfig().isEnableCompactionMemControl()) {
+ if (this.performer instanceof ReadChunkCompactionPerformer) {
+ 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) {
@@ -112,6 +129,7 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
if (!tsFileManager.isAllowCompaction()) {
return true;
}
+
long startTime = System.currentTimeMillis();
// get resource of target file
String dataDirectory =
selectedTsFileResourceList.get(0).getTsFile().getParent();
@@ -320,6 +338,8 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
isSequence());
}
} finally {
+ SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
+
SystemInfo.getInstance().decreaseCompactionFileNumCost(selectedTsFileResourceList.size());
releaseAllLocksAndResetStatus();
}
return isSuccess;
@@ -456,9 +476,24 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
return false;
}
}
+ if (innerSpaceEstimator != null) {
+ memoryCost =
innerSpaceEstimator.estimateInnerCompactionMemory(selectedTsFileResourceList);
+ }
+ SystemInfo.getInstance().addCompactionMemoryCost(memoryCost, 60);
+
SystemInfo.getInstance().addCompactionFileNum(selectedTsFileResourceList.size(),
60);
} catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ LOGGER.warn("Interrupted when allocating memory for compaction", e);
+ Thread.currentThread().interrupt();
+ } else if (e instanceof CompactionMemoryNotEnoughException) {
+ LOGGER.info("No enough memory for current compaction task {}", this,
e);
+ } else if (e instanceof CompactionFileCountExceededException) {
+ LOGGER.info("No enough file num for current compaction task {}", this,
e);
+ SystemInfo.getInstance().resetCompactionMemoryCost(memoryCost);
+ }
+ resetCompactionCandidateStatusForAllSourceFiles();
releaseAllLocksAndResetStatus();
- throw e;
+ return false;
}
return true;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java
index 3d53867232b..c4450637344 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractCompactionEstimator.java
@@ -52,7 +52,8 @@ public abstract class AbstractCompactionEstimator {
List<TsFileResource> seqResources, TsFileResource unseqResource) throws
IOException;
/** Estimate the memory cost of compacting the source files in inner space
compaction task. */
- public abstract long estimateInnerCompactionMemory(List<TsFileResource>
resources);
+ public abstract long estimateInnerCompactionMemory(List<TsFileResource>
resources)
+ throws IOException;
/**
* Construct a new or get an existing TsFileSequenceReader of a TsFile.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
index cc34bd28cc4..a5309aaff30 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/AbstractInnerSpaceEstimator.java
@@ -19,9 +19,14 @@
package
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import
org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
/**
@@ -29,11 +34,78 @@ import java.util.List;
* its corresponding implementation.
*/
public abstract class AbstractInnerSpaceEstimator extends
AbstractCompactionEstimator {
- public abstract long estimateInnerCompactionMemory(List<TsFileResource>
resources);
+ protected IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+
+ public long estimateInnerCompactionMemory(List<TsFileResource> resources)
throws IOException {
+ InnerCompactionTaskInfo taskInfo =
calculatingCompactionTaskInfo(resources);
+ long cost = calculatingMetadataMemoryCost(taskInfo);
+ cost += calculatingDataMemoryCost(taskInfo);
+ return cost;
+ }
+
+ public abstract long calculatingMetadataMemoryCost(InnerCompactionTaskInfo
taskInfo);
+
+ public abstract long calculatingDataMemoryCost(InnerCompactionTaskInfo
taskInfo);
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");
}
+
+ protected InnerCompactionTaskInfo
calculatingCompactionTaskInfo(List<TsFileResource> resources)
+ throws IOException {
+ List<FileInfo> fileInfoList = new ArrayList<>();
+ for (TsFileResource resource : resources) {
+ TsFileSequenceReader reader = getFileReader(resource);
+ FileInfo fileInfo =
CompactionEstimateUtils.getSeriesAndDeviceChunkNum(reader);
+ fileInfoList.add(fileInfo);
+ }
+ return new InnerCompactionTaskInfo(resources, fileInfoList);
+ }
+
+ protected static class InnerCompactionTaskInfo {
+ private final List<FileInfo> fileInfoList;
+ private int maxConcurrentSeriesNum = 1;
+ private long maxChunkMetadataSize = 0;
+ private int maxChunkMetadataNumInDevice = 0;
+ private long modificationFileSize = 0;
+
+ protected InnerCompactionTaskInfo(List<TsFileResource> resources,
List<FileInfo> fileInfoList) {
+ this.fileInfoList = fileInfoList;
+ for (TsFileResource resource : resources) {
+ ModificationFile modificationFile = resource.getModFile();
+ if (modificationFile.exists()) {
+ modificationFileSize += modificationFile.getSize();
+ }
+ }
+ for (FileInfo fileInfo : fileInfoList) {
+ maxConcurrentSeriesNum =
+ Math.max(maxConcurrentSeriesNum,
fileInfo.maxAlignedSeriesNumInDevice);
+ maxChunkMetadataNumInDevice =
+ Math.max(maxChunkMetadataNumInDevice, fileInfo.maxDeviceChunkNum);
+ maxChunkMetadataSize = Math.max(maxChunkMetadataSize,
fileInfo.averageChunkMetadataSize);
+ }
+ }
+
+ public int getMaxChunkMetadataNumInDevice() {
+ return maxChunkMetadataNumInDevice;
+ }
+
+ public long getMaxChunkMetadataSize() {
+ return maxChunkMetadataSize;
+ }
+
+ public List<FileInfo> getFileInfoList() {
+ return fileInfoList;
+ }
+
+ public int getMaxConcurrentSeriesNum() {
+ return maxConcurrentSeriesNum;
+ }
+
+ public long getModificationFileSize() {
+ return modificationFileSize;
+ }
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/CompactionEstimateUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/CompactionEstimateUtils.java
new file mode 100644
index 00000000000..fe6f24fcfb4
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/CompactionEstimateUtils.java
@@ -0,0 +1,96 @@
+/*
+ * 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.storageengine.dataregion.compaction.selector.estimator;
+
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.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 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);
+ }
+ 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/FastCompactionInnerCompactionEstimator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/FastCompactionInnerCompactionEstimator.java
new file mode 100644
index 00000000000..9de3f58af9b
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/FastCompactionInnerCompactionEstimator.java
@@ -0,0 +1,52 @@
+/*
+ * 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.storageengine.dataregion.compaction.selector.estimator;
+
+public class FastCompactionInnerCompactionEstimator extends
AbstractInnerSpaceEstimator {
+
+ /**
+ * The metadata algorithm is: maxChunkMetaDataSize * maxChunkNumber *
fileSize * maxSeriesNumber
+ *
+ * @return estimate metadata memory cost
+ */
+ @Override
+ public long calculatingMetadataMemoryCost(InnerCompactionTaskInfo taskInfo) {
+ return taskInfo.getFileInfoList().size()
+ * taskInfo.getMaxChunkMetadataNumInDevice()
+ * taskInfo.getMaxChunkMetadataSize()
+ * Math.max(config.getSubCompactionTaskNum(),
taskInfo.getMaxConcurrentSeriesNum());
+ }
+
+ /**
+ * The data algorithm is: (targetChunkSize * fileSize * compressionRatio *
maxSeriesNumber) +
+ * modsFileSize
+ *
+ * @return estimate data memory cost
+ */
+ @Override
+ public long calculatingDataMemoryCost(InnerCompactionTaskInfo taskInfo) {
+ long cost =
+ config.getTargetChunkSize()
+ * taskInfo.getFileInfoList().size()
+ * Math.max(config.getSubCompactionTaskNum(),
taskInfo.getMaxConcurrentSeriesNum());
+ cost += taskInfo.getModificationFileSize();
+ return cost;
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/FileInfo.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/FileInfo.java
new file mode 100644
index 00000000000..7ff1ae9be3d
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/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.storageengine.dataregion.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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadChunkInnerCompactionEstimator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadChunkInnerCompactionEstimator.java
new file mode 100644
index 00000000000..02924768952
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadChunkInnerCompactionEstimator.java
@@ -0,0 +1,59 @@
+/*
+ * 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.storageengine.dataregion.compaction.selector.estimator;
+
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo;
+
+public class ReadChunkInnerCompactionEstimator extends
AbstractInnerSpaceEstimator {
+
+ @Override
+ public long calculatingMetadataMemoryCost(InnerCompactionTaskInfo taskInfo) {
+ long cost = 0;
+ // add ChunkMetadata size of MultiTsFileDeviceIterator
+ cost +=
+ 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(InnerCompactionTaskInfo taskInfo) {
+ // add max target chunk size and max source chunk size
+ long cost =
+ 2
+ * taskInfo.getMaxConcurrentSeriesNum()
+ * IoTDBDescriptor.getInstance().getConfig().getTargetChunkSize();
+
+ // add modification file size
+ cost += taskInfo.getModificationFileSize();
+ return cost;
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadPointCrossCompactionEstimator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadPointCrossCompactionEstimator.java
index 318420a831f..d8908cfd0e6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadPointCrossCompactionEstimator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/estimator/ReadPointCrossCompactionEstimator.java
@@ -22,7 +22,6 @@ package
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimat
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
-import org.apache.iotdb.tsfile.file.metadata.TimeseriesMetadata;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.slf4j.Logger;
@@ -31,7 +30,6 @@ 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 =
@@ -77,18 +75,7 @@ public class ReadPointCrossCompactionEstimator extends
AbstractCrossSpaceEstimat
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;
+ return CompactionEstimateUtils.addReadLock(allResources);
}
private void releaseReadLock(List<TsFileResource> seqResources,
TsFileResource unseqResource) {
@@ -104,7 +91,7 @@ public class ReadPointCrossCompactionEstimator extends
AbstractCrossSpaceEstimat
*/
private long calculateReadingUnseqFile(TsFileResource unseqResource) throws
IOException {
TsFileSequenceReader reader = getFileReader(unseqResource);
- FileInfo fileInfo = getSeriesAndDeviceChunkNum(reader);
+ FileInfo fileInfo =
CompactionEstimateUtils.getSeriesAndDeviceChunkNum(reader);
// it is max aligned series num of one device when tsfile contains aligned
series,
// else is sub compaction task num.
int concurrentSeriesNum =
@@ -141,7 +128,7 @@ public class ReadPointCrossCompactionEstimator extends
AbstractCrossSpaceEstimat
long cost = 0;
for (TsFileResource seqResource : seqResources) {
TsFileSequenceReader reader = getFileReader(seqResource);
- FileInfo fileInfo = getSeriesAndDeviceChunkNum(reader);
+ FileInfo fileInfo =
CompactionEstimateUtils.getSeriesAndDeviceChunkNum(reader);
// it is max aligned series num of one device when tsfile contains
aligned series,
// else is sub compaction task num.
int concurrentSeriesNum =
@@ -207,66 +194,4 @@ public class ReadPointCrossCompactionEstimator extends
AbstractCrossSpaceEstimat
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
- *
- * @throws IOException if io errors occurred
- */
- 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
- private int totalChunkNum = 0;
- // max chunk num of one timeseries in this tsfile
- private 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.
- private int maxAlignedSeriesNumInDevice = -1;
- // max chunk num of one device in this tsfile
- @SuppressWarnings("squid:S1068")
- private 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/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionSchedulerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionSchedulerTest.java
index 198e865dd07..d417eab7f7f 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionSchedulerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionSchedulerTest.java
@@ -101,6 +101,7 @@ public class CompactionSchedulerTest {
.getConfig()
.setInnerUnseqCompactionPerformer(InnerUnseqCompactionPerformer.READ_POINT);
IoTDBDescriptor.getInstance().getConfig().setMinCrossCompactionUnseqFileLevel(0);
+
IoTDBDescriptor.getInstance().getConfig().setEnableCompactionMemControl(false);
CompactionTaskManager.getInstance().start();
while (CompactionTaskManager.getInstance().getExecutingTaskCount() > 0) {
try {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTaskMemCostEstimatorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTaskMemCostEstimatorTest.java
new file mode 100644
index 00000000000..fc0e5d3d4c4
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionTaskMemCostEstimatorTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.storageengine.dataregion.compaction.utils;
+
+import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator.FastCompactionInnerCompactionEstimator;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.estimator.ReadChunkInnerCompactionEstimator;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+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(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
FastCompactionInnerCompactionEstimator().estimateInnerCompactionMemory(tsFileList);
+ 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);
+ }
+}
diff --git
a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
index 56602d07b09..b1742330c2c 100644
---
a/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
+++
b/iotdb-core/tsfile/src/main/java/org/apache/iotdb/tsfile/read/TsFileSequenceReader.java
@@ -228,12 +228,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. */
