You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2021/07/27 11:57:02 UTC
[iotdb] branch rel/0.12 updated (18f0144 -> 9be4d2f)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a change to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git.
from 18f0144 [To rel/0.12] Cherry pick from autoai (#3637)
new 87fb3d9 fix bug in compaction (#3626)
new 9be4d2f replace synchronized with write lock in cross compaction selection (#3633)
The 2 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.
Summary of changes:
.../compaction/CompactionMergeTaskPoolManager.java | 31 +++--
.../db/engine/compaction/TsFileManagement.java | 148 +++++++++++----------
.../level/LevelCompactionTsFileManagement.java | 92 ++++++++-----
.../db/engine/merge/manage/MergeResource.java | 3 +-
.../merge/selector/MaxFileMergeFileSelector.java | 15 ++-
.../db/engine/storagegroup/TsFileResource.java | 2 +-
.../compaction/LevelCompactionRecoverTest.java | 13 +-
7 files changed, 170 insertions(+), 134 deletions(-)
[iotdb] 02/02: replace synchronized with write lock in cross
compaction selection (#3633)
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9be4d2fdbc78fad1d82c93d6522443ba5ad9dbda
Author: liuxuxin <37...@users.noreply.github.com>
AuthorDate: Tue Jul 27 19:54:23 2021 +0800
replace synchronized with write lock in cross compaction selection (#3633)
---
.../db/engine/compaction/TsFileManagement.java | 148 +++++++++++----------
.../db/engine/merge/manage/MergeResource.java | 3 +-
.../merge/selector/MaxFileMergeFileSelector.java | 15 ++-
.../db/engine/storagegroup/TsFileResource.java | 2 +-
4 files changed, 87 insertions(+), 81 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
index d997e04..dde54a7 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileManagement.java
@@ -193,7 +193,7 @@ public abstract class TsFileManagement {
}
}
- public synchronized boolean merge(
+ public boolean merge(
boolean fullMerge,
List<TsFileResource> seqMergeList,
List<TsFileResource> unSeqMergeList,
@@ -209,89 +209,93 @@ public abstract class TsFileManagement {
}
}
isUnseqMerging = true;
-
- if (seqMergeList.isEmpty()) {
- logger.info("{} no seq files to be merged", storageGroupName);
- isUnseqMerging = false;
- return false;
- }
-
- if (unSeqMergeList.isEmpty()) {
- logger.info("{} no unseq files to be merged", storageGroupName);
- isUnseqMerging = false;
- return false;
- }
-
- if (unSeqMergeList.size() > maxOpenFileNumInEachUnseqCompaction) {
- logger.info(
- "{} too much unseq files to be merged, reduce it to {}",
- storageGroupName,
- maxOpenFileNumInEachUnseqCompaction);
- unSeqMergeList = unSeqMergeList.subList(0, maxOpenFileNumInEachUnseqCompaction);
- }
-
- long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget();
- long timeLowerBound = System.currentTimeMillis() - dataTTL;
- MergeResource mergeResource = new MergeResource(seqMergeList, unSeqMergeList, timeLowerBound);
-
- IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource);
+ writeLock();
try {
- List[] mergeFiles = fileSelector.select();
- if (mergeFiles.length == 0) {
- logger.info(
- "{} cannot select merge candidates under the budget {}", storageGroupName, budget);
+ if (seqMergeList.isEmpty()) {
+ logger.info("{} no seq files to be merged", storageGroupName);
isUnseqMerging = false;
return false;
}
- // avoid pending tasks holds the metadata and streams
- mergeResource.clear();
- String taskName = storageGroupName + "-" + System.currentTimeMillis();
- // do not cache metadata until true candidates are chosen, or too much metadata will be
- // cached during selection
- mergeResource.setCacheDeviceMeta(true);
-
- for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) {
- tsFileResource.setMerging(true);
- }
- for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) {
- tsFileResource.setMerging(true);
+
+ if (unSeqMergeList.isEmpty()) {
+ logger.info("{} no unseq files to be merged", storageGroupName);
+ isUnseqMerging = false;
+ return false;
}
- mergeStartTime = System.currentTimeMillis();
- MergeTask mergeTask =
- new MergeTask(
- mergeResource,
- storageGroupDir,
- this::mergeEndAction,
- taskName,
- fullMerge,
- fileSelector.getConcurrentMergeNum(),
- storageGroupName);
- mergingModification =
- new ModificationFile(storageGroupDir + File.separator + MERGING_MODIFICATION_FILE_NAME);
- MergeManager.getINSTANCE().submitMainTask(mergeTask);
- if (logger.isInfoEnabled()) {
+ if (unSeqMergeList.size() > maxOpenFileNumInEachUnseqCompaction) {
logger.info(
- "{} submits a merge task {}, merging {} seqFiles, {} unseqFiles",
+ "{} too much unseq files to be merged, reduce it to {}",
storageGroupName,
- taskName,
- mergeFiles[0].size(),
- mergeFiles[1].size());
+ maxOpenFileNumInEachUnseqCompaction);
+ unSeqMergeList = unSeqMergeList.subList(0, maxOpenFileNumInEachUnseqCompaction);
}
- // wait until unseq merge has finished
- while (isUnseqMerging) {
- try {
- Thread.sleep(200);
- } catch (InterruptedException e) {
- logger.error("{} [Compaction] shutdown", storageGroupName, e);
- Thread.currentThread().interrupt();
+
+ long budget = IoTDBDescriptor.getInstance().getConfig().getMergeMemoryBudget();
+ long timeLowerBound = System.currentTimeMillis() - dataTTL;
+ MergeResource mergeResource = new MergeResource(seqMergeList, unSeqMergeList, timeLowerBound);
+
+ IMergeFileSelector fileSelector = getMergeFileSelector(budget, mergeResource);
+ try {
+ List[] mergeFiles = fileSelector.select();
+ if (mergeFiles.length == 0) {
+ logger.info(
+ "{} cannot select merge candidates under the budget {}", storageGroupName, budget);
+ isUnseqMerging = false;
return false;
}
+ // avoid pending tasks holds the metadata and streams
+ mergeResource.clear();
+ String taskName = storageGroupName + "-" + System.currentTimeMillis();
+ // do not cache metadata until true candidates are chosen, or too much metadata will be
+ // cached during selection
+ mergeResource.setCacheDeviceMeta(true);
+
+ for (TsFileResource tsFileResource : mergeResource.getSeqFiles()) {
+ tsFileResource.setMerging(true);
+ }
+ for (TsFileResource tsFileResource : mergeResource.getUnseqFiles()) {
+ tsFileResource.setMerging(true);
+ }
+
+ mergeStartTime = System.currentTimeMillis();
+ MergeTask mergeTask =
+ new MergeTask(
+ mergeResource,
+ storageGroupDir,
+ this::mergeEndAction,
+ taskName,
+ fullMerge,
+ fileSelector.getConcurrentMergeNum(),
+ storageGroupName);
+ mergingModification =
+ new ModificationFile(storageGroupDir + File.separator + MERGING_MODIFICATION_FILE_NAME);
+ MergeManager.getINSTANCE().submitMainTask(mergeTask);
+ if (logger.isInfoEnabled()) {
+ logger.info(
+ "{} submits a merge task {}, merging {} seqFiles, {} unseqFiles",
+ storageGroupName,
+ taskName,
+ mergeFiles[0].size(),
+ mergeFiles[1].size());
+ }
+ // wait until unseq merge has finished
+ while (isUnseqMerging) {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ logger.error("{} [Compaction] shutdown", storageGroupName, e);
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ return true;
+ } catch (MergeException | IOException e) {
+ logger.error("{} cannot select file for merge", storageGroupName, e);
+ return false;
}
- return true;
- } catch (MergeException | IOException e) {
- logger.error("{} cannot select file for merge", storageGroupName, e);
- return false;
+ } finally {
+ writeUnlock();
}
}
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 df1add1..bb00989 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
@@ -81,7 +81,8 @@ public class MergeResource {
private boolean filterResource(TsFileResource res) {
return res.getTsFile().exists()
&& !res.isDeleted()
- && (!res.isClosed() || res.stillLives(timeLowerBound));
+ && (!res.isClosed() || res.stillLives(timeLowerBound))
+ && !res.isMerging();
}
public MergeResource(
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/selector/MaxFileMergeFileSelector.java
index 3deb763..d55eaab 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/selector/MaxFileMergeFileSelector.java
@@ -155,7 +155,7 @@ public class MaxFileMergeFileSelector implements IMergeFileSelector {
if (seqSelectedNum != resource.getSeqFiles().size()) {
selectOverlappedSeqFiles(unseqFile);
}
- boolean isClosed = checkClosed(unseqFile);
+ boolean isClosed = checkChoosable(unseqFile);
if (!isClosed) {
tmpSelectedSeqFiles.clear();
unseqIndex++;
@@ -206,18 +206,19 @@ public class MaxFileMergeFileSelector implements IMergeFileSelector {
return false;
}
- private boolean checkClosed(TsFileResource unseqFile) {
- boolean isClosed = unseqFile.isClosed();
- if (!isClosed) {
+ private boolean checkChoosable(TsFileResource unseqFile) {
+ boolean choosable = unseqFile.isClosed() && !unseqFile.isMerging();
+ if (!choosable) {
return false;
}
for (Integer seqIdx : tmpSelectedSeqFiles) {
- if (!resource.getSeqFiles().get(seqIdx).isClosed()) {
- isClosed = false;
+ if (!resource.getSeqFiles().get(seqIdx).isClosed()
+ || resource.getSeqFiles().get(seqIdx).isMerging()) {
+ choosable = false;
break;
}
}
- return isClosed;
+ return choosable;
}
private void selectOverlappedSeqFiles(TsFileResource unseqFile) {
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 16cdb54..6ba652c 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
@@ -522,7 +522,7 @@ public class TsFileResource {
this.deleted = deleted;
}
- boolean isMerging() {
+ public boolean isMerging() {
return isMerging;
}
[iotdb] 01/02: fix bug in compaction (#3626)
Posted by qi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 87fb3d9b69b8b41e54ee6d2b50ad397aeb2eb1cd
Author: liuxuxin <37...@users.noreply.github.com>
AuthorDate: Tue Jul 27 19:52:01 2021 +0800
fix bug in compaction (#3626)
---
.../compaction/CompactionMergeTaskPoolManager.java | 31 +++++---
.../level/LevelCompactionTsFileManagement.java | 92 +++++++++++++---------
.../compaction/LevelCompactionRecoverTest.java | 13 ++-
3 files changed, 83 insertions(+), 53 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionMergeTaskPoolManager.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionMergeTaskPoolManager.java
index 9f7ff5a..e347300 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionMergeTaskPoolManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionMergeTaskPoolManager.java
@@ -34,11 +34,10 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.Collections;
-import java.util.Iterator;
+import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
@@ -54,8 +53,9 @@ public class CompactionMergeTaskPoolManager implements IService {
LoggerFactory.getLogger(CompactionMergeTaskPoolManager.class);
private static final CompactionMergeTaskPoolManager INSTANCE =
new CompactionMergeTaskPoolManager();
- private ScheduledExecutorService pool;
- private Map<String, Set<Future<Void>>> storageGroupTasks = new ConcurrentHashMap<>();
+ private ScheduledExecutorService scheduledPool;
+ private ExecutorService pool;
+ private Map<String, List<Future<Void>>> storageGroupTasks = new ConcurrentHashMap<>();
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static ConcurrentHashMap<String, Boolean> sgCompactionStatus = new ConcurrentHashMap<>();
@@ -68,6 +68,10 @@ public class CompactionMergeTaskPoolManager implements IService {
public void start() {
if (pool == null) {
this.pool =
+ IoTDBThreadPoolFactory.newFixedThreadPool(
+ IoTDBDescriptor.getInstance().getConfig().getCompactionThreadNum(),
+ ThreadName.COMPACTION_SERVICE.getName());
+ this.scheduledPool =
IoTDBThreadPoolFactory.newScheduledThreadPool(
IoTDBDescriptor.getInstance().getConfig().getCompactionThreadNum(),
ThreadName.COMPACTION_SERVICE.getName());
@@ -79,6 +83,7 @@ public class CompactionMergeTaskPoolManager implements IService {
public void stop() {
if (pool != null) {
pool.shutdownNow();
+ scheduledPool.shutdownNow();
logger.info("Waiting for task pool to shut down");
waitTermination();
storageGroupTasks.clear();
@@ -88,6 +93,7 @@ public class CompactionMergeTaskPoolManager implements IService {
@Override
public void waitAndStop(long milliseconds) {
if (pool != null) {
+ awaitTermination(scheduledPool, milliseconds);
awaitTermination(pool, milliseconds);
logger.info("Waiting for task pool to shut down");
waitTermination();
@@ -143,6 +149,7 @@ public class CompactionMergeTaskPoolManager implements IService {
}
}
pool = null;
+ scheduledPool = null;
storageGroupTasks.clear();
logger.info("CompactionManager stopped");
}
@@ -168,17 +175,15 @@ public class CompactionMergeTaskPoolManager implements IService {
* corresponding storage group.
*/
public void abortCompaction(String storageGroup) {
- Set<Future<Void>> subTasks =
- storageGroupTasks.getOrDefault(storageGroup, Collections.emptySet());
- Iterator<Future<Void>> subIterator = subTasks.iterator();
- while (subIterator.hasNext()) {
- Future<Void> next = subIterator.next();
+ List<Future<Void>> subTasks =
+ storageGroupTasks.getOrDefault(storageGroup, Collections.emptyList());
+ for (Future<Void> next : subTasks) {
if (!next.isDone() && !next.isCancelled()) {
next.cancel(true);
sgCompactionStatus.put(storageGroup, false);
}
- subIterator.remove();
}
+ subTasks.clear();
}
public synchronized void clearCompactionStatus(String storageGroupName) {
@@ -190,7 +195,7 @@ public class CompactionMergeTaskPoolManager implements IService {
}
public void init(Runnable function) {
- pool.scheduleWithFixedDelay(
+ scheduledPool.scheduleWithFixedDelay(
function, 1000, config.getCompactionInterval(), TimeUnit.MILLISECONDS);
}
@@ -206,7 +211,7 @@ public class CompactionMergeTaskPoolManager implements IService {
sgCompactionStatus.put(storageGroup, true);
Future<Void> future = pool.submit(storageGroupCompactionTask);
storageGroupTasks
- .computeIfAbsent(storageGroup, k -> new ConcurrentSkipListSet<>())
+ .computeIfAbsent(storageGroup, k -> new CopyOnWriteArrayList<>())
.add(future);
}
}
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 f740c6b..5f260f3 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
@@ -472,6 +472,13 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
} else {
// get tsfile resource from list, as they have been recovered in StorageGroupProcessor
TsFileResource targetResource = getRecoverTsFileResource(targetFile, isSeq);
+ if (targetResource == null) {
+ // new file already merged but old file not deleted
+ targetResource = getTsFileResource(targetFile, isSeq);
+ if (targetResource == null) {
+ throw new IOException();
+ }
+ }
long timePartition = targetResource.getTimePartition();
List<TsFileResource> sourceTsFileResources = new ArrayList<>();
for (String file : sourceFileList) {
@@ -484,6 +491,7 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
}
int level = TsFileResource.getMergeLevel(new File(sourceFileList.get(0)).getName());
RestorableTsFileIOWriter writer = new RestorableTsFileIOWriter(target);
+ List<Modification> modifications = new ArrayList<>();
// if not complete compaction, resume merge
if (writer.hasCrashed()) {
if (offset > 0) {
@@ -492,7 +500,6 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
writer.close();
CompactionLogger compactionLogger =
new CompactionLogger(storageGroupDir, storageGroupName);
- List<Modification> modifications = new ArrayList<>();
CompactionUtils.merge(
targetResource,
sourceTsFileResources,
@@ -501,31 +508,39 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
deviceSet,
isSeq,
modifications);
- // complete compaction and delete source file
- writeLock();
- try {
- if (Thread.currentThread().isInterrupted()) {
- throw new InterruptedException(
- String.format("%s [Compaction] abort", storageGroupName));
- }
- int targetLevel = TsFileResource.getMergeLevel(targetResource.getTsFile().getName());
- if (isSeq) {
- sequenceTsFileResources.get(timePartition).get(targetLevel).add(targetResource);
- sequenceRecoverTsFileResources.clear();
- } else {
- unSequenceTsFileResources.get(timePartition).get(targetLevel).add(targetResource);
- unSequenceRecoverTsFileResources.clear();
- }
- deleteLevelFilesInList(timePartition, sourceTsFileResources, level, isSeq);
- } finally {
- writeUnlock();
- }
- deleteLevelFilesInDisk(sourceTsFileResources);
- renameLevelFilesMods(modifications, sourceTsFileResources, targetResource);
compactionLogger.close();
+ // complete compaction and add target tsfile
+ int targetLevel = TsFileResource.getMergeLevel(targetResource.getTsFile().getName());
+ if (isSeq) {
+ sequenceTsFileResources.get(timePartition).get(targetLevel).add(targetResource);
+ sequenceRecoverTsFileResources.clear();
+ } else {
+ unSequenceTsFileResources.get(timePartition).get(targetLevel).add(targetResource);
+ unSequenceRecoverTsFileResources.clear();
+ }
} else {
+ // complete compaction, just close writer
writer.close();
}
+ // complete compaction, delete source files
+ writeLock();
+ try {
+ if (Thread.currentThread().isInterrupted()) {
+ throw new InterruptedException(
+ String.format("%s [Compaction] abort", storageGroupName));
+ }
+ deleteLevelFilesInList(timePartition, sourceTsFileResources, level, isSeq);
+ } finally {
+ writeUnlock();
+ }
+ for (TsFileResource tsFileResource : sourceTsFileResources) {
+ logger.warn(
+ "{} recover storage group delete source file {}",
+ storageGroupName,
+ tsFileResource.getTsFile().getName());
+ }
+ deleteLevelFilesInDisk(sourceTsFileResources);
+ renameLevelFilesMods(modifications, sourceTsFileResources, targetResource);
}
}
} catch (IOException | IllegalPathException | InterruptedException e) {
@@ -664,8 +679,10 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
Long.MAX_VALUE);
} else {
compactionLogger = new CompactionLogger(storageGroupDir, storageGroupName);
+ List<TsFileResource> toMergeTsFiles =
+ mergeResources.get(i).subList(0, currMaxFileNumInEachLevel);
// log source file list and target file for recover
- for (TsFileResource mergeResource : mergeResources.get(i)) {
+ for (TsFileResource mergeResource : toMergeTsFiles) {
mergeResource.setMerging(true);
compactionLogger.logFile(SOURCE_NAME, mergeResource.getTsFile());
}
@@ -673,8 +690,6 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
TsFileResource.modifyTsFileNameMergeCnt(mergeResources.get(i).get(0).getTsFile());
compactionLogger.logSequence(sequence);
compactionLogger.logFile(TARGET_NAME, newLevelFile);
- List<TsFileResource> toMergeTsFiles =
- mergeResources.get(i).subList(0, currMaxFileNumInEachLevel);
logger.info(
"{} [Compaction] merge level-{}'s {} TsFiles to next level",
storageGroupName,
@@ -789,21 +804,24 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
private TsFileResource getRecoverTsFileResource(String filePath, boolean isSeq)
throws IOException {
- if (isSeq) {
- for (TsFileResource tsFileResource : sequenceRecoverTsFileResources) {
- if (Files.isSameFile(tsFileResource.getTsFile().toPath(), new File(filePath).toPath())) {
- return tsFileResource;
+ try {
+ if (isSeq) {
+ for (TsFileResource tsFileResource : sequenceRecoverTsFileResources) {
+ if (Files.isSameFile(tsFileResource.getTsFile().toPath(), new File(filePath).toPath())) {
+ return tsFileResource;
+ }
}
- }
- } else {
- for (TsFileResource tsFileResource : unSequenceRecoverTsFileResources) {
- if (Files.isSameFile(tsFileResource.getTsFile().toPath(), new File(filePath).toPath())) {
- return tsFileResource;
+ } else {
+ for (TsFileResource tsFileResource : unSequenceRecoverTsFileResources) {
+ if (Files.isSameFile(tsFileResource.getTsFile().toPath(), new File(filePath).toPath())) {
+ return tsFileResource;
+ }
}
}
+ } catch (IOException e) {
+ logger.error("cannot get tsfile resource path: {}", filePath);
}
- logger.error("cannot get tsfile resource path: {}", filePath);
- throw new IOException();
+ return null;
}
private TsFileResource getTsFileResource(String filePath, boolean isSeq) {
@@ -868,6 +886,8 @@ public class LevelCompactionTsFileManagement extends TsFileManagement {
if (targetFilePath != null) {
File targetFile = new File(targetFilePath);
if (targetFile.exists()) {
+ logger.error(
+ "{} restore delete target file {} ", storageGroupName, targetFile.getName());
targetFile.delete();
}
}
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
index 7adffc2..cc2cb39 100644
--- 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
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.engine.compaction;
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.compaction.level.LevelCompactionTsFileManagement;
import org.apache.iotdb.db.engine.compaction.utils.CompactionLogger;
@@ -139,7 +140,7 @@ public class LevelCompactionRecoverTest extends LevelCompactionTest {
true,
new ArrayList<>());
compactionLogger.close();
- levelCompactionTsFileManagement.addRecover(targetTsFileResource, true);
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
levelCompactionTsFileManagement.recover();
context = new QueryContext();
path =
@@ -254,7 +255,7 @@ public class LevelCompactionRecoverTest extends LevelCompactionTest {
}
logStream.close();
- levelCompactionTsFileManagement.addRecover(targetTsFileResource, true);
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
levelCompactionTsFileManagement.recover();
context = new QueryContext();
path =
@@ -408,6 +409,9 @@ public class LevelCompactionRecoverTest extends LevelCompactionTest {
@Test
public void testCompactionMergeRecoverMergeFinishedUnseq()
throws IOException, IllegalPathException {
+ int prevUnseqLevelNum = IoTDBDescriptor.getInstance().getConfig().getUnseqLevelNum();
+ IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(2);
+
LevelCompactionTsFileManagement levelCompactionTsFileManagement =
new LevelCompactionTsFileManagement(COMPACTION_TEST_SG, tempSGDir.getPath());
levelCompactionTsFileManagement.addAll(seqResources, true);
@@ -466,7 +470,7 @@ public class LevelCompactionRecoverTest extends LevelCompactionTest {
false,
new ArrayList<>());
compactionLogger.close();
- levelCompactionTsFileManagement.addRecover(targetTsFileResource, false);
+ levelCompactionTsFileManagement.add(targetTsFileResource, false);
levelCompactionTsFileManagement.recover();
context = new QueryContext();
path =
@@ -493,6 +497,7 @@ public class LevelCompactionRecoverTest extends LevelCompactionTest {
}
}
assertEquals(500, count);
+ IoTDBDescriptor.getInstance().getConfig().setUnseqLevelNum(prevUnseqLevelNum);
}
/** compaction recover merge start just log source file */
@@ -671,7 +676,7 @@ public class LevelCompactionRecoverTest extends LevelCompactionTest {
new HashSet<>(),
true,
new ArrayList<>());
- levelCompactionTsFileManagement.addRecover(targetTsFileResource, true);
+ levelCompactionTsFileManagement.add(targetTsFileResource, true);
compactionLogger.close();
levelCompactionTsFileManagement.recover();
QueryContext context = new QueryContext();