You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2019/10/24 01:29:41 UTC
[incubator-iotdb] branch dev_new_merge_strategy created (now
f51a9ef)
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a change to branch dev_new_merge_strategy
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git.
at f51a9ef temp save
This branch includes the following new commits:
new f51a9ef temp save
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-iotdb] 01/01: temp save
Posted by ji...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch dev_new_merge_strategy
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit f51a9ef4637d7b274c01ecb19f6b72bd8beef1b7
Author: jt <jt...@163.com>
AuthorDate: Thu Oct 24 09:29:13 2019 +0800
temp save
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +-
.../{selector => }/IFileQueryMemMeasurement.java | 4 +-
.../merge/{selector => }/IMergeFileSelector.java | 2 +-
.../merge/{selector => }/IMergePathSelector.java | 2 +-
.../db/engine/merge/{task => }/MergeCallback.java | 2 +-
.../merge/{ => inplace}/recover/LogAnalyzer.java | 6 +-
.../merge/{ => inplace}/recover/MergeLogger.java | 2 +-
.../selector/MaxFileMergeFileSelector.java | 4 +-
.../selector/MaxSeriesMergeFileSelector.java | 2 +-
.../{ => inplace}/selector/MergeFileStrategy.java | 2 +-
.../{ => inplace}/selector/NaivePathSelector.java | 3 +-
.../merge/{ => inplace}/task/MergeFileTask.java | 6 +-
.../{ => inplace}/task/MergeMultiChunkTask.java | 21 +-
.../engine/merge/{ => inplace}/task/MergeTask.java | 5 +-
.../merge/{ => inplace}/task/RecoverMergeTask.java | 11 +-
.../iotdb/db/engine/merge/manage/MergeManager.java | 2 +-
.../db/engine/merge/manage/MergeResource.java | 9 +-
.../merge/{ => squeeze}/recover/LogAnalyzer.java | 23 +-
.../merge/{ => squeeze}/recover/MergeLogger.java | 8 +-
.../selector/MaxFileMergeFileSelector.java | 7 +-
.../selector/MaxSeriesMergeFileSelector.java | 5 +-
.../{ => squeeze}/selector/MergeFileStrategy.java | 2 +-
.../{ => squeeze}/selector/NaivePathSelector.java | 3 +-
.../task/MergeSeriesTask.java} | 315 +++++++++------------
.../task/SqueezeMergeTask.java} | 81 ++----
.../engine/storagegroup/StorageGroupProcessor.java | 17 +-
.../engine/merge/MaxFileMergeFileSelectorTest.java | 3 +-
.../merge/MaxSeriesMergeFileSelectorTest.java | 2 +-
.../apache/iotdb/db/engine/merge/MergeLogTest.java | 2 +-
.../iotdb/db/engine/merge/MergePerfTest.java | 2 +-
.../iotdb/db/engine/merge/MergeTaskTest.java | 2 +-
.../tsfile/common/constant/TsFileConstant.java | 1 +
.../iotdb/tsfile/write/chunk/ChunkBuffer.java | 3 +-
.../iotdb/tsfile/write/chunk/ChunkWriterImpl.java | 26 +-
.../iotdb/tsfile/write/chunk/IChunkWriter.java | 2 +
.../iotdb/tsfile/write/writer/TsFileIOWriter.java | 3 +
36 files changed, 289 insertions(+), 303 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index d0e9109..c5cd266 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -25,7 +25,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import org.apache.iotdb.db.engine.merge.selector.MergeFileStrategy;
+import org.apache.iotdb.db.engine.merge.inplace.selector.MergeFileStrategy;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.db.service.TSServiceImpl;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IFileQueryMemMeasurement.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/IFileQueryMemMeasurement.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IFileQueryMemMeasurement.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/IFileQueryMemMeasurement.java
index 7883f1a..4bec629 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IFileQueryMemMeasurement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/IFileQueryMemMeasurement.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge;
import java.io.IOException;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -30,6 +30,6 @@ import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
* @return
* @throws IOException
*/
-interface IFileQueryMemMeasurement {
+public interface IFileQueryMemMeasurement {
long measure(TsFileResource resource) throws IOException;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IMergeFileSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/IMergeFileSelector.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IMergeFileSelector.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/IMergeFileSelector.java
index d452dff..e85c4df 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IMergeFileSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/IMergeFileSelector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge;
import java.util.List;
import org.apache.iotdb.db.exception.MergeException;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IMergePathSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/IMergePathSelector.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IMergePathSelector.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/IMergePathSelector.java
index a5ed5b0..683a45c 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/IMergePathSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/IMergePathSelector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge;
import java.util.Iterator;
import java.util.List;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeCallback.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/MergeCallback.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeCallback.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/MergeCallback.java
index d783abf..d67bb06 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeCallback.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/MergeCallback.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.task;
+package org.apache.iotdb.db.engine.merge;
import java.io.File;
import java.util.List;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/recover/LogAnalyzer.java
similarity index 98%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/recover/LogAnalyzer.java
index 04f364b..9b1f957 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/recover/LogAnalyzer.java
@@ -17,9 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.recover;
+package org.apache.iotdb.db.engine.merge.inplace.recover;
-import static org.apache.iotdb.db.engine.merge.recover.MergeLogger.*;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.*;
import java.io.BufferedReader;
import java.io.File;
@@ -30,7 +30,7 @@ import java.util.Map.Entry;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.merge.inplace.task.MergeTask;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.metadata.MManager;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/recover/MergeLogger.java
similarity index 98%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/recover/MergeLogger.java
index 986bacb..a7ab9d0 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/recover/MergeLogger.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.recover;
+package org.apache.iotdb.db.engine.merge.inplace.recover;
import java.io.BufferedWriter;
import java.io.File;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MaxFileMergeFileSelector.java
similarity index 98%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MaxFileMergeFileSelector.java
index 612a758..a060634 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MaxFileMergeFileSelector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.inplace.selector;
import java.io.IOException;
import java.util.ArrayList;
@@ -28,6 +28,8 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.IFileQueryMemMeasurement;
+import org.apache.iotdb.db.engine.merge.IMergeFileSelector;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MergeException;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxSeriesMergeFileSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MaxSeriesMergeFileSelector.java
similarity index 98%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxSeriesMergeFileSelector.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MaxSeriesMergeFileSelector.java
index 5209f48..3b710f2 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxSeriesMergeFileSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MaxSeriesMergeFileSelector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.inplace.selector;
import java.io.IOException;
import java.util.Collections;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergeFileStrategy.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MergeFileStrategy.java
similarity index 93%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergeFileStrategy.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MergeFileStrategy.java
index 874a91b..637ff72 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergeFileStrategy.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/MergeFileStrategy.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.inplace.selector;
public enum MergeFileStrategy {
MAX_SERIES_NUM,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/NaivePathSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/NaivePathSelector.java
similarity index 92%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/NaivePathSelector.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/NaivePathSelector.java
index 073a248..161e4a4 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/NaivePathSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/selector/NaivePathSelector.java
@@ -17,10 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.inplace.selector;
import java.util.List;
import java.util.NoSuchElementException;
+import org.apache.iotdb.db.engine.merge.IMergePathSelector;
import org.apache.iotdb.tsfile.read.common.Path;
public class NaivePathSelector implements IMergePathSelector {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeFileTask.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeFileTask.java
index e559c8e..cb74a32 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeFileTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeFileTask.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.task;
+package org.apache.iotdb.db.engine.merge.inplace.task;
import java.io.IOException;
import java.util.List;
@@ -28,7 +28,7 @@ import org.apache.iotdb.db.engine.cache.DeviceMetaDataCache;
import org.apache.iotdb.db.engine.cache.TsFileMetaDataCache;
import org.apache.iotdb.db.engine.merge.manage.MergeContext;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
+import org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.tsfile.exception.write.TsFileNotCompleteException;
import org.apache.iotdb.tsfile.file.metadata.ChunkGroupMetaData;
@@ -226,7 +226,7 @@ class MergeFileTask {
if (metaData.getStartTime() == startTime) {
Chunk chunk = reader.readMemChunk(metaData);
fileWriter.writeChunk(chunk, metaData);
- maxVersion = metaData.getVersion() > maxVersion ? metaData.getVersion() : maxVersion;
+ maxVersion = Math.max(metaData.getVersion(), maxVersion);
context.incTotalPointWritten(metaData.getNumOfPoints());
break;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeMultiChunkTask.java
similarity index 96%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeMultiChunkTask.java
index 7bf6488..992e5ba 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeMultiChunkTask.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.task;
+package org.apache.iotdb.db.engine.merge.inplace.task;
import static org.apache.iotdb.db.utils.MergeUtils.writeBatchPoint;
import static org.apache.iotdb.db.utils.MergeUtils.writeTVPair;
@@ -35,9 +35,9 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.merge.manage.MergeContext;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
-import org.apache.iotdb.db.engine.merge.selector.IMergePathSelector;
-import org.apache.iotdb.db.engine.merge.selector.NaivePathSelector;
+import org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger;
+import org.apache.iotdb.db.engine.merge.IMergePathSelector;
+import org.apache.iotdb.db.engine.merge.inplace.selector.NaivePathSelector;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.query.reader.IPointReader;
@@ -171,12 +171,12 @@ class MergeMultiChunkTask {
boolean isLastFile = seqFileIdx + 1 == resource.getSeqFiles().size();
TsFileSequenceReader fileSequenceReader = resource.getFileReader(currTsFile);
- List<Modification>[] modifications = new List[currMergingPaths.size()];
+ List<Modification> modifications;
List<ChunkMetaData>[] seqChunkMeta = new List[currMergingPaths.size()];
for (int i = 0; i < currMergingPaths.size(); i++) {
- modifications[i] = resource.getModifications(currTsFile, currMergingPaths.get(i));
+ modifications = resource.getModifications(currTsFile, currMergingPaths.get(i));
seqChunkMeta[i] = resource.queryChunkMetadata(currMergingPaths.get(i), currTsFile);
- modifyChunkMetaData(seqChunkMeta[i], modifications[i]);
+ modifyChunkMetaData(seqChunkMeta[i], modifications);
}
List<Integer> unskippedPathIndices = filterNoDataPaths(seqChunkMeta, seqFileIdx);
@@ -367,8 +367,7 @@ class MergeMultiChunkTask {
mergedChunkNum.incrementAndGet();
} else {
// 3.2 SK is overflowed, uncompress sequence chunk and merge with unseq chunk, then write
- unclosedChunkPoint += writeChunkWithUnseq(chunk, chunkWriter, unseqReader,
- currMeta.getEndTime(), pathIdx);
+ unclosedChunkPoint += writeChunkWithUnseq(chunk, chunkWriter, unseqReader, pathIdx);
mergedChunkNum.incrementAndGet();
}
@@ -398,15 +397,13 @@ class MergeMultiChunkTask {
return ptWritten;
}
- private int writeChunkWithUnseq(Chunk chunk, IChunkWriter chunkWriter, IPointReader unseqReader,
- long chunkLimitTime, int pathIdx) throws IOException {
+ private int writeChunkWithUnseq(Chunk chunk, IChunkWriter chunkWriter, IPointReader unseqReader, int pathIdx) throws IOException {
int cnt = 0;
ChunkReader chunkReader = new ChunkReaderWithoutFilter(chunk);
while (chunkReader.hasNextBatch()) {
BatchData batchData = chunkReader.nextBatch();
cnt += mergeWriteBatch(batchData, chunkWriter, unseqReader, pathIdx);
}
- cnt += writeRemainingUnseq(chunkWriter, unseqReader, chunkLimitTime, pathIdx);
return cnt;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeTask.java
similarity index 97%
copy from server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeTask.java
copy to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeTask.java
index 39bbb02..d22a55d 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeTask.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.task;
+package org.apache.iotdb.db.engine.merge.inplace.task;
import java.io.File;
import java.io.IOException;
@@ -25,9 +25,10 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
+import org.apache.iotdb.db.engine.merge.MergeCallback;
import org.apache.iotdb.db.engine.merge.manage.MergeContext;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
+import org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.metadata.MManager;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/RecoverMergeTask.java
similarity index 96%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/RecoverMergeTask.java
index 1f3ebe7..84c4fab 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/RecoverMergeTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/RecoverMergeTask.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.task;
+package org.apache.iotdb.db.engine.merge.inplace.task;
import java.io.File;
import java.io.FileInputStream;
@@ -28,10 +28,11 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.engine.merge.recover.LogAnalyzer;
-import org.apache.iotdb.db.engine.merge.recover.LogAnalyzer.Status;
-import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
-import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.MergeCallback;
+import org.apache.iotdb.db.engine.merge.inplace.recover.LogAnalyzer;
+import org.apache.iotdb.db.engine.merge.inplace.recover.LogAnalyzer.Status;
+import org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger;
+import org.apache.iotdb.db.engine.merge.inplace.selector.MaxSeriesMergeFileSelector;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.utils.MergeUtils;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeManager.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeManager.java
index ee43724..a6d8432 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeManager.java
@@ -28,7 +28,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
-import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.merge.inplace.task.MergeTask;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.ServiceType;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
index 941e572..7372b44 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/manage/MergeResource.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
+import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,7 +46,7 @@ import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
-import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+import static org.apache.iotdb.db.engine.merge.inplace.task.MergeTask.MERGE_SUFFIX;
/**
* MergeResource manages files and caches of readers, writers, MeasurementSchemas and
@@ -257,4 +258,10 @@ public class MergeResource {
measurementSchemaMap.put(measurementSchema.getMeasurementId(), measurementSchema);
}
}
+
+ public void flushChunks(TsFileIOWriter writer) throws IOException {
+ for (IChunkWriter chunkWriter : chunkWriterCache.values()) {
+ chunkWriter.writeToFileWriter(writer);
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/recover/LogAnalyzer.java
similarity index 90%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/recover/LogAnalyzer.java
index 04f364b..04ffb6d 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/LogAnalyzer.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/recover/LogAnalyzer.java
@@ -17,20 +17,30 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.recover;
+package org.apache.iotdb.db.engine.merge.squeeze.recover;
-import static org.apache.iotdb.db.engine.merge.recover.MergeLogger.*;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_ALL_TS_END;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_END;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_MERGE_END;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_MERGE_START;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_SEQ_FILES;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_START;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_TIMESERIES;
+import static org.apache.iotdb.db.engine.merge.inplace.recover.MergeLogger.STR_UNSEQ_FILES;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
-
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.engine.merge.inplace.task.MergeTask;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.task.MergeTask;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.metadata.MManager;
@@ -60,7 +70,8 @@ import org.slf4j.LoggerFactory;
*/
public class LogAnalyzer {
- private static final Logger logger = LoggerFactory.getLogger(LogAnalyzer.class);
+ private static final Logger logger = LoggerFactory.getLogger(
+ LogAnalyzer.class);
private MergeResource resource;
private String taskName;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/recover/MergeLogger.java
similarity index 94%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/recover/MergeLogger.java
index 986bacb..0c152ff 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/recover/MergeLogger.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/recover/MergeLogger.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.recover;
+package org.apache.iotdb.db.engine.merge.squeeze.recover;
import java.io.BufferedWriter;
import java.io.File;
@@ -129,4 +129,10 @@ public class MergeLogger {
logStream.newLine();
logStream.flush();
}
+
+ public void logNewFile(TsFileResource resource) throws IOException {
+ logStream.write(resource.getFile().getAbsolutePath());
+ logStream.newLine();
+ logStream.flush();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MaxFileMergeFileSelector.java
similarity index 97%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MaxFileMergeFileSelector.java
index 612a758..f509fa9 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxFileMergeFileSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MaxFileMergeFileSelector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.squeeze.selector;
import java.io.IOException;
import java.util.ArrayList;
@@ -28,6 +28,8 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.IFileQueryMemMeasurement;
+import org.apache.iotdb.db.engine.merge.IMergeFileSelector;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MergeException;
@@ -42,7 +44,8 @@ import org.slf4j.LoggerFactory;
*/
public class MaxFileMergeFileSelector implements IMergeFileSelector {
- private static final Logger logger = LoggerFactory.getLogger(MaxFileMergeFileSelector.class);
+ private static final Logger logger = LoggerFactory.getLogger(
+ MaxFileMergeFileSelector.class);
private static final String LOG_FILE_COST = "Memory cost of file {} is {}";
MergeResource resource;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxSeriesMergeFileSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MaxSeriesMergeFileSelector.java
similarity index 95%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxSeriesMergeFileSelector.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MaxSeriesMergeFileSelector.java
index 5209f48..621bdd6 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MaxSeriesMergeFileSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MaxSeriesMergeFileSelector.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.squeeze.selector;
import java.io.IOException;
import java.util.Collections;
@@ -35,7 +35,8 @@ import org.slf4j.LoggerFactory;
public class MaxSeriesMergeFileSelector extends MaxFileMergeFileSelector {
public static final int MAX_SERIES_NUM = 1024;
- private static final Logger logger = LoggerFactory.getLogger(MaxSeriesMergeFileSelector.class);
+ private static final Logger logger = LoggerFactory.getLogger(
+ MaxSeriesMergeFileSelector.class);
private List<TsFileResource> lastSelectedSeqFiles = Collections.emptyList();
private List<TsFileResource> lastSelectedUnseqFiles = Collections.emptyList();
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergeFileStrategy.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MergeFileStrategy.java
similarity index 93%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergeFileStrategy.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MergeFileStrategy.java
index 874a91b..28fd6ff 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/MergeFileStrategy.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/MergeFileStrategy.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.squeeze.selector;
public enum MergeFileStrategy {
MAX_SERIES_NUM,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/NaivePathSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/NaivePathSelector.java
similarity index 92%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/selector/NaivePathSelector.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/NaivePathSelector.java
index 073a248..6fa9bca 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/selector/NaivePathSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/selector/NaivePathSelector.java
@@ -17,10 +17,11 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.selector;
+package org.apache.iotdb.db.engine.merge.squeeze.selector;
import java.util.List;
import java.util.NoSuchElementException;
+import org.apache.iotdb.db.engine.merge.IMergePathSelector;
import org.apache.iotdb.tsfile.read.common.Path;
public class NaivePathSelector implements IMergePathSelector {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/MergeSeriesTask.java
similarity index 53%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/MergeSeriesTask.java
index 7bf6488..89796f8 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeMultiChunkTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/MergeSeriesTask.java
@@ -17,12 +17,16 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.merge.task;
+package org.apache.iotdb.db.engine.merge.squeeze.task;
+import static org.apache.iotdb.db.engine.merge.squeeze.task.SqueezeMergeTask.MERGE_SUFFIX;
import static org.apache.iotdb.db.utils.MergeUtils.writeBatchPoint;
import static org.apache.iotdb.db.utils.MergeUtils.writeTVPair;
import static org.apache.iotdb.db.utils.QueryUtils.modifyChunkMetaData;
+import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SEPARATOR;
+import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
+import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
@@ -30,14 +34,13 @@ import java.util.List;
import java.util.PriorityQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
-import java.util.concurrent.atomic.AtomicInteger;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.merge.IMergePathSelector;
import org.apache.iotdb.db.engine.merge.manage.MergeContext;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
-import org.apache.iotdb.db.engine.merge.selector.IMergePathSelector;
-import org.apache.iotdb.db.engine.merge.selector.NaivePathSelector;
+import org.apache.iotdb.db.engine.merge.squeeze.recover.MergeLogger;
+import org.apache.iotdb.db.engine.merge.squeeze.selector.NaivePathSelector;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.query.reader.IPointReader;
@@ -45,6 +48,7 @@ import org.apache.iotdb.db.utils.MergeUtils;
import org.apache.iotdb.db.utils.MergeUtils.MetaListEntry;
import org.apache.iotdb.db.utils.TimeValuePair;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.read.common.Chunk;
@@ -53,14 +57,15 @@ import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReader;
import org.apache.iotdb.tsfile.read.reader.chunk.ChunkReaderWithoutFilter;
import org.apache.iotdb.tsfile.write.chunk.IChunkWriter;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
+import org.apache.iotdb.tsfile.write.schema.Schema;
import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
-import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-class MergeMultiChunkTask {
+class MergeSeriesTask {
- private static final Logger logger = LoggerFactory.getLogger(MergeMultiChunkTask.class);
+ private static final Logger logger = LoggerFactory.getLogger(
+ MergeSeriesTask.class);
private static int minChunkPointNum = IoTDBDescriptor.getInstance().getConfig()
.getChunkMergePointThreshold();
@@ -70,31 +75,31 @@ class MergeMultiChunkTask {
private String taskName;
private MergeResource resource;
private TimeValuePair[] currTimeValuePairs;
- private boolean fullMerge;
private MergeContext mergeContext;
- private AtomicInteger mergedChunkNum = new AtomicInteger();
- private AtomicInteger unmergedChunkNum = new AtomicInteger();
private int mergedSeriesCnt;
private double progress;
private int concurrentMergeSeriesNum;
private List<Path> currMergingPaths = new ArrayList<>();
- MergeMultiChunkTask(MergeContext context, String taskName, MergeLogger mergeLogger,
- MergeResource mergeResource, boolean fullMerge, List<Path> unmergedSeries,
+ private RestorableTsFileIOWriter newFileWriter;
+ private TsFileResource newResource;
+ private String currDevice = null;
+
+ MergeSeriesTask(MergeContext context, String taskName, MergeLogger mergeLogger,
+ MergeResource mergeResource, List<Path> unmergedSeries,
int concurrentMergeSeriesNum) {
this.mergeContext = context;
this.taskName = taskName;
this.mergeLogger = mergeLogger;
this.resource = mergeResource;
- this.fullMerge = fullMerge;
this.unmergedSeries = unmergedSeries;
this.concurrentMergeSeriesNum = concurrentMergeSeriesNum;
}
- void mergeSeries() throws IOException {
+ TsFileResource mergeSeries() throws IOException {
if (logger.isInfoEnabled()) {
logger.info("{} starts to merge {} series", taskName, unmergedSeries.size());
}
@@ -114,11 +119,21 @@ class MergeMultiChunkTask {
logMergeProgress();
}
}
+
+ newFileWriter.endFile(new Schema(newFileWriter.getKnownSchema()));
+ newResource.serialize();
+ // the new file is ready to replace the old ones, write logs so we will not need to start from
+ // the beginning after system failure
+ mergeLogger.logAllTsEnd();
+ mergeLogger.logNewFile(newResource);
+
if (logger.isInfoEnabled()) {
logger.info("{} all series are merged after {}ms", taskName,
System.currentTimeMillis() - startTime);
}
mergeLogger.logAllTsEnd();
+
+ return newResource;
}
private void logMergeProgress() {
@@ -132,7 +147,6 @@ class MergeMultiChunkTask {
}
private void mergePaths() throws IOException {
- mergeLogger.logTSStart(currMergingPaths);
IPointReader[] unseqReaders;
unseqReaders = resource.getUnseqReaders(currMergingPaths);
currTimeValuePairs = new TimeValuePair[currMergingPaths.size()];
@@ -142,10 +156,49 @@ class MergeMultiChunkTask {
}
}
- for (int i = 0; i < resource.getSeqFiles().size(); i++) {
+ createNewFileWriter();
+
+ // merge data of the current paths in each seq file
+ for (int i = 0; i < resource.getSeqFiles().size(); i ++) {
pathsMergeOneFile(i, unseqReaders);
}
- mergeLogger.logTSEnd();
+
+ // merge the remaining data in unseq files
+ // this should not change the device max time, because time range of unseq files is always
+ // covered by that of seq files
+ for (int i = 0; i < currMergingPaths.size(); i++) {
+ Path path = currMergingPaths.get(i);
+ IChunkWriter chunkWriter = resource.getChunkWriter(resource.getSchema(path.getMeasurement()));
+ writeRemainingUnseq(chunkWriter, unseqReaders[i], i);
+ chunkWriter.writeToFileWriter(newFileWriter);
+ }
+ newFileWriter.endChunkGroup(0);
+ }
+
+ private void writeRemainingUnseq(IChunkWriter chunkWriter,
+ IPointReader unseqReader, int pathIdx) throws IOException {
+ while (currTimeValuePairs[pathIdx] != null) {
+ writeTVPair(currTimeValuePairs[pathIdx], chunkWriter);
+ unseqReader.next();
+ currTimeValuePairs[pathIdx] = unseqReader.hasNext() ? unseqReader.current() : null;
+ }
+ }
+
+ private void createNewFileWriter() throws IOException {
+ // use the minimum version as the version of the new file
+ long currFileVersion = Long.MAX_VALUE;
+ File parent = null;
+ for (TsFileResource seqFile : resource.getSeqFiles()) {
+ long fileVersion =
+ Long.parseLong(seqFile.getFile().getName()
+ .replace(TSFILE_SUFFIX, "").split(TSFILE_SEPARATOR)[1]);
+ currFileVersion = Math.min(currFileVersion, fileVersion);
+ parent = parent == null ? seqFile.getFile().getParentFile() : parent;
+ }
+ File newFile = FSFactoryProducer.getFSFactory().getFile(parent,
+ System.currentTimeMillis() + TSFILE_SEPARATOR + currFileVersion + TSFILE_SUFFIX + MERGE_SUFFIX);
+ newFileWriter = new RestorableTsFileIOWriter(newFile);
+ newResource = new TsFileResource(newFile);
}
private void pathsMergeOneFile(int seqFileIdx, IPointReader[] unseqReaders)
@@ -157,10 +210,6 @@ class MergeMultiChunkTask {
return;
}
- for (Path path : currMergingPaths) {
- mergeContext.getUnmergedChunkStartTimes().get(currTsFile).put(path, new ArrayList<>());
- }
-
// if this TsFile receives data later than fileLimitTime, it will overlap the next TsFile,
// which is forbidden
for (TimeValuePair timeValuePair : currTimeValuePairs) {
@@ -168,57 +217,53 @@ class MergeMultiChunkTask {
currDeviceMinTime = timeValuePair.getTimestamp();
}
}
- boolean isLastFile = seqFileIdx + 1 == resource.getSeqFiles().size();
TsFileSequenceReader fileSequenceReader = resource.getFileReader(currTsFile);
- List<Modification>[] modifications = new List[currMergingPaths.size()];
+ List<Modification> modifications;
List<ChunkMetaData>[] seqChunkMeta = new List[currMergingPaths.size()];
for (int i = 0; i < currMergingPaths.size(); i++) {
- modifications[i] = resource.getModifications(currTsFile, currMergingPaths.get(i));
+ modifications = resource.getModifications(currTsFile, currMergingPaths.get(i));
seqChunkMeta[i] = resource.queryChunkMetadata(currMergingPaths.get(i), currTsFile);
- modifyChunkMetaData(seqChunkMeta[i], modifications[i]);
+ modifyChunkMetaData(seqChunkMeta[i], modifications);
}
- List<Integer> unskippedPathIndices = filterNoDataPaths(seqChunkMeta, seqFileIdx);
- if (unskippedPathIndices.isEmpty()) {
+ if (allPathEmpty(seqChunkMeta)) {
return;
}
- RestorableTsFileIOWriter mergeFileWriter = resource.getMergeFileWriter(currTsFile);
for (Path path : currMergingPaths) {
MeasurementSchema schema = resource.getSchema(path.getMeasurement());
- mergeFileWriter.addSchema(schema);
+ newFileWriter.addSchema(schema);
}
- // merge unseq data with seq data in this file or small chunks in this file into a larger chunk
- mergeFileWriter.startChunkGroup(deviceId);
- boolean dataWritten = mergeChunks(seqChunkMeta, isLastFile, fileSequenceReader, unseqReaders,
- mergeFileWriter, currTsFile);
- if (dataWritten) {
- mergeFileWriter.endChunkGroup(0);
- mergeLogger.logFilePosition(mergeFileWriter.getFile());
- currTsFile.getStartTimeMap().put(deviceId, currDeviceMinTime);
+ // start merging a device
+ if (!deviceId.equals(currDevice)) {
+ if (currDevice != null) {
+ // flush previous device
+ resource.flushChunks(newFileWriter);
+ newFileWriter.endChunkGroup(0);
+
+ }
+ newFileWriter.startChunkGroup(deviceId);
+ currDevice = deviceId;
+ newResource.getStartTimeMap().put(deviceId, currDeviceMinTime);
}
+ // merge unseq data with seq data in this file or small chunks in this file into a larger chunk
+ long maxTime = mergeChunks(seqChunkMeta, fileSequenceReader, unseqReaders);
+ newResource.updateEndTime(deviceId, maxTime);
}
- private List<Integer> filterNoDataPaths(List[] seqChunkMeta, int seqFileIdx) {
- // if the last seqFile does not contains this series but the unseqFiles do, data of this
- // series should also be written into a new chunk
- List<Integer> ret = new ArrayList<>();
- for (int i = 0; i < currMergingPaths.size(); i++) {
- if (seqChunkMeta[i].isEmpty()
- && !(seqFileIdx + 1 == resource.getSeqFiles().size() && currTimeValuePairs[i] != null)) {
- continue;
+ private boolean allPathEmpty(List[] seqChunkMeta) {
+ for (int i = 0; i < seqChunkMeta.length; i++) {
+ if (!seqChunkMeta[i].isEmpty() || currTimeValuePairs[i] != null) {
+ return false;
}
- ret.add(i);
}
- return ret;
+ return true;
}
- private boolean mergeChunks(List<ChunkMetaData>[] seqChunkMeta, boolean isLastFile,
- TsFileSequenceReader reader, IPointReader[] unseqReaders,
- RestorableTsFileIOWriter mergeFileWriter, TsFileResource currFile)
+ private long mergeChunks(List<ChunkMetaData>[] seqChunkMeta,
+ TsFileSequenceReader reader, IPointReader[] unseqReaders)
throws IOException {
- int[] ptWrittens = new int[seqChunkMeta.length];
int mergeChunkSubTaskNum = IoTDBDescriptor.getInstance().getConfig().getMergeChunkSubThreadNum();
PriorityQueue<MetaListEntry>[] chunkMetaHeaps = new PriorityQueue[mergeChunkSubTaskNum];
for (int i = 0; i < mergeChunkSubTaskNum; i++) {
@@ -234,41 +279,30 @@ class MergeMultiChunkTask {
chunkMetaHeaps[idx % mergeChunkSubTaskNum].add(entry);
idx++;
- ptWrittens[i] = 0;
}
- mergedChunkNum.set(0);
- unmergedChunkNum.set(0);
- List<Future> futures = new ArrayList<>();
+ List<Future<Long>> futures = new ArrayList<>();
for (int i = 0; i < mergeChunkSubTaskNum; i++) {
int finalI = i;
- futures.add(MergeManager.getINSTANCE().submitChunkSubTask(() -> {
- mergeChunkHeap(chunkMetaHeaps[finalI], ptWrittens, reader, mergeFileWriter, unseqReaders, currFile,
- isLastFile);
- return null;
- }));
+ futures.add(MergeManager.getINSTANCE().submitChunkSubTask(() -> mergeChunkHeap(chunkMetaHeaps[finalI], reader, unseqReaders)));
}
+ long maxTime = Long.MIN_VALUE;
for (int i = 0; i < mergeChunkSubTaskNum; i++) {
try {
- futures.get(i).get();
+ Long heapMaxTimeStamp = futures.get(i).get();
+ maxTime = Math.max(maxTime, heapMaxTimeStamp);
} catch (InterruptedException | ExecutionException e) {
throw new IOException(e);
}
}
-
- // add merge and unmerged chunk statistic
- mergeContext.getMergedChunkCnt().compute(currFile, (tsFileResource, anInt) -> anInt == null ?
- mergedChunkNum.get() : anInt + mergedChunkNum.get());
- mergeContext.getUnmergedChunkCnt().compute(currFile, (tsFileResource, anInt) -> anInt == null ?
- unmergedChunkNum.get() : anInt + unmergedChunkNum.get());
-
- return mergedChunkNum.get() > 0;
+ return maxTime;
}
- private void mergeChunkHeap(PriorityQueue<MetaListEntry> chunkMetaHeap, int[] ptWrittens, TsFileSequenceReader reader,
- RestorableTsFileIOWriter mergeFileWriter, IPointReader[] unseqReaders, TsFileResource currFile,
- boolean isLastFile) throws IOException {
+ private long mergeChunkHeap(PriorityQueue<MetaListEntry> chunkMetaHeap,
+ TsFileSequenceReader reader, IPointReader[] unseqReaders)
+ throws IOException {
+ long maxTime = Long.MIN_VALUE;
while (!chunkMetaHeap.isEmpty()) {
MetaListEntry metaListEntry = chunkMetaHeap.poll();
ChunkMetaData currMeta = metaListEntry.current();
@@ -279,140 +313,64 @@ class MergeMultiChunkTask {
IChunkWriter chunkWriter = resource.getChunkWriter(measurementSchema);
boolean chunkOverflowed = MergeUtils.isChunkOverflowed(currTimeValuePairs[pathIdx], currMeta);
- boolean chunkTooSmall = MergeUtils
- .isChunkTooSmall(ptWrittens[pathIdx], currMeta, isLastChunk, minChunkPointNum);
+ boolean chunkTooSmall =
+ chunkWriter.getPtNum() > 0 || currMeta.getNumOfPoints() < minChunkPointNum;
Chunk chunk;
synchronized (reader) {
chunk = reader.readMemChunk(currMeta);
}
- ptWrittens[pathIdx] = mergeChunkV2(currMeta, chunkOverflowed, chunkTooSmall, chunk,
- ptWrittens[pathIdx], pathIdx, mergeFileWriter, unseqReaders[pathIdx], chunkWriter,
- currFile);
+
+ long maxMergedTime = mergeChunkV2(currMeta, chunkOverflowed, chunkTooSmall, chunk, pathIdx,
+ unseqReaders[pathIdx], chunkWriter);
+ maxTime = Math.max(maxMergedTime, maxTime);
if (!isLastChunk) {
metaListEntry.next();
chunkMetaHeap.add(metaListEntry);
- } else {
- // this only happens when the seqFiles do not contain this series, otherwise the remaining
- // data will be merged with the last chunk in the seqFiles
- if (isLastFile && currTimeValuePairs[pathIdx] != null) {
- ptWrittens[pathIdx] += writeRemainingUnseq(chunkWriter, unseqReaders[pathIdx], Long.MAX_VALUE,
- pathIdx);
- mergedChunkNum.incrementAndGet();
- }
- // the last merged chunk may still be smaller than the threshold, flush it anyway
- if (ptWrittens[pathIdx] > 0) {
- synchronized (mergeFileWriter) {
- chunkWriter.writeToFileWriter(mergeFileWriter);
- }
- }
}
}
+ return maxTime;
}
- /**
- * merge a sequence chunk SK
- *
- * 1. no need to write the chunk to .merge file when:
- * isn't full merge &
- * there isn't unclosed chunk before &
- * SK is big enough &
- * SK isn't overflowed &
- * SK isn't modified
- *
- *
- * 2. write SK to .merge.file without compressing when:
- * is full merge &
- * there isn't unclosed chunk before &
- * SK is big enough &
- * SK isn't overflowed &
- * SK isn't modified
- *
- * 3. other cases: need to unCompress the chunk and write
- * 3.1 SK isn't overflowed
- * 3.2 SK is overflowed
- *
- */
- private int mergeChunkV2(ChunkMetaData currMeta, boolean chunkOverflowed,
- boolean chunkTooSmall,Chunk chunk, int lastUnclosedChunkPoint, int pathIdx,
- TsFileIOWriter mergeFileWriter, IPointReader unseqReader,
- IChunkWriter chunkWriter, TsFileResource currFile) throws IOException {
-
- int unclosedChunkPoint = lastUnclosedChunkPoint;
- boolean chunkModified = currMeta.getDeletedAt() > Long.MIN_VALUE;
-
- // no need to write the chunk to .merge file
- if (!fullMerge && lastUnclosedChunkPoint == 0 && !chunkTooSmall && !chunkOverflowed && !chunkModified) {
- unmergedChunkNum.incrementAndGet();
- mergeContext.getUnmergedChunkStartTimes().get(currFile).get(currMergingPaths.get(pathIdx))
- .add(currMeta.getStartTime());
- return 0;
- }
+ private long mergeChunkV2(ChunkMetaData currMeta, boolean chunkOverflowed,
+ boolean chunkTooSmall, Chunk chunk, int pathIdx,
+ IPointReader unseqReader, IChunkWriter chunkWriter) throws IOException {
- // write SK to .merge.file without compressing
- if (fullMerge && lastUnclosedChunkPoint == 0 && !chunkTooSmall && !chunkOverflowed && !chunkModified) {
- synchronized (mergeFileWriter) {
- mergeFileWriter.writeChunk(chunk, currMeta);
+ // write the chunk to .merge.file without compressing
+ if (chunkWriter.getPtNum() == 0 && !chunkTooSmall && !chunkOverflowed) {
+ synchronized (newFileWriter) {
+ newFileWriter.writeChunk(chunk, currMeta);
}
- mergeContext.incTotalPointWritten(currMeta.getNumOfPoints());
- mergeContext.incTotalChunkWritten();
- mergedChunkNum.incrementAndGet();
- return 0;
+ return currMeta.getEndTime();
}
- // 3.1 SK isn't overflowed, just uncompress and write sequence chunk
- if (!chunkOverflowed) {
- unclosedChunkPoint += MergeUtils.writeChunkWithoutUnseq(chunk, chunkWriter);
- mergedChunkNum.incrementAndGet();
- } else {
- // 3.2 SK is overflowed, uncompress sequence chunk and merge with unseq chunk, then write
- unclosedChunkPoint += writeChunkWithUnseq(chunk, chunkWriter, unseqReader,
- currMeta.getEndTime(), pathIdx);
- mergedChunkNum.incrementAndGet();
- }
+ // uncompress and write the chunk
+ writeChunkWithUnseq(chunk, chunkWriter, unseqReader, pathIdx);
- // update points written statistics
- mergeContext.incTotalPointWritten(unclosedChunkPoint - lastUnclosedChunkPoint);
- if (minChunkPointNum > 0 && unclosedChunkPoint >= minChunkPointNum
- || unclosedChunkPoint > 0 && minChunkPointNum < 0) {
+ // check chunk size for flush and update points written statistics
+ if (minChunkPointNum > 0 && chunkWriter.getPtNum() >= minChunkPointNum
+ || chunkWriter.getPtNum() > 0 && minChunkPointNum < 0) {
// the new chunk's size is large enough and it should be flushed
- synchronized (mergeFileWriter) {
- chunkWriter.writeToFileWriter(mergeFileWriter);
+ synchronized (newFileWriter) {
+ mergeContext.incTotalPointWritten(chunkWriter.getPtNum());
+ chunkWriter.writeToFileWriter(newFileWriter);
}
- unclosedChunkPoint = 0;
- }
- return unclosedChunkPoint;
- }
-
- private int writeRemainingUnseq(IChunkWriter chunkWriter,
- IPointReader unseqReader, long timeLimit, int pathIdx) throws IOException {
- int ptWritten = 0;
- while (currTimeValuePairs[pathIdx] != null
- && currTimeValuePairs[pathIdx].getTimestamp() < timeLimit) {
- writeTVPair(currTimeValuePairs[pathIdx], chunkWriter);
- ptWritten++;
- unseqReader.next();
- currTimeValuePairs[pathIdx] = unseqReader.hasNext() ? unseqReader.current() : null;
}
- return ptWritten;
+ return currMeta.getEndTime();
}
- private int writeChunkWithUnseq(Chunk chunk, IChunkWriter chunkWriter, IPointReader unseqReader,
- long chunkLimitTime, int pathIdx) throws IOException {
- int cnt = 0;
+ private void writeChunkWithUnseq(Chunk chunk, IChunkWriter chunkWriter, IPointReader unseqReader,
+ int pathIdx) throws IOException {
ChunkReader chunkReader = new ChunkReaderWithoutFilter(chunk);
while (chunkReader.hasNextBatch()) {
BatchData batchData = chunkReader.nextBatch();
- cnt += mergeWriteBatch(batchData, chunkWriter, unseqReader, pathIdx);
+ mergeWriteBatch(batchData, chunkWriter, unseqReader, pathIdx);
}
- cnt += writeRemainingUnseq(chunkWriter, unseqReader, chunkLimitTime, pathIdx);
- return cnt;
}
- private int mergeWriteBatch(BatchData batchData, IChunkWriter chunkWriter,
+ private void mergeWriteBatch(BatchData batchData, IChunkWriter chunkWriter,
IPointReader unseqReader, int pathIdx) throws IOException {
- int cnt = 0;
for (int i = 0; i < batchData.length(); i++) {
long time = batchData.getTimeByIndex(i);
// merge data in batch and data in unseqReader
@@ -427,14 +385,11 @@ class MergeMultiChunkTask {
}
unseqReader.next();
currTimeValuePairs[pathIdx] = unseqReader.hasNext() ? unseqReader.current() : null;
- cnt++;
}
// unseq point.time > sequence point.time, write seq point
if (!overwriteSeqPoint) {
writeBatchPoint(batchData, i, chunkWriter);
- cnt++;
}
}
- return cnt;
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/SqueezeMergeTask.java
similarity index 60%
rename from server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeTask.java
rename to server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/SqueezeMergeTask.java
index 39bbb02..5d026d1 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/task/MergeTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/SqueezeMergeTask.java
@@ -1,23 +1,4 @@
-/*
- * 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.merge.task;
+package org.apache.iotdb.db.engine.merge.squeeze.task;
import java.io.File;
import java.io.IOException;
@@ -27,7 +8,8 @@ import java.util.List;
import java.util.concurrent.Callable;
import org.apache.iotdb.db.engine.merge.manage.MergeContext;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.recover.MergeLogger;
+import org.apache.iotdb.db.engine.merge.MergeCallback;
+import org.apache.iotdb.db.engine.merge.squeeze.recover.MergeLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.metadata.MManager;
@@ -37,48 +19,40 @@ import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- * MergeTask merges given seqFiles and unseqFiles into new ones, which basically consists of three
- * steps: 1. rewrite overflowed, modified or small-sized chunks into temp merge files
- * 2. move the merged chunks in the temp files back to the seqFiles or move the unmerged
- * chunks in the seqFiles into temp files and replace the seqFiles with the temp files.
- * 3. remove unseqFiles
- */
-public class MergeTask implements Callable<Void> {
+public class SqueezeMergeTask implements Callable<Void> {
- public static final String MERGE_SUFFIX = ".merge";
- private static final Logger logger = LoggerFactory.getLogger(MergeTask.class);
+ static final String MERGE_SUFFIX = ".merge";
+ private static final Logger logger = LoggerFactory.getLogger(SqueezeMergeTask.class);
- MergeResource resource;
- String storageGroupSysDir;
- String storageGroupName;
- MergeLogger mergeLogger;
- MergeContext mergeContext = new MergeContext();
+ private MergeResource resource;
+ private String storageGroupSysDir;
+ private String storageGroupName;
+ private MergeLogger mergeLogger;
+ private MergeContext mergeContext = new MergeContext();
private MergeCallback callback;
- int concurrentMergeSeriesNum;
- String taskName;
- boolean fullMerge;
+ private int concurrentMergeSeriesNum;
+ private String taskName;
- MergeTask(List<TsFileResource> seqFiles,
+ private TsFileResource newResource;
+
+ SqueezeMergeTask(List<TsFileResource> seqFiles,
List<TsFileResource> unseqFiles, String storageGroupSysDir, MergeCallback callback,
- String taskName, boolean fullMerge, String storageGroupName) {
+ String taskName, String storageGroupName) {
this.resource = new MergeResource(seqFiles, unseqFiles);
this.storageGroupSysDir = storageGroupSysDir;
this.callback = callback;
this.taskName = taskName;
- this.fullMerge = fullMerge;
this.concurrentMergeSeriesNum = 1;
this.storageGroupName = storageGroupName;
}
- public MergeTask(MergeResource mergeResource, String storageGroupSysDir, MergeCallback callback,
- String taskName, boolean fullMerge, int concurrentMergeSeriesNum, String storageGroupName) {
+ public SqueezeMergeTask(MergeResource mergeResource, String storageGroupSysDir, MergeCallback callback,
+ String taskName, int concurrentMergeSeriesNum, String storageGroupName) {
this.resource = mergeResource;
this.storageGroupSysDir = storageGroupSysDir;
this.callback = callback;
this.taskName = taskName;
- this.fullMerge = fullMerge;
this.concurrentMergeSeriesNum = concurrentMergeSeriesNum;
this.storageGroupName = storageGroupName;
}
@@ -92,7 +66,8 @@ public class MergeTask implements Callable<Void> {
cleanUp(false);
// call the callback to make sure the StorageGroup exit merging status, but passing 2
// empty file lists to avoid files being deleted.
- callback.call(Collections.emptyList(), Collections.emptyList(), new File(storageGroupSysDir, MergeLogger.MERGE_LOG_NAME));
+ callback.call(
+ Collections.emptyList(), Collections.emptyList(), new File(storageGroupSysDir, MergeLogger.MERGE_LOG_NAME));
throw e;
}
return null;
@@ -122,13 +97,9 @@ public class MergeTask implements Callable<Void> {
mergeLogger.logMergeStart();
- MergeMultiChunkTask mergeChunkTask = new MergeMultiChunkTask(mergeContext, taskName, mergeLogger, resource,
- fullMerge, unmergedSeries, concurrentMergeSeriesNum);
- mergeChunkTask.mergeSeries();
-
- MergeFileTask mergeFileTask = new MergeFileTask(taskName, mergeContext, mergeLogger, resource,
- resource.getSeqFiles());
- mergeFileTask.mergeFiles();
+ MergeSeriesTask mergeChunkTask = new MergeSeriesTask(mergeContext, taskName, mergeLogger, resource
+ ,unmergedSeries, concurrentMergeSeriesNum);
+ newResource = mergeChunkTask.mergeSeries();
cleanUp(true);
if (logger.isInfoEnabled()) {
@@ -145,7 +116,7 @@ public class MergeTask implements Callable<Void> {
}
}
- void cleanUp(boolean executeCallback) throws IOException {
+ private void cleanUp(boolean executeCallback) throws IOException {
logger.info("{} is cleaning up", taskName);
resource.clear();
@@ -169,4 +140,4 @@ public class MergeTask implements Callable<Void> {
logFile.delete();
}
}
-}
+}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 06bbb9e..28abc83 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -18,8 +18,9 @@
*/
package org.apache.iotdb.db.engine.storagegroup;
-import static org.apache.iotdb.db.engine.merge.task.MergeTask.MERGE_SUFFIX;
+import static org.apache.iotdb.db.engine.merge.inplace.task.MergeTask.MERGE_SUFFIX;
import static org.apache.iotdb.db.engine.storagegroup.TsFileResource.TEMP_SUFFIX;
+import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SEPARATOR;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
import java.io.File;
@@ -41,12 +42,12 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
import org.apache.iotdb.db.engine.merge.manage.MergeManager;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.selector.IMergeFileSelector;
-import org.apache.iotdb.db.engine.merge.selector.MaxFileMergeFileSelector;
-import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
-import org.apache.iotdb.db.engine.merge.selector.MergeFileStrategy;
-import org.apache.iotdb.db.engine.merge.task.MergeTask;
-import org.apache.iotdb.db.engine.merge.task.RecoverMergeTask;
+import org.apache.iotdb.db.engine.merge.IMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.inplace.selector.MaxFileMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.inplace.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.inplace.selector.MergeFileStrategy;
+import org.apache.iotdb.db.engine.merge.inplace.task.MergeTask;
+import org.apache.iotdb.db.engine.merge.inplace.task.RecoverMergeTask;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.modification.ModificationFile;
@@ -495,7 +496,7 @@ public class StorageGroupProcessor {
fsFactory.getFile(baseDir, storageGroupName).mkdirs();
String filePath = baseDir + File.separator + storageGroupName + File.separator +
- System.currentTimeMillis() + "-" + versionController.nextVersion() + TSFILE_SUFFIX;
+ System.currentTimeMillis() + TSFILE_SEPARATOR + versionController.nextVersion() + TSFILE_SUFFIX;
if (sequence) {
return new TsFileProcessor(storageGroupName, fsFactory.getFile(filePath),
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxFileMergeFileSelectorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxFileMergeFileSelectorTest.java
index 0f795b5..a380c53 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxFileMergeFileSelectorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxFileMergeFileSelectorTest.java
@@ -24,8 +24,7 @@ import static org.junit.Assert.assertEquals;
import java.io.IOException;
import java.util.List;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.selector.MaxFileMergeFileSelector;
-import org.apache.iotdb.db.engine.merge.selector.IMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.inplace.selector.MaxFileMergeFileSelector;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MergeException;
import org.junit.Test;
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxSeriesMergeFileSelectorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxSeriesMergeFileSelectorTest.java
index 2669159..3a55c38 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxSeriesMergeFileSelectorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MaxSeriesMergeFileSelectorTest.java
@@ -24,7 +24,7 @@ import static org.junit.Assert.assertEquals;
import java.io.IOException;
import java.util.List;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.selector.MaxSeriesMergeFileSelector;
+import org.apache.iotdb.db.engine.merge.inplace.selector.MaxSeriesMergeFileSelector;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MergeException;
import org.junit.Test;
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java
index 6fe1c3e..ac1210d 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeLogTest.java
@@ -29,7 +29,7 @@ import java.io.IOException;
import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.merge.inplace.task.MergeTask;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.exception.PathErrorException;
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergePerfTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergePerfTest.java
index c36236c..67082fd 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergePerfTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergePerfTest.java
@@ -25,7 +25,7 @@ import java.util.List;
import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.merge.inplace.task.MergeTask;
import org.apache.iotdb.db.metadata.MManager;
import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
index 33e0acc..dc9d6e1 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/MergeTaskTest.java
@@ -27,7 +27,7 @@ import java.util.Collections;
import org.apache.commons.io.FileUtils;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.merge.manage.MergeResource;
-import org.apache.iotdb.db.engine.merge.task.MergeTask;
+import org.apache.iotdb.db.engine.merge.inplace.task.MergeTask;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.exception.MetadataErrorException;
import org.apache.iotdb.db.exception.PathErrorException;
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java
index 594a613..9a3d2b0 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/common/constant/TsFileConstant.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.tsfile.common.constant;
public class TsFileConstant {
public static final String TSFILE_SUFFIX = ".tsfile";
+ public static final String TSFILE_SEPARATOR = "-";
public static final String TSFILE_HOME = "TSFILE_HOME";
public static final String TSFILE_CONF = "TSFILE_CONF";
public static final String PATH_SEPARATOR = ".";
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkBuffer.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkBuffer.java
index 4345505..39023ad 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkBuffer.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkBuffer.java
@@ -172,7 +172,8 @@ public class ChunkBuffer {
public long writeAllPagesOfSeriesToTsFile(TsFileIOWriter writer, Statistics<?> statistics)
throws IOException {
if (minTimestamp == Long.MIN_VALUE) {
- LOG.error("Write page error, {}, minTime:{}, maxTime:{}", schema, minTimestamp, maxTimestamp);
+ LOG.debug("No data in the chunk buffer");
+ return 0;
}
// start to write this column chunk
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java
index 295f4fb..c63e8de 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ChunkWriterImpl.java
@@ -20,7 +20,6 @@ package org.apache.iotdb.tsfile.write.chunk;
import java.io.IOException;
import java.math.BigDecimal;
-import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.exception.write.PageException;
import org.apache.iotdb.tsfile.file.header.ChunkHeader;
@@ -79,6 +78,7 @@ public class ChunkWriterImpl implements IChunkWriter {
private long minTimestamp = Long.MIN_VALUE;
private MeasurementSchema measurementSchema;
+ private int ptNum;
/**
* constructor of ChunkWriterImpl.
@@ -117,6 +117,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long time, long value) {
this.time = time;
++valueCountInOnePage;
+ ++ptNum;
dataPageWriter.write(time, value);
pageStatistics.updateStats(value);
if (minTimestamp == Long.MIN_VALUE) {
@@ -129,6 +130,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long time, int value) {
this.time = time;
++valueCountInOnePage;
+ ++ptNum;
dataPageWriter.write(time, value);
pageStatistics.updateStats(value);
if (minTimestamp == Long.MIN_VALUE) {
@@ -141,6 +143,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long time, boolean value) {
this.time = time;
++valueCountInOnePage;
+ ++ptNum;
dataPageWriter.write(time, value);
pageStatistics.updateStats(value);
if (minTimestamp == Long.MIN_VALUE) {
@@ -153,6 +156,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long time, float value) {
this.time = time;
++valueCountInOnePage;
+ ++ptNum;
dataPageWriter.write(time, value);
pageStatistics.updateStats(value);
if (minTimestamp == Long.MIN_VALUE) {
@@ -165,6 +169,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long time, double value) {
this.time = time;
++valueCountInOnePage;
+ ++ptNum;
dataPageWriter.write(time, value);
pageStatistics.updateStats(value);
if (minTimestamp == Long.MIN_VALUE) {
@@ -177,6 +182,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long time, BigDecimal value) {
this.time = time;
++valueCountInOnePage;
+ ++ptNum;
dataPageWriter.write(time, value);
pageStatistics.updateStats(value);
if (minTimestamp == Long.MIN_VALUE) {
@@ -189,6 +195,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long time, Binary value) {
this.time = time;
++valueCountInOnePage;
+ ++ptNum;
dataPageWriter.write(time, value);
pageStatistics.updateStats(value);
if (minTimestamp == Long.MIN_VALUE) {
@@ -201,6 +208,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long[] timestamps, int[] values, int batchSize) {
this.time = timestamps[batchSize - 1];
valueCountInOnePage += batchSize;
+ ptNum += batchSize;
if (minTimestamp == Long.MIN_VALUE) {
minTimestamp = timestamps[0];
}
@@ -213,6 +221,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long[] timestamps, long[] values, int batchSize) {
this.time = timestamps[batchSize - 1];
valueCountInOnePage += batchSize;
+ ptNum += batchSize;
if (minTimestamp == Long.MIN_VALUE) {
minTimestamp = timestamps[0];
}
@@ -225,6 +234,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long[] timestamps, boolean[] values, int batchSize) {
this.time = timestamps[batchSize - 1];
valueCountInOnePage += batchSize;
+ ptNum += batchSize;
if (minTimestamp == Long.MIN_VALUE) {
minTimestamp = timestamps[0];
}
@@ -237,6 +247,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long[] timestamps, float[] values, int batchSize) {
this.time = timestamps[batchSize - 1];
valueCountInOnePage += batchSize;
+ ptNum += batchSize;
if (minTimestamp == Long.MIN_VALUE) {
minTimestamp = timestamps[0];
}
@@ -249,6 +260,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long[] timestamps, double[] values, int batchSize) {
this.time = timestamps[batchSize - 1];
valueCountInOnePage += batchSize;
+ ptNum += batchSize;
if (minTimestamp == Long.MIN_VALUE) {
minTimestamp = timestamps[0];
}
@@ -261,6 +273,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long[] timestamps, BigDecimal[] values, int batchSize) {
this.time = timestamps[batchSize - 1];
valueCountInOnePage += batchSize;
+ ptNum += batchSize;
if (minTimestamp == Long.MIN_VALUE) {
minTimestamp = timestamps[0];
}
@@ -273,6 +286,7 @@ public class ChunkWriterImpl implements IChunkWriter {
public void write(long[] timestamps, Binary[] values, int batchSize) {
this.time = timestamps[batchSize - 1];
valueCountInOnePage += batchSize;
+ ptNum += batchSize;
if (minTimestamp == Long.MIN_VALUE) {
minTimestamp = timestamps[0];
}
@@ -342,10 +356,13 @@ public class ChunkWriterImpl implements IChunkWriter {
@Override
public void writeToFileWriter(TsFileIOWriter tsfileWriter) throws IOException {
sealCurrentPage();
- chunkBuffer.writeAllPagesOfSeriesToTsFile(tsfileWriter, chunkStatistics);
+ if (chunkBuffer.writeAllPagesOfSeriesToTsFile(tsfileWriter, chunkStatistics) == 0) {
+ return;
+ }
chunkBuffer.reset();
// reset series_statistics
this.chunkStatistics = Statistics.getStatsByType(dataType);
+ ptNum = 0;
}
@Override
@@ -380,4 +397,9 @@ public class ChunkWriterImpl implements IChunkWriter {
public TSDataType getDataType() {
return dataType;
}
+
+ @Override
+ public int getPtNum() {
+ return ptNum;
+ }
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkWriter.java
index 3f7f762..fb0ff32 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/IChunkWriter.java
@@ -124,4 +124,6 @@ public interface IChunkWriter {
int getNumOfPages();
TSDataType getDataType();
+
+ int getPtNum();
}
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
index fffa6bd..37cfb42 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/writer/TsFileIOWriter.java
@@ -161,6 +161,9 @@ public class TsFileIOWriter {
* end chunk and write some log.
*/
public void endChunkGroup(long version) throws IOException {
+ if (currentChunkGroupMetaData.getChunkMetaDataList().isEmpty()) {
+ return;
+ }
long dataSize = out.getPosition() - currentChunkGroupMetaData.getStartOffsetOfChunkGroup();
ChunkGroupFooter chunkGroupFooter = new ChunkGroupFooter(
currentChunkGroupMetaData.getDeviceID(),