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 2022/03/06 09:50:14 UTC
[iotdb] branch master updated: [IOTDB-2633]Unification of compaction log formats (#5157)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new d2c21e3 [IOTDB-2633]Unification of compaction log formats (#5157)
d2c21e3 is described below
commit d2c21e32476be8cf611c9d9e3627a6227e6458ec
Author: 周沛辰 <45...@users.noreply.github.com>
AuthorDate: Sun Mar 6 17:48:45 2022 +0800
[IOTDB-2633]Unification of compaction log formats (#5157)
---
.../db/engine/compaction/TsFileIdentifier.java | 41 +-
.../CrossSpaceCompactionExceptionHandler.java | 71 ++--
.../RewriteCrossSpaceCompactionLogAnalyzer.java | 150 --------
.../task/RewriteCrossCompactionRecoverTask.java | 38 +-
.../task/RewriteCrossSpaceCompactionTask.java | 29 +-
.../SizeTieredCompactionRecoverTask.java | 102 +++--
.../inner/sizetiered/SizeTieredCompactionTask.java | 28 +-
.../inner/utils/InnerSpaceCompactionUtils.java | 3 +-
.../utils/SizeTieredCompactionLogAnalyzer.java | 101 -----
.../inner/utils/SizeTieredCompactionLogger.java | 87 -----
.../compaction/task/CompactionRecoverTask.java | 7 +-
.../utils/log/CompactionLogAnalyzer.java | 163 ++++++++
.../log/CompactionLogger.java} | 55 ++-
.../storagegroup/VirtualStorageGroupProcessor.java | 4 +-
.../cross/CrossSpaceCompactionExceptionTest.java | 73 ++--
.../RewriteCrossSpaceCompactionRecoverTest.java | 279 +++++++++++---
.../compaction/inner/InnerCompactionLogTest.java | 4 +-
.../inner/InnerSpaceCompactionExceptionTest.java | 101 +++--
.../inner/InnerSpaceCompactionUtilsOldTest.java | 14 +-
.../SizeTieredCompactionRecoverTest.java | 393 ++++++++++++++-----
...eCrossSpaceCompactionRecoverCompatibleTest.java | 414 +++++++++++++++++++++
.../SizeTieredCompactionRecoverCompatibleTest.java | 8 +-
.../recover/SizeTieredCompactionRecoverTest.java | 83 ++---
.../compaction/utils/CompactionClearUtils.java | 7 +-
24 files changed, 1440 insertions(+), 815 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java
index e5221c3..a011c63 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/TsFileIdentifier.java
@@ -43,11 +43,17 @@ public class TsFileIdentifier {
public static final int LOGICAL_SG_OFFSET_IN_PATH = 4;
public static final int SEQUENCE_OFFSET_IN_PATH = 5;
- public static final int LOGICAL_SG_OFFSET_IN_LOG = 0;
- public static final int VIRTUAL_SG_OFFSET_IN_LOG = 1;
- public static final int TIME_PARTITION_OFFSET_IN_LOG = 2;
- public static final int FILE_NAME_OFFSET_IN_LOG = 3;
- public static final int SEQUENCE_OFFSET_IN_LOG = 4;
+ public static final int SEQUENCE_OFFSET_IN_LOG = 0;
+ public static final int LOGICAL_SG_OFFSET_IN_LOG = 1;
+ public static final int VIRTUAL_SG_OFFSET_IN_LOG = 2;
+ public static final int TIME_PARTITION_OFFSET_IN_LOG = 3;
+ public static final int FILE_NAME_OFFSET_IN_LOG = 4;
+
+ private static final int LOGICAL_SG_OFFSET_IN_LOG_FROM_OLD = 0;
+ private static final int VIRTUAL_SG_OFFSET_IN_LOG_FROM_OLD = 1;
+ private static final int TIME_PARTITION_OFFSET_IN_LOG_FROM_OLD = 2;
+ private static final int FILE_NAME_OFFSET_IN_LOG_FROM_OLD = 3;
+ private static final int SEQUENCE_OFFSET_IN_LOG_FROM_OLD = 4;
private TsFileIdentifier(
String logicalStorageGroupName,
@@ -94,7 +100,8 @@ public class TsFileIdentifier {
/**
* This function generates an instance of CompactionFileIdentifier by parsing the info string of a
- * tsfile(usually recorded in a compaction.log), such as “root.test.sg 0 0 true 1-1-0-0.tsfile"
+ * tsfile(usually recorded in a compaction.log), such as “sequence root.test.sg 0 0
+ * 0-0-0-0.tsfile"
*/
public static TsFileIdentifier getFileIdentifierFromInfoString(String infoString) {
String[] splittedFileInfo = infoString.split(INFO_SEPARATOR);
@@ -113,8 +120,9 @@ public class TsFileIdentifier {
/**
* This function generates an instance of CompactionFileIdentifier by parsing the old info string
- * from previous version (<0.13) of a tsfile(usually recorded in a compaction.log), such as
- * “root.test.sg 0 0 1-1-0-0.tsfile true"
+ * from previous version (<0.13) of a tsfile(usually recorded in a compaction.log). Such as
+ * “root.test.sg 0 0 1-1-0-0.tsfile true" from old cross space compaction log and "root.test.sg 0
+ * 0 1-1-0-0.tsfile sequence" from old inner space compaction log.
*/
public static TsFileIdentifier getFileIdentifierFromOldInfoString(String oldInfoString) {
String[] splittedFileInfo = oldInfoString.split(INFO_SEPARATOR);
@@ -126,26 +134,27 @@ public class TsFileIdentifier {
oldInfoString));
}
return new TsFileIdentifier(
- splittedFileInfo[LOGICAL_SG_OFFSET_IN_LOG],
- splittedFileInfo[VIRTUAL_SG_OFFSET_IN_LOG],
- splittedFileInfo[TIME_PARTITION_OFFSET_IN_LOG],
- Boolean.parseBoolean(splittedFileInfo[SEQUENCE_OFFSET_IN_LOG]),
- splittedFileInfo[FILE_NAME_OFFSET_IN_LOG]);
+ splittedFileInfo[LOGICAL_SG_OFFSET_IN_LOG_FROM_OLD],
+ splittedFileInfo[VIRTUAL_SG_OFFSET_IN_LOG_FROM_OLD],
+ splittedFileInfo[TIME_PARTITION_OFFSET_IN_LOG_FROM_OLD],
+ splittedFileInfo[SEQUENCE_OFFSET_IN_LOG_FROM_OLD].equals("true")
+ || splittedFileInfo[SEQUENCE_OFFSET_IN_LOG_FROM_OLD].equals("sequence"),
+ splittedFileInfo[FILE_NAME_OFFSET_IN_LOG_FROM_OLD]);
}
@Override
public String toString() {
return String.format(
"%s%s%s%s%s%s%s%s%s",
+ sequence ? "sequence" : "unsequence",
+ INFO_SEPARATOR,
logicalStorageGroupName,
INFO_SEPARATOR,
virtualStorageGroupId,
INFO_SEPARATOR,
timePartitionId,
INFO_SEPARATOR,
- filename,
- INFO_SEPARATOR,
- sequence ? "sequence" : "unsequence");
+ filename);
}
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionHandler.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionHandler.java
index 6782e81..3da6de5 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionHandler.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionHandler.java
@@ -21,12 +21,11 @@ package org.apache.iotdb.db.engine.compaction.cross;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.engine.compaction.CompactionUtils;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogAnalyzer;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResourceList;
import org.apache.iotdb.db.rescon.TsFileResourceManager;
+import org.apache.iotdb.tsfile.utils.TsFileUtils;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
@@ -34,7 +33,6 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
@@ -65,11 +63,10 @@ public class CrossSpaceCompactionExceptionHandler {
boolean handleSuccess = true;
- List<TsFileResource> lostSeqFiles = new ArrayList<>();
- List<TsFileResource> lostUnseqFiles = new ArrayList<>();
+ List<TsFileResource> lostSourceFiles = new ArrayList<>();
- boolean allSeqFilesExist = checkAllSourceFileExists(seqResourceList, lostSeqFiles);
- boolean allUnseqFilesExist = checkAllSourceFileExists(unseqResourceList, lostUnseqFiles);
+ boolean allSeqFilesExist = checkAllSourceFileExists(seqResourceList, lostSourceFiles);
+ boolean allUnseqFilesExist = checkAllSourceFileExists(unseqResourceList, lostSourceFiles);
if (allSeqFilesExist && allUnseqFilesExist) {
// all source files exists, remove target file and recover memory
@@ -84,7 +81,11 @@ public class CrossSpaceCompactionExceptionHandler {
} else {
handleSuccess =
handleWhenSomeSourceFilesLost(
- storageGroup, seqResourceList, unseqResourceList, logFile);
+ storageGroup,
+ seqResourceList,
+ unseqResourceList,
+ targetResourceList,
+ lostSourceFiles);
}
if (!handleSuccess) {
@@ -188,40 +189,10 @@ public class CrossSpaceCompactionExceptionHandler {
String storageGroup,
List<TsFileResource> seqFileList,
List<TsFileResource> unseqFileList,
- File logFile)
+ List<TsFileResource> targetFileList,
+ List<TsFileResource> lostSourceFiles)
throws IOException {
- long magicStringLength =
- RewriteCrossSpaceCompactionLogger.MAGIC_STRING.getBytes(StandardCharsets.UTF_8).length;
- long fileLength = logFile.length();
-
- if (fileLength < 2 * magicStringLength) {
- // the log length is less than twice the Magic String length
- // it means the compaction has not finished yet
- LOGGER.error(
- "{} [Compaction][ExceptionHandler] the compaction log length is less than twice "
- + "the MagicString length",
- storageGroup);
- return false;
- }
-
- // read head magic string in compaction log
- RewriteCrossSpaceCompactionLogAnalyzer logAnalyzer =
- new RewriteCrossSpaceCompactionLogAnalyzer(logFile);
- logAnalyzer.analyze();
- if (!logAnalyzer.isFirstMagicStringExisted()) {
- LOGGER.error(
- "{} [Compaction][ExceptionHandler] the head magic string in compaction log is incorrect,"
- + " failed to handle exception",
- storageGroup);
- return false;
- }
-
- // read tail string in compaction log
- if (!logAnalyzer.isEndMagicStringExisted()) {
- LOGGER.error(
- "{} [Compaction][ExceptionHandler] the tail magic string in compaction log is incorrect,"
- + " failed to handle exception",
- storageGroup);
+ if (!checkIsTargetFilesComplete(targetFileList, lostSourceFiles, storageGroup)) {
return false;
}
@@ -240,4 +211,22 @@ public class CrossSpaceCompactionExceptionHandler {
return true;
}
+
+ public static boolean checkIsTargetFilesComplete(
+ List<TsFileResource> targetResources,
+ List<TsFileResource> lostSourceResources,
+ String fullStorageGroupName)
+ throws IOException {
+ for (TsFileResource targetResource : targetResources) {
+ if (!TsFileUtils.isTsFileComplete(targetResource.getTsFile())) {
+ LOGGER.error(
+ "{} [Compaction][ExceptionHandler] target file {} is not complete, and some source files {} is lost, do nothing. Set allowCompaction to false",
+ fullStorageGroupName,
+ targetResource,
+ lostSourceResources);
+ return false;
+ }
+ }
+ return true;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/recover/RewriteCrossSpaceCompactionLogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/recover/RewriteCrossSpaceCompactionLogAnalyzer.java
deleted file mode 100644
index e6a884c..0000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/recover/RewriteCrossSpaceCompactionLogAnalyzer.java
+++ /dev/null
@@ -1,150 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.db.engine.compaction.cross.rewrite.recover;
-
-import org.apache.iotdb.db.conf.IoTDBConstant;
-import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
-import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.InvalidPropertiesFormatException;
-import java.util.List;
-
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.MAGIC_STRING;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_SEQ_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_TARGET_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_UNSEQ_FILES;
-
-public class RewriteCrossSpaceCompactionLogAnalyzer {
-
- private final File logFile;
- private final List<TsFileIdentifier> sourceFileInfos = new ArrayList<>();
- private final List<TsFileIdentifier> targetFileInfos = new ArrayList<>();
- private boolean isFirstMagicStringExisted = false;
- private boolean isEndMagicStringExisted = false;
- private boolean isLogFromOld = false;
-
- public RewriteCrossSpaceCompactionLogAnalyzer(File logFile) {
- this.logFile = logFile;
- }
-
- /** @return analyze (source file list, target file) */
- public void analyze() throws IOException {
- String currLine;
- try (BufferedReader bufferedReader = new BufferedReader(new FileReader(logFile))) {
- if ((currLine = bufferedReader.readLine()) != null) {
- switch (currLine) {
- case MAGIC_STRING:
- // compaction log of version 0.13
- isFirstMagicStringExisted = true;
- analyzeLog(bufferedReader);
- break;
- case STR_SEQ_FILES:
- // compaction log of version < 0.13
- isLogFromOld = true;
- analyzeOldLog(bufferedReader);
- break;
- default:
- throw new InvalidPropertiesFormatException(
- "Unsupported string in cross space log :" + logFile.getAbsolutePath());
- }
- }
- }
- }
-
- /** Analyze cross space compaction log of version 0.13. */
- private void analyzeLog(BufferedReader bufferedReader) throws IOException {
- String currLine;
- boolean isTargetFile = false;
- while ((currLine = bufferedReader.readLine()) != null) {
- switch (currLine) {
- case MAGIC_STRING:
- isEndMagicStringExisted = true;
- break;
- case STR_TARGET_FILES:
- isTargetFile = true;
- break;
- case STR_SEQ_FILES:
- case STR_UNSEQ_FILES:
- isTargetFile = false;
- break;
- default:
- analyzeFilePath(isTargetFile, currLine);
- break;
- }
- }
- }
-
- /** Analyze cross space compaction log of previous version (<0.13). */
- private void analyzeOldLog(BufferedReader bufferedReader) throws IOException {
- String currLine;
- boolean isSeqSource = true;
- while ((currLine = bufferedReader.readLine()) != null) {
- if (currLine.equals(STR_UNSEQ_FILES)) {
- isSeqSource = false;
- continue;
- }
- analyzeOldFilePath(isSeqSource, currLine);
- }
- }
-
- private void analyzeFilePath(boolean isTargetFile, String filePath) {
- if (isTargetFile) {
- targetFileInfos.add(TsFileIdentifier.getFileIdentifierFromInfoString(filePath));
- } else {
- sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromInfoString(filePath));
- }
- }
-
- private void analyzeOldFilePath(boolean isSeqSource, String oldFilePath) {
- sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromOldInfoString(oldFilePath));
- if (isSeqSource) {
- String targetFilePath =
- oldFilePath.replace(
- TsFileConstant.TSFILE_SUFFIX,
- TsFileConstant.TSFILE_SUFFIX
- + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD);
- targetFileInfos.add(TsFileIdentifier.getFileIdentifierFromOldInfoString(targetFilePath));
- }
- }
-
- public List<TsFileIdentifier> getSourceFileInfos() {
- return sourceFileInfos;
- }
-
- public List<TsFileIdentifier> getTargetFileInfos() {
- return targetFileInfos;
- }
-
- public boolean isEndMagicStringExisted() {
- return isEndMagicStringExisted;
- }
-
- public boolean isFirstMagicStringExisted() {
- return isFirstMagicStringExisted;
- }
-
- public boolean isLogFromOld() {
- return isLogFromOld;
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossCompactionRecoverTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossCompactionRecoverTask.java
index 1cabb2e..9fd2c7d 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossCompactionRecoverTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossCompactionRecoverTask.java
@@ -22,9 +22,10 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.CompactionUtils;
import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogAnalyzer;
import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogAnalyzer;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
@@ -81,9 +82,13 @@ public class RewriteCrossCompactionRecoverTask extends RewriteCrossSpaceCompacti
"{} [Compaction][Recover] cross space compaction log file {} exists, start to recover it",
fullStorageGroupName,
compactionLogFile);
- RewriteCrossSpaceCompactionLogAnalyzer logAnalyzer =
- new RewriteCrossSpaceCompactionLogAnalyzer(compactionLogFile);
- logAnalyzer.analyze();
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ if (isOldLog()) {
+ // log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
@@ -170,14 +175,14 @@ public class RewriteCrossCompactionRecoverTask extends RewriteCrossSpaceCompacti
.replace(
IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
TsFileConstant.TSFILE_SUFFIX));
- TsFileResource targetResource;
+ TsFileResource targetResource = null;
if (tmpTargetFile != null) {
targetResource = new TsFileResource(tmpTargetFile);
- } else {
+ } else if (targetFile != null) {
targetResource = new TsFileResource(targetFile);
}
- if (!targetResource.remove()) {
+ if (targetResource != null && !targetResource.remove()) {
// failed to remove tmp target tsfile
// system should not carry out the subsequent compaction in case of data redundant
LOGGER.warn(
@@ -217,6 +222,18 @@ public class RewriteCrossCompactionRecoverTask extends RewriteCrossSpaceCompacti
File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
if (sourceFile != null) {
remainSourceTsFileResources.add(new TsFileResource(sourceFile));
+ } else {
+ // if source file does not exist, its resource file may still exist, so delete it.
+ File resourceFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + TsFileResource.RESOURCE_SUFFIX);
+ if (resourceFile != null && !resourceFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ resourceFile);
+ handleSuccess = false;
+ }
}
// delete .compaction.mods file and .mods file of all source files
File compactionModFile =
@@ -397,7 +414,7 @@ public class RewriteCrossCompactionRecoverTask extends RewriteCrossSpaceCompacti
return true;
}
- public void appendCompactionModificationsFromOld(
+ public static void appendCompactionModificationsFromOld(
TsFileResource resource, ModificationFile compactionModsFile) throws IOException {
if (compactionModsFile != null) {
@@ -439,4 +456,9 @@ public class RewriteCrossCompactionRecoverTask extends RewriteCrossSpaceCompacti
public boolean checkValidAndSetMerging() {
return compactionLogFile.exists();
}
+
+ /** Return whether compaction log file is from previous version (<0.13). */
+ private boolean isOldLog() {
+ return compactionLogFile.getName().equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java
index 313bf9f..f06011f 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java
@@ -22,8 +22,8 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.engine.compaction.CompactionUtils;
import org.apache.iotdb.db.engine.compaction.cross.AbstractCrossSpaceCompactionTask;
import org.apache.iotdb.db.engine.compaction.cross.CrossSpaceCompactionExceptionHandler;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger;
import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -43,11 +43,8 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.apache.iotdb.db.conf.IoTDBConstant.PATH_SEPARATOR;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.MAGIC_STRING;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_SEQ_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_TARGET_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_UNSEQ_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SOURCE_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_TARGET_FILES;
public class RewriteCrossSpaceCompactionTask extends AbstractCrossSpaceCompactionTask {
private static final Logger logger =
@@ -134,25 +131,21 @@ public class RewriteCrossSpaceCompactionTask extends AbstractCrossSpaceCompactio
selectedSeqTsFileResourceList.get(0).getTsFile().getParent()
+ File.separator
+ targetTsfileResourceList.get(0).getTsFile().getName()
- + PATH_SEPARATOR
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
- try (RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(logFile)) {
+ try (CompactionLogger compactionLogger = new CompactionLogger(logFile)) {
// print the path of the temporary file first for priority check during recovery
+ compactionLogger.logFiles(selectedSeqTsFileResourceList, STR_SOURCE_FILES);
+ compactionLogger.logFiles(selectedUnSeqTsFileResourceList, STR_SOURCE_FILES);
compactionLogger.logFiles(targetTsfileResourceList, STR_TARGET_FILES);
- compactionLogger.logFiles(selectedSeqTsFileResourceList, STR_SEQ_FILES);
- compactionLogger.logFiles(selectedUnSeqTsFileResourceList, STR_UNSEQ_FILES);
- CompactionUtils.compact(
- selectedSeqTsFileResourceList, selectedUnSeqTsFileResourceList, targetTsfileResourceList);
-
- CompactionUtils.moveTargetFile(targetTsfileResourceList, false, fullStorageGroupName);
-
// indicates that the cross compaction is complete and the result can be reused during a
// restart recovery
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
+ CompactionUtils.compact(
+ selectedSeqTsFileResourceList, selectedUnSeqTsFileResourceList, targetTsfileResourceList);
+
+ CompactionUtils.moveTargetFile(targetTsfileResourceList, false, fullStorageGroupName);
CompactionUtils.combineModsInCompaction(
selectedSeqTsFileResourceList, selectedUnSeqTsFileResourceList, targetTsfileResourceList);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java
index 78d709d..97ae5c1 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTask.java
@@ -20,11 +20,13 @@ package org.apache.iotdb.db.engine.compaction.inner.sizetiered;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+import org.apache.iotdb.db.engine.compaction.cross.rewrite.task.RewriteCrossCompactionRecoverTask;
import org.apache.iotdb.db.engine.compaction.inner.InnerSpaceCompactionExceptionHandler;
import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
-import org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogAnalyzer;
import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogAnalyzer;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -37,6 +39,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
@@ -98,11 +101,15 @@ public class SizeTieredCompactionRecoverTask extends SizeTieredCompactionTask {
logicalStorageGroupName,
virtualStorageGroup,
compactionLogFile);
- SizeTieredCompactionLogAnalyzer logAnalyzer =
- new SizeTieredCompactionLogAnalyzer(compactionLogFile);
- logAnalyzer.analyze();
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ if (isOldLog()) {
+ // log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
- TsFileIdentifier targetFileIdentifier = logAnalyzer.getTargetFileInfo();
+ TsFileIdentifier targetFileIdentifier = logAnalyzer.getTargetFileInfos().get(0);
// compaction log file is incomplete
if (targetFileIdentifier == null || sourceFileIdentifiers.isEmpty()) {
@@ -123,27 +130,39 @@ public class SizeTieredCompactionRecoverTask extends SizeTieredCompactionTask {
}
}
+ TsFileResource targetResource = null;
+ // xxx.target
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
+ }
+
if (isAllSourcesFileExisted) {
- // xxx.target
- File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
- // xxx.tsfile
- File targetFile =
- getFileFromDataDirs(
- targetFileIdentifier
- .getFilePath()
- .replace(
- IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
- TsFileConstant.TSFILE_SUFFIX));
- TsFileResource targetResource;
- if (tmpTargetFile != null) {
- targetResource = new TsFileResource(tmpTargetFile);
- } else {
- targetResource = new TsFileResource(targetFile);
- }
List<TsFileResource> sourceResources = new ArrayList<>();
for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
sourceResources.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
}
+ if (targetResource == null) {
+ // target file may not exist
+ LOGGER.info(
+ "{}-{} [Compaction][Recover] Target file {} does not exist.",
+ logicalStorageGroupName,
+ virtualStorageGroup,
+ targetFileIdentifier.getFilePath());
+ // delete compaction mods files
+ CompactionUtils.deleteCompactionModsFile(sourceResources, Collections.emptyList());
+ return;
+ }
handleSuccess =
InnerSpaceCompactionExceptionHandler.handleWhenAllSourceFilesExist(
fullStorageGroupName,
@@ -154,6 +173,9 @@ public class SizeTieredCompactionRecoverTask extends SizeTieredCompactionTask {
true);
} else {
handleSuccess = handleWithoutAllSourceFilesExist(sourceFileIdentifiers);
+ if (logAnalyzer.isLogFromOld()) {
+ appendCompactionModificationsFromOld(targetResource);
+ }
}
}
} catch (IOException e) {
@@ -211,6 +233,19 @@ public class SizeTieredCompactionRecoverTask extends SizeTieredCompactionTask {
File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
if (sourceFile != null) {
remainSourceTsFileResources.add(new TsFileResource(sourceFile));
+ } else {
+ // if source file does not exist, its resource file may still exist, so delete it.
+ File resourceFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + TsFileResource.RESOURCE_SUFFIX);
+ if (resourceFile != null && !resourceFile.delete()) {
+ LOGGER.error(
+ "{}-{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ logicalStorageGroupName,
+ virtualStorageGroup,
+ resourceFile);
+ handleSuccess = false;
+ }
}
// delete .compaction.mods file and .mods file of all source files
File compactionModFile =
@@ -261,4 +296,29 @@ public class SizeTieredCompactionRecoverTask extends SizeTieredCompactionTask {
}
return null;
}
+
+ /**
+ * Append compaction modifications from previous version (<0.13) to target mods file.
+ *
+ * @param targetResource
+ */
+ private void appendCompactionModificationsFromOld(TsFileResource targetResource)
+ throws IOException {
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists()) {
+ ModificationFile compactionModsFile =
+ new ModificationFile(compactionModsFileFromOld.getPath());
+ RewriteCrossCompactionRecoverTask.appendCompactionModificationsFromOld(
+ targetResource, compactionModsFile);
+ }
+ }
+
+ /** Return whether compaction log file is from previous version (<0.13). */
+ private boolean isOldLog() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
index 854fe42..fb6bfbe 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
@@ -23,8 +23,8 @@ import org.apache.iotdb.db.engine.compaction.CompactionUtils;
import org.apache.iotdb.db.engine.compaction.inner.AbstractInnerSpaceCompactionTask;
import org.apache.iotdb.db.engine.compaction.inner.InnerSpaceCompactionExceptionHandler;
import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
-import org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger;
import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -40,9 +40,6 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.SOURCE_INFO;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.TARGET_INFO;
-
/**
* SizeTiredCompactionTask compact several inner space files selected by {@link
* SizeTieredCompactionSelector} into one file.
@@ -104,16 +101,16 @@ public class SizeTieredCompactionTask extends AbstractInnerSpaceCompactionTask {
dataDirectory
+ File.separator
+ targetTsFileResource.getTsFile().getName()
- + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger sizeTieredCompactionLogger = null;
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger sizeTieredCompactionLogger = null;
try {
- sizeTieredCompactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
-
- for (TsFileResource resource : selectedTsFileResourceList) {
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, resource.getTsFile());
- }
- sizeTieredCompactionLogger.logSequence(sequence);
- sizeTieredCompactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
+ sizeTieredCompactionLogger = new CompactionLogger(logFile);
+ sizeTieredCompactionLogger.logFiles(
+ selectedTsFileResourceList, CompactionLogger.STR_SOURCE_FILES);
+ sizeTieredCompactionLogger.logFiles(
+ Collections.singletonList(targetTsFileResource), CompactionLogger.STR_TARGET_FILES);
+ LOGGER.info("{} [SizeTiredCompactionTask] Close the logger", fullStorageGroupName);
+ sizeTieredCompactionLogger.close();
LOGGER.info(
"{} [Compaction] compaction with {}", fullStorageGroupName, selectedTsFileResourceList);
@@ -133,11 +130,6 @@ public class SizeTieredCompactionTask extends AbstractInnerSpaceCompactionTask {
InnerSpaceCompactionUtils.combineModsInCompaction(
selectedTsFileResourceList, targetTsFileResource);
- LOGGER.info(
- "{} [SizeTiredCompactionTask] compact finish, close the logger, edit the tsFileResourceList",
- fullStorageGroupName);
- sizeTieredCompactionLogger.close();
-
if (Thread.currentThread().isInterrupted()) {
throw new InterruptedException(
String.format("%s [Compaction] abort", fullStorageGroupName));
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/InnerSpaceCompactionUtils.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/InnerSpaceCompactionUtils.java
index fab7ba7..432cc54 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/InnerSpaceCompactionUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/InnerSpaceCompactionUtils.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.engine.compaction.cross.CrossCompactionStrategy;
import org.apache.iotdb.db.engine.compaction.cross.rewrite.manage.CrossSpaceCompactionResource;
import org.apache.iotdb.db.engine.compaction.cross.rewrite.selector.ICrossSpaceMergeFileSelector;
import org.apache.iotdb.db.engine.compaction.cross.rewrite.selector.RewriteCompactionFileSelector;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
@@ -245,7 +246,7 @@ public class InnerSpaceCompactionUtils {
File timePartitionDir = new File(directory);
if (timePartitionDir.exists()) {
return timePartitionDir.listFiles(
- (dir, name) -> name.endsWith(SizeTieredCompactionLogger.COMPACTION_LOG_NAME));
+ (dir, name) -> name.endsWith(CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX));
} else {
return new File[0];
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/SizeTieredCompactionLogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/SizeTieredCompactionLogAnalyzer.java
deleted file mode 100644
index 9948091..0000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/SizeTieredCompactionLogAnalyzer.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.engine.compaction.inner.utils;
-
-import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.*;
-
-public class SizeTieredCompactionLogAnalyzer {
-
- private File logFile;
- private List<String> sourceFiles = new ArrayList<>();
- private List<TsFileIdentifier> sourceFileInfos = new ArrayList<>();
- private TsFileIdentifier targetFileInfo = null;
- private String targetFile = null;
- private boolean isSeq = false;
-
- public SizeTieredCompactionLogAnalyzer(File logFile) {
- this.logFile = logFile;
- }
-
- /** @return analyze (source file list, target file) */
- public void analyze() throws IOException {
- String currLine;
- try (BufferedReader bufferedReader = new BufferedReader(new FileReader(logFile))) {
- while ((currLine = bufferedReader.readLine()) != null) {
- switch (currLine) {
- // we parse the compaction log with file path to keep compatible with 0.12
- case SOURCE_NAME:
- currLine = bufferedReader.readLine();
- sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromFilePath(currLine));
- break;
- case TARGET_NAME:
- currLine = bufferedReader.readLine();
- targetFileInfo = TsFileIdentifier.getFileIdentifierFromFilePath(currLine);
- break;
- case SOURCE_INFO:
- currLine = bufferedReader.readLine();
- sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromInfoString(currLine));
- break;
- case TARGET_INFO:
- currLine = bufferedReader.readLine();
- targetFileInfo = TsFileIdentifier.getFileIdentifierFromInfoString(currLine);
- break;
- case SEQUENCE_NAME:
- isSeq = true;
- break;
- case UNSEQUENCE_NAME:
- isSeq = false;
- break;
- default:
- break;
- }
- }
- }
- }
-
- public List<String> getSourceFiles() {
- return sourceFiles;
- }
-
- public List<TsFileIdentifier> getSourceFileInfos() {
- return sourceFileInfos;
- }
-
- public TsFileIdentifier getTargetFileInfo() {
- return targetFileInfo;
- }
-
- public String getTargetFile() {
- return targetFile;
- }
-
- public boolean isSeq() {
- return isSeq;
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/SizeTieredCompactionLogger.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/SizeTieredCompactionLogger.java
deleted file mode 100644
index bc8d36b..0000000
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/utils/SizeTieredCompactionLogger.java
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.engine.compaction.inner.utils;
-
-import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
-import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
-
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-
-public class SizeTieredCompactionLogger {
-
- public static final String COMPACTION_LOG_NAME = ".compaction.log";
- public static final String SOURCE_NAME = "source";
- public static final String TARGET_NAME = "target";
- public static final String SOURCE_INFO = "info-source";
- public static final String TARGET_INFO = "info-target";
- public static final String SEQUENCE_NAME = "sequence";
- public static final String UNSEQUENCE_NAME = "unsequence";
-
- private BufferedWriter logStream;
-
- public SizeTieredCompactionLogger(String storageGroupDir, String storageGroupName)
- throws IOException {
- logStream =
- new BufferedWriter(
- new FileWriter(
- SystemFileFactory.INSTANCE.getFile(
- storageGroupDir, storageGroupName + COMPACTION_LOG_NAME),
- true));
- }
-
- public SizeTieredCompactionLogger(String logFile) throws IOException {
- logStream =
- new BufferedWriter(new FileWriter(SystemFileFactory.INSTANCE.getFile(logFile), true));
- }
-
- public void close() throws IOException {
- logStream.close();
- }
-
- public void logFileInfo(String prefix, File file) throws IOException {
- logStream.write(prefix);
- logStream.newLine();
- logStream.write(
- TsFileIdentifier.getFileIdentifierFromFilePath(file.getAbsolutePath()).toString());
- logStream.newLine();
- logStream.flush();
- }
-
- public void logFile(String prefix, File file) throws IOException {
- logStream.write(prefix);
- logStream.newLine();
- logStream.write(file.getPath());
- logStream.newLine();
- logStream.flush();
- }
-
- public void logSequence(boolean isSeq) throws IOException {
- if (isSeq) {
- logStream.write(SEQUENCE_NAME);
- } else {
- logStream.write(UNSEQUENCE_NAME);
- }
- logStream.newLine();
- logStream.flush();
- }
-}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
index 7b6b816..f4d19bf 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.engine.compaction.task;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.DirectoryManager;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.slf4j.Logger;
@@ -79,8 +79,7 @@ public class CompactionRecoverTask {
continue;
}
File[] compactionLogs =
- RewriteCrossSpaceCompactionLogger.findCrossSpaceCompactionLogs(
- timePartitionDir.getPath());
+ CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
for (File compactionLog : compactionLogs) {
logger.info("calling cross compaction task");
IoTDBDescriptor.getInstance()
@@ -104,7 +103,7 @@ public class CompactionRecoverTask {
new File(
tsFileManager.getStorageGroupDir()
+ File.separator
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME_FEOM_OLD);
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
if (mergeLogFromOldVersion.exists()) {
logger.info("calling cross compaction task to recover from previous version.");
IoTDBDescriptor.getInstance()
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/log/CompactionLogAnalyzer.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/log/CompactionLogAnalyzer.java
new file mode 100644
index 0000000..24f8902
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/log/CompactionLogAnalyzer.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.engine.compaction.utils.log;
+
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.SEQUENCE_NAME_FROM_OLD;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_MERGE_START_FROM_OLD;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SEQ_FILES_FROM_OLD;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SOURCE_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SOURCE_FILES_FROM_OLD;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_TARGET_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_TARGET_FILES_FROM_OLD;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_UNSEQ_FILES_FROM_OLD;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.UNSEQUENCE_NAME_FROM_OLD;
+
+public class CompactionLogAnalyzer {
+
+ private final File logFile;
+ private final List<TsFileIdentifier> sourceFileInfos = new ArrayList<>();
+ private final List<TsFileIdentifier> targetFileInfos = new ArrayList<>();
+ private boolean isLogFromOld = false;
+
+ public CompactionLogAnalyzer(File logFile) {
+ this.logFile = logFile;
+ }
+
+ /** @return analyze (source file list, target file) */
+ public void analyze() throws IOException {
+ String currLine;
+ try (BufferedReader bufferedReader = new BufferedReader(new FileReader(logFile))) {
+ while ((currLine = bufferedReader.readLine()) != null) {
+ String fileInfo;
+ if (currLine.startsWith(STR_SOURCE_FILES)) {
+ fileInfo = currLine.replace(STR_SOURCE_FILES + TsFileIdentifier.INFO_SEPARATOR, "");
+ analyzeFilePath(false, fileInfo);
+ } else {
+ fileInfo = currLine.replace(STR_TARGET_FILES + TsFileIdentifier.INFO_SEPARATOR, "");
+ analyzeFilePath(true, fileInfo);
+ }
+ }
+ }
+ }
+
+ /** Analyze inner space compaction log of previous version (<0.13). */
+ public void analyzeOldInnerCompactionLog() throws IOException {
+ isLogFromOld = true;
+ String currLine;
+ try (BufferedReader bufferedReader = new BufferedReader(new FileReader(logFile))) {
+ while ((currLine = bufferedReader.readLine()) != null) {
+ switch (currLine) {
+ case STR_SOURCE_FILES_FROM_OLD:
+ currLine = bufferedReader.readLine();
+ sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromOldInfoString(currLine));
+ break;
+ case STR_TARGET_FILES_FROM_OLD:
+ currLine = bufferedReader.readLine();
+ targetFileInfos.add(TsFileIdentifier.getFileIdentifierFromOldInfoString(currLine));
+ break;
+ case STR_SOURCE_FILES:
+ currLine = bufferedReader.readLine();
+ sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromFilePath(currLine));
+ break;
+ case STR_TARGET_FILES:
+ currLine = bufferedReader.readLine();
+ targetFileInfos.add(TsFileIdentifier.getFileIdentifierFromFilePath(currLine));
+ break;
+ case SEQUENCE_NAME_FROM_OLD:
+ case UNSEQUENCE_NAME_FROM_OLD:
+ break;
+ default:
+ break;
+ }
+ }
+ }
+ }
+
+ /** Analyze cross space compaction log of previous version (<0.13). */
+ public void analyzeOldCrossCompactionLog() throws IOException {
+ isLogFromOld = true;
+ String currLine;
+ boolean isSeqSource = true;
+ try (BufferedReader bufferedReader = new BufferedReader(new FileReader(logFile))) {
+ while ((currLine = bufferedReader.readLine()) != null) {
+ if (currLine.equals(STR_UNSEQ_FILES_FROM_OLD)) {
+ isSeqSource = false;
+ continue;
+ } else if (currLine.equals(STR_SEQ_FILES_FROM_OLD)) {
+ isSeqSource = true;
+ continue;
+ } else if (currLine.equals(STR_MERGE_START_FROM_OLD)) {
+ break;
+ }
+ analyzeOldFilePath(isSeqSource, currLine);
+ }
+ }
+ }
+
+ private void analyzeFilePath(boolean isTargetFile, String filePath) {
+ if (isTargetFile) {
+ targetFileInfos.add(TsFileIdentifier.getFileIdentifierFromInfoString(filePath));
+ } else {
+ sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromInfoString(filePath));
+ }
+ }
+
+ private void analyzeOldFilePath(boolean isSeqSource, String oldFilePath) {
+ if (oldFilePath.startsWith("root")) {
+ sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromOldInfoString(oldFilePath));
+ } else {
+ sourceFileInfos.add(TsFileIdentifier.getFileIdentifierFromFilePath(oldFilePath));
+ }
+ if (isSeqSource) {
+ String targetFilePath =
+ oldFilePath.replace(
+ TsFileConstant.TSFILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD);
+ if (oldFilePath.startsWith("root")) {
+ targetFileInfos.add(TsFileIdentifier.getFileIdentifierFromOldInfoString(targetFilePath));
+ } else {
+ targetFileInfos.add(TsFileIdentifier.getFileIdentifierFromFilePath(targetFilePath));
+ }
+ }
+ }
+
+ public List<TsFileIdentifier> getSourceFileInfos() {
+ return sourceFileInfos;
+ }
+
+ public List<TsFileIdentifier> getTargetFileInfos() {
+ return targetFileInfos;
+ }
+
+ public boolean isLogFromOld() {
+ return isLogFromOld;
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/recover/RewriteCrossSpaceCompactionLogger.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/log/CompactionLogger.java
similarity index 51%
rename from server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/recover/RewriteCrossSpaceCompactionLogger.java
rename to server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/log/CompactionLogger.java
index b92af94..6181bce 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/recover/RewriteCrossSpaceCompactionLogger.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/utils/log/CompactionLogger.java
@@ -17,10 +17,9 @@
* under the License.
*/
-package org.apache.iotdb.db.engine.compaction.cross.rewrite.recover;
+package org.apache.iotdb.db.engine.compaction.utils.log;
import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.manage.CrossSpaceCompactionResource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import java.io.BufferedWriter;
@@ -30,21 +29,27 @@ import java.io.IOException;
import java.util.List;
/** MergeLogger records the progress of a merge in file "merge.log" as text lines. */
-public class RewriteCrossSpaceCompactionLogger implements AutoCloseable {
+public class CompactionLogger implements AutoCloseable {
- public static final String COMPACTION_LOG_NAME = "cross-compaction.log";
- public static final String COMPACTION_LOG_NAME_FEOM_OLD = "merge.log";
+ public static final String CROSS_COMPACTION_LOG_NAME_SUFFIX = ".cross-compaction.log";
+ public static final String CROSS_COMPACTION_LOG_NAME_FROM_OLD = "merge.log";
+ public static final String INNER_COMPACTION_LOG_NAME_SUFFIX = ".inner-compaction.log";
- public static final String STR_SEQ_FILES = "seqFiles";
- public static final String STR_TARGET_FILES = "targetFiles";
- public static final String STR_UNSEQ_FILES = "unseqFiles";
- public static final String MAGIC_STRING = "crossSpaceCompaction";
+ public static final String STR_SOURCE_FILES = "source";
+ public static final String STR_TARGET_FILES = "target";
+
+ public static final String STR_SOURCE_FILES_FROM_OLD = "info-source";
+ public static final String STR_TARGET_FILES_FROM_OLD = "info-target";
+ public static final String STR_SEQ_FILES_FROM_OLD = "seqFiles";
+ public static final String STR_UNSEQ_FILES_FROM_OLD = "unseqFiles";
+ public static final String SEQUENCE_NAME_FROM_OLD = "sequence";
+ public static final String UNSEQUENCE_NAME_FROM_OLD = "unsequence";
+ public static final String STR_MERGE_START_FROM_OLD = "merge start";
private BufferedWriter logStream;
- public RewriteCrossSpaceCompactionLogger(File logFile) throws IOException {
+ public CompactionLogger(File logFile) throws IOException {
logStream = new BufferedWriter(new FileWriter(logFile, true));
- logStringInfo(MAGIC_STRING);
}
@Override
@@ -52,25 +57,14 @@ public class RewriteCrossSpaceCompactionLogger implements AutoCloseable {
logStream.close();
}
- public void logStringInfo(String logInfo) throws IOException {
- logStream.write(logInfo);
- logStream.newLine();
- logStream.flush();
- }
-
- public void logFiles(CrossSpaceCompactionResource resource) throws IOException {
- logFiles(resource.getSeqFiles(), STR_SEQ_FILES);
- logFiles(resource.getUnseqFiles(), STR_UNSEQ_FILES);
- }
-
- public void logFiles(List<TsFileResource> seqFiles, String flag) throws IOException {
- logStream.write(flag);
- logStream.newLine();
- for (TsFileResource tsFileResource : seqFiles) {
+ public void logFiles(List<TsFileResource> tsFiles, String flag) throws IOException {
+ for (TsFileResource tsFileResource : tsFiles) {
logStream.write(
- TsFileIdentifier.getFileIdentifierFromFilePath(
- tsFileResource.getTsFile().getAbsolutePath())
- .toString());
+ flag
+ + TsFileIdentifier.INFO_SEPARATOR
+ + TsFileIdentifier.getFileIdentifierFromFilePath(
+ tsFileResource.getTsFile().getAbsolutePath())
+ .toString());
logStream.newLine();
}
logStream.flush();
@@ -79,7 +73,8 @@ public class RewriteCrossSpaceCompactionLogger implements AutoCloseable {
public static File[] findCrossSpaceCompactionLogs(String directory) {
File timePartitionDir = new File(directory);
if (timePartitionDir.exists()) {
- return timePartitionDir.listFiles((dir, name) -> name.endsWith(COMPACTION_LOG_NAME));
+ return timePartitionDir.listFiles(
+ (dir, name) -> name.endsWith(CROSS_COMPACTION_LOG_NAME_SUFFIX));
} else {
return new File[0];
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
index fca0ab7..1738870 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/VirtualStorageGroupProcessor.java
@@ -122,7 +122,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.regex.Pattern;
import static org.apache.iotdb.db.conf.IoTDBConstant.FILE_NAME_SEPARATOR;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.COMPACTION_LOG_NAME;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX;
import static org.apache.iotdb.db.engine.storagegroup.TsFileResource.TEMP_SUFFIX;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
@@ -654,7 +654,7 @@ public class VirtualStorageGroupProcessor {
FSFactoryProducer.getFSFactory()
.getFile(
storageGroupSysDir.getAbsolutePath(),
- logicalStorageGroupName + COMPACTION_LOG_NAME);
+ logicalStorageGroupName + INNER_COMPACTION_LOG_NAME_SUFFIX);
if (logFile.exists()) {
IoTDBDescriptor.getInstance()
.getConfig()
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionTest.java
index 20e96ec..52fe364 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/CrossSpaceCompactionExceptionTest.java
@@ -22,9 +22,9 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.AbstractCompactionTest;
import org.apache.iotdb.db.engine.compaction.CompactionUtils;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger;
import org.apache.iotdb.db.engine.compaction.utils.CompactionConfigRestorer;
import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -45,10 +45,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.MAGIC_STRING;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_SEQ_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_TARGET_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_UNSEQ_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SOURCE_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_TARGET_FILES;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
public class CrossSpaceCompactionExceptionTest extends AbstractCompactionTest {
@@ -85,16 +83,13 @@ public class CrossSpaceCompactionExceptionTest extends AbstractCompactionTest {
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
CrossSpaceCompactionExceptionHandler.handleException(
COMPACTION_TEST_SG,
@@ -157,16 +152,13 @@ public class CrossSpaceCompactionExceptionTest extends AbstractCompactionTest {
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
CrossSpaceCompactionExceptionHandler.handleException(
@@ -230,14 +222,12 @@ public class CrossSpaceCompactionExceptionTest extends AbstractCompactionTest {
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
for (TsFileResource resource : seqResources) {
@@ -250,7 +240,6 @@ public class CrossSpaceCompactionExceptionTest extends AbstractCompactionTest {
tsFileManager.getSequenceListByTimePartition(0).keepOrderInsert(resource);
}
seqResources.get(0).getTsFile().delete();
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
CrossSpaceCompactionExceptionHandler.handleException(
COMPACTION_TEST_SG,
@@ -313,17 +302,14 @@ public class CrossSpaceCompactionExceptionTest extends AbstractCompactionTest {
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
@@ -428,17 +414,14 @@ public class CrossSpaceCompactionExceptionTest extends AbstractCompactionTest {
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionRecoverTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionRecoverTest.java
index f8a799e..aa38a15 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionRecoverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCrossSpaceCompactionRecoverTest.java
@@ -23,9 +23,9 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.AbstractCompactionTest;
import org.apache.iotdb.db.engine.compaction.CompactionTaskManager;
import org.apache.iotdb.db.engine.compaction.CompactionUtils;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger;
import org.apache.iotdb.db.engine.compaction.cross.rewrite.task.RewriteCrossCompactionRecoverTask;
import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
@@ -46,11 +46,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.MAGIC_STRING;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_SEQ_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_TARGET_FILES;
-import static org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger.STR_UNSEQ_FILES;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.COMPACTION_LOG_NAME;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SOURCE_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_TARGET_FILES;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTest {
@@ -85,19 +83,90 @@ public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTe
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
new RewriteCrossCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
+ "0",
+ 0,
+ compactionLogFile,
+ CompactionTaskManager.currentTaskNum,
+ tsFileManager)
+ .call();
+ // all source file should still exist
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(
+ new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(
+ new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
+ }
+ // tmp target file, target file and target resource file should be deleted
+ for (TsFileResource resource : targetResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(
+ new File(
+ resource
+ .getTsFilePath()
+ .replace(
+ IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX))
+ .exists());
+ Assert.assertFalse(
+ new File(
+ resource
+ .getTsFilePath()
+ .replace(
+ IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX)
+ + TsFileResource.RESOURCE_SUFFIX)
+ .exists());
+ }
+ }
+
+ @Test
+ public void testRecoverWithAllSourceFilesExistedAndSomeTargetFilesNotExist() throws Exception {
+ registerTimeseriesInMManger(4, 5, false);
+ createFiles(2, 2, 3, 300, 0, 0, 50, 50, false, true);
+ createFiles(2, 4, 5, 300, 700, 700, 50, 50, false, true);
+ createFiles(3, 3, 4, 200, 20, 10020, 30, 30, false, false);
+ createFiles(2, 1, 5, 100, 450, 20450, 0, 0, false, false);
+ TsFileManager tsFileManager = new TsFileManager(COMPACTION_TEST_SG, "0", SEQ_DIRS.getPath());
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+
+ List<TsFileResource> targetResources =
+ CompactionFileGeneratorUtils.getCrossCompactionTargetTsFileResources(seqResources);
+ File compactionLogFile =
+ new File(
+ SEQ_DIRS,
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
+ CompactionUtils.compact(seqResources, unseqResources, targetResources);
+ // Target files may not exist
+ for (int i = 0; i < targetResources.size(); i++) {
+ if (i < 2) {
+ targetResources.get(i).removeResourceFile();
+ } else {
+ targetResources.get(i).remove();
+ }
+ }
+ compactionLogger.close();
+ new RewriteCrossCompactionRecoverTask(
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -154,20 +223,17 @@ public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTe
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
new RewriteCrossCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -224,21 +290,18 @@ public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTe
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
seqResources.get(0).getTsFile().delete();
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
new RewriteCrossCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -295,17 +358,14 @@ public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTe
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
@@ -325,10 +385,9 @@ public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTe
}
CompactionUtils.combineModsInCompaction(seqResources, unseqResources, targetResources);
seqResources.get(0).remove();
- compactionLogger.close();
new RewriteCrossCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -400,16 +459,13 @@ public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTe
File compactionLogFile =
new File(
SEQ_DIRS,
- seqResources.get(0).getTsFile().getName()
- + "."
- + RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
- RewriteCrossSpaceCompactionLogger compactionLogger =
- new RewriteCrossSpaceCompactionLogger(compactionLogFile);
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
- compactionLogger.logFiles(seqResources, STR_SEQ_FILES);
- compactionLogger.logFiles(unseqResources, STR_UNSEQ_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
CompactionUtils.compact(seqResources, unseqResources, targetResources);
- compactionLogger.logStringInfo(MAGIC_STRING);
compactionLogger.close();
CompactionUtils.moveTargetFile(targetResources, false, COMPACTION_TEST_SG);
for (int i = 0; i < seqResources.size(); i++) {
@@ -431,7 +487,126 @@ public class RewriteCrossSpaceCompactionRecoverTest extends AbstractCompactionTe
CompactionUtils.combineModsInCompaction(seqResources, unseqResources, targetResources);
new RewriteCrossCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
+ "0",
+ 0,
+ compactionLogFile,
+ CompactionTaskManager.currentTaskNum,
+ tsFileManager)
+ .call();
+ // all source file should still exist
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(
+ new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(
+ new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
+ }
+ // tmp target file, target file and target resource file should be deleted
+ for (TsFileResource resource : targetResources) {
+ // xxx.tsfile should not exist
+ Assert.assertFalse(resource.getTsFile().exists());
+ // xxx.merge should not exist
+ Assert.assertFalse(
+ new File(
+ resource
+ .getTsFilePath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX,
+ IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX))
+ .exists());
+ // xxx.tsfile.resource should not exist
+ Assert.assertFalse(
+ new File(resource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
+
+ // mods file of the target file should not exist
+ Assert.assertFalse(resource.getModFile().exists());
+ }
+
+ // all compaction mods file of each source file should not exist
+ for (int i = 0; i < seqResources.size(); i++) {
+ seqResources.get(i).resetModFile();
+ ModificationFile f = seqResources.get(i).getCompactionModFile();
+ Assert.assertFalse(f.exists());
+ }
+ for (int i = 0; i < unseqResources.size(); i++) {
+ unseqResources.get(i).resetModFile();
+ Assert.assertFalse(unseqResources.get(i).getCompactionModFile().exists());
+ }
+
+ // all mods file of each source file should exist
+ for (TsFileResource resource : seqResources) {
+ resource.resetModFile();
+ Assert.assertTrue(resource.getModFile().exists());
+ Assert.assertEquals(1, resource.getModFile().getModifications().size());
+ }
+ for (TsFileResource resource : unseqResources) {
+ resource.resetModFile();
+ Assert.assertTrue(resource.getModFile().exists());
+ Assert.assertEquals(1, resource.getModFile().getModifications().size());
+ }
+
+ // compaction log file should not exist
+ Assert.assertFalse(compactionLogFile.exists());
+
+ Assert.assertTrue(tsFileManager.isAllowCompaction());
+ }
+
+ @Test
+ public void testRecoverWithAllSourcesFileAndCompactonModFileExistAndSomeTargetFilesNotExist()
+ throws Exception {
+ registerTimeseriesInMManger(4, 5, false);
+ createFiles(2, 2, 3, 300, 0, 0, 50, 50, false, true);
+ createFiles(2, 4, 5, 300, 700, 700, 50, 50, false, true);
+ createFiles(3, 3, 4, 200, 20, 10020, 30, 30, false, false);
+ createFiles(2, 1, 5, 100, 450, 20450, 0, 0, false, false);
+ TsFileManager tsFileManager = new TsFileManager(COMPACTION_TEST_SG, "0", SEQ_DIRS.getPath());
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+
+ List<TsFileResource> targetResources =
+ CompactionFileGeneratorUtils.getCrossCompactionTargetTsFileResources(seqResources);
+ File compactionLogFile =
+ new File(
+ SEQ_DIRS,
+ targetResources.get(0).getTsFile().getName()
+ + CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ compactionLogger.logFiles(targetResources, STR_TARGET_FILES);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(unseqResources, STR_SOURCE_FILES);
+ CompactionUtils.compact(seqResources, unseqResources, targetResources);
+ // Target files may not exist
+ for (int i = 0; i < targetResources.size(); i++) {
+ if (i < 2) {
+ targetResources.get(i).removeResourceFile();
+ } else {
+ targetResources.get(i).remove();
+ }
+ }
+ compactionLogger.close();
+ for (int i = 0; i < seqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d0" + PATH_SEPARATOR + "s0",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), true);
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
+ }
+ for (int i = 0; i < unseqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d1" + PATH_SEPARATOR + "s1",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, unseqResources.get(i), true);
+ CompactionFileGeneratorUtils.generateMods(deleteMap, unseqResources.get(i), false);
+ }
+
+ new RewriteCrossCompactionRecoverTask(
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerCompactionLogTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerCompactionLogTest.java
index b4e2e97..bac5394 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerCompactionLogTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerCompactionLogTest.java
@@ -35,7 +35,7 @@ import org.junit.Test;
import java.io.File;
import java.io.IOException;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.COMPACTION_LOG_NAME;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX;
import static org.junit.Assert.assertFalse;
public class InnerCompactionLogTest extends InnerCompactionTest {
@@ -78,7 +78,7 @@ public class InnerCompactionLogTest extends InnerCompactionTest {
}
File logFile =
FSFactoryProducer.getFSFactory()
- .getFile(tempSGDir.getPath(), COMPACTION_TEST_SG + COMPACTION_LOG_NAME);
+ .getFile(tempSGDir.getPath(), COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX);
assertFalse(logFile.exists());
}
}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionTest.java
index 7d2bb87..f2e8111 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionExceptionTest.java
@@ -20,8 +20,8 @@ package org.apache.iotdb.db.engine.compaction.inner;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
-import org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger;
import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
@@ -36,6 +36,7 @@ import java.io.File;
import java.io.FileOutputStream;
import java.nio.channels.FileChannel;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -55,14 +56,12 @@ public class InnerSpaceCompactionExceptionTest extends AbstractInnerSpaceCompact
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
+ compactionLogger.logFiles(
+ Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
InnerSpaceCompactionUtils.compact(targetResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
try (FileOutputStream os = new FileOutputStream(targetResource.getTsFile(), true);
@@ -102,14 +101,12 @@ public class InnerSpaceCompactionExceptionTest extends AbstractInnerSpaceCompact
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
+ compactionLogger.logFiles(
+ Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
InnerSpaceCompactionUtils.compact(targetResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
compactionLogger.close();
@@ -145,14 +142,12 @@ public class InnerSpaceCompactionExceptionTest extends AbstractInnerSpaceCompact
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
+ compactionLogger.logFiles(
+ Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
InnerSpaceCompactionUtils.compact(targetResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
for (TsFileResource resource : seqResources) {
@@ -199,14 +194,12 @@ public class InnerSpaceCompactionExceptionTest extends AbstractInnerSpaceCompact
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
+ compactionLogger.logFiles(
+ Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
InnerSpaceCompactionUtils.compact(targetResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
seqResources.get(0).remove();
@@ -251,14 +244,12 @@ public class InnerSpaceCompactionExceptionTest extends AbstractInnerSpaceCompact
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
+ compactionLogger.logFiles(
+ Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
InnerSpaceCompactionUtils.compact(targetResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
for (int i = 0; i < seqResources.size(); i++) {
@@ -316,14 +307,12 @@ public class InnerSpaceCompactionExceptionTest extends AbstractInnerSpaceCompact
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
+ compactionLogger.logFiles(
+ Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
@@ -374,14 +363,12 @@ public class InnerSpaceCompactionExceptionTest extends AbstractInnerSpaceCompact
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
+ compactionLogger.logFiles(
+ Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
for (int i = 0; i < seqResources.size(); i++) {
Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
deleteMap.put(
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionUtilsOldTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionUtilsOldTest.java
index 15f9374..1097f48 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionUtilsOldTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/InnerSpaceCompactionUtilsOldTest.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.engine.compaction.inner;
import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.constant.TestConstant;
import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
-import org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -42,7 +42,6 @@ import java.io.File;
import java.io.IOException;
import java.util.Collections;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.SOURCE_INFO;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -102,13 +101,10 @@ public class InnerSpaceCompactionUtilsOldTest extends InnerCompactionTest {
if (targetFile.exists()) {
assertTrue(targetFile.delete());
}
- SizeTieredCompactionLogger sizeTieredCompactionLogger =
- new SizeTieredCompactionLogger(
- targetTsFileResource.getTsFilePath().concat(".compaction.log"));
- for (TsFileResource resource : seqResources) {
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, resource.getTsFile());
- }
- sizeTieredCompactionLogger.logSequence(true);
+ CompactionLogger sizeTieredCompactionLogger =
+ new CompactionLogger(
+ new File(targetTsFileResource.getTsFilePath().concat(".compaction.log")));
+ sizeTieredCompactionLogger.logFiles(seqResources, CompactionLogger.STR_SOURCE_FILES);
InnerSpaceCompactionUtils.compact(targetTsFileResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetTsFileResource, COMPACTION_TEST_SG);
sizeTieredCompactionLogger.close();
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTest.java
index 9482036..718a965 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionRecoverTest.java
@@ -25,9 +25,9 @@ import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
import org.apache.iotdb.db.engine.compaction.CompactionTaskManager;
import org.apache.iotdb.db.engine.compaction.inner.AbstractInnerSpaceCompactionTest;
import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
-import org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger;
import org.apache.iotdb.db.engine.compaction.utils.CompactionConfigRestorer;
import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
@@ -59,13 +59,14 @@ import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.COMPACTION_LOG_NAME;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.SOURCE_INFO;
-import static org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger.TARGET_INFO;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SOURCE_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_TARGET_FILES;
import static org.junit.Assert.assertEquals;
public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactionTest {
@@ -133,15 +134,15 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
seqResources.get(0).getTsFile().getParent()
+ File.separator
+ targetTsFileResource.getTsFile().getName()
- + COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger =
- new SizeTieredCompactionLogger(compactionLogFile.getPath());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- compactionLogger.logSequence(true);
+ + INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
deleteFileIfExists(targetTsFileResource.getTsFile());
- compactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
InnerSpaceCompactionUtils.compact(
targetTsFileResource, new ArrayList<>(seqResources.subList(0, 3)));
compactionLogger.close();
@@ -157,7 +158,7 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
new BufferedWriter(
new FileWriter(
SystemFileFactory.INSTANCE.getFile(
- tempSGDir.getPath(), COMPACTION_TEST_SG + COMPACTION_LOG_NAME),
+ tempSGDir.getPath(), COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX),
false));
for (int i = 0; i < logs.size() - 1; i++) {
logStream.write(logs.get(i));
@@ -173,7 +174,7 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
tsFileManager.addForRecover(targetTsFileResource, true);
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -261,20 +262,141 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
seqResources.get(0).getTsFile().getParent()
+ File.separator
+ targetTsFileResource.getTsFile().getName()
- + COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger =
- new SizeTieredCompactionLogger(compactionLogFile.getPath());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- compactionLogger.logSequence(true);
+ + INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
deleteFileIfExists(targetTsFileResource.getTsFile());
- compactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
InnerSpaceCompactionUtils.compact(
targetTsFileResource, new ArrayList<>(seqResources.subList(0, 3)));
compactionLogger.close();
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
+ "0",
+ 0,
+ compactionLogFile,
+ tempSGDir.getAbsolutePath(),
+ true,
+ CompactionTaskManager.currentTaskNum,
+ tsFileManager)
+ .call();
+ // all source file should still exist
+ Assert.assertTrue(seqResources.get(0).getTsFile().exists());
+ Assert.assertTrue(seqResources.get(1).getTsFile().exists());
+ Assert.assertTrue(seqResources.get(2).getTsFile().exists());
+ // tmp target file, target file and target resource file should be deleted
+ Assert.assertFalse(targetTsFileResource.getTsFile().exists());
+ Assert.assertFalse(
+ new File(
+ targetTsFileResource
+ .getTsFilePath()
+ .replace(
+ IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX))
+ .exists());
+ Assert.assertFalse(
+ new File(targetTsFileResource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
+
+ path =
+ SchemaTestUtils.getMeasurementPath(
+ deviceIds[0]
+ + TsFileConstant.PATH_SEPARATOR
+ + measurementSchemas[0].getMeasurementId());
+ System.out.println(tsFileManager.getTsFileList(true));
+ tsFilesReader =
+ new SeriesRawDataBatchReader(
+ path,
+ measurementSchemas[0].getType(),
+ EnvironmentUtils.TEST_QUERY_CONTEXT,
+ tsFileManager.getTsFileList(true),
+ new ArrayList<>(),
+ null,
+ null,
+ true);
+ count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i), batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ tsFilesReader.close();
+ closeTsFileSequenceReader();
+ assertEquals(500, count);
+ }
+
+ @Test
+ public void testRecoverWithAllSourceFilesExistedAndTargetFileNotExist() throws Exception {
+ TsFileManager tsFileManager =
+ new TsFileManager(COMPACTION_TEST_SG, "0", tempSGDir.getAbsolutePath());
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+ MeasurementPath path =
+ SchemaTestUtils.getMeasurementPath(
+ deviceIds[0]
+ + TsFileConstant.PATH_SEPARATOR
+ + measurementSchemas[0].getMeasurementId());
+ IBatchReader tsFilesReader =
+ new SeriesRawDataBatchReader(
+ path,
+ measurementSchemas[0].getType(),
+ EnvironmentUtils.TEST_QUERY_CONTEXT,
+ tsFileManager.getTsFileList(true),
+ new ArrayList<>(),
+ null,
+ null,
+ true);
+ int count = 0;
+ while (tsFilesReader.hasNextBatch()) {
+ BatchData batchData = tsFilesReader.nextBatch();
+ for (int i = 0; i < batchData.length(); i++) {
+ assertEquals(batchData.getTimeByIndex(i), batchData.getDoubleByIndex(i), 0.001);
+ count++;
+ }
+ }
+ tsFilesReader.close();
+ closeTsFileSequenceReader();
+ assertEquals(500, count);
+
+ TsFileResource targetTsFileResource =
+ new TsFileResource(
+ new File(
+ SEQ_DIRS
+ + File.separator.concat(
+ 0
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 1
+ + IoTDBConstant.FILE_NAME_SEPARATOR
+ + 0
+ + IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
+ File compactionLogFile =
+ new File(
+ seqResources.get(0).getTsFile().getParent()
+ + File.separator
+ + targetTsFileResource.getTsFile().getName()
+ + INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
+ deleteFileIfExists(targetTsFileResource.getTsFile());
+ InnerSpaceCompactionUtils.compact(
+ targetTsFileResource, new ArrayList<>(seqResources.subList(0, 3)));
+ // target file may not exist
+ targetTsFileResource.remove();
+ compactionLogger.close();
+ new SizeTieredCompactionRecoverTask(
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -380,15 +502,15 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
seqResources.get(0).getTsFile().getParent()
+ File.separator
+ targetTsFileResource.getTsFile().getName()
- + COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger =
- new SizeTieredCompactionLogger(compactionLogFile.getPath());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- compactionLogger.logSequence(true);
+ + INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
deleteFileIfExists(targetTsFileResource.getTsFile());
- compactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
InnerSpaceCompactionUtils.compact(
targetTsFileResource, new ArrayList<>(seqResources.subList(0, 3)));
InnerSpaceCompactionUtils.moveTargetFile(targetTsFileResource, COMPACTION_TEST_SG);
@@ -396,7 +518,7 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
seqResources.get(0).remove();
compactionLogger.close();
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -465,14 +587,11 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
InnerSpaceCompactionUtils.compact(targetResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
for (int i = 0; i < seqResources.size(); i++) {
@@ -487,7 +606,84 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
compactionLogger.close();
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
+ "0",
+ 0,
+ logFile,
+ tempSGDir.getAbsolutePath(),
+ true,
+ CompactionTaskManager.currentTaskNum,
+ tsFileManager)
+ .call();
+ // all source file should exist
+ for (int i = 0; i < seqResources.size(); i++) {
+ Assert.assertTrue(seqResources.get(i).getTsFile().exists());
+ Assert.assertTrue(seqResources.get(i).resourceFileExists());
+ }
+
+ // tmp target file, target file and target resource file should be deleted
+ Assert.assertFalse(targetResource.getTsFile().exists());
+ Assert.assertFalse(
+ new File(
+ targetResource
+ .getTsFilePath()
+ .replace(
+ IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX))
+ .exists());
+ Assert.assertFalse(
+ new File(targetResource.getTsFilePath() + TsFileResource.RESOURCE_SUFFIX).exists());
+
+ // all compaction mods file of each source file should not exist
+ for (int i = 0; i < seqResources.size(); i++) {
+ Assert.assertFalse(seqResources.get(i).getCompactionModFile().exists());
+ }
+
+ // all mods file of each source file should exist
+ for (int i = 0; i < seqResources.size(); i++) {
+ seqResources.get(i).resetModFile();
+ Assert.assertTrue(seqResources.get(i).getModFile().exists());
+ Assert.assertEquals(1, seqResources.get(i).getModFile().getModifications().size());
+ }
+
+ // mods file of the target file should not exist
+ Assert.assertFalse(targetResource.getModFile().exists());
+
+ // compaction log file should not exist
+ Assert.assertFalse(logFile.exists());
+
+ Assert.assertTrue(tsFileManager.isAllowCompaction());
+ }
+
+ @Test
+ public void testRecoverWithAllSourcesFileAndCompactonModFileExistAndTargetFileNotExist()
+ throws Exception {
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+ TsFileResource targetResource =
+ TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
+ File logFile =
+ new File(
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ InnerSpaceCompactionUtils.compact(targetResource, seqResources);
+ // target file may not exist
+ targetResource.remove();
+ for (int i = 0; i < seqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ deviceIds[0] + "." + measurementSchemas[0].getMeasurementId(),
+ new Pair<>(i * ptNum, i * ptNum + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), true);
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
+ }
+ compactionLogger.close();
+
+ new SizeTieredCompactionRecoverTask(
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
logFile,
@@ -548,14 +744,11 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
TsFileNameGenerator.getInnerCompactionTargetFileResource(seqResources, true);
File logFile =
new File(
- targetResource.getTsFile().getPath() + SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger = new SizeTieredCompactionLogger(logFile.getPath());
- for (TsFileResource source : seqResources) {
- compactionLogger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, source.getTsFile());
- }
- compactionLogger.logSequence(true);
- compactionLogger.logFileInfo(
- SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
+ targetResource.getTsFile().getPath()
+ + CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(logFile);
+ compactionLogger.logFiles(seqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
InnerSpaceCompactionUtils.compact(targetResource, seqResources);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
for (int i = 0; i < seqResources.size(); i++) {
@@ -571,7 +764,7 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
compactionLogger.close();
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
logFile,
@@ -667,22 +860,22 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
seqResources.get(0).getTsFile().getParent()
+ File.separator
+ targetTsFileResource.getTsFile().getName()
- + COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger =
- new SizeTieredCompactionLogger(compactionLogFile.getPath());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- compactionLogger.logSequence(true);
+ + INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
deleteFileIfExists(targetTsFileResource.getTsFile());
- compactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
InnerSpaceCompactionUtils.compact(
targetTsFileResource, new ArrayList<>(seqResources.subList(0, 3)));
compactionLogger.close();
InnerSpaceCompactionUtils.moveTargetFile(targetTsFileResource, COMPACTION_TEST_SG);
tsFileManager.add(targetTsFileResource, true);
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -772,15 +965,15 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
seqResources.get(0).getTsFile().getParent()
+ File.separator
+ targetTsFileResource.getTsFile().getName()
- + COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger =
- new SizeTieredCompactionLogger(compactionLogFile.getPath());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- compactionLogger.logSequence(true);
+ + INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
deleteFileIfExists(targetTsFileResource.getTsFile());
- compactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
InnerSpaceCompactionUtils.compact(
targetTsFileResource, new ArrayList<>(seqResources.subList(0, 3)));
InnerSpaceCompactionUtils.moveTargetFile(targetTsFileResource, COMPACTION_TEST_SG);
@@ -791,7 +984,7 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
}
tsFileManager.add(targetTsFileResource, true);
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -880,15 +1073,15 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
seqResources.get(0).getTsFile().getParent()
+ File.separator
+ targetTsFileResource.getTsFile().getName()
- + COMPACTION_LOG_NAME);
- SizeTieredCompactionLogger compactionLogger =
- new SizeTieredCompactionLogger(compactionLogFile.getPath());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- compactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- compactionLogger.logSequence(true);
+ + INNER_COMPACTION_LOG_NAME_SUFFIX);
+ CompactionLogger compactionLogger = new CompactionLogger(compactionLogFile);
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ compactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
deleteFileIfExists(targetTsFileResource.getTsFile());
- compactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
InnerSpaceCompactionUtils.compact(
targetTsFileResource, new ArrayList<>(seqResources.subList(0, 3)));
compactionLogger.close();
@@ -900,7 +1093,7 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
}
tsFileManager.add(targetTsFileResource, true);
new SizeTieredCompactionRecoverTask(
- COMPACTION_LOG_NAME,
+ INNER_COMPACTION_LOG_NAME_SUFFIX,
"0",
0,
compactionLogFile,
@@ -944,11 +1137,16 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
throws IOException, MetadataException {
tsFileManager.addAll(seqResources, true);
tsFileManager.addAll(unseqResources, false);
- SizeTieredCompactionLogger sizeTieredCompactionLogger =
- new SizeTieredCompactionLogger(tempSGDir.getPath(), COMPACTION_TEST_SG);
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
+ CompactionLogger sizeTieredCompactionLogger =
+ new CompactionLogger(
+ new File(
+ tempSGDir.getPath(),
+ COMPACTION_TEST_SG + COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX));
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ sizeTieredCompactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
sizeTieredCompactionLogger.close();
MeasurementPath path =
SchemaTestUtils.getMeasurementPath(
@@ -984,12 +1182,14 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
throws IOException, MetadataException {
tsFileManager.addAll(seqResources, true);
tsFileManager.addAll(unseqResources, false);
- SizeTieredCompactionLogger sizeTieredCompactionLogger =
- new SizeTieredCompactionLogger(tempSGDir.getPath(), COMPACTION_TEST_SG);
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- sizeTieredCompactionLogger.logSequence(true);
+ CompactionLogger sizeTieredCompactionLogger =
+ new CompactionLogger(
+ new File(tempSGDir.getPath(), COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX));
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ sizeTieredCompactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
sizeTieredCompactionLogger.close();
MeasurementPath path =
SchemaTestUtils.getMeasurementPath(
@@ -1024,12 +1224,14 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
public void testCompactionMergeRecoverMergeStart() throws IOException, MetadataException {
tsFileManager.addAll(seqResources, true);
tsFileManager.addAll(unseqResources, false);
- SizeTieredCompactionLogger sizeTieredCompactionLogger =
- new SizeTieredCompactionLogger(tempSGDir.getPath(), COMPACTION_TEST_SG);
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(0).getTsFile());
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(1).getTsFile());
- sizeTieredCompactionLogger.logFileInfo(SOURCE_INFO, seqResources.get(2).getTsFile());
- sizeTieredCompactionLogger.logSequence(true);
+ CompactionLogger sizeTieredCompactionLogger =
+ new CompactionLogger(
+ new File(tempSGDir.getPath(), COMPACTION_TEST_SG + INNER_COMPACTION_LOG_NAME_SUFFIX));
+ List<TsFileResource> tmpSeqResources = new ArrayList<>();
+ tmpSeqResources.add(seqResources.get(0));
+ tmpSeqResources.add(seqResources.get(1));
+ tmpSeqResources.add(seqResources.get(2));
+ sizeTieredCompactionLogger.logFiles(tmpSeqResources, STR_SOURCE_FILES);
TsFileResource targetTsFileResource =
new TsFileResource(
new File(
@@ -1043,7 +1245,8 @@ public class SizeTieredCompactionRecoverTest extends AbstractInnerSpaceCompactio
+ IoTDBConstant.FILE_NAME_SEPARATOR
+ 0
+ IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX)));
- sizeTieredCompactionLogger.logFileInfo(TARGET_INFO, targetTsFileResource.getTsFile());
+ sizeTieredCompactionLogger.logFiles(
+ Collections.singletonList(targetTsFileResource), STR_TARGET_FILES);
tsFileManager.addForRecover(targetTsFileResource, true);
sizeTieredCompactionLogger.close();
MeasurementPath path =
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/RewriteCrossSpaceCompactionRecoverCompatibleTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/RewriteCrossSpaceCompactionRecoverCompatibleTest.java
new file mode 100644
index 0000000..51dda8a
--- /dev/null
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/RewriteCrossSpaceCompactionRecoverCompatibleTest.java
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.engine.compaction.recover;
+
+import org.apache.iotdb.db.conf.IoTDBConstant;
+import org.apache.iotdb.db.engine.compaction.AbstractCompactionTest;
+import org.apache.iotdb.db.engine.compaction.CompactionTaskManager;
+import org.apache.iotdb.db.engine.compaction.cross.rewrite.task.RewriteCrossCompactionRecoverTask;
+import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.modification.Deletion;
+import org.apache.iotdb.db.engine.modification.ModificationFile;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.metadata.path.PartialPath;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
+import org.apache.iotdb.tsfile.utils.Pair;
+
+import com.google.common.io.Files;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
+
+public class RewriteCrossSpaceCompactionRecoverCompatibleTest extends AbstractCompactionTest {
+ private String TEST_SG = "root.compactionTest";
+ private final String oldThreadName = Thread.currentThread().getName();
+
+ @Override
+ @Before
+ public void setUp() throws IOException, MetadataException, WriteProcessException {
+ super.setUp();
+ Thread.currentThread().setName("pool-1-IoTDB-Compaction-1");
+ }
+
+ @Override
+ @After
+ public void tearDown() throws StorageEngineException, IOException {
+ super.tearDown();
+ Thread.currentThread().setName(oldThreadName);
+ }
+
+ @Test
+ public void testCompatibleWithAllSourceFilesExistWithFileInfo() throws Exception {
+ createFiles(6, 2, 3, 100, 0, 0, 50, 50, false, true);
+ createFiles(3, 2, 3, 50, 0, 0, 50, 50, false, false);
+ registerTimeseriesInMManger(2, 3, false);
+ List<TsFileResource> tmpTargetResources = new ArrayList<>();
+ for (TsFileResource resource : seqResources) {
+ File mergeFile =
+ new File(
+ resource.getTsFilePath() + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD);
+ mergeFile.createNewFile();
+ tmpTargetResources.add(new TsFileResource(mergeFile));
+ }
+
+ File logFile =
+ new File(SEQ_DIRS.getPath(), CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ BufferedWriter logWriter = new BufferedWriter(new FileWriter(logFile));
+ logWriter.write("seqFiles\n");
+ for (TsFileResource tsFileResource : seqResources) {
+ logWriter.write(
+ String.format(TEST_SG + " 0 0 %s true\n", tsFileResource.getTsFile().getName()));
+ }
+ logWriter.write("unseqFiles\n");
+ for (TsFileResource tsFileResource : unseqResources) {
+ logWriter.write(
+ String.format(TEST_SG + " 0 0 %s false\n", tsFileResource.getTsFile().getName()));
+ }
+ logWriter.close();
+
+ for (int i = 0; i < seqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d0" + PATH_SEPARATOR + "s0",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
+ }
+ for (int i = 0; i < unseqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d1" + PATH_SEPARATOR + "s1",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, unseqResources.get(i), false);
+ }
+ ModificationFile mergeMods =
+ new ModificationFile(
+ SEQ_DIRS + File.separator + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ mergeMods.write(new Deletion(new PartialPath("root.d1.s1"), 100, 0, 100));
+ mergeMods.close();
+
+ TsFileManager tsFileManager = new TsFileManager(TEST_SG, "0", SEQ_DIRS.getPath());
+ new RewriteCrossCompactionRecoverTask(
+ TEST_SG, "0", 0, logFile, CompactionTaskManager.currentTaskNum, tsFileManager)
+ .call();
+
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(resource.resourceFileExists());
+ Assert.assertTrue(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(resource.resourceFileExists());
+ Assert.assertTrue(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : tmpTargetResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ }
+ for (TsFileResource resource :
+ CompactionFileGeneratorUtils.getCrossCompactionTargetTsFileResources(seqResources)) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ }
+
+ Assert.assertFalse(logFile.exists());
+ Assert.assertFalse(mergeMods.exists());
+ }
+
+ @Test
+ public void testCompatibleWithSomeSourceFilesLostWithFileInfo() throws Exception {
+ createFiles(6, 2, 3, 100, 0, 0, 50, 50, false, true);
+ createFiles(3, 2, 3, 50, 0, 0, 50, 50, false, false);
+ registerTimeseriesInMManger(2, 3, false);
+ List<TsFileResource> tmpTargetResources = new ArrayList<>();
+ for (TsFileResource resource : seqResources) {
+ File mergeFile =
+ new File(
+ resource.getTsFilePath() + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD);
+ Files.copy(resource.getTsFile(), mergeFile);
+ tmpTargetResources.add(new TsFileResource(mergeFile));
+ }
+
+ File logFile =
+ new File(SEQ_DIRS.getPath(), CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ BufferedWriter logWriter = new BufferedWriter(new FileWriter(logFile));
+ logWriter.write("seqFiles\n");
+ for (TsFileResource tsFileResource : seqResources) {
+ logWriter.write(
+ String.format(TEST_SG + " 0 0 %s true\n", tsFileResource.getTsFile().getName()));
+ }
+ logWriter.write("unseqFiles\n");
+ for (TsFileResource tsFileResource : unseqResources) {
+ logWriter.write(
+ String.format(TEST_SG + " 0 0 %s false\n", tsFileResource.getTsFile().getName()));
+ }
+ logWriter.close();
+
+ // First seq files lost
+ seqResources.get(0).delete();
+
+ for (int i = 0; i < seqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d0" + PATH_SEPARATOR + "s0",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
+ }
+ for (int i = 0; i < unseqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d1" + PATH_SEPARATOR + "s1",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, unseqResources.get(i), false);
+ }
+ ModificationFile mergeMods =
+ new ModificationFile(
+ SEQ_DIRS + File.separator + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ mergeMods.write(new Deletion(new PartialPath("root.d1.s1"), 100, 0, 100));
+ mergeMods.close();
+
+ TsFileManager tsFileManager = new TsFileManager(TEST_SG, "0", SEQ_DIRS.getPath());
+ new RewriteCrossCompactionRecoverTask(
+ TEST_SG, "0", 0, logFile, CompactionTaskManager.currentTaskNum, tsFileManager)
+ .call();
+
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ Assert.assertFalse(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ Assert.assertFalse(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : tmpTargetResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ }
+ for (TsFileResource seqResource :
+ CompactionFileGeneratorUtils.getCrossCompactionTargetTsFileResources(seqResources)) {
+ TsFileResource resource =
+ new TsFileResource(
+ new File(
+ seqResource
+ .getTsFilePath()
+ .replace(
+ IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX)));
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(resource.resourceFileExists());
+ Assert.assertTrue(resource.getModFile().exists());
+ }
+
+ Assert.assertFalse(logFile.exists());
+ Assert.assertFalse(mergeMods.exists());
+ }
+
+ @Test
+ public void testCompatibleWithAllSourceFilesExistWithFilePath() throws Exception {
+ createFiles(6, 2, 3, 100, 0, 0, 50, 50, false, true);
+ createFiles(3, 2, 3, 50, 0, 0, 50, 50, false, false);
+ registerTimeseriesInMManger(2, 3, false);
+ List<TsFileResource> tmpTargetResources = new ArrayList<>();
+ for (TsFileResource resource : seqResources) {
+ File mergeFile =
+ new File(
+ resource.getTsFilePath() + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD);
+ mergeFile.createNewFile();
+ tmpTargetResources.add(new TsFileResource(mergeFile));
+ }
+
+ File logFile =
+ new File(SEQ_DIRS.getPath(), CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ BufferedWriter logWriter = new BufferedWriter(new FileWriter(logFile));
+ logWriter.write("seqFiles\n");
+ for (TsFileResource tsFileResource : seqResources) {
+ logWriter.write(tsFileResource.getTsFile().getAbsolutePath());
+ logWriter.write("\n");
+ }
+ logWriter.write("unseqFiles\n");
+ for (TsFileResource tsFileResource : unseqResources) {
+ logWriter.write(tsFileResource.getTsFile().getAbsolutePath());
+ logWriter.write("\n");
+ }
+ logWriter.write(CompactionLogger.STR_MERGE_START_FROM_OLD);
+ logWriter.write("\n");
+ logWriter.write(seqResources.get(0).getTsFile().getAbsolutePath() + " " + 100);
+ logWriter.close();
+
+ for (int i = 0; i < seqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d0" + PATH_SEPARATOR + "s0",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
+ }
+ for (int i = 0; i < unseqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d1" + PATH_SEPARATOR + "s1",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, unseqResources.get(i), false);
+ }
+ ModificationFile mergeMods =
+ new ModificationFile(
+ SEQ_DIRS + File.separator + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ mergeMods.write(new Deletion(new PartialPath("root.d1.s1"), 100, 0, 100));
+ mergeMods.close();
+
+ TsFileManager tsFileManager = new TsFileManager(TEST_SG, "0", SEQ_DIRS.getPath());
+ new RewriteCrossCompactionRecoverTask(
+ TEST_SG, "0", 0, logFile, CompactionTaskManager.currentTaskNum, tsFileManager)
+ .call();
+
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(resource.resourceFileExists());
+ Assert.assertTrue(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(resource.resourceFileExists());
+ Assert.assertTrue(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : tmpTargetResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ }
+ for (TsFileResource resource :
+ CompactionFileGeneratorUtils.getCrossCompactionTargetTsFileResources(seqResources)) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ }
+
+ Assert.assertFalse(logFile.exists());
+ Assert.assertFalse(mergeMods.exists());
+ }
+
+ @Test
+ public void testCompatibleWithSomeSourceFilesLostWithFilePath() throws Exception {
+ createFiles(6, 2, 3, 100, 0, 0, 50, 50, false, true);
+ createFiles(3, 2, 3, 50, 0, 0, 50, 50, false, false);
+ registerTimeseriesInMManger(2, 3, false);
+ List<TsFileResource> tmpTargetResources = new ArrayList<>();
+ for (TsFileResource resource : seqResources) {
+ File mergeFile =
+ new File(
+ resource.getTsFilePath() + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD);
+ Files.copy(resource.getTsFile(), mergeFile);
+ tmpTargetResources.add(new TsFileResource(mergeFile));
+ }
+
+ File logFile =
+ new File(SEQ_DIRS.getPath(), CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ BufferedWriter logWriter = new BufferedWriter(new FileWriter(logFile));
+ logWriter.write("seqFiles\n");
+ for (TsFileResource tsFileResource : seqResources) {
+ logWriter.write(tsFileResource.getTsFile().getAbsolutePath());
+ logWriter.write("\n");
+ }
+ logWriter.write("unseqFiles\n");
+ for (TsFileResource tsFileResource : unseqResources) {
+ logWriter.write(tsFileResource.getTsFile().getAbsolutePath());
+ logWriter.write("\n");
+ }
+ logWriter.write(CompactionLogger.STR_MERGE_START_FROM_OLD);
+ logWriter.write("\n");
+ logWriter.write(seqResources.get(0).getTsFile().getAbsolutePath() + " " + 100);
+ logWriter.close();
+
+ // First seq files lost
+ seqResources.get(0).delete();
+
+ for (int i = 0; i < seqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d0" + PATH_SEPARATOR + "s0",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, seqResources.get(i), false);
+ }
+ for (int i = 0; i < unseqResources.size(); i++) {
+ Map<String, Pair<Long, Long>> deleteMap = new HashMap<>();
+ deleteMap.put(
+ COMPACTION_TEST_SG + PATH_SEPARATOR + "d1" + PATH_SEPARATOR + "s1",
+ new Pair(i * 10L, i * 10L + 10));
+ CompactionFileGeneratorUtils.generateMods(deleteMap, unseqResources.get(i), false);
+ }
+ ModificationFile mergeMods =
+ new ModificationFile(
+ SEQ_DIRS + File.separator + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ mergeMods.write(new Deletion(new PartialPath("root.d1.s1"), 100, 0, 100));
+ mergeMods.close();
+
+ TsFileManager tsFileManager = new TsFileManager(TEST_SG, "0", SEQ_DIRS.getPath());
+ new RewriteCrossCompactionRecoverTask(
+ TEST_SG, "0", 0, logFile, CompactionTaskManager.currentTaskNum, tsFileManager)
+ .call();
+
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ Assert.assertFalse(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ Assert.assertFalse(resource.getModFile().exists());
+ }
+ for (TsFileResource resource : tmpTargetResources) {
+ Assert.assertFalse(resource.getTsFile().exists());
+ Assert.assertFalse(resource.resourceFileExists());
+ }
+ for (TsFileResource seqResource :
+ CompactionFileGeneratorUtils.getCrossCompactionTargetTsFileResources(seqResources)) {
+ TsFileResource resource =
+ new TsFileResource(
+ new File(
+ seqResource
+ .getTsFilePath()
+ .replace(
+ IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX)));
+ Assert.assertTrue(resource.getTsFile().exists());
+ Assert.assertTrue(resource.resourceFileExists());
+ Assert.assertTrue(resource.getModFile().exists());
+ }
+
+ Assert.assertFalse(logFile.exists());
+ Assert.assertFalse(mergeMods.exists());
+ }
+}
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java
index e9bd9f8..e666d37 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverCompatibleTest.java
@@ -68,9 +68,7 @@ public class SizeTieredCompactionRecoverCompatibleTest extends AbstractCompactio
targetFile.close();
File logFile =
- new File(
- targetResource.getTsFile().getParent(),
- targetResource.getTsFile().getName() + ".compaction.log");
+ new File(targetResource.getTsFile().getParent(), "root.compactionTest" + ".compaction.log");
BufferedWriter logWriter = new BufferedWriter(new FileWriter(logFile));
for (TsFileResource tsFileResource : seqResources) {
logWriter.write(
@@ -121,9 +119,7 @@ public class SizeTieredCompactionRecoverCompatibleTest extends AbstractCompactio
targetFile.close();
File logFile =
- new File(
- targetResource.getTsFile().getParent(),
- targetResource.getTsFile().getName() + ".compaction.log");
+ new File(targetResource.getTsFile().getParent(), "root.compactionTest" + ".compaction.log");
BufferedWriter logWriter = new BufferedWriter(new FileWriter(logFile));
for (TsFileResource tsFileResource : seqResources) {
logWriter.write(String.format("source\n%s\n", tsFileResource.getTsFile().getAbsolutePath()));
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java
index 98eec21..1c73f0c 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/recover/SizeTieredCompactionRecoverTest.java
@@ -24,9 +24,9 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.constant.TestConstant;
import org.apache.iotdb.db.engine.compaction.inner.sizetiered.SizeTieredCompactionRecoverTask;
import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
-import org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger;
import org.apache.iotdb.db.engine.compaction.utils.CompactionConfigRestorer;
import org.apache.iotdb.db.engine.compaction.utils.CompactionFileGeneratorUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
@@ -64,6 +64,9 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_SOURCE_FILES;
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.STR_TARGET_FILES;
+
public class SizeTieredCompactionRecoverTest {
static final String COMPACTION_TEST_SG = "root.compactionTest";
@@ -230,12 +233,10 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, resource.getTsFile());
- }
- logger.logFileInfo(SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
- logger.close();
+ CompactionLogger compactionLogger = new CompactionLogger(new File(logFilePath));
+ compactionLogger.logFiles(sourceFiles, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ compactionLogger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
SizeTieredCompactionRecoverTask recoverTask =
@@ -289,12 +290,10 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, resource.getTsFile());
- }
- logger.logFileInfo(SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
- logger.close();
+ CompactionLogger compactionLogger = new CompactionLogger(new File(logFilePath));
+ compactionLogger.logFiles(sourceFiles, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ compactionLogger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
FileOutputStream targetStream = new FileOutputStream(targetResource.getTsFile(), true);
@@ -353,11 +352,9 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFile(SizeTieredCompactionLogger.SOURCE_NAME, resource.getTsFile());
- }
- logger.logFile(SizeTieredCompactionLogger.TARGET_NAME, targetResource.getTsFile());
+ CompactionLogger logger = new CompactionLogger(new File(logFilePath));
+ logger.logFiles(sourceFiles, CompactionLogger.STR_SOURCE_FILES);
+ logger.logFiles(Collections.singletonList(targetResource), CompactionLogger.STR_TARGET_FILES);
logger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
@@ -412,12 +409,10 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFile(SizeTieredCompactionLogger.SOURCE_NAME, resource.getTsFile());
- }
- logger.logFile(SizeTieredCompactionLogger.TARGET_NAME, targetResource.getTsFile());
- logger.close();
+ CompactionLogger compactionLogger = new CompactionLogger(new File(logFilePath));
+ compactionLogger.logFiles(sourceFiles, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ compactionLogger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
FileOutputStream targetStream = new FileOutputStream(targetResource.getTsFile(), true);
@@ -479,12 +474,10 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, resource.getTsFile());
- }
- logger.logFileInfo(SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
- logger.close();
+ CompactionLogger compactionLogger = new CompactionLogger(new File(logFilePath));
+ compactionLogger.logFiles(sourceFiles, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ compactionLogger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
long sizeOfTargetFile = targetResource.getTsFileSize();
@@ -584,12 +577,10 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFileInfo(SizeTieredCompactionLogger.SOURCE_INFO, resource.getTsFile());
- }
- logger.logFileInfo(SizeTieredCompactionLogger.TARGET_INFO, targetResource.getTsFile());
- logger.close();
+ CompactionLogger compactionLogger = new CompactionLogger(new File(logFilePath));
+ compactionLogger.logFiles(sourceFiles, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ compactionLogger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
FileOutputStream targetStream = new FileOutputStream(targetResource.getTsFile(), true);
@@ -686,12 +677,10 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFile(SizeTieredCompactionLogger.SOURCE_NAME, resource.getTsFile());
- }
- logger.logFile(SizeTieredCompactionLogger.TARGET_NAME, targetResource.getTsFile());
- logger.close();
+ CompactionLogger compactionLogger = new CompactionLogger(new File(logFilePath));
+ compactionLogger.logFiles(sourceFiles, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ compactionLogger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
long sizeOfTargetFile = targetResource.getTsFileSize();
@@ -791,12 +780,10 @@ public class SizeTieredCompactionRecoverTest {
}
TsFileResource targetResource =
TsFileNameGenerator.getInnerCompactionTargetFileResource(sourceFiles, true);
- SizeTieredCompactionLogger logger = new SizeTieredCompactionLogger(logFilePath);
- for (TsFileResource resource : sourceFiles) {
- logger.logFile(SizeTieredCompactionLogger.SOURCE_NAME, resource.getTsFile());
- }
- logger.logFile(SizeTieredCompactionLogger.TARGET_NAME, targetResource.getTsFile());
- logger.close();
+ CompactionLogger compactionLogger = new CompactionLogger(new File(logFilePath));
+ compactionLogger.logFiles(sourceFiles, STR_SOURCE_FILES);
+ compactionLogger.logFiles(Collections.singletonList(targetResource), STR_TARGET_FILES);
+ compactionLogger.close();
InnerSpaceCompactionUtils.compact(targetResource, sourceFiles);
InnerSpaceCompactionUtils.moveTargetFile(targetResource, COMPACTION_TEST_SG);
FileOutputStream targetStream = new FileOutputStream(targetResource.getTsFile(), true);
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionClearUtils.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionClearUtils.java
index 9141b43..c28a708 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionClearUtils.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/utils/CompactionClearUtils.java
@@ -24,8 +24,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.cache.BloomFilterCache;
import org.apache.iotdb.db.engine.cache.ChunkCache;
import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
-import org.apache.iotdb.db.engine.compaction.cross.rewrite.recover.RewriteCrossSpaceCompactionLogger;
-import org.apache.iotdb.db.engine.compaction.inner.utils.SizeTieredCompactionLogger;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
import org.apache.iotdb.db.engine.modification.ModificationFile;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.query.control.FileReaderManager;
@@ -47,8 +46,8 @@ public class CompactionClearUtils {
deleteAllFilesInOneDirBySuffix("target", ModificationFile.FILE_SUFFIX);
deleteAllFilesInOneDirBySuffix("target", IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX);
deleteAllFilesInOneDirBySuffix("target", IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX);
- deleteAllFilesInOneDirBySuffix("target", SizeTieredCompactionLogger.COMPACTION_LOG_NAME);
- deleteAllFilesInOneDirBySuffix("target", RewriteCrossSpaceCompactionLogger.COMPACTION_LOG_NAME);
+ deleteAllFilesInOneDirBySuffix("target", CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX);
+ deleteAllFilesInOneDirBySuffix("target", CompactionLogger.CROSS_COMPACTION_LOG_NAME_SUFFIX);
// clean cache
if (IoTDBDescriptor.getInstance().getConfig().isMetaDataCacheEnable()) {
ChunkCache.getInstance().clear();