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 09:44:07 UTC
[incubator-iotdb] branch dev_new_merge_strategy updated: complete
SqueezeMergeTask
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
The following commit(s) were added to refs/heads/dev_new_merge_strategy by this push:
new ccde793 complete SqueezeMergeTask
ccde793 is described below
commit ccde7939814ff7d1ad59e0d47066464548527d1c
Author: jt <jt...@163.com>
AuthorDate: Thu Oct 24 17:43:40 2019 +0800
complete SqueezeMergeTask
---
.../iotdb/db/engine/merge/MergeCallback.java | 5 +-
.../db/engine/merge/inplace/task/MergeTask.java | 5 +-
.../merge/squeeze/task/SqueezeMergeTask.java | 16 +--
.../engine/storagegroup/StorageGroupProcessor.java | 128 +++++++++++++++------
.../apache/iotdb/db/engine/merge/MergeLogTest.java | 2 +-
5 files changed, 106 insertions(+), 50 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/MergeCallback.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/MergeCallback.java
index d67bb06..099251b 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/MergeCallback.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/MergeCallback.java
@@ -35,6 +35,9 @@ public interface MergeCallback {
* 4. exit merging status
* @param seqFiles
* @param unseqFiles
+ * @param newFile the new file generated by merging all seqFiles and unseqFiles, it will be
+ * null for inplace strategy
*/
- void call(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles, File logFile);
+ void call(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles, File logFile,
+ TsFileResource newFile);
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeTask.java
index d22a55d..19c7934 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/inplace/task/MergeTask.java
@@ -93,7 +93,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), null);
throw e;
}
return null;
@@ -165,7 +166,7 @@ public class MergeTask implements Callable<Void> {
if (executeCallback) {
// make sure merge.log is not deleted until unseqFiles are cleared so that when system
// reboots, the undeleted files can be deleted again
- callback.call(resource.getSeqFiles(), resource.getUnseqFiles(), logFile);
+ callback.call(resource.getSeqFiles(), resource.getUnseqFiles(), logFile, null);
} else {
logFile.delete();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/SqueezeMergeTask.java b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/SqueezeMergeTask.java
index 5d026d1..339a8c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/SqueezeMergeTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/merge/squeeze/task/SqueezeMergeTask.java
@@ -36,17 +36,6 @@ public class SqueezeMergeTask implements Callable<Void> {
private TsFileResource newResource;
- SqueezeMergeTask(List<TsFileResource> seqFiles,
- List<TsFileResource> unseqFiles, String storageGroupSysDir, MergeCallback callback,
- String taskName, String storageGroupName) {
- this.resource = new MergeResource(seqFiles, unseqFiles);
- this.storageGroupSysDir = storageGroupSysDir;
- this.callback = callback;
- this.taskName = taskName;
- this.concurrentMergeSeriesNum = 1;
- this.storageGroupName = storageGroupName;
- }
-
public SqueezeMergeTask(MergeResource mergeResource, String storageGroupSysDir, MergeCallback callback,
String taskName, int concurrentMergeSeriesNum, String storageGroupName) {
this.resource = mergeResource;
@@ -67,7 +56,8 @@ public class SqueezeMergeTask implements Callable<Void> {
// 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));
+ Collections.emptyList(), Collections.emptyList(), new File(storageGroupSysDir,
+ MergeLogger.MERGE_LOG_NAME), null);
throw e;
}
return null;
@@ -135,7 +125,7 @@ public class SqueezeMergeTask implements Callable<Void> {
if (executeCallback) {
// make sure merge.log is not deleted until unseqFiles are cleared so that when system
// reboots, the undeleted files can be deleted again
- callback.call(resource.getSeqFiles(), resource.getUnseqFiles(), logFile);
+ callback.call(resource.getSeqFiles(), resource.getUnseqFiles(), logFile, newResource);
} else {
logFile.delete();
}
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 28abc83..64aae8e 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
@@ -1,4 +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
@@ -241,7 +241,7 @@ public class StorageGroupProcessor {
}
}
- private List<TsFileResource> getAllFiles(List<String> folders) throws IOException {
+ private List<TsFileResource> getAllFiles(List<String> folders) {
List<File> tsFiles = new ArrayList<>();
for (String baseDir : folders) {
File fileFolder = fsFactory.getFile(baseDir, storageGroupName);
@@ -303,13 +303,13 @@ public class StorageGroupProcessor {
// TsFileNameComparator compares TsFiles by the version number in its name
// ({systemTime}-{versionNum}.tsfile)
- public int compareFileName(File o1, File o2) {
+ private int compareFileName(File o1, File o2) {
String[] items1 = o1.getName().replace(TSFILE_SUFFIX, "").split("-");
String[] items2 = o2.getName().replace(TSFILE_SUFFIX, "").split("-");
- if (Long.valueOf(items1[0]) - Long.valueOf(items2[0]) == 0) {
- return Long.compare(Long.valueOf(items1[1]), Long.valueOf(items2[1]));
+ if (Long.parseLong(items1[0]) - Long.parseLong(items2[0]) == 0) {
+ return Long.compare(Long.parseLong(items1[1]), Long.parseLong(items2[1]));
} else {
- return Long.compare(Long.valueOf(items1[0]), Long.valueOf(items2[0]));
+ return Long.compare(Long.parseLong(items1[0]), Long.parseLong(items2[0]));
}
}
@@ -826,8 +826,7 @@ public class StorageGroupProcessor {
/**
* put the memtable back to the MemTablePool and make the metadata in writer visible
*/
- // TODO please consider concurrency with query and insert method.
- public void closeUnsealedTsFileProcessor(
+ private void closeUnsealedTsFileProcessor(
TsFileProcessor tsFileProcessor) throws TsFileProcessorException {
closeQueryLock.writeLock().lock();
try {
@@ -910,32 +909,10 @@ public class StorageGroupProcessor {
}
}
- protected void mergeEndAction(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles,
- File mergeLog) {
- logger.info("{} a merge task is ending...", storageGroupName);
+ private void handleInplaceMerge(List<TsFileResource> seqFiles,
+ List<TsFileResource> unseqFiles, File mergeLog) {
- if (unseqFiles.isEmpty()) {
- // merge runtime exception arose, just end this merge
- isMerging = false;
- logger.info("{} a merge task abnormally ends", storageGroupName);
- return;
- }
-
- mergeLock.writeLock().lock();
- try {
- unSequenceFileList.removeAll(unseqFiles);
- } finally {
- mergeLock.writeLock().unlock();
- }
-
- for (TsFileResource unseqFile : unseqFiles) {
- unseqFile.getMergeQueryLock().writeLock().lock();
- try {
- unseqFile.remove();
- } finally {
- unseqFile.getMergeQueryLock().writeLock().unlock();
- }
- }
+ removeUnseqFiles(unseqFiles);
for (int i = 0; i < seqFiles.size(); i++) {
TsFileResource seqFile = seqFiles.get(i);
@@ -975,6 +952,91 @@ public class StorageGroupProcessor {
logger.info("{} a merge task ends", storageGroupName);
}
+ private void handleSqueezeMerge(
+ List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles, File mergeLog,
+ TsFileResource newFile) {
+
+ // make sure no queries are holding the seqFiles
+ for (TsFileResource seqFile : seqFiles) {
+ seqFile.getMergeQueryLock().writeLock().lock();
+ }
+ // block new queries and insertions to prevent the seqFiles from changing
+ writeLock();
+ mergeLock.writeLock().lock();;
+ try {
+ removeUnseqFiles(unseqFiles);
+ // insert the new file into a proper place
+ List<TsFileResource> newSeqFiles = new ArrayList<>();
+ int startIdx = sequenceFileList.indexOf(seqFiles.get(0));
+ int endIdx = sequenceFileList.indexOf(seqFiles.get(seqFiles.size() - 1));
+ for (int i = 0; i < startIdx; i++) {
+ newSeqFiles.add(sequenceFileList.get(i));
+ }
+ newSeqFiles.add(newFile);
+ for (int i = endIdx + 1; i < sequenceFileList.size(); i++) {
+ newSeqFiles.add(sequenceFileList.get(i));
+ }
+ sequenceFileList = newSeqFiles;
+ // move modifications generated during merge into the new file
+ if (mergingModification != null) {
+ logger.debug("{} is updating the merged file's modification file", storageGroupName);
+ for (Modification modification : mergingModification.getModifications()) {
+ newFile.getModFile().write(modification);
+ }
+ mergingModification.remove();
+ mergingModification = null;
+ }
+ // remove old seqFiles
+ for (TsFileResource seqFile : seqFiles) {
+ seqFile.remove();
+ seqFile.getMergeQueryLock().writeLock().unlock();
+ }
+ mergeLog.delete();
+ } catch (IOException e) {
+ logger.error("{} fails to do the after merge action,", storageGroupName, e);
+ } finally {
+ isMerging = false;
+ writeUnlock();
+ mergeLock.writeLock().unlock();
+ logger.info("{} a merge task ends", storageGroupName);
+ }
+ }
+
+ private void removeUnseqFiles(List<TsFileResource> unseqFiles) {
+ mergeLock.writeLock().lock();
+ try {
+ unSequenceFileList.removeAll(unseqFiles);
+ } finally {
+ mergeLock.writeLock().unlock();
+ }
+ for (TsFileResource unseqFile : unseqFiles) {
+ unseqFile.getMergeQueryLock().writeLock().lock();
+ try {
+ unseqFile.remove();
+ } finally {
+ unseqFile.getMergeQueryLock().writeLock().unlock();
+ }
+ }
+ }
+
+ void mergeEndAction(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles,
+ File mergeLog, TsFileResource newFile) {
+ logger.info("{} a merge task is ending...", storageGroupName);
+
+ if (unseqFiles.isEmpty()) {
+ // merge runtime exception arose, just end this merge
+ isMerging = false;
+ logger.info("{} a merge task abnormally ends", storageGroupName);
+ return;
+ }
+
+ if (newFile == null) {
+ handleInplaceMerge(seqFiles, unseqFiles, mergeLog);
+ } else {
+ handleSqueezeMerge(seqFiles, unseqFiles, mergeLog, newFile);
+ }
+ }
+
public TsFileProcessor getWorkSequenceTsFileProcessor() {
return workSequenceTsFileProcessor;
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 ac1210d..0c525c8 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
@@ -65,7 +65,7 @@ public class MergeLogTest extends MergeTest {
}
private void testCallBack(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles,
- File mergeLog) {
+ File mergeLog, TsFileResource newFile) {
int lineCnt = 0;
try (BufferedReader bufferedReader = new BufferedReader(new FileReader(mergeLog))) {
String line;