This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch rel/0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.11 by this push:
new 98b5503 cherry pick compaction modification to 0.11 (#2376)
98b5503 is described below
commit 98b550357cb6cf8926a6409512d0f541d5af7635
Author: zhanglingzhe0820 <[email protected]>
AuthorDate: Tue Dec 29 23:06:00 2020 +0800
cherry pick compaction modification to 0.11 (#2376)
---
docs/zh/SystemDesign/StorageEngine/Compaction.md | 10 +-
.../level/LevelCompactionTsFileManagement.java | 10 +-
.../no/NoCompactionTsFileManagement.java | 10 +-
.../compaction/utils/CompactionLogAnalyzer.java | 19 +-
.../engine/compaction/utils/CompactionLogger.java | 16 +-
.../engine/compaction/utils/CompactionUtils.java | 76 +-
.../db/engine/storagegroup/TsFileResource.java | 20 +-
.../java/org/apache/iotdb/db/utils/QueryUtils.java | 1 -
.../engine/compaction/LevelCompactionLogTest.java | 82 ++
.../compaction/LevelCompactionMergeTest.java | 142 +++
.../compaction/LevelCompactionRecoverTest.java | 507 +++++++++
.../compaction/LevelCompactionSelectorTest.java | 73 ++
.../db/engine/compaction/LevelCompactionTest.java | 208 ++++
.../LevelCompactionTsFileManagementTest.java | 119 ++
.../NoCompactionTsFileManagementTest.java | 126 +++
.../iotdb/db/integration/IoTDBCompactionIT.java | 155 ---
.../db/integration/IoTDBLevelCompactionIT.java | 1180 ++++++++++++++++++++
17 files changed, 2533 insertions(+), 221 deletions(-)
diff --git a/docs/zh/SystemDesign/StorageEngine/Compaction.md
b/docs/zh/SystemDesign/StorageEngine/Compaction.md
index 3898ff7..add0a3a 100644
--- a/docs/zh/SystemDesign/StorageEngine/Compaction.md
+++ b/docs/zh/SystemDesign/StorageEngine/Compaction.md
@@ -70,7 +70,9 @@
* 生成合并日志 .compaction.log
* 记录是层级合并
* 记录目标文件
- * 进行合并（记录日志 device - offset）
+ * 进行合并（每写完一个 device,记录日志 device - offset）
+ * 生成 .resource 文件
+ * writer endFile
* 记录完成合并
* 加写锁
* 从磁盘删掉待合并的文件，并从正式文件列表中移除
@@ -81,13 +83,15 @@
* 如果日志文件存在
* 如果是全局合并（把所有小文件合并到最后一层）
- * 如果合并没结束
+ * 如果合并没结束且目标文件还未封口
* 截断文件
* 继续全局合并
+ * 删除原文件及其对应列表
* 如果是层级合并
- * 如果合并没结束
+ * 如果合并没结束且目标文件还未封口
* 截断文件
* 继续层级合并
+ * 删除原文件及其对应列表
* 如果日志文件不存在
* 无需恢复
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
index 33e4e11..2f279d3 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
@@ -401,7 +401,7 @@ public class LevelCompactionTsFileManagement extends
TsFileManagement {
deleteLevelFilesInList(timePartition, sourceTsFileResources, level,
isSeq);
}
}
- } catch (IOException e) {
+ } catch (IOException | IllegalPathException e) {
logger.error("recover level tsfile management error ", e);
} finally {
if (logFile.exists()) {
@@ -418,16 +418,16 @@ public class LevelCompactionTsFileManagement extends
TsFileManagement {
if (isSeq) {
for (int level = 0; level <
sequenceTsFileResources.get(timePartition).size();
level++) {
- SortedSet<TsFileResource> currLevelMergeFile = sequenceTsFileResources
- .get(timePartition).get(level);
+ SortedSet<TsFileResource> currLevelMergeFile =
sequenceTsFileResources.get(timePartition)
+ .get(level);
deleteLevelFilesInDisk(currLevelMergeFile);
deleteLevelFilesInList(timePartition, currLevelMergeFile, level,
isSeq);
}
} else {
for (int level = 0; level <
unSequenceTsFileResources.get(timePartition).size();
level++) {
- SortedSet<TsFileResource> currLevelMergeFile = sequenceTsFileResources
- .get(timePartition).get(level);
+ SortedSet<TsFileResource> currLevelMergeFile =
sequenceTsFileResources.get(timePartition)
+ .get(level);
deleteLevelFilesInDisk(currLevelMergeFile);
deleteLevelFilesInList(timePartition, currLevelMergeFile, level,
isSeq);
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java
index d80a41a..1e130a0 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/no/NoCompactionTsFileManagement.java
@@ -34,9 +34,13 @@ public class NoCompactionTsFileManagement extends
TsFileManagement {
// includes sealed and unsealed sequence TsFiles
private TreeSet<TsFileResource> sequenceFileTreeSet = new TreeSet<>(
(o1, o2) -> {
- int rangeCompare =
Long.compare(Long.parseLong(o1.getTsFile().getParentFile().getName()),
- Long.parseLong(o2.getTsFile().getParentFile().getName()));
- return rangeCompare == 0 ? compareFileName(o1.getTsFile(),
o2.getTsFile()) : rangeCompare;
+ try {
+ int rangeCompare =
Long.compare(Long.parseLong(o1.getTsFile().getParentFile().getName()),
+ Long.parseLong(o2.getTsFile().getParentFile().getName()));
+ return rangeCompare == 0 ? compareFileName(o1.getTsFile(),
o2.getTsFile()) : rangeCompare;
+ } catch (NumberFormatException e) {
+ return compareFileName(o1.getTsFile(), o2.getTsFile());
+ }
});
// includes sealed and unsealed unSequence TsFiles
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogAnalyzer.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogAnalyzer.java
index ee32afd..9b2d8a6 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogAnalyzer.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogAnalyzer.java
@@ -20,7 +20,6 @@
package org.apache.iotdb.db.engine.compaction.utils;
import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.FULL_MERGE;
-import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.MERGE_FINISHED;
import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.SEQUENCE_NAME;
import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.SOURCE_NAME;
import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.TARGET_NAME;
@@ -37,10 +36,9 @@ import java.util.Set;
public class CompactionLogAnalyzer {
- public static final String STR_DEVICE_OFFSET_SEPERATOR = " ";
+ public static final String STR_DEVICE_OFFSET_SEPARATOR = " ";
private File logFile;
- private boolean isMergeFinished = false;
private Set<String> deviceSet = new HashSet<>();
private long offset = 0;
private List<String> sourceFiles = new ArrayList<>();
@@ -69,9 +67,6 @@ public class CompactionLogAnalyzer {
currLine = bufferedReader.readLine();
targetFile = currLine;
break;
- case MERGE_FINISHED:
- isMergeFinished = true;
- break;
case FULL_MERGE:
fullMerge = true;
break;
@@ -82,21 +77,15 @@ public class CompactionLogAnalyzer {
isSeq = false;
break;
default:
- if (currLine.contains(STR_DEVICE_OFFSET_SEPERATOR)) {
- String[] resultList =
currLine.split(STR_DEVICE_OFFSET_SEPERATOR);
- deviceSet.add(resultList[0]);
- offset = Long.parseLong(resultList[1]);
- }
+ int separatorIndex =
currLine.lastIndexOf(STR_DEVICE_OFFSET_SEPARATOR);
+ deviceSet.add(currLine.substring(0, separatorIndex));
+ offset = Long.parseLong(currLine.substring(separatorIndex + 1));
break;
}
}
}
}
- public boolean isMergeFinished() {
- return isMergeFinished;
- }
-
public Set<String> getDeviceSet() {
return deviceSet;
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogger.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogger.java
index 692ef24..1145d98 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogger.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionLogger.java
@@ -19,8 +19,6 @@
package org.apache.iotdb.db.engine.compaction.utils;
-import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogAnalyzer.STR_DEVICE_OFFSET_SEPERATOR;
-
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
@@ -50,7 +48,7 @@ public class CompactionLogger {
}
public void logDevice(String device, long offset) throws IOException {
- logStream.write(device + STR_DEVICE_OFFSET_SEPERATOR + offset);
+ logStream.write(device + CompactionLogAnalyzer.STR_DEVICE_OFFSET_SEPARATOR
+ offset);
logStream.newLine();
logStream.flush();
}
@@ -63,18 +61,6 @@ public class CompactionLogger {
logStream.flush();
}
- public void logMergeFinish() throws IOException {
- logStream.write(MERGE_FINISHED);
- logStream.newLine();
- logStream.flush();
- }
-
- public void logFullMerge() throws IOException {
- logStream.write(FULL_MERGE);
- logStream.newLine();
- logStream.flush();
- }
-
public void logSequence(boolean isSeq) throws IOException {
if (isSeq) {
logStream.write(SEQUENCE_NAME);
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
index 8152cc8..bd31319 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/CompactionUtils.java
@@ -20,14 +20,15 @@
package org.apache.iotdb.db.engine.compaction.utils;
import static org.apache.iotdb.db.utils.MergeUtils.writeTVPair;
+import static org.apache.iotdb.db.utils.QueryUtils.modifyChunkMetaData;
import com.google.common.util.concurrent.RateLimiter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.LinkedHashMap;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -35,7 +36,10 @@ import java.util.Set;
import java.util.TreeMap;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
+import org.apache.iotdb.db.engine.modification.Modification;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.service.IoTDB;
@@ -66,13 +70,18 @@ public class CompactionUtils {
}
private static Pair<ChunkMetadata, Chunk> readByAppendMerge(
- Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap)
throws IOException {
+ Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap,
+ Map<String, List<Modification>> modificationCache, PartialPath
seriesPath)
+ throws IOException {
ChunkMetadata newChunkMetadata = null;
Chunk newChunk = null;
for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry :
readerChunkMetadataMap
.entrySet()) {
- for (ChunkMetadata chunkMetadata : entry.getValue()) {
- Chunk chunk = entry.getKey().readMemChunk(chunkMetadata);
+ TsFileSequenceReader reader = entry.getKey();
+ List<ChunkMetadata> chunkMetadataList = entry.getValue();
+ modifyChunkMetaDataWithCache(reader, chunkMetadataList,
modificationCache, seriesPath);
+ for (ChunkMetadata chunkMetadata : chunkMetadataList) {
+ Chunk chunk = reader.readMemChunk(chunkMetadata);
if (newChunkMetadata == null) {
newChunkMetadata = chunkMetadata;
newChunk = chunk;
@@ -87,12 +96,14 @@ public class CompactionUtils {
private static long readByDeserializeMerge(
Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap,
long maxVersion,
- Map<Long, TimeValuePair> timeValuePairMap)
+ Map<Long, TimeValuePair> timeValuePairMap, Map<String,
List<Modification>> modificationCache,
+ PartialPath seriesPath)
throws IOException {
for (Entry<TsFileSequenceReader, List<ChunkMetadata>> entry :
readerChunkMetadataMap
.entrySet()) {
TsFileSequenceReader reader = entry.getKey();
List<ChunkMetadata> chunkMetadataList = entry.getValue();
+ modifyChunkMetaDataWithCache(reader, chunkMetadataList,
modificationCache, seriesPath);
for (ChunkMetadata chunkMetadata : chunkMetadataList) {
maxVersion = Math.max(chunkMetadata.getVersion(), maxVersion);
IChunkReader chunkReader = new ChunkReaderByTimestamp(
@@ -112,9 +123,12 @@ public class CompactionUtils {
private static long writeByAppendMerge(long maxVersion, String device,
RateLimiter compactionWriteRateLimiter,
- Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadatasMap,
- TsFileResource targetResource, RestorableTsFileIOWriter writer) throws
IOException {
- Pair<ChunkMetadata, Chunk> chunkPair =
readByAppendMerge(readerChunkMetadatasMap);
+ Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>> entry,
+ TsFileResource targetResource, RestorableTsFileIOWriter writer,
+ Map<String, List<Modification>> modificationCache)
+ throws IOException, IllegalPathException {
+ Pair<ChunkMetadata, Chunk> chunkPair = readByAppendMerge(entry.getValue(),
+ modificationCache, new PartialPath(device, entry.getKey()));
ChunkMetadata newChunkMetadata = chunkPair.left;
Chunk newChunk = chunkPair.right;
if (newChunkMetadata != null && newChunk != null) {
@@ -132,16 +146,20 @@ public class CompactionUtils {
private static long writeByDeserializeMerge(long maxVersion, String device,
RateLimiter compactionRateLimiter,
Entry<String, Map<TsFileSequenceReader, List<ChunkMetadata>>> entry,
- TsFileResource targetResource, RestorableTsFileIOWriter writer) throws
IOException {
+ TsFileResource targetResource, RestorableTsFileIOWriter writer,
+ Map<String, List<Modification>> modificationCache) throws IOException,
IllegalPathException {
Map<Long, TimeValuePair> timeValuePairMap = new TreeMap<>();
- maxVersion = readByDeserializeMerge(entry.getValue(), maxVersion,
timeValuePairMap);
- Iterator<List<ChunkMetadata>> chunkMetadataListIterator =
entry.getValue().values()
- .iterator();
- if (!chunkMetadataListIterator.hasNext()) {
- return maxVersion;
+ Map<TsFileSequenceReader, List<ChunkMetadata>> readerChunkMetadataMap =
entry.getValue();
+ maxVersion = readByDeserializeMerge(readerChunkMetadataMap, maxVersion,
timeValuePairMap,
+ modificationCache, new PartialPath(device, entry.getKey()));
+ boolean isChunkMetadataEmpty = true;
+ for (List<ChunkMetadata> chunkMetadataList :
readerChunkMetadataMap.values()) {
+ if (!chunkMetadataList.isEmpty()) {
+ isChunkMetadataEmpty = false;
+ break;
+ }
}
- List<ChunkMetadata> chunkMetadataList = chunkMetadataListIterator.next();
- if (chunkMetadataList.isEmpty()) {
+ if (isChunkMetadataEmpty) {
return maxVersion;
}
IChunkWriter chunkWriter;
@@ -192,9 +210,10 @@ public class CompactionUtils {
public static void merge(TsFileResource targetResource,
List<TsFileResource> tsFileResources, String storageGroup,
CompactionLogger compactionLogger,
- Set<String> devices, boolean sequence) throws IOException {
+ Set<String> devices, boolean sequence) throws IOException,
IllegalPathException {
RestorableTsFileIOWriter writer = new
RestorableTsFileIOWriter(targetResource.getTsFile());
Map<String, TsFileSequenceReader> tsFileSequenceReaderMap = new
HashMap<>();
+ Map<String, List<Modification>> modificationCache = new HashMap<>();
RateLimiter compactionWriteRateLimiter =
MergeManager.getINSTANCE().getMergeWriteRateLimiter();
Set<String> tsFileDevicesMap = getTsFileDevicesSet(tsFileResources,
tsFileSequenceReaderMap,
storageGroup);
@@ -241,7 +260,7 @@ public class CompactionUtils {
.entrySet()) {
maxVersion = writeByDeserializeMerge(maxVersion, device,
compactionWriteRateLimiter,
entry,
- targetResource, writer);
+ targetResource, writer, modificationCache);
}
writer.endChunkGroup();
writer.writeVersion(maxVersion);
@@ -263,13 +282,13 @@ public class CompactionUtils {
logger.debug("{} [Compaction] page enough large, use append
merge", storageGroup);
// append page in chunks, so we do not have to deserialize a chunk
maxVersion = writeByAppendMerge(maxVersion, device,
compactionWriteRateLimiter,
- readerChunkMetadatasMap, targetResource, writer);
+ entry, targetResource, writer, modificationCache);
} else {
logger
.debug("{} [Compaction] page too small, use deserialize
merge", storageGroup);
// we have to deserialize chunks to merge pages
maxVersion = writeByDeserializeMerge(maxVersion, device,
compactionWriteRateLimiter,
- entry, targetResource, writer);
+ entry, targetResource, writer, modificationCache);
}
}
writer.endChunkGroup();
@@ -312,4 +331,21 @@ public class CompactionUtils {
}
});
}
+
+ private static void modifyChunkMetaDataWithCache(TsFileSequenceReader reader,
+ List<ChunkMetadata> chunkMetadataList, Map<String, List<Modification>>
modificationCache,
+ PartialPath seriesPath) {
+ List<Modification> modifications =
+ modificationCache.computeIfAbsent(reader.getFileName(),
+ fileName -> new LinkedList<>(
+ new ModificationFile(fileName + ModificationFile.FILE_SUFFIX)
+ .getModifications()));
+ List<Modification> seriesModifications = new LinkedList<>();
+ for (Modification modification : modifications) {
+ if (modification.getPath().matchFullPath(seriesPath)) {
+ seriesModifications.add(modification);
+ }
+ }
+ modifyChunkMetaData(chunkMetadataList, seriesModifications);
+ }
}
\ No newline at end of file
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index eef20dd..f136732 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -722,11 +722,15 @@ public class TsFileResource {
* make sure Either the deviceToIndex is not empty Or the path contains a
partition folder
*/
public long getTimePartition() {
- if (deviceToIndex != null && !deviceToIndex.isEmpty()) {
- return
StorageEngine.getTimePartition(startTimes[deviceToIndex.values().iterator().next()]);
+ try {
+ if (deviceToIndex != null && !deviceToIndex.isEmpty()) {
+ return
StorageEngine.getTimePartition(startTimes[deviceToIndex.values().iterator().next()]);
+ }
+ String[] splits = FilePathUtils.splitTsFilePath(this);
+ return Long.parseLong(splits[splits.length - 2]);
+ } catch (NumberFormatException e) {
+ return 0;
}
- String[] splits = FilePathUtils.splitTsFilePath(this);
- return Long.parseLong(splits[splits.length - 2]);
}
/**
@@ -876,6 +880,14 @@ public class TsFileResource {
}
}
+ /**
+ * For merge, the index range of the new file should be the union of all
files' in this merge.
+ */
+ public void updatePlanIndexes(TsFileResource another) {
+ maxPlanIndex = Math.max(maxPlanIndex, another.maxPlanIndex);
+ minPlanIndex = Math.min(minPlanIndex, another.minPlanIndex);
+ }
+
public boolean isPlanIndexOverlap(TsFileResource another) {
return another.maxPlanIndex >= this.minPlanIndex &&
another.minPlanIndex <= this.maxPlanIndex;
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java
b/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java
index 7e90db2..6d38744 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java
@@ -45,7 +45,6 @@ public class QueryUtils {
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
public static void modifyChunkMetaData(List<ChunkMetadata> chunkMetaData,
List<Modification> modifications) {
- int modIndex = 0;
for (int metaIndex = 0; metaIndex < chunkMetaData.size(); metaIndex++) {
ChunkMetadata metaData = chunkMetaData.get(metaIndex);
for (Modification modification : modifications) {
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionLogTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionLogTest.java
new file mode 100644
index 0000000..a6d1647
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionLogTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.COMPACTION_LOG_NAME;
+import static org.junit.Assert.assertFalse;
+
+import java.io.File;
+import java.io.IOException;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.constant.TestConstant;
+import
org.apache.iotdb.db.engine.compaction.TsFileManagement.CompactionMergeTask;
+import
org.apache.iotdb.db.engine.compaction.level.LevelCompactionTsFileManagement;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class LevelCompactionLogTest extends LevelCompactionTest {
+
+ File tempSGDir;
+ boolean compactionMergeWorking = false;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
+ super.setUp();
+ tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
+ tempSGDir.mkdirs();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ FileUtils.deleteDirectory(tempSGDir);
+ }
+
+ @Test
+ public void testCompactionLog() {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ levelCompactionTsFileManagement.forkCurrentFileList(0);
+ CompactionMergeTask compactionMergeTask =
levelCompactionTsFileManagement.new CompactionMergeTask(
+ this::closeCompactionMergeCallBack, 0);
+ compactionMergeWorking = true;
+ compactionMergeTask.run();
+ while (compactionMergeWorking) {
+ //wait
+ }
+ File logFile = FSFactoryProducer.getFSFactory()
+ .getFile(tempSGDir.getPath(), COMPACTION_TEST_SG +
COMPACTION_LOG_NAME);
+ assertFalse(logFile.exists());
+ }
+
+ /**
+ * close compaction merge callback, to release some locks
+ */
+ private void closeCompactionMergeCallBack() {
+ this.compactionMergeWorking = false;
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionMergeTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionMergeTest.java
new file mode 100644
index 0000000..8970a4c
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionMergeTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.constant.TestConstant;
+import
org.apache.iotdb.db.engine.compaction.TsFileManagement.CompactionMergeTask;
+import
org.apache.iotdb.db.engine.compaction.level.LevelCompactionTsFileManagement;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.reader.IBatchReader;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class LevelCompactionMergeTest extends LevelCompactionTest {
+
+ File tempSGDir;
+ boolean compactionMergeWorking = false;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
+ super.setUp();
+ tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
+ tempSGDir.mkdirs();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ FileUtils.deleteDirectory(tempSGDir);
+ }
+
+ /**
+ * just compaction once
+ */
+ @Test
+ public void testCompactionMergeOnce() throws IllegalPathException,
IOException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ levelCompactionTsFileManagement.forkCurrentFileList(0);
+ CompactionMergeTask compactionMergeTask =
levelCompactionTsFileManagement.new CompactionMergeTask(
+ this::closeCompactionMergeCallBack, 0);
+ compactionMergeWorking = true;
+ compactionMergeTask.run();
+ while (compactionMergeWorking) {
+ //wait
+ }
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ }
+ }
+ }
+
+ /**
+ * just compaction stable list
+ */
+ @Test
+ public void testCompactionMergeStableList() throws IllegalPathException,
IOException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(2);
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ levelCompactionTsFileManagement.forkCurrentFileList(0);
+ CompactionMergeTask compactionMergeTask =
levelCompactionTsFileManagement.new CompactionMergeTask(
+ this::closeCompactionMergeCallBack, 0);
+ compactionMergeWorking = true;
+ compactionMergeTask.run();
+ while (compactionMergeWorking) {
+ //wait
+ }
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null,
+ true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ count++;
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ }
+ }
+ assertEquals(500, count);
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * close compaction merge callback, to release some locks
+ */
+ private void closeCompactionMergeCallBack() {
+ this.compactionMergeWorking = false;
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRecoverTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRecoverTest.java
new file mode 100644
index 0000000..4ee7880
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionRecoverTest.java
@@ -0,0 +1,507 @@
+/*
+ * 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;
+
+import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.COMPACTION_LOG_NAME;
+import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.SOURCE_NAME;
+import static
org.apache.iotdb.db.engine.compaction.utils.CompactionLogger.TARGET_NAME;
+import static org.junit.Assert.assertEquals;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.constant.TestConstant;
+import
org.apache.iotdb.db.engine.compaction.level.LevelCompactionTsFileManagement;
+import org.apache.iotdb.db.engine.compaction.utils.CompactionLogger;
+import org.apache.iotdb.db.engine.compaction.utils.CompactionUtils;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.IllegalPathException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.query.context.QueryContext;
+import org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.common.BatchData;
+import org.apache.iotdb.tsfile.read.reader.IBatchReader;
+import org.apache.iotdb.tsfile.write.writer.TsFileOutput;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class LevelCompactionRecoverTest extends LevelCompactionTest {
+
+ File tempSGDir;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
+ super.setUp();
+ tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
+ tempSGDir.mkdirs();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ FileUtils.deleteDirectory(tempSGDir);
+ }
+
+ /**
+ * compaction recover merge finished
+ */
+ @Test
+ public void testCompactionMergeRecoverMergeFinished() throws IOException,
IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.logSequence(true);
+ TsFileResource targetTsFileResource = new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 0 + IoTDBConstant.FILE_NAME_SEPARATOR + 0 +
IoTDBConstant.FILE_NAME_SEPARATOR + 1
+ + ".tsfile")));
+ compactionLogger.logFile(TARGET_NAME, targetTsFileResource.getTsFile());
+ CompactionUtils.merge(targetTsFileResource, new
ArrayList<>(seqResources.subList(0, 3)),
+ COMPACTION_TEST_SG, compactionLogger, new HashSet<>(), true);
+ compactionLogger.close();
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
+ levelCompactionTsFileManagement.recover();
+ context = new QueryContext();
+ path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+
+ /**
+ * compaction recover merge finished, delete one offset
+ */
+ @Test
+ public void testCompactionMergeRecoverMergeFinishedAndDeleteOneOffset()
+ throws IOException, IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.logSequence(true);
+ TsFileResource targetTsFileResource = new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 0 + IoTDBConstant.FILE_NAME_SEPARATOR + 0 +
IoTDBConstant.FILE_NAME_SEPARATOR + 1
+ + ".tsfile")));
+ compactionLogger.logFile(TARGET_NAME, targetTsFileResource.getTsFile());
+ CompactionUtils.merge(targetTsFileResource, new
ArrayList<>(seqResources.subList(0, 3)),
+ COMPACTION_TEST_SG, compactionLogger, new HashSet<>(), true);
+ compactionLogger.close();
+
+ BufferedReader logReader = new BufferedReader(
+ new FileReader(SystemFileFactory.INSTANCE.getFile(tempSGDir.getPath(),
+ COMPACTION_TEST_SG + COMPACTION_LOG_NAME)));
+ List<String> logs = new ArrayList<>();
+ String line;
+ while ((line = logReader.readLine()) != null) {
+ logs.add(line);
+ }
+ logReader.close();
+ BufferedWriter logStream = new BufferedWriter(
+ new FileWriter(SystemFileFactory.INSTANCE.getFile(tempSGDir.getPath(),
+ COMPACTION_TEST_SG + COMPACTION_LOG_NAME), false));
+ for (int i = 0; i < logs.size() - 1; i++) {
+ logStream.write(logs.get(i));
+ logStream.newLine();
+ }
+ logStream.close();
+
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
+ levelCompactionTsFileManagement.recover();
+ context = new QueryContext();
+ path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+
+ /**
+ * compaction recover merge finished, delete one device - offset
+ */
+ @Test
+ public void
testCompactionMergeRecoverMergeFinishedAndDeleteOneDeviceWithOffset()
+ throws IOException, IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.logSequence(true);
+ TsFileResource targetTsFileResource = new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 0 + IoTDBConstant.FILE_NAME_SEPARATOR + 0 +
IoTDBConstant.FILE_NAME_SEPARATOR + 1
+ + ".tsfile")));
+ compactionLogger.logFile(TARGET_NAME, targetTsFileResource.getTsFile());
+ CompactionUtils.merge(targetTsFileResource, new
ArrayList<>(seqResources.subList(0, 3)),
+ COMPACTION_TEST_SG, compactionLogger, new HashSet<>(), true);
+ compactionLogger.close();
+
+ BufferedReader logReader = new BufferedReader(
+ new FileReader(SystemFileFactory.INSTANCE.getFile(tempSGDir.getPath(),
+ COMPACTION_TEST_SG + COMPACTION_LOG_NAME)));
+ List<String> logs = new ArrayList<>();
+ String line;
+ while ((line = logReader.readLine()) != null) {
+ logs.add(line);
+ }
+ logReader.close();
+ BufferedWriter logStream = new BufferedWriter(
+ new FileWriter(SystemFileFactory.INSTANCE.getFile(tempSGDir.getPath(),
+ COMPACTION_TEST_SG + COMPACTION_LOG_NAME), false));
+ for (int i = 0; i < logs.size() - 1; i++) {
+ logStream.write(logs.get(i));
+ logStream.newLine();
+ }
+ logStream.close();
+
+ TsFileOutput out = FSFactoryProducer.getFileOutputFactory()
+ .getTsFileOutput(targetTsFileResource.getTsFile().getPath(), true);
+ out.truncate(Long.parseLong(logs.get(logs.size() - 1).split(" ")[1]) - 1);
+ out.close();
+
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
+ levelCompactionTsFileManagement.recover();
+ context = new QueryContext();
+ path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+
+ /**
+ * compaction recover merge finished,unseq
+ */
+ @Test
+ public void testCompactionMergeRecoverMergeFinishedUnseq()
+ throws IOException, IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(seqResources, false);
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.logSequence(false);
+ TsFileResource targetTsFileResource = new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 0 + IoTDBConstant.FILE_NAME_SEPARATOR + 0 +
IoTDBConstant.FILE_NAME_SEPARATOR + 1
+ + ".tsfile")));
+ compactionLogger.logFile(TARGET_NAME, targetTsFileResource.getTsFile());
+ CompactionUtils.merge(targetTsFileResource, new
ArrayList<>(seqResources.subList(0, 3)),
+ COMPACTION_TEST_SG, compactionLogger, new HashSet<>(), false);
+ compactionLogger.close();
+ levelCompactionTsFileManagement.add(targetTsFileResource, false);
+ levelCompactionTsFileManagement.recover();
+ context = new QueryContext();
+ path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(false), new
ArrayList<>(), null, null, true);
+ count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+
+ /**
+ * compaction recover merge start just log source file
+ */
+ @Test
+ public void testCompactionMergeRecoverMergeStartSourceLog()
+ throws IOException, IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.close();
+ levelCompactionTsFileManagement.recover();
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+
+ /**
+ * compaction recover merge start just log source file and sequence flag
+ */
+ @Test
+ public void testCompactionMergeRecoverMergeStartSequenceLog()
+ throws IOException, IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.logSequence(true);
+ compactionLogger.close();
+ levelCompactionTsFileManagement.recover();
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+
+ /**
+ * compaction recover merge start target file logged
+ */
+ @Test
+ public void testCompactionMergeRecoverMergeStart() throws IOException,
IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.logSequence(true);
+ TsFileResource targetTsFileResource = new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 0 + IoTDBConstant.FILE_NAME_SEPARATOR + 0 +
IoTDBConstant.FILE_NAME_SEPARATOR + 1
+ + ".tsfile")));
+ compactionLogger.logFile(TARGET_NAME, targetTsFileResource.getTsFile());
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
+ compactionLogger.close();
+ levelCompactionTsFileManagement.recover();
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+
+ /**
+ * compaction recover merge finished but no finish log
+ */
+ @Test
+ public void testCompactionMergeRecoverMergeFinishedNoLog()
+ throws IOException, IllegalPathException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ CompactionLogger compactionLogger = new
CompactionLogger(tempSGDir.getPath(),
+ COMPACTION_TEST_SG);
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(0).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(1).getTsFile());
+ compactionLogger.logFile(SOURCE_NAME, seqResources.get(2).getTsFile());
+ compactionLogger.logSequence(true);
+ TsFileResource targetTsFileResource = new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 0 + IoTDBConstant.FILE_NAME_SEPARATOR + 0 +
IoTDBConstant.FILE_NAME_SEPARATOR + 1
+ + ".tsfile")));
+ compactionLogger.logFile(TARGET_NAME, targetTsFileResource.getTsFile());
+ CompactionUtils.merge(targetTsFileResource, new
ArrayList<>(seqResources.subList(0, 3)),
+ COMPACTION_TEST_SG, compactionLogger, new HashSet<>(), true);
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
+ compactionLogger.close();
+ levelCompactionTsFileManagement.recover();
+ QueryContext context = new QueryContext();
+ PartialPath path = new PartialPath(
+ deviceIds[0] + TsFileConstant.PATH_SEPARATOR +
measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader = new SeriesRawDataBatchReader(path,
measurementSchemas[0].getType(),
+ context,
+ levelCompactionTsFileManagement.getTsFileList(true), new
ArrayList<>(), null, null, true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i),
batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ assertEquals(500, count);
+ }
+}
+
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionSelectorTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionSelectorTest.java
new file mode 100644
index 0000000..e272ad2
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionSelectorTest.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;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.constant.TestConstant;
+import
org.apache.iotdb.db.engine.compaction.level.LevelCompactionTsFileManagement;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class LevelCompactionSelectorTest extends LevelCompactionTest {
+
+ File tempSGDir;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
+ super.setUp();
+ tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
+ tempSGDir.mkdirs();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ FileUtils.deleteDirectory(tempSGDir);
+ }
+
+ /**
+ * just compaction once
+ */
+ @Test
+ public void testCompactionSelector() throws NoSuchFieldException,
IllegalAccessException {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ levelCompactionTsFileManagement.addAll(seqResources, true);
+ levelCompactionTsFileManagement.addAll(unseqResources, false);
+ levelCompactionTsFileManagement.forkCurrentFileList(0);
+ Field fieldForkedSequenceTsFileResources =
LevelCompactionTsFileManagement.class
+ .getDeclaredField("forkedSequenceTsFileResources");
+ fieldForkedSequenceTsFileResources.setAccessible(true);
+ List<TsFileResource> forkedSequenceTsFileResources =
(List<TsFileResource>) fieldForkedSequenceTsFileResources
+ .get(levelCompactionTsFileManagement);
+ assertEquals(2, forkedSequenceTsFileResources.size());
+ }
+}
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTest.java
new file mode 100644
index 0000000..7fbc509
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTest.java
@@ -0,0 +1,208 @@
+/*
+ * 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;
+
+import static org.apache.iotdb.db.conf.IoTDBConstant.PATH_SEPARATOR;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.constant.TestConstant;
+import org.apache.iotdb.db.engine.cache.ChunkCache;
+import org.apache.iotdb.db.engine.cache.ChunkMetadataCache;
+import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
+import org.apache.iotdb.db.engine.merge.manage.MergeManager;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.metadata.PartialPath;
+import org.apache.iotdb.db.query.control.FileReaderManager;
+import org.apache.iotdb.db.service.IoTDB;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.common.Path;
+import org.apache.iotdb.tsfile.write.TsFileWriter;
+import org.apache.iotdb.tsfile.write.record.TSRecord;
+import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
+import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.Before;
+
+abstract class LevelCompactionTest {
+
+ static final String COMPACTION_TEST_SG = "root.compactionTest";
+
+ int seqFileNum = 6;
+ int unseqFileNum = 0;
+ int measurementNum = 10;
+ int deviceNum = 10;
+ long ptNum = 100;
+ long flushInterval = 20;
+ TSEncoding encoding = TSEncoding.PLAIN;
+
+ String[] deviceIds;
+ MeasurementSchema[] measurementSchemas;
+
+ List<TsFileResource> seqResources = new ArrayList<>();
+ List<TsFileResource> unseqResources = new ArrayList<>();
+
+ private int prevMergeChunkThreshold;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
+ IoTDB.metaManager.init();
+ prevMergeChunkThreshold =
+
IoTDBDescriptor.getInstance().getConfig().getMergeChunkPointNumberThreshold();
+
IoTDBDescriptor.getInstance().getConfig().setMergeChunkPointNumberThreshold(-1);
+ prepareSeries();
+ prepareFiles(seqFileNum, unseqFileNum);
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ removeFiles();
+ seqResources.clear();
+ unseqResources.clear();
+ IoTDBDescriptor.getInstance().getConfig()
+ .setMergeChunkPointNumberThreshold(prevMergeChunkThreshold);
+ ChunkCache.getInstance().clear();
+ ChunkMetadataCache.getInstance().clear();
+ TimeSeriesMetadataCache.getInstance().clear();
+ IoTDB.metaManager.clear();
+ EnvironmentUtils.cleanAllDir();
+ }
+
+ private void prepareSeries() throws MetadataException {
+ measurementSchemas = new MeasurementSchema[measurementNum];
+ for (int i = 0; i < measurementNum; i++) {
+ measurementSchemas[i] = new MeasurementSchema("sensor" + i,
TSDataType.DOUBLE,
+ encoding, CompressionType.UNCOMPRESSED);
+ }
+ deviceIds = new String[deviceNum];
+ for (int i = 0; i < deviceNum; i++) {
+ deviceIds[i] = COMPACTION_TEST_SG + PATH_SEPARATOR + "device" + i;
+ }
+ IoTDB.metaManager.setStorageGroup(new PartialPath(COMPACTION_TEST_SG));
+ for (String device : deviceIds) {
+ for (MeasurementSchema measurementSchema : measurementSchemas) {
+ PartialPath devicePath = new PartialPath(device);
+ IoTDB.metaManager.createTimeseries(
+ devicePath.concatNode(measurementSchema.getMeasurementId()),
measurementSchema
+ .getType(), measurementSchema.getEncodingType(),
measurementSchema.getCompressor(),
+ Collections.emptyMap());
+ }
+ }
+ }
+
+ void prepareFiles(int seqFileNum, int unseqFileNum)
+ throws IOException, WriteProcessException {
+ for (int i = 0; i < seqFileNum; i++) {
+ File file = new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ i + IoTDBConstant.FILE_NAME_SEPARATOR + i +
IoTDBConstant.FILE_NAME_SEPARATOR + 0
+ + ".tsfile"));
+ TsFileResource tsFileResource = new TsFileResource(file);
+ tsFileResource.setClosed(true);
+ tsFileResource.updatePlanIndexes((long) i);
+ seqResources.add(tsFileResource);
+ prepareFile(tsFileResource, i * ptNum, ptNum, 0);
+ }
+ for (int i = 0; i < unseqFileNum; i++) {
+ File file = new File(TestConstant.BASE_OUTPUT_PATH.concat(
+ (10000 + i) + IoTDBConstant.FILE_NAME_SEPARATOR + (10000 + i)
+ + IoTDBConstant.FILE_NAME_SEPARATOR + 0 + ".tsfile"));
+ TsFileResource tsFileResource = new TsFileResource(file);
+ tsFileResource.setClosed(true);
+ tsFileResource.updatePlanIndexes((long) (i + seqFileNum));
+ unseqResources.add(tsFileResource);
+ prepareFile(tsFileResource, i * ptNum, ptNum * (i + 1) / unseqFileNum,
10000);
+ }
+
+ File file = new File(TestConstant.BASE_OUTPUT_PATH.concat(
+ unseqFileNum + IoTDBConstant.FILE_NAME_SEPARATOR + unseqFileNum
+ + IoTDBConstant.FILE_NAME_SEPARATOR + 0 + ".tsfile"));
+ TsFileResource tsFileResource = new TsFileResource(file);
+ tsFileResource.setClosed(true);
+ tsFileResource.updatePlanIndexes((long) (seqFileNum + unseqFileNum));
+ unseqResources.add(tsFileResource);
+ prepareFile(tsFileResource, 0, ptNum * unseqFileNum, 20000);
+ }
+
+ private void removeFiles() throws IOException {
+ for (TsFileResource tsFileResource : seqResources) {
+ if (tsFileResource.getTsFile().exists()) {
+ tsFileResource.remove();
+ }
+ }
+ for (TsFileResource tsFileResource : unseqResources) {
+ if (tsFileResource.getTsFile().exists()) {
+ tsFileResource.remove();
+ }
+ }
+ File[] files =
FSFactoryProducer.getFSFactory().listFilesBySuffix("target", ".tsfile");
+ for (File file : files) {
+ file.delete();
+ }
+ File[] resourceFiles = FSFactoryProducer.getFSFactory()
+ .listFilesBySuffix("target", ".resource");
+ for (File resourceFile : resourceFiles) {
+ resourceFile.delete();
+ }
+ FileReaderManager.getInstance().closeAndRemoveAllOpenedReaders();
+ FileReaderManager.getInstance().stop();
+ }
+
+ void prepareFile(TsFileResource tsFileResource, long timeOffset, long ptNum,
+ long valueOffset)
+ throws IOException, WriteProcessException {
+ TsFileWriter fileWriter = new TsFileWriter(tsFileResource.getTsFile());
+ for (String deviceId : deviceIds) {
+ for (MeasurementSchema measurementSchema : measurementSchemas) {
+ fileWriter.registerTimeseries(
+ new Path(deviceId, measurementSchema.getMeasurementId()),
measurementSchema);
+ }
+ }
+ for (long i = timeOffset; i < timeOffset + ptNum; i++) {
+ for (int j = 0; j < deviceNum; j++) {
+ TSRecord record = new TSRecord(i, deviceIds[j]);
+ for (int k = 0; k < measurementNum; k++) {
+
record.addTuple(DataPoint.getDataPoint(measurementSchemas[k].getType(),
+ measurementSchemas[k].getMeasurementId(), String.valueOf(i +
valueOffset)));
+ }
+ fileWriter.write(record);
+ tsFileResource.updateStartTime(deviceIds[j], i);
+ tsFileResource.updateEndTime(deviceIds[j], i);
+ }
+ if ((i + 1) % flushInterval == 0) {
+ fileWriter.flushAllChunkGroups();
+ }
+ }
+ fileWriter.close();
+ }
+
+}
\ No newline at end of file
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTsFileManagementTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTsFileManagementTest.java
new file mode 100644
index 0000000..6b51613
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/LevelCompactionTsFileManagementTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Iterator;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.constant.TestConstant;
+import
org.apache.iotdb.db.engine.compaction.level.LevelCompactionTsFileManagement;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class LevelCompactionTsFileManagementTest extends LevelCompactionTest {
+
+ File tempSGDir;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
+ super.setUp();
+ tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
+ tempSGDir.mkdirs();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ FileUtils.deleteDirectory(tempSGDir);
+ }
+
+ /**
+ * just compaction once
+ */
+ @Test
+ public void testAddRemoveAndIterator() {
+ LevelCompactionTsFileManagement levelCompactionTsFileManagement = new
LevelCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ for (TsFileResource tsFileResource : seqResources) {
+ levelCompactionTsFileManagement.add(tsFileResource, true);
+ }
+ levelCompactionTsFileManagement.addAll(seqResources, false);
+ assertEquals(6,
levelCompactionTsFileManagement.getTsFileList(true).size());
+ assertEquals(6,
levelCompactionTsFileManagement.getTsFileList(false).size());
+ assertEquals(6, levelCompactionTsFileManagement.size(true));
+ assertEquals(6, levelCompactionTsFileManagement.size(false));
+ assertTrue(levelCompactionTsFileManagement.contains(seqResources.get(0),
true));
+ assertFalse(levelCompactionTsFileManagement.contains(new
TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 0
+ + ".tsfile"))), false));
+ assertTrue(levelCompactionTsFileManagement.contains(seqResources.get(0),
false));
+ assertFalse(levelCompactionTsFileManagement.contains(new
TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 0
+ + ".tsfile"))), false));
+ assertFalse(levelCompactionTsFileManagement.isEmpty(true));
+ assertFalse(levelCompactionTsFileManagement.isEmpty(false));
+ levelCompactionTsFileManagement
+ .remove(levelCompactionTsFileManagement.getTsFileList(true).get(0),
true);
+ levelCompactionTsFileManagement
+ .remove(levelCompactionTsFileManagement.getTsFileList(false).get(0),
false);
+ assertEquals(5,
levelCompactionTsFileManagement.getTsFileList(true).size());
+ levelCompactionTsFileManagement
+ .removeAll(levelCompactionTsFileManagement.getTsFileList(false),
false);
+ assertEquals(0,
levelCompactionTsFileManagement.getTsFileList(false).size());
+ long count = 0;
+ Iterator<TsFileResource> iterator =
levelCompactionTsFileManagement.getIterator(true);
+ while (iterator.hasNext()) {
+ iterator.next();
+ count++;
+ }
+ assertEquals(5, count);
+ levelCompactionTsFileManagement
+ .removeAll(levelCompactionTsFileManagement.getTsFileList(true), true);
+ assertEquals(0,
levelCompactionTsFileManagement.getTsFileList(true).size());
+ assertTrue(levelCompactionTsFileManagement.isEmpty(true));
+ assertTrue(levelCompactionTsFileManagement.isEmpty(false));
+ levelCompactionTsFileManagement.add(new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 10
+ + ".tsfile"))), true);
+ levelCompactionTsFileManagement.add(new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 10
+ + ".tsfile"))), false);
+ assertEquals(1, levelCompactionTsFileManagement.size(true));
+ assertEquals(1, levelCompactionTsFileManagement.size(false));
+ levelCompactionTsFileManagement.clear();
+ assertEquals(0, levelCompactionTsFileManagement.size(true));
+ assertEquals(0, levelCompactionTsFileManagement.size(false));
+ }
+}
\ No newline at end of file
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/NoCompactionTsFileManagementTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/NoCompactionTsFileManagementTest.java
new file mode 100644
index 0000000..f99ec45
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/NoCompactionTsFileManagementTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Iterator;
+import org.apache.commons.io.FileUtils;
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.constant.TestConstant;
+import
org.apache.iotdb.db.engine.compaction.TsFileManagement.CompactionMergeTask;
+import org.apache.iotdb.db.engine.compaction.no.NoCompactionTsFileManagement;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class NoCompactionTsFileManagementTest extends LevelCompactionTest {
+
+ File tempSGDir;
+
+ @Before
+ public void setUp() throws IOException, WriteProcessException,
MetadataException {
+ super.setUp();
+ tempSGDir = new File(TestConstant.BASE_OUTPUT_PATH.concat("tempSG"));
+ tempSGDir.mkdirs();
+ }
+
+ @After
+ public void tearDown() throws IOException, StorageEngineException {
+ super.tearDown();
+ FileUtils.deleteDirectory(tempSGDir);
+ }
+
+ /**
+ * just compaction once
+ */
+ @Test
+ public void testAddRemoveAndIterator() {
+ NoCompactionTsFileManagement noCompactionTsFileManagement = new
NoCompactionTsFileManagement(
+ COMPACTION_TEST_SG, tempSGDir.getPath());
+ for (TsFileResource tsFileResource : seqResources) {
+ noCompactionTsFileManagement.add(tsFileResource, true);
+ }
+ noCompactionTsFileManagement.addAll(seqResources, false);
+ assertEquals(6, noCompactionTsFileManagement.getTsFileList(true).size());
+ assertEquals(6, noCompactionTsFileManagement.getTsFileList(false).size());
+ assertEquals(6, noCompactionTsFileManagement.size(true));
+ assertEquals(6, noCompactionTsFileManagement.size(false));
+ assertTrue(noCompactionTsFileManagement.contains(seqResources.get(0),
true));
+ assertFalse(noCompactionTsFileManagement.contains(new TsFileResource(new
File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 0
+ + ".tsfile"))), false));
+ assertTrue(noCompactionTsFileManagement.contains(seqResources.get(0),
false));
+ assertFalse(noCompactionTsFileManagement.contains(new TsFileResource(new
File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 0
+ + ".tsfile"))), false));
+ assertFalse(noCompactionTsFileManagement.isEmpty(true));
+ assertFalse(noCompactionTsFileManagement.isEmpty(false));
+ noCompactionTsFileManagement
+ .remove(noCompactionTsFileManagement.getTsFileList(true).get(0), true);
+ noCompactionTsFileManagement
+ .remove(noCompactionTsFileManagement.getTsFileList(false).get(0),
false);
+ assertEquals(5, noCompactionTsFileManagement.getTsFileList(true).size());
+ noCompactionTsFileManagement
+ .removeAll(noCompactionTsFileManagement.getTsFileList(false), false);
+ assertEquals(0, noCompactionTsFileManagement.getTsFileList(false).size());
+ long count = 0;
+ Iterator<TsFileResource> iterator =
noCompactionTsFileManagement.getIterator(true);
+ while (iterator.hasNext()) {
+ iterator.next();
+ count++;
+ }
+ assertEquals(5, count);
+ noCompactionTsFileManagement
+ .removeAll(noCompactionTsFileManagement.getTsFileList(true), true);
+ assertEquals(0, noCompactionTsFileManagement.getTsFileList(true).size());
+ assertTrue(noCompactionTsFileManagement.isEmpty(true));
+ assertTrue(noCompactionTsFileManagement.isEmpty(false));
+ noCompactionTsFileManagement.add(new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 10
+ + ".tsfile"))), true);
+ noCompactionTsFileManagement.add(new TsFileResource(new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ 10 + IoTDBConstant.FILE_NAME_SEPARATOR + 10 +
IoTDBConstant.FILE_NAME_SEPARATOR + 10
+ + ".tsfile"))), false);
+ noCompactionTsFileManagement.forkCurrentFileList(0);
+ noCompactionTsFileManagement.recover();
+ CompactionMergeTask compactionMergeTask = noCompactionTsFileManagement.new
CompactionMergeTask(
+ () -> {
+ }, 0);
+ compactionMergeTask.run();
+ assertEquals(1, noCompactionTsFileManagement.size(true));
+ assertEquals(1, noCompactionTsFileManagement.size(false));
+ noCompactionTsFileManagement.clear();
+ assertEquals(0, noCompactionTsFileManagement.size(true));
+ assertEquals(0, noCompactionTsFileManagement.size(false));
+ }
+}
\ No newline at end of file
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCompactionIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCompactionIT.java
deleted file mode 100644
index e9779da..0000000
---
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBCompactionIT.java
+++ /dev/null
@@ -1,155 +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.integration;
-
-import static org.junit.Assert.assertEquals;
-
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.utils.EnvironmentUtils;
-import org.apache.iotdb.jdbc.Config;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class IoTDBCompactionIT {
-
- private int prevSeqLevelFileNum;
- private int prevSeqLevelNum;
-
- @Before
- public void setUp() throws Exception {
- EnvironmentUtils.closeStatMonitor();
- prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
- prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
- IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
- IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
- EnvironmentUtils.envSetUp();
- Class.forName(Config.JDBC_DRIVER_NAME);
- }
-
- @After
- public void tearDown() throws Exception {
- EnvironmentUtils.cleanEnv();
-
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
- IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
- }
-
- @Test
- public void test() throws SQLException {
- try (Connection connection = DriverManager
- .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
- Statement statement = connection.createStatement()) {
- statement.execute("SET STORAGE GROUP TO root.compactionTest");
- for (int i = 1; i <= 3; i++) {
- try {
- statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
- + "ENCODING=PLAIN");
- } catch (SQLException e) {
- // ignore
- }
- }
-
- for (int i = 0; i < 32; i++) {
- statement
- .execute(
- String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
- + "%d,%d)", i, i + 1, i + 2, i + 3));
- statement.execute("FLUSH");
- }
-
- int cnt;
- try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
- cnt = 0;
- while (resultSet.next()) {
- long time = resultSet.getLong("Time");
- long s1 = resultSet.getLong("root.compactionTest.s1");
- long s2 = resultSet.getLong("root.compactionTest.s2");
- long s3 = resultSet.getLong("root.compactionTest.s3");
- assertEquals(time + 1, s1);
- assertEquals(time + 2, s2);
- assertEquals(time + 3, s3);
- cnt++;
- }
- }
- assertEquals(32, cnt);
- }
- }
-
- @Test
- public void testAppendMergeAfterDeserializeMerge() throws SQLException {
- boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
- .isEnableUnseqCompaction();
- IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(false);
- try (Connection connection = DriverManager
- .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
- Statement statement = connection.createStatement()) {
- statement.execute("SET STORAGE GROUP TO root.compactionTest");
- try {
- statement.execute("CREATE TIMESERIES root.compactionTest.s1 WITH
DATATYPE=INT64");
- } catch (SQLException e) {
- // ignore
- }
-
- long pageSize = 100;
- long timestamp = 1;
-
- for (long row = 0; row < 10000; row++) {
- statement
- .execute(
- String.format("INSERT INTO root.compactionTest(timestamp,s1)
VALUES (%d,%d)",
- timestamp, 1));
- if (row % pageSize == 0) {
- statement.execute("FLUSH");
- }
- timestamp++;
- }
-
- timestamp = 8322;
-
- for (long row = 0; row < 2400; row++) {
- statement
- .execute(
- String.format("INSERT INTO root.compactionTest(timestamp,s1)
VALUES (%d,%d)",
- timestamp, 1));
- if (row % pageSize == 0) {
- statement.execute("FLUSH");
- }
- timestamp++;
- }
-
- int cnt;
- try (ResultSet resultSet = statement
- .executeQuery("SELECT COUNT(s1) FROM root.compactionTest")) {
- cnt = 0;
- while (resultSet.next()) {
- System.out.println(resultSet.getLong(1));
- assertEquals(10721, resultSet.getLong(1));
- cnt++;
- }
- }
- assertEquals(1, cnt);
- }
-
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
- }
-}
diff --git
a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBLevelCompactionIT.java
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBLevelCompactionIT.java
new file mode 100644
index 0000000..761aa26
--- /dev/null
+++
b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBLevelCompactionIT.java
@@ -0,0 +1,1180 @@
+/*
+ * 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.integration;
+
+import static org.junit.Assert.assertEquals;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Random;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.compaction.CompactionStrategy;
+import org.apache.iotdb.db.utils.EnvironmentUtils;
+import org.apache.iotdb.jdbc.Config;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class IoTDBLevelCompactionIT {
+
+ CompactionStrategy prevCompactionStrategy;
+
+ @Before
+ public void setUp() throws Exception {
+ EnvironmentUtils.closeStatMonitor();
+ prevCompactionStrategy = IoTDBDescriptor.getInstance().getConfig()
+ .getCompactionStrategy();
+ IoTDBDescriptor.getInstance().getConfig()
+ .setCompactionStrategy(CompactionStrategy.LEVEL_COMPACTION);
+ EnvironmentUtils.envSetUp();
+ Class.forName(Config.JDBC_DRIVER_NAME);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvironmentUtils.cleanEnv();
+
IoTDBDescriptor.getInstance().getConfig().setCompactionStrategy(prevCompactionStrategy);
+ }
+
+ /**
+ * test compaction files num > MAX_FILE_NUM_IN_LEVEL * MAX_LEVEL_NUM
+ */
+ @Test
+ public void test() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ int flushCount = 32;
+ for (int i = 0; i < flushCount; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(flushCount, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test compaction first use deserialize merge and then use append merge
+ */
+ @Test
+ public void testAppendMergeAfterDeserializeMerge() throws SQLException {
+ boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
+ .isEnableUnseqCompaction();
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(false);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s1 WITH
DATATYPE=INT64");
+ } catch (SQLException e) {
+ // ignore
+ }
+
+ long pageSize = 100;
+ long timestamp = 1;
+
+ for (long row = 0; row < 10000; row++) {
+ statement
+ .execute(
+ String.format("INSERT INTO root.compactionTest(timestamp,s1)
VALUES (%d,%d)",
+ timestamp, 1));
+ if (row % pageSize == 0) {
+ statement.execute("FLUSH");
+ }
+ timestamp++;
+ }
+
+ timestamp = 8322;
+
+ for (long row = 0; row < 2400; row++) {
+ statement
+ .execute(
+ String.format("INSERT INTO root.compactionTest(timestamp,s1)
VALUES (%d,%d)",
+ timestamp, 1));
+ if (row % pageSize == 0) {
+ statement.execute("FLUSH");
+ }
+ timestamp++;
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement
+ .executeQuery("SELECT COUNT(s1) FROM root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ System.out.println(resultSet.getLong(1));
+ assertEquals(10721, resultSet.getLong(1));
+ cnt++;
+ }
+ }
+ assertEquals(1, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
+ }
+
+ /**
+ * test compaction first use append merge and then use deserialize merge
+ */
+ @Test
+ public void testDeserializeMergeAfterAppendMerge() throws SQLException {
+ boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
+ .isEnableUnseqCompaction();
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(false);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s1 WITH
DATATYPE=INT64");
+ } catch (SQLException e) {
+ // ignore
+ }
+
+ long pageSize = 100;
+ long timestamp = 1;
+
+ int prevMergePagePointNumberThreshold =
IoTDBDescriptor.getInstance().getConfig()
+ .getMergePagePointNumberThreshold();
+
IoTDBDescriptor.getInstance().getConfig().setMergePagePointNumberThreshold(1);
+
+ for (long row = 0; row < 10000; row++) {
+ statement
+ .execute(
+ String.format("INSERT INTO root.compactionTest(timestamp,s1)
VALUES (%d,%d)",
+ timestamp, 1));
+ if (row % pageSize == 0) {
+ statement.execute("FLUSH");
+ }
+ timestamp++;
+ }
+
+ timestamp = 8322;
+
+
IoTDBDescriptor.getInstance().getConfig().setMergePagePointNumberThreshold(10000000);
+
+ for (long row = 0; row < 2400; row++) {
+ statement
+ .execute(
+ String.format("INSERT INTO root.compactionTest(timestamp,s1)
VALUES (%d,%d)",
+ timestamp, 1));
+ if (row % pageSize == 0) {
+ statement.execute("FLUSH");
+ }
+ timestamp++;
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement
+ .executeQuery("SELECT COUNT(s1) FROM root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ System.out.println(resultSet.getLong(1));
+ assertEquals(10721, resultSet.getLong(1));
+ cnt++;
+ }
+ }
+ assertEquals(1, cnt);
+ IoTDBDescriptor.getInstance().getConfig()
+ .setMergePagePointNumberThreshold(prevMergePagePointNumberThreshold);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
+ }
+
+ private void testCompactionNoUnseq(int mergeCount) throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 0; i < mergeCount; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(mergeCount, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test compaction just once, no unseq
+ */
+ @Test
+ public void testCompactionOnceNoUnseq() throws SQLException {
+ this.testCompactionNoUnseq(2);
+ }
+
+ /**
+ * test compaction to just once, with unseq
+ */
+ @Test
+ public void testCompactionOnceWithUnseq() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", 1, 2, 3, 4));
+ statement.execute("FLUSH");
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", 0, 1, 2, 3));
+ statement.execute("FLUSH");
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(2, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test compaction to second level once, no unseq
+ */
+ @Test
+ public void testCompactionToSecondLevelNoUnseq() throws SQLException {
+ this.testCompactionNoUnseq(4);
+ }
+
+ /**
+ * test compaction to second level once, with unseq
+ */
+ @Test
+ public void testCompactionToSecondLevelWithUnseq() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 1; i < 3; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 1; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 3; i < 5; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(5, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test compaction to second level once, with unseq, disable unseqCompaction
+ */
+ @Test
+ public void testCompactionToSecondLevelWithUnseqDisableUnseqCompaction()
throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
+ .isEnableUnseqCompaction();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 1; i < 3; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 1; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 3; i < 5; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(5, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
+ }
+
+ /**
+ * test compaction to stable level once, No unseq
+ */
+ @Test
+ public void testCompactionToStableLevelNoUnseq() throws SQLException {
+ this.testCompactionNoUnseq(8);
+ }
+
+ /**
+ * test compaction to stable level once, with unseq
+ */
+ @Test
+ public void testCompactionToStableLevelWithUnseq() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 4; i < 8; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 4; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 8; i < 12; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(12, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test compaction to stable level once, with unseq, disable unseqCompaction
+ */
+ @Test
+ public void testCompactionToStableLevelWithUnseqDisableUnseqCompaction()
throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
+ .isEnableUnseqCompaction();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 4; i < 8; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 4; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 8; i < 12; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(12, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
+ }
+
+ /**
+ * test seq max level num = 0
+ */
+ @Test
+ public void testCompactionSeqMaxLevelNumError0() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(0);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 4; i < 8; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 4; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 8; i < 12; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(12, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test seq max level num = -1
+ */
+ @Test
+ public void testCompactionSeqMaxLevelNumError1() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(-1);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 4; i < 8; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 4; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 8; i < 12; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(12, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test seq file num in each level = 0
+ */
+ @Test
+ public void testCompactionSeqFileNumInEachLevelError0() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(0);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 4; i < 8; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 4; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 8; i < 12; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(12, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test seq max level num = -1
+ */
+ @Test
+ public void testCompactionSeqFileNumInEachLevelError1() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(-1);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 4; i < 8; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 4; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 8; i < 12; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(12, cnt);
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+ }
+
+ /**
+ * test compaction with unseq compaction
+ */
+ @Test
+ public void testCompactionWithUnseqCompaction() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ int prevUnSeqLevelFileNum = IoTDBDescriptor.getInstance().getConfig()
+ .getUnseqFileNumInEachLevel();
+ int prevUnSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getUnseqLevelNum();
+ boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
+ .isEnableUnseqCompaction();
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(false);
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ IoTDBDescriptor.getInstance().getConfig().setUnseqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 10000; i < 10001; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 50; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 10001; i < 10005; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(55, cnt);
+
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(true);
+ for (int i = 10010; i < 10055; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s1 = resultSet.getLong("root.compactionTest.s1");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 1, s1);
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(100, cnt);
+ }
+
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+
IoTDBDescriptor.getInstance().getConfig().setUnseqFileNumInEachLevel(prevUnSeqLevelFileNum);
+
IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(prevUnSeqLevelNum);
+ }
+
+ /**
+ * test compaction with deletion timeseries
+ */
+ @Test
+ public void testCompactionWithDeletionTimeseries() throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ int prevUnSeqLevelFileNum = IoTDBDescriptor.getInstance().getConfig()
+ .getUnseqFileNumInEachLevel();
+ int prevUnSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getUnseqLevelNum();
+ boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
+ .isEnableUnseqCompaction();
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(false);
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ IoTDBDescriptor.getInstance().getConfig().setUnseqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 0; i < 1; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+ statement.execute("DELETE timeseries root.compactionTest.s1");
+
+ for (int i = 1; i < 2; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s2,s3) VALUES (%d,"
+ + "%d,%d)", i, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(2, cnt);
+ }
+
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+
IoTDBDescriptor.getInstance().getConfig().setUnseqFileNumInEachLevel(prevUnSeqLevelFileNum);
+
IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(prevUnSeqLevelNum);
+ }
+
+ /**
+ * test compaction with deletion timeseries and create different type
+ */
+ @Test
+ public void testCompactionWithDeletionTimeseriesAndCreateDifferentTypeTest()
throws SQLException {
+ int prevSeqLevelFileNum =
IoTDBDescriptor.getInstance().getConfig().getSeqFileNumInEachLevel();
+ int prevSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getSeqLevelNum();
+ int prevUnSeqLevelFileNum = IoTDBDescriptor.getInstance().getConfig()
+ .getUnseqFileNumInEachLevel();
+ int prevUnSeqLevelNum =
IoTDBDescriptor.getInstance().getConfig().getUnseqLevelNum();
+ boolean prevEnableUnseqCompaction =
IoTDBDescriptor.getInstance().getConfig()
+ .isEnableUnseqCompaction();
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(false);
+ IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(3);
+ IoTDBDescriptor.getInstance().getConfig().setUnseqFileNumInEachLevel(2);
+ IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(3);
+ try (Connection connection = DriverManager
+ .getConnection(Config.IOTDB_URL_PREFIX + "127.0.0.1:6667/", "root",
"root");
+ Statement statement = connection.createStatement()) {
+ statement.execute("SET STORAGE GROUP TO root.compactionTest");
+ for (int i = 1; i <= 3; i++) {
+ try {
+ statement.execute("CREATE TIMESERIES root.compactionTest.s" + i + "
WITH DATATYPE=INT64,"
+ + "ENCODING=PLAIN");
+ } catch (SQLException e) {
+ // ignore
+ }
+ }
+
+ for (int i = 10000; i < 10001; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ for (int i = 0; i < 50; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ if (i % 2 == 1) {
+ statement.execute("FLUSH");
+ }
+ }
+ statement.execute("FLUSH");
+ statement.execute("DELETE timeseries root.compactionTest.s1");
+ statement.execute(
+ "create timeseries root.compactionTest.s1 with datatype=FLOAT,
encoding=PLAIN");
+
+ for (int i = 10001; i < 10005; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ int cnt;
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(55, cnt);
+
+ IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(true);
+ for (int i = 10010; i < 10055; i++) {
+ statement
+ .execute(
+ String.format("INSERT INTO
root.compactionTest(timestamp,s1,s2,s3) VALUES (%d,%d,"
+ + "%d,%d)", i, i + 1, i + 2, i + 3));
+ statement.execute("FLUSH");
+ }
+
+ try (ResultSet resultSet = statement.executeQuery("SELECT * FROM
root.compactionTest")) {
+ cnt = 0;
+ while (resultSet.next()) {
+ long time = resultSet.getLong("Time");
+ long s2 = resultSet.getLong("root.compactionTest.s2");
+ long s3 = resultSet.getLong("root.compactionTest.s3");
+ assertEquals(time + 2, s2);
+ assertEquals(time + 3, s3);
+ cnt++;
+ }
+ }
+ assertEquals(100, cnt);
+ }
+
+
IoTDBDescriptor.getInstance().getConfig().setEnableUnseqCompaction(prevEnableUnseqCompaction);
+
IoTDBDescriptor.getInstance().getConfig().setSeqFileNumInEachLevel(prevSeqLevelFileNum);
+ IoTDBDescriptor.getInstance().getConfig().setSeqLevelNum(prevSeqLevelNum);
+
IoTDBDescriptor.getInstance().getConfig().setUnseqFileNumInEachLevel(prevUnSeqLevelFileNum);
+
IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(prevUnSeqLevelNum);
+ }
+}
