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 5c7b113f6ee [To rel/1.1] cp commits from master to solve the mods file
OOM (#10928)
5c7b113f6ee is described below
commit 5c7b113f6eeee38d45609d72eecda1611a45553b
Author: Zhijia Cao <[email protected]>
AuthorDate: Wed Aug 23 17:54:02 2023 +0800
[To rel/1.1] cp commits from master to solve the mods file OOM (#10928)
---
.../apache/iotdb/commons/conf/IoTDBConstant.java | 4 +
.../execute/recover/CompactionRecoverManager.java | 56 ++++++-
.../execute/task/InnerSpaceCompactionTask.java | 12 ++
.../DefaultCompactionTaskComparatorImpl.java | 7 +
.../ReadPointCrossCompactionEstimator.java | 15 +-
.../impl/SizeTieredCompactionSelector.java | 72 +++++---
.../iotdb/db/engine/modification/Deletion.java | 10 ++
.../db/engine/modification/ModificationFile.java | 88 +++++++++-
.../iotdb/db/engine/storagegroup/DataRegion.java | 2 +
.../iotdb/db/query/context/QueryContext.java | 38 +----
.../compaction/CompactionTaskComparatorTest.java | 39 +++++
.../inner/InnerSpaceCompactionSelectorTest.java | 33 ++++
.../engine/modification/ModificationFileTest.java | 182 +++++++++++++++++++++
.../engine/storagegroup/FakedTsFileResource.java | 1 +
.../iotdb/tsfile/read/common/TimeRangeTest.java | 41 +++++
15 files changed, 529 insertions(+), 71 deletions(-)
diff --git
a/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
b/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
index 628fdbf8a23..3dc6842a511 100644
---
a/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
+++
b/node-commons/src/main/java/org/apache/iotdb/commons/conf/IoTDBConstant.java
@@ -257,6 +257,10 @@ public class IoTDBConstant {
// compaction mods of previous version (<0.13)
public static final String COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD =
"merge.mods";
+ public static final String SETTLE_SUFFIX = ".settle";
+ public static final String MODS_SETTLE_FILE_SUFFIX = ".mods.settle";
+ public static final String BLANK = "";
+
// write ahead log
public static final String WAL_FILE_PREFIX = "_";
public static final String WAL_FILE_SUFFIX = ".wal";
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java
index b51ff61268d..8b202ee997b 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/recover/CompactionRecoverManager.java
@@ -28,9 +28,16 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.List;
import java.util.regex.Pattern;
+import java.util.stream.Stream;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.BLANK;
+import static
org.apache.iotdb.commons.conf.IoTDBConstant.MODS_SETTLE_FILE_SUFFIX;
+import static org.apache.iotdb.commons.conf.IoTDBConstant.SETTLE_SUFFIX;
import static
org.apache.iotdb.db.engine.compaction.execute.utils.log.CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD;
import static
org.apache.iotdb.db.engine.compaction.execute.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
@@ -87,18 +94,51 @@ public class CompactionRecoverManager {
||
!Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
continue;
}
- File[] compactionLogs =
- CompactionLogger.findCompactionLogs(isInnerSpace,
timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("Calling compaction recover task.");
- new CompactionRecoverTask(
- logicalStorageGroupName, dataRegionId, tsFileManager,
compactionLog, isInnerSpace)
- .doCompaction();
- }
+ // recover temporary files generated during compacted
+ recoverCompaction(isInnerSpace, timePartitionDir);
+
+ // recover temporary files generated during .mods file settled
+ recoverModSettleFile(timePartitionDir.toPath());
}
}
}
+ public void recoverModSettleFile(Path timePartitionDir) {
+ try (Stream<Path> settlesStream = Files.list(timePartitionDir)) {
+ settlesStream
+ .filter(path -> path.toString().endsWith(MODS_SETTLE_FILE_SUFFIX))
+ .forEach(
+ modsSettle -> {
+ Path originModFile =
+ modsSettle.resolveSibling(
+
modsSettle.getFileName().toString().replace(SETTLE_SUFFIX, BLANK));
+ try {
+ if (Files.exists(originModFile)) {
+ Files.deleteIfExists(modsSettle);
+ } else {
+ Files.move(modsSettle, originModFile);
+ }
+ } catch (IOException e) {
+ logger.error(
+ "recover mods file error on delete origin file or rename
mods settle,", e);
+ }
+ });
+ } catch (IOException e) {
+ logger.error("recover mods file error on list files:{}",
timePartitionDir, e);
+ }
+ }
+
+ public void recoverCompaction(boolean isInnerSpace, File timePartitionDir) {
+ File[] compactionLogs =
+ CompactionLogger.findCompactionLogs(isInnerSpace,
timePartitionDir.getPath());
+ for (File compactionLog : compactionLogs) {
+ logger.info("Calling compaction recover task.");
+ new CompactionRecoverTask(
+ logicalStorageGroupName, dataRegionId, tsFileManager,
compactionLog, isInnerSpace)
+ .doCompaction();
+ }
+ }
+
/** Check whether there is old compaction log from previous version (<0.13)
and recover it. */
private void recoverCompactionBefore013(boolean isInnerSpace) {
String oldLogName =
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 6cd8875d184..2219fc8d08b 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
@@ -46,6 +46,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
public class InnerSpaceCompactionTask extends AbstractCompactionTask {
@@ -66,6 +67,8 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
protected boolean[] isHoldingReadLock;
protected boolean[] isHoldingWriteLock;
+ protected long maxModsFileSize;
+
public InnerSpaceCompactionTask(
long timePartition,
TsFileManager tsFileManager,
@@ -339,6 +342,7 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
sumOfCompactionCount = 0;
maxFileVersion = -1L;
maxCompactionCount = -1;
+ maxModsFileSize = 0;
if (selectedTsFileResourceList == null) {
return;
}
@@ -354,6 +358,10 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
if (fileName.getVersion() > maxFileVersion) {
maxFileVersion = fileName.getVersion();
}
+ if (!Objects.isNull(resource.getModFile())) {
+ long modsFileSize = resource.getModFile().getSize();
+ maxModsFileSize = Math.max(maxModsFileSize, modsFileSize);
+ }
} catch (IOException e) {
LOGGER.warn("Fail to get the tsfile name of {}", resource.getTsFile(),
e);
}
@@ -380,6 +388,10 @@ public class InnerSpaceCompactionTask extends
AbstractCompactionTask {
return maxFileVersion;
}
+ public long getMaxModsFileSize() {
+ return maxModsFileSize;
+ }
+
@Override
public String toString() {
return storageGroupName
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
index 6ef8d7f2b24..24dcc5d1d43 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
@@ -62,6 +62,13 @@ public class DefaultCompactionTaskComparatorImpl implements
ICompactionTaskCompa
public int compareInnerSpaceCompactionTask(
InnerSpaceCompactionTask o1, InnerSpaceCompactionTask o2) {
+
+ // if max mods file size of o1 and o2 are different
+ // we prefer to execute task with greater mods file
+ if (o1.getMaxModsFileSize() != o2.getMaxModsFileSize()) {
+ return o2.getMaxModsFileSize() > o1.getMaxModsFileSize() ? 1 : -1;
+ }
+
// if the sum of compaction count of the selected files are different
// we prefer to execute task with smaller compaction count
// this can reduce write amplification
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
index 13bcce13277..9891a48291d 100644
---
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
@@ -116,9 +116,15 @@ public class ReadPointCrossCompactionEstimator extends
AbstractCrossSpaceEstimat
return 0;
}
// it means the max size of a timeseries in this file when reading all of
its chunk into memory.
- return compressionRatio
- * concurrentSeriesNum
- * (unseqResource.getTsFileSize() * fileInfo.maxSeriesChunkNum /
fileInfo.totalChunkNum);
+
+ long resourceFileSize =
+ compressionRatio
+ * concurrentSeriesNum
+ * (unseqResource.getTsFileSize() * fileInfo.maxSeriesChunkNum /
fileInfo.totalChunkNum);
+
+ // add mod file size
+ long modFileSize = unseqResource.getModFile().getSize();
+ return resourceFileSize + modFileSize;
}
/**
@@ -162,6 +168,9 @@ public class ReadPointCrossCompactionEstimator extends
AbstractCrossSpaceEstimat
cost += seqFileCost;
maxCostOfReadingSeqFile = seqFileCost;
}
+
+ // add mod file size
+ cost += seqResource.getModFile().getSize();
}
return cost;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/SizeTieredCompactionSelector.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/SizeTieredCompactionSelector.java
index 805abe2f1d8..4a327ba84e3 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/SizeTieredCompactionSelector.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/impl/SizeTieredCompactionSelector.java
@@ -25,6 +25,7 @@ import
org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager;
import
org.apache.iotdb.db.engine.compaction.schedule.comparator.ICompactionTaskComparator;
import org.apache.iotdb.db.engine.compaction.selector.IInnerSeqSpaceSelector;
import org.apache.iotdb.db.engine.compaction.selector.IInnerUnseqSpaceSelector;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -40,6 +41,7 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
+import java.util.Objects;
import java.util.PriorityQueue;
/**
@@ -64,6 +66,7 @@ public class SizeTieredCompactionSelector
protected boolean sequence;
protected TsFileManager tsFileManager;
protected boolean hasNextTimePartition;
+ private static final long MODS_FILE_SIZE_THRESHOLD = 1024 * 1024 * 50L;
public SizeTieredCompactionSelector(
String storageGroupName,
@@ -88,27 +91,23 @@ public class SizeTieredCompactionSelector
* longer search for higher layers), otherwise it will return true.
*
* @param level the level to be searched
- * @param taskPriorityQueue it stores the batches of files to be compacted
and the total size of
- * each batch
* @return return whether to continue the search to higher levels
* @throws IOException
*/
- private boolean selectLevelTask(
- int level, PriorityQueue<Pair<List<TsFileResource>, Long>>
taskPriorityQueue)
- throws IOException {
- boolean shouldContinueToSearch = true;
+ @SuppressWarnings({"squid:S3776", "squid:S135"})
+ private List<Pair<List<TsFileResource>, Long>> selectSingleLevel(int level)
throws IOException {
List<TsFileResource> selectedFileList = new ArrayList<>();
long selectedFileSize = 0L;
long targetCompactionFileSize = config.getTargetCompactionFileSize();
+ List<Pair<List<TsFileResource>, Long>> taskList = new ArrayList<>();
for (TsFileResource currentFile : tsFileResources) {
TsFileNameGenerator.TsFileName currentName =
TsFileNameGenerator.getTsFileName(currentFile.getTsFile().getName());
if (currentName.getInnerCompactionCnt() != level) {
// meet files of another level
if (selectedFileList.size() > 1) {
- taskPriorityQueue.add(new Pair<>(new ArrayList<>(selectedFileList),
selectedFileSize));
- shouldContinueToSearch = false;
+ taskList.add(new Pair<>(new ArrayList<>(selectedFileList),
selectedFileSize));
}
selectedFileList = new ArrayList<>();
selectedFileSize = 0L;
@@ -132,8 +131,7 @@ public class SizeTieredCompactionSelector
|| selectedFileList.size() >=
config.getMaxInnerCompactionCandidateFileNum()) {
// submit the task
if (selectedFileList.size() > 1) {
- taskPriorityQueue.add(new Pair<>(new ArrayList<>(selectedFileList),
selectedFileSize));
- shouldContinueToSearch = false;
+ taskList.add(new Pair<>(new ArrayList<>(selectedFileList),
selectedFileSize));
}
selectedFileList = new ArrayList<>();
selectedFileSize = 0L;
@@ -143,17 +141,19 @@ public class SizeTieredCompactionSelector
// if next time partition exists
// submit a merge task even it does not meet the requirement for file num
or file size
if (hasNextTimePartition && selectedFileList.size() > 1) {
- taskPriorityQueue.add(new Pair<>(new ArrayList<>(selectedFileList),
selectedFileSize));
- shouldContinueToSearch = false;
+ taskList.add(new Pair<>(new ArrayList<>(selectedFileList),
selectedFileSize));
}
- return shouldContinueToSearch;
+ return taskList;
}
/**
- * This method searches for a batch of files to be compacted from layer 0 to
the highest layer. If
- * there are more than a batch of files to be merged on a certain layer, it
does not search to
- * higher layers. It creates a compaction thread for each batch of files and
put it into the
- * candidateCompactionTaskQueue of the {@link CompactionTaskManager}.
+ * This method is used to select a batch of files to be merged. There are
two ways to select
+ * files.If the first method selects the appropriate file, the second method
is not executed. The
+ * first one is based on the mods file corresponding to the file. We will
preferentially select
+ * file with mods file larger than 50M. The second way is based on the file
layer from layer 0 to
+ * the highest layer. If there are more than a batch of files to be merged
on a certain layer, it
+ * does not search to higher layers. It creates a compaction thread for each
batch of files and
+ * put it into the candidateCompactionTaskQueue of the {@link
CompactionTaskManager}.
*
* @return Returns whether the file was found and submits the merge task
*/
@@ -163,12 +163,14 @@ public class SizeTieredCompactionSelector
PriorityQueue<Pair<List<TsFileResource>, Long>> taskPriorityQueue =
new PriorityQueue<>(new SizeTieredCompactionTaskComparator());
try {
- int maxLevel = searchMaxFileLevel();
- for (int currentLevel = 0; currentLevel <= maxLevel; currentLevel++) {
- if (!selectLevelTask(currentLevel, taskPriorityQueue)) {
- break;
- }
+ // preferentially select files based on mods file size
+ taskPriorityQueue.addAll(selectMaxModsFileTask());
+
+ // if a suitable file is not selected in the first step, select the file
at the tsfile level
+ if (taskPriorityQueue.isEmpty()) {
+ taskPriorityQueue.addAll(selectLevelTask());
}
+
List<List<TsFileResource>> taskList = new LinkedList<>();
while (taskPriorityQueue.size() > 0) {
List<TsFileResource> resources = taskPriorityQueue.poll().left;
@@ -181,6 +183,32 @@ public class SizeTieredCompactionSelector
return Collections.emptyList();
}
+ private List<Pair<List<TsFileResource>, Long>> selectLevelTask() throws
IOException {
+ List<Pair<List<TsFileResource>, Long>> taskList = new ArrayList<>();
+ int maxLevel = searchMaxFileLevel();
+ for (int currentLevel = 0; currentLevel <= maxLevel; currentLevel++) {
+ List<Pair<List<TsFileResource>, Long>> singleLevelTask =
selectSingleLevel(currentLevel);
+ if (!singleLevelTask.isEmpty()) {
+ taskList.addAll(singleLevelTask);
+ break;
+ }
+ }
+ return taskList;
+ }
+
+ private List<Pair<List<TsFileResource>, Long>> selectMaxModsFileTask() {
+ List<Pair<List<TsFileResource>, Long>> taskList = new ArrayList<>();
+ for (TsFileResource tsFileResource : tsFileResources) {
+ ModificationFile modFile = tsFileResource.getModFile();
+ if (!Objects.isNull(modFile) && modFile.getSize() >
MODS_FILE_SIZE_THRESHOLD) {
+ taskList.add(
+ new Pair<>(Collections.singletonList(tsFileResource),
tsFileResource.getTsFileSize()));
+ LOGGER.debug("select tsfile {},the mod file size is {}",
tsFileResource, modFile.getSize());
+ }
+ }
+ return taskList;
+ }
+
private int searchMaxFileLevel() throws IOException {
int maxLevel = -1;
for (TsFileResource currentFile : tsFileResources) {
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java
b/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java
index be03f2d17ba..e2942f7d1fa 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/modification/Deletion.java
@@ -44,6 +44,10 @@ public class Deletion extends Modification implements
Cloneable {
public Deletion(PartialPath path, long fileOffset, long endTime) {
super(Type.DELETION, path, fileOffset);
this.timeRange = new TimeRange(Long.MIN_VALUE, endTime);
+ this.timeRange.setLeftClose(false);
+ if (endTime == Long.MAX_VALUE) {
+ this.timeRange.setRightClose(false);
+ }
}
/**
@@ -56,6 +60,12 @@ public class Deletion extends Modification implements
Cloneable {
public Deletion(PartialPath path, long fileOffset, long startTime, long
endTime) {
super(Type.DELETION, path, fileOffset);
this.timeRange = new TimeRange(startTime, endTime);
+ if (startTime == Long.MIN_VALUE) {
+ this.timeRange.setLeftClose(false);
+ }
+ if (endTime == Long.MAX_VALUE) {
+ this.timeRange.setRightClose(false);
+ }
}
public long getStartTime() {
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java
b/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java
index 6c9c063f9e2..80646e62fcb 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/modification/ModificationFile.java
@@ -36,8 +36,13 @@ import java.io.IOException;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
+import java.util.Map;
import java.util.Random;
+import java.util.Set;
+import java.util.stream.Collectors;
/**
* ModificationFile stores the Modifications of a TsFile or unseq file in
another file in the same
@@ -47,6 +52,7 @@ public class ModificationFile implements AutoCloseable {
private static final Logger logger =
LoggerFactory.getLogger(ModificationFile.class);
public static final String FILE_SUFFIX = ".mods";
+ public static final String COMPACT_SUFFIX = ".settle";
public static final String COMPACTION_FILE_SUFFIX = ".compaction.mods";
// whether to verify the last line, it may be incomplete in extreme cases
@@ -57,6 +63,9 @@ public class ModificationFile implements AutoCloseable {
private String filePath;
private Random random = new Random();
+ private static final long COMPACT_THRESHOLD = 1024 * 1024;
+
+ private boolean hasCompacted = false;
/**
* Construct a ModificationFile using a file as its storage.
*
@@ -124,7 +133,10 @@ public class ModificationFile implements AutoCloseable {
public void remove() throws IOException {
close();
- FSFactoryProducer.getFSFactory().getFile(filePath).delete();
+ boolean deleted =
FSFactoryProducer.getFSFactory().getFile(filePath).delete();
+ if (!deleted) {
+ logger.warn("Delete ModificationFile {} failed.", filePath);
+ }
}
public boolean exists() {
@@ -177,4 +189,78 @@ public class ModificationFile implements AutoCloseable {
return 0;
}
}
+
+ public void compact() {
+ long originFileSize = getSize();
+ if (originFileSize > COMPACT_THRESHOLD && !hasCompacted) {
+ Map<String, List<Modification>> pathModificationMap =
+
getModifications().stream().collect(Collectors.groupingBy(Modification::getPathString));
+ String newModsFileName = filePath + COMPACT_SUFFIX;
+ try (ModificationFile compactedModificationFile = new
ModificationFile(newModsFileName)) {
+ Set<Map.Entry<String, List<Modification>>> modificationsEntrySet =
+ pathModificationMap.entrySet();
+ for (Map.Entry<String, List<Modification>> modificationEntry :
modificationsEntrySet) {
+ List<Modification> settledModifications =
sortAndMerge(modificationEntry.getValue());
+ for (Modification settledModification : settledModifications) {
+ compactedModificationFile.write(settledModification);
+ }
+ }
+ } catch (IOException e) {
+ logger.error("compact mods file exception of {}", filePath, e);
+ }
+
+ try {
+ // remove origin mods file
+ this.remove();
+ // rename new mods file to origin name
+ Files.move(new File(newModsFileName).toPath(), new
File(filePath).toPath());
+ logger.info("{} settle successful", filePath);
+
+ if (getSize() > COMPACT_THRESHOLD) {
+ logger.warn(
+ "After the mod file is settled, the file size is still greater
than 1M,the size of the file before settle is {},after settled the file size is
{}",
+ originFileSize,
+ getSize());
+ }
+ } catch (IOException e) {
+ logger.error("remove origin file or rename new mods file error.", e);
+ }
+ hasCompacted = true;
+ }
+ }
+
+ public static List<Modification> sortAndMerge(List<Modification>
modifications) {
+ modifications.sort(
+ (o1, o2) -> {
+ if (!o1.getType().equals(o2.getType())) {
+ return o1.getType().compareTo(o2.getType());
+ } else if (!o1.getPath().equals(o2.getPath())) {
+ return o1.getPath().compareTo(o2.getPath());
+ } else if (o1.getFileOffset() != o2.getFileOffset()) {
+ return (int) (o1.getFileOffset() - o2.getFileOffset());
+ } else {
+ if (o1.getType() == Modification.Type.DELETION) {
+ Deletion del1 = (Deletion) o1;
+ Deletion del2 = (Deletion) o2;
+ return del1.getTimeRange().compareTo(del2.getTimeRange());
+ }
+ throw new IllegalArgumentException();
+ }
+ });
+ List<Modification> result = new ArrayList<>();
+ if (!modifications.isEmpty()) {
+ Deletion current = ((Deletion) modifications.get(0)).clone();
+ for (int i = 1; i < modifications.size(); i++) {
+ Deletion del = (Deletion) modifications.get(i);
+ if (current.intersects(del)) {
+ current.merge(del);
+ } else {
+ result.add(current);
+ current = del.clone();
+ }
+ }
+ result.add(current);
+ }
+ return result;
+ }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
index f9fe55e2d62..2b6e2b27e91 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/DataRegion.java
@@ -2055,6 +2055,8 @@ public class DataRegion implements IDataRegionForQuery {
tsFileResource.getModFile().write(deletion);
// remember to close mod file
tsFileResource.getModFile().close();
+ // if file length greater than 1M,execute compact.
+ modFile.compact();
if (!modFileExists) {
TsFileMetricManager.getInstance().increaseModFileNum(1);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java
b/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java
index 7fabfc378e6..db3bc4e3f1c 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/context/QueryContext.java
@@ -22,7 +22,6 @@ package org.apache.iotdb.db.query.context;
import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.path.PatternTreeMap;
-import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.metadata.path.PatternTreeMapFactory;
@@ -113,45 +112,10 @@ public class QueryContext {
}
fileModCache.put(modFile.getFilePath(), allModifications);
}
- return sortAndMerge(allModifications.getOverlapped(path));
+ return
ModificationFile.sortAndMerge(allModifications.getOverlapped(path));
});
}
- private List<Modification> sortAndMerge(List<Modification> modifications) {
- modifications.sort(
- (o1, o2) -> {
- if (!o1.getType().equals(o2.getType())) {
- return o1.getType().compareTo(o2.getType());
- } else if (!o1.getPath().equals(o2.getPath())) {
- return o1.getPath().compareTo(o2.getPath());
- } else if (o1.getFileOffset() != o2.getFileOffset()) {
- return (int) (o1.getFileOffset() - o2.getFileOffset());
- } else {
- if (o1.getType() == Modification.Type.DELETION) {
- Deletion del1 = (Deletion) o1;
- Deletion del2 = (Deletion) o2;
- return del1.getTimeRange().compareTo(del2.getTimeRange());
- }
- throw new IllegalArgumentException();
- }
- });
- List<Modification> result = new ArrayList<>();
- if (!modifications.isEmpty()) {
- Deletion current = ((Deletion) modifications.get(0)).clone();
- for (int i = 1; i < modifications.size(); i++) {
- Deletion del = (Deletion) modifications.get(i);
- if (current.intersects(del)) {
- current.merge(del);
- } else {
- result.add(current);
- current = del.clone();
- }
- }
- result.add(current);
- }
- return result;
- }
-
/**
* Find the modifications of all aligned 'paths' in 'modFile'. If they are
not in the cache, read
* them from 'modFile' and put then into the cache.
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
index 8dd67723c13..9772626f02d 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.engine.compaction;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import
org.apache.iotdb.db.engine.compaction.execute.performer.impl.FastCompactionPerformer;
import
org.apache.iotdb.db.engine.compaction.execute.performer.impl.ReadPointCompactionPerformer;
@@ -28,6 +30,7 @@ import
org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager;
import
org.apache.iotdb.db.engine.compaction.schedule.comparator.DefaultCompactionTaskComparatorImpl;
import
org.apache.iotdb.db.engine.compaction.schedule.constant.CompactionPriority;
import org.apache.iotdb.db.engine.compaction.utils.CompactionConfigRestorer;
+import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
@@ -41,7 +44,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -374,6 +379,40 @@ public class CompactionTaskComparatorTest {
}
}
+ @Test
+ public void testCompareByMaxModsFileSize()
+ throws InterruptedException, IllegalPathException, IOException {
+ for (int i = 0; i < 100; ++i) {
+ List<TsFileResource> resources = new ArrayList<>();
+ for (int j = i; j < 100; ++j) {
+ resources.add(
+ new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile",
i + j, i + j)), j));
+ }
+ FakedInnerSpaceCompactionTask innerTask =
+ new FakedInnerSpaceCompactionTask(
+ "fakeSg", 0, tsFileManager, taskNum, true, resources, 0);
+ compactionTaskQueue.put(innerTask);
+ }
+
+ String targetFileName = "101-101-0-0.tsfile";
+ FakedTsFileResource fakedTsFileResource =
+ new FakedTsFileResource(new File(targetFileName), 100);
+ fakedTsFileResource.getModFile().write(new Deletion(new
PartialPath("root.test.d1"), 1, 1));
+ compactionTaskQueue.put(
+ new FakedInnerSpaceCompactionTask(
+ "fakeSg",
+ 0,
+ tsFileManager,
+ taskNum,
+ true,
+ Collections.singletonList(fakedTsFileResource),
+ 0));
+ FakedInnerSpaceCompactionTask task = (FakedInnerSpaceCompactionTask)
compactionTaskQueue.take();
+ Assert.assertEquals(
+ targetFileName,
task.getSelectedTsFileResourceList().get(0).getTsFile().getName());
+ fakedTsFileResource.getModFile().remove();
+ }
+
private static class FakedInnerSpaceCompactionTask extends
InnerSpaceCompactionTask {
public FakedInnerSpaceCompactionTask(
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionSelectorTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionSelectorTest.java
index 307283e71ca..b4dfa07b2d9 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionSelectorTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionSelectorTest.java
@@ -19,12 +19,15 @@
package org.apache.iotdb.db.engine.compaction.inner;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.AbstractCompactionTest;
import
org.apache.iotdb.db.engine.compaction.execute.performer.impl.FastCompactionPerformer;
import
org.apache.iotdb.db.engine.compaction.execute.task.InnerSpaceCompactionTask;
import org.apache.iotdb.db.engine.compaction.schedule.CompactionTaskManager;
import
org.apache.iotdb.db.engine.compaction.selector.impl.SizeTieredCompactionSelector;
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceStatus;
import org.apache.iotdb.db.exception.StorageEngineException;
@@ -41,6 +44,8 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
+import static
org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
+
public class InnerSpaceCompactionSelectorTest extends AbstractCompactionTest {
@Before
public void setUp()
@@ -623,4 +628,32 @@ public class InnerSpaceCompactionSelectorTest extends
AbstractCompactionTest {
Assert.fail();
}
}
+
+ @Test
+ public void testSelectWhenModsFileGreaterThan50M()
+ throws IOException, MetadataException, WriteProcessException {
+ createFiles(6, 2, 3, 50, 0, 10000, 50, 50, false, true);
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+
+ TsFileResource tsFileResource = seqResources.get(0);
+
+ ModificationFile modFile = tsFileResource.getModFile();
+
+ while (modFile.getSize() < 1024 * 1024 * 50) {
+ modFile.write(
+ new Deletion(
+ new PartialPath(COMPACTION_TEST_SG + PATH_SEPARATOR + "**"),
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ }
+
+ SizeTieredCompactionSelector selector =
+ new SizeTieredCompactionSelector("", "", 0, true, tsFileManager);
+ // copy candidate source file list
+ List<TsFileResource> resources =
tsFileManager.getOrCreateSequenceListByTimePartition(0);
+ List<List<TsFileResource>> taskResource =
selector.selectInnerSpaceTask(resources);
+ Assert.assertEquals(1, taskResource.size());
+ modFile.remove();
+ }
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/modification/ModificationFileTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/modification/ModificationFileTest.java
index fa17cbd4e30..47109c4f146 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/modification/ModificationFileTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/modification/ModificationFileTest.java
@@ -21,14 +21,19 @@ package org.apache.iotdb.db.engine.modification;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.constant.TestConstant;
+import
org.apache.iotdb.db.engine.compaction.execute.recover.CompactionRecoverManager;
+import org.junit.Assert;
import org.junit.Test;
import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
+import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class ModificationFileTest {
@@ -99,4 +104,181 @@ public class ModificationFileTest {
new File(tempFileName).delete();
}
}
+
+ // test if file size greater than 1M.
+ @Test
+ public void testCompact01() {
+ String tempFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact01.mods");
+ long time = 1000;
+ try (ModificationFile modificationFile = new
ModificationFile(tempFileName)) {
+ while (modificationFile.getSize() < 1024 * 1024) {
+ modificationFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "d1"}),
+ 1000,
+ Long.MIN_VALUE,
+ time += 5000));
+ }
+ modificationFile.compact();
+ List<Modification> modificationList = new
ArrayList<>(modificationFile.getModifications());
+ assertEquals(1, modificationList.size());
+
+ Deletion deletion = (Deletion) modificationList.get(0);
+ assertEquals(time, deletion.getEndTime());
+ assertEquals(Long.MIN_VALUE, deletion.getStartTime());
+ } catch (IOException e) {
+ fail(e.getMessage());
+ } finally {
+ new File(tempFileName).delete();
+ }
+ }
+
+ // test if file size less than 1M.
+ @Test
+ public void testCompact02() {
+ String tempFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact02.mods");
+ long time = 1000;
+ try (ModificationFile modificationFile = new
ModificationFile(tempFileName)) {
+ while (modificationFile.getSize() < 1024 * 100) {
+ modificationFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "d1"}),
+ 1000,
+ Long.MIN_VALUE,
+ time += 5000));
+ }
+ modificationFile.compact();
+ List<Modification> modificationList = new
ArrayList<>(modificationFile.getModifications());
+ assertTrue(modificationList.size() > 1);
+ } catch (IOException e) {
+ fail(e.getMessage());
+ } finally {
+ new File(tempFileName).delete();
+ }
+ }
+
+ // test if file size greater than 1M.
+ @Test
+ public void testCompact03() {
+ String tempFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact03.mods");
+ try (ModificationFile modificationFile = new
ModificationFile(tempFileName)) {
+ while (modificationFile.getSize() < 1024 * 1024) {
+ modificationFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "d1"}),
+ 1000,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ }
+ modificationFile.compact();
+ List<Modification> modificationList = new
ArrayList<>(modificationFile.getModifications());
+ assertEquals(1, modificationList.size());
+
+ Deletion deletion = (Deletion) modificationList.get(0);
+ assertEquals(Long.MAX_VALUE, deletion.getEndTime());
+ assertEquals(Long.MIN_VALUE, deletion.getStartTime());
+ } catch (IOException e) {
+ fail(e.getMessage());
+ } finally {
+ new File(tempFileName).delete();
+ }
+ }
+
+ @Test
+ public void testCompact04() {
+ String tempFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact04.mods");
+ try (ModificationFile modificationFile = new
ModificationFile(tempFileName)) {
+ long time = 0;
+ while (modificationFile.getSize() < 1024 * 1024) {
+ for (int i = 0; i < 5; i++) {
+ modificationFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "d1"}),
+ 1000,
+ Long.MIN_VALUE,
+ time += 5000));
+ modificationFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "*"}),
+ 1000,
+ Long.MIN_VALUE,
+ time += 5000));
+ }
+ }
+ modificationFile.compact();
+ List<Modification> modificationList = new
ArrayList<>(modificationFile.getModifications());
+ assertEquals(2, modificationList.size());
+ } catch (IOException e) {
+ fail(e.getMessage());
+ } finally {
+ new File(tempFileName).delete();
+ }
+ }
+
+ // test mods file and mods settle file both exists
+ @Test
+ public void testRecover01() {
+ String modsFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact01.mods");
+ String modsSettleFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact01.mods.settle");
+
+ try (ModificationFile modsFile = new ModificationFile(modsFileName);
+ ModificationFile modsSettleFile = new
ModificationFile(modsSettleFileName)) {
+
+ modsFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "d1"}),
+ 1000,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ modsSettleFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "d1"}),
+ 1000,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+
+ modsFile.close();
+ modsSettleFile.close();
+ new CompactionRecoverManager(null, null, null)
+ .recoverModSettleFile(new
File(TestConstant.BASE_OUTPUT_PATH).toPath());
+ Assert.assertTrue(modsFile.exists());
+ Assert.assertFalse(modsSettleFile.exists());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ Files.delete(new File(modsFileName).toPath());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ // test only mods settle file exists
+ @Test
+ public void testRecover02() {
+ String modsSettleFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact02.mods.settle");
+ String originModsFileName =
TestConstant.BASE_OUTPUT_PATH.concat("compact02.mods");
+ try (ModificationFile modsSettleFile = new
ModificationFile(modsSettleFileName)) {
+ modsSettleFile.write(
+ new Deletion(
+ new PartialPath(new String[] {"root", "sg", "d1"}),
+ 1000,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE));
+ modsSettleFile.close();
+ new CompactionRecoverManager(null, null, null)
+ .recoverModSettleFile(new
File(TestConstant.BASE_OUTPUT_PATH).toPath());
+ Assert.assertFalse(modsSettleFile.exists());
+ Assert.assertTrue(new File(originModsFileName).exists());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ Files.delete(new File(originModsFileName).toPath());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FakedTsFileResource.java
b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FakedTsFileResource.java
index d8535833372..c3f755d2790 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FakedTsFileResource.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/FakedTsFileResource.java
@@ -33,6 +33,7 @@ public class FakedTsFileResource extends TsFileResource {
private String fakeTsfileName;
public FakedTsFileResource(long tsFileSize, String name) {
+ super(new File(name));
this.timeIndex = new FileTimeIndex();
this.tsFileSize = tsFileSize;
fakeTsfileName = name;
diff --git
a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/common/TimeRangeTest.java
b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/common/TimeRangeTest.java
index e1a6d5271df..cd929d24479 100644
---
a/tsfile/src/test/java/org/apache/iotdb/tsfile/read/common/TimeRangeTest.java
+++
b/tsfile/src/test/java/org/apache/iotdb/tsfile/read/common/TimeRangeTest.java
@@ -448,4 +448,45 @@ public class TimeRangeTest {
Assert.assertTrue(new TimeRange(0, 3).compareTo(new TimeRange(1, 2)) < 0);
Assert.assertTrue(new TimeRange(5, 6).compareTo(new TimeRange(5, 6)) == 0);
}
+
+ @Test
+ /*
+ * test min is Long.MIN_VALUE
+ */
+ public void intersect8() {
+ TimeRange r1 = new TimeRange(Long.MIN_VALUE, 3);
+ r1.setLeftClose(false);
+ TimeRange r2 = new TimeRange(Long.MIN_VALUE, 5);
+ r2.setLeftClose(false);
+ assertTrue(r1.intersects(r2));
+ assertTrue(r2.intersects(r1));
+ }
+
+ @Test
+ /*
+ * test max is Long.MAX_VALUE
+ */
+ public void intersect9() {
+ TimeRange r1 = new TimeRange(1, Long.MAX_VALUE);
+ r1.setRightClose(false);
+ TimeRange r2 = new TimeRange(3, Long.MAX_VALUE);
+ r2.setRightClose(false);
+ assertTrue(r1.intersects(r2));
+ assertTrue(r2.intersects(r1));
+ }
+
+ @Test
+ /*
+ * test min is Long.MIN_VALUE and max is Long.MAX_VALUE
+ */
+ public void intersect10() {
+ TimeRange r1 = new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ r1.setLeftClose(false);
+ r1.setRightClose(false);
+ TimeRange r2 = new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ r2.setLeftClose(false);
+ r2.setRightClose(false);
+ assertTrue(r1.intersects(r2));
+ assertTrue(r2.intersects(r1));
+ }
}
