You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/03/22 07:18:04 UTC
[GitHub] [iotdb] choubenson opened a new pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
choubenson opened a new pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316
Previously, there are four sets of code for restart recovery and exception handling for inner space compaction and cross space compaction, which needs to be refactored.
Now, cross space compaction and inner space compaction use one compaction exception handler and one compaction recover task, which is "org.apache.iotdb.db.engine.compaction.task.CompactionExceptionHandler" and "org.apache.iotdb.db.engine.compaction.task.CompactionRecoverTask"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r832950998
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
Review comment:
Here the nested class needs to call the method of the outer class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] qiaojialin commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
qiaojialin commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r833125271
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverHelper.java
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverHelper searches compaction log and call {@link CompactionRecoverTask} to execute
+ * the recover process for all compaction task sequentially, including InnerCompactionTask in
+ * sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverHelper {
Review comment:
+1 for Manager
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
Review comment:
```suggestion
public class CompactionRecoverTask {
```
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
##########
@@ -202,13 +201,29 @@ protected void doCompaction() throws Exception {
if (sizeTieredCompactionLogger != null) {
sizeTieredCompactionLogger.close();
}
- InnerSpaceCompactionExceptionHandler.handleException(
- fullStorageGroupName,
- logFile,
- targetTsFileResource,
- selectedTsFileResourceList,
- tsFileManager,
- tsFileResourceList);
+ if (isSequence()) {
+ CompactionExceptionHandler.handleException(
Review comment:
Only the position of these two parameters?
selectedTsFileResourceList,
Collections.emptyList(),
Oh, I know, it represents sequence list and unsequence list
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r834986936
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,483 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.utils.TsFileUtils;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
+ private final String fullStorageGroupName;
+ private final TsFileManager tsFileManager;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ TsFileManager tsFileManager,
+ File logFile,
+ boolean isInnerSpace) {
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
+ this.fullStorageGroupName = logicalStorageGroupName + "-" + virtualStorageGroupName;
this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isInnerCompactionLogBefore013()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isCrossCompactionLogBefore013()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistBefore013(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(targetFileIdentifiers, sourceFileIdentifiers);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostBefore013(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithSomeSourceFilesLost(targetFileIdentifiers, sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers, List<TsFileIdentifier> sourceFileIdentifiers) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(
+ List<TsFileIdentifier> targetFileIdentifiers, List<TsFileIdentifier> sourceFileIdentifiers)
+ throws IOException {
+ // some source files have been deleted, while target file must exist and complete.
+ if (!checkIsTargetFilesComplete(targetFileIdentifiers)) {
+ return false;
+ }
+
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
+ }
+ }
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
+ }
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ private boolean checkIsTargetFilesComplete(List<TsFileIdentifier> targetFileIdentifiers)
+ throws IOException {
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ if (targetFile == null
+ || !TsFileUtils.isTsFileComplete(new TsFileResource(targetFile).getTsFile())) {
+ LOGGER.error(
+ "{} [Compaction][ExceptionHandler] target file {} is not complete, and some source files is lost, do nothing. Set allowCompaction to false",
+ fullStorageGroupName,
+ targetFileIdentifier.getFilePath());
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
+ return false;
}
}
+ return true;
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isCrossCompactionLogBefore013() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isInnerCompactionLogBefore013() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistBefore013(
+ List<TsFileIdentifier> targetFileIdentifiers) {
+ // delete tmp target file
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile.merge
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * 1. If target file does not exist, then move .merge file to target file <br>
+ * 2. If target resource file does not exist, then serialize it. <br>
+ * 3. Append merging modification to target mods file and delete merging mods file. <br>
+ * 4. Delete source files and .merge file. <br>
+ */
+ private boolean handleCrossCompactionWithSomeSourceFilesLostBefore013(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers) {
+ try {
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ List<TsFileResource> targetFileResources = new ArrayList<>();
+ for (int i = 0; i < sourceFileIdentifiers.size(); i++) {
+ TsFileIdentifier sourceFileIdentifier = sourceFileIdentifiers.get(i);
+ if (sourceFileIdentifier.isSequence()) {
+ File tmpTargetFile = targetFileIdentifiers.get(i).getFileFromDataDirs();
+ File targetFile = null;
+
+ // move tmp target file to target file if not exist
+ if (tmpTargetFile != null) {
+ // move tmp target file to target file
+ String sourceFilePath =
+ tmpTargetFile
+ .getPath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX);
+ targetFile = TsFileNameGenerator.increaseCrossCompactionCnt(new File(sourceFilePath));
+ FSFactoryProducer.getFSFactory().moveFile(tmpTargetFile, targetFile);
+ } else {
+ // target file must exist
+ File file =
+ TsFileNameGenerator.increaseCrossCompactionCnt(
+ new File(
+ targetFileIdentifiers
+ .get(i)
+ .getFilePath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX)));
+
+ targetFile = getFileFromDataDirs(file.getPath());
+ }
+ if (targetFile == null) {
+ LOGGER.error(
+ "{} [Compaction][Recover] target file of source seq file {} does not exist (<0.13).",
+ fullStorageGroupName,
+ sourceFileIdentifier.getFilePath());
+ return false;
+ }
+
+ // serialize target resource file if not exist
+ TsFileResource targetResource = new TsFileResource(targetFile);
+ if (!targetResource.resourceFileExists()) {
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(targetFile.getAbsolutePath())) {
+ FileLoaderUtils.updateTsFileResource(reader, targetResource);
+ }
+ targetResource.serialize();
+ }
+
+ targetFileResources.add(targetResource);
+
+ // append compaction modifications to target mods file and delete compaction mods file
+ if (compactionModsFileFromOld.exists()) {
+ ModificationFile compactionModsFile =
+ new ModificationFile(compactionModsFileFromOld.getPath());
+ appendCompactionModificationsBefore013(targetResource, compactionModsFile);
+ }
+
+ // delete tmp target file
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
Review comment:
Resolved. And it is better to check and delete the source file, resource file and mods file separately rather than using `TsFileResource.remove()` , because the tsfile may be deleted, and the resource file and mods file may have not been deleted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r831930696
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverHelper.java
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverHelper searches compaction log and call {@link CompactionRecoverTask} to execute
+ * the recover process for all compaction task sequentially, including InnerCompactionTask in
+ * sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverHelper {
Review comment:
Name of `CompactionRecoverHelper ` feels more like a util class, how about rename to `CompactionRecoveryManager`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r831943804
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
Review comment:
Looks like the nested class is unnecessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r832972816
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverHelper.java
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverHelper searches compaction log and call {@link CompactionRecoverTask} to execute
+ * the recover process for all compaction task sequentially, including InnerCompactionTask in
+ * sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverHelper {
+ private static final Logger logger =
+ LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private TsFileManager tsFileManager;
+ private String logicalStorageGroupName;
+ private String virtualStorageGroupId;
+
+ public CompactionRecoverHelper(
+ TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ this.tsFileManager = tsFileManager;
+ this.logicalStorageGroupName = logicalStorageGroupName;
+ this.virtualStorageGroupId = virtualStorageGroupId;
+ }
+
+ public void recoverInnerSpaceCompaction(boolean isSequence) throws Exception {
+ // search compaction log for SizeTieredCompaction
+ List<String> dirs;
+ if (isSequence) {
+ dirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
+ } else {
+ dirs = DirectoryManager.getInstance().getAllUnSequenceFileFolders();
+ }
+ for (String dir : dirs) {
+ File storageGroupDir =
+ new File(
+ dir
+ + File.separator
+ + logicalStorageGroupName
+ + File.separator
+ + virtualStorageGroupId);
+ if (!storageGroupDir.exists()) {
+ return;
+ }
+ File[] timePartitionDirs = storageGroupDir.listFiles();
+ if (timePartitionDirs == null) {
+ return;
+ }
+ for (File timePartitionDir : timePartitionDirs) {
+ if (!timePartitionDir.isDirectory()
+ || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
+ continue;
+ }
+ File[] compactionLogs =
+ InnerSpaceCompactionUtils.findInnerSpaceCompactionLogs(timePartitionDir.getPath());
+ for (File compactionLog : compactionLogs) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ Long.parseLong(
+ timePartitionDir
+ .getPath()
+ .substring(timePartitionDir.getPath().lastIndexOf(File.separator) + 1)),
+ compactionLog,
+ timePartitionDir.getPath(),
+ isSequence,
+ tsFileManager)
+ .call();
+ }
+ }
+ }
+
+ // search compaction log for old LevelCompaction
+ File logFile =
+ FSFactoryProducer.getFSFactory()
+ .getFile(
+ tsFileManager.getStorageGroupDir(),
+ logicalStorageGroupName + INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD);
+ if (logFile.exists()) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ -1,
+ logFile,
+ logFile.getParent(),
+ isSequence,
+ tsFileManager)
+ .call();
+ }
+ }
+
+ public void recoverCrossSpaceCompaction() throws Exception {
+ logger.info("recovering cross compaction");
+ recoverCrossCompactionFromOldVersion();
+ recoverCrossCompaction();
+ logger.info("try to synchronize CompactionScheduler");
+ }
+
+ private void recoverCrossCompaction() throws Exception {
+ List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
+ for (String dir : sequenceDirs) {
+ File storageGroupDir =
+ new File(
+ dir
+ + File.separator
+ + logicalStorageGroupName
+ + File.separator
+ + virtualStorageGroupId);
+ if (!storageGroupDir.exists()) {
+ return;
+ }
+ File[] timePartitionDirs = storageGroupDir.listFiles();
+ if (timePartitionDirs == null) {
+ return;
+ }
+ for (File timePartitionDir : timePartitionDirs) {
+ if (!timePartitionDir.isDirectory()
+ || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
+ continue;
+ }
+ File[] compactionLogs =
+ CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
+ for (File compactionLog : compactionLogs) {
+ logger.info("calling cross compaction task");
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getCrossCompactionStrategy()
+ .getCompactionRecoverTask(
+ logicalStorageGroupName,
+ virtualStorageGroupId,
+ Long.parseLong(timePartitionDir.getName()),
+ compactionLog,
+ tsFileManager)
+ .call();
+ }
+ }
+ }
+ }
+
+ private void recoverCrossCompactionFromOldVersion() throws Exception {
Review comment:
Resolved
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
Review comment:
Resolved
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
Review comment:
Resolved
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistFromOld(
Review comment:
Resolved
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistFromOld(
+ List<TsFileIdentifier> targetFileIdentifiers) {
+ // delete tmp target file
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile.merge
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * 1. If target file does not exist, then move .merge file to target file <br>
+ * 2. If target resource file does not exist, then serialize it. <br>
+ * 3. Append merging modification to target mods file and delete merging mods file. <br>
+ * 4. Delete source files and .merge file. <br>
+ */
+ private boolean handleCrossCompactionWithSomeSourceFilesLostFromOld(
Review comment:
Resolved
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistFromOld(
+ List<TsFileIdentifier> targetFileIdentifiers) {
+ // delete tmp target file
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile.merge
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * 1. If target file does not exist, then move .merge file to target file <br>
+ * 2. If target resource file does not exist, then serialize it. <br>
+ * 3. Append merging modification to target mods file and delete merging mods file. <br>
+ * 4. Delete source files and .merge file. <br>
+ */
+ private boolean handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers) {
+ try {
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ List<TsFileResource> targetFileResources = new ArrayList<>();
+ for (int i = 0; i < sourceFileIdentifiers.size(); i++) {
+ TsFileIdentifier sourceFileIdentifier = sourceFileIdentifiers.get(i);
+ if (sourceFileIdentifier.isSequence()) {
+ File tmpTargetFile = targetFileIdentifiers.get(i).getFileFromDataDirs();
+ File targetFile = null;
+
+ // move tmp target file to target file if not exist
+ if (tmpTargetFile != null) {
+ // move tmp target file to target file
+ String sourceFilePath =
+ tmpTargetFile
+ .getPath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX);
+ targetFile = TsFileNameGenerator.increaseCrossCompactionCnt(new File(sourceFilePath));
+ FSFactoryProducer.getFSFactory().moveFile(tmpTargetFile, targetFile);
+ } else {
+ // target file must exist
+ File file =
+ TsFileNameGenerator.increaseCrossCompactionCnt(
+ new File(
+ targetFileIdentifiers
+ .get(i)
+ .getFilePath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX)));
+
+ targetFile = getFileFromDataDirs(file.getPath());
+ }
+ if (targetFile == null) {
+ LOGGER.error(
+ "{} [Compaction][Recover] target file of source seq file {} does not exist (<0.13).",
+ fullStorageGroupName,
+ sourceFileIdentifier.getFilePath());
+ return false;
+ }
+
+ // serialize target resource file if not exist
+ TsFileResource targetResource = new TsFileResource(targetFile);
+ if (!targetResource.resourceFileExists()) {
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(targetFile.getAbsolutePath())) {
+ FileLoaderUtils.updateTsFileResource(reader, targetResource);
+ }
+ targetResource.serialize();
+ }
+
+ targetFileResources.add(targetResource);
+
+ // append compaction modifications to target mods file and delete compaction mods file
+ if (compactionModsFileFromOld.exists()) {
+ ModificationFile compactionModsFile =
+ new ModificationFile(compactionModsFileFromOld.getPath());
+ appendCompactionModificationsFromOld(targetResource, compactionModsFile);
+ }
+
+ // delete tmp target file
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+
+ // delete source tsfile
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ sourceFile.delete();
+ }
+
+ // delete source resource file
+ sourceFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + TsFileResource.RESOURCE_SUFFIX);
+ if (sourceFile != null) {
+ sourceFile.delete();
+ }
+
+ // delete source mods file
+ sourceFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (sourceFile != null) {
+ sourceFile.delete();
+ }
+ }
+
+ // delete compaction mods file
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to handle with some source files lost from old version.",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+
+ return true;
+ }
+
+ private void appendCompactionModificationsFromOld(
Review comment:
Resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r833377488
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
Review comment:
Resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r831943804
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
Review comment:
Nested class is unnecessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r831919283
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
##########
@@ -202,13 +201,29 @@ protected void doCompaction() throws Exception {
if (sizeTieredCompactionLogger != null) {
sizeTieredCompactionLogger.close();
}
- InnerSpaceCompactionExceptionHandler.handleException(
- fullStorageGroupName,
- logFile,
- targetTsFileResource,
- selectedTsFileResourceList,
- tsFileManager,
- tsFileResourceList);
+ if (isSequence()) {
+ CompactionExceptionHandler.handleException(
Review comment:
any difference between the two `CompactionExceptionHandler.handleException()` call?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r834980963
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverManager.java
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverManager searches compaction log and call {@link CompactionRecoverTask} to
+ * execute the recover process for all compaction task sequentially, including InnerCompactionTask
+ * in sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverManager {
+ private static final Logger logger =
+ LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final TsFileManager tsFileManager;
+ private final String logicalStorageGroupName;
+ private final String virtualStorageGroupId;
+
+ public CompactionRecoverManager(
+ TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ this.tsFileManager = tsFileManager;
+ this.logicalStorageGroupName = logicalStorageGroupName;
+ this.virtualStorageGroupId = virtualStorageGroupId;
+ }
+
+ public void recoverInnerSpaceCompaction(boolean isSequence) {
+ // search compaction log for SizeTieredCompaction
+ List<String> dirs;
+ if (isSequence) {
+ dirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
+ } else {
+ dirs = DirectoryManager.getInstance().getAllUnSequenceFileFolders();
+ }
+ for (String dir : dirs) {
+ File storageGroupDir =
+ new File(
+ dir
+ + File.separator
+ + logicalStorageGroupName
+ + File.separator
+ + virtualStorageGroupId);
+ if (!storageGroupDir.exists()) {
+ return;
+ }
+ File[] timePartitionDirs = storageGroupDir.listFiles();
+ if (timePartitionDirs == null) {
+ return;
+ }
+ for (File timePartitionDir : timePartitionDirs) {
+ if (!timePartitionDir.isDirectory()
+ || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
+ continue;
+ }
+ File[] compactionLogs =
+ InnerSpaceCompactionUtils.findInnerSpaceCompactionLogs(timePartitionDir.getPath());
+ for (File compactionLog : compactionLogs) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ compactionLog,
+ tsFileManager)
+ .doCompaction();
+ }
+ }
+ }
+
+ // search compaction log for old LevelCompaction
+ File logFile =
+ FSFactoryProducer.getFSFactory()
+ .getFile(
+ tsFileManager.getStorageGroupDir(),
+ logicalStorageGroupName + INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD);
+ if (logFile.exists()) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ logFile,
+ tsFileManager)
+ .doCompaction();
+ }
+ }
+
+ public void recoverCrossSpaceCompaction() {
+ logger.info("recovering cross compaction");
+ recoverCrossCompactionBefore013();
Review comment:
Resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] THUMarkLau commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
THUMarkLau commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r834020060
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,483 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.utils.TsFileUtils;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
+ private final String fullStorageGroupName;
+ private final TsFileManager tsFileManager;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ TsFileManager tsFileManager,
+ File logFile,
+ boolean isInnerSpace) {
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
+ this.fullStorageGroupName = logicalStorageGroupName + "-" + virtualStorageGroupName;
this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ public void doCompaction() {
+ boolean handleSuccess = true;
Review comment:
better to rename it to `recoverSuccess`
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/CrossCompactionStrategy.java
##########
@@ -59,22 +59,16 @@ public AbstractCrossSpaceCompactionTask getCompactionTask(
}
}
- public AbstractCrossSpaceCompactionTask getCompactionRecoverTask(
+ public CompactionRecoverTask getCompactionRecoverTask(
String logicalStorageGroupName,
String virtualStorageGroupName,
- long timePartitionId,
File logFile,
TsFileManager tsFileManager) {
switch (this) {
case REWRITE_COMPACTION:
default:
- return new RewriteCrossCompactionRecoverTask(
Review comment:
Since both inner space and cross space compaction recovery is executed in `CompactionRecoverTask`, why not just new a instance of `CompactionRecoverTask` in the recover process rather than use a factory method
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,483 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.utils.TsFileUtils;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
+ private final String fullStorageGroupName;
+ private final TsFileManager tsFileManager;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ TsFileManager tsFileManager,
+ File logFile,
+ boolean isInnerSpace) {
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
+ this.fullStorageGroupName = logicalStorageGroupName + "-" + virtualStorageGroupName;
this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isInnerCompactionLogBefore013()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isCrossCompactionLogBefore013()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistBefore013(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(targetFileIdentifiers, sourceFileIdentifiers);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostBefore013(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithSomeSourceFilesLost(targetFileIdentifiers, sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers, List<TsFileIdentifier> sourceFileIdentifiers) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(
+ List<TsFileIdentifier> targetFileIdentifiers, List<TsFileIdentifier> sourceFileIdentifiers)
+ throws IOException {
+ // some source files have been deleted, while target file must exist and complete.
+ if (!checkIsTargetFilesComplete(targetFileIdentifiers)) {
+ return false;
+ }
+
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
+ }
+ }
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
+ }
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ private boolean checkIsTargetFilesComplete(List<TsFileIdentifier> targetFileIdentifiers)
+ throws IOException {
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ if (targetFile == null
+ || !TsFileUtils.isTsFileComplete(new TsFileResource(targetFile).getTsFile())) {
+ LOGGER.error(
+ "{} [Compaction][ExceptionHandler] target file {} is not complete, and some source files is lost, do nothing. Set allowCompaction to false",
+ fullStorageGroupName,
+ targetFileIdentifier.getFilePath());
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
+ return false;
}
}
+ return true;
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isCrossCompactionLogBefore013() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isInnerCompactionLogBefore013() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistBefore013(
+ List<TsFileIdentifier> targetFileIdentifiers) {
+ // delete tmp target file
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile.merge
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * 1. If target file does not exist, then move .merge file to target file <br>
+ * 2. If target resource file does not exist, then serialize it. <br>
+ * 3. Append merging modification to target mods file and delete merging mods file. <br>
+ * 4. Delete source files and .merge file. <br>
+ */
+ private boolean handleCrossCompactionWithSomeSourceFilesLostBefore013(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers) {
+ try {
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ List<TsFileResource> targetFileResources = new ArrayList<>();
+ for (int i = 0; i < sourceFileIdentifiers.size(); i++) {
+ TsFileIdentifier sourceFileIdentifier = sourceFileIdentifiers.get(i);
+ if (sourceFileIdentifier.isSequence()) {
+ File tmpTargetFile = targetFileIdentifiers.get(i).getFileFromDataDirs();
+ File targetFile = null;
+
+ // move tmp target file to target file if not exist
+ if (tmpTargetFile != null) {
+ // move tmp target file to target file
+ String sourceFilePath =
+ tmpTargetFile
+ .getPath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX);
+ targetFile = TsFileNameGenerator.increaseCrossCompactionCnt(new File(sourceFilePath));
+ FSFactoryProducer.getFSFactory().moveFile(tmpTargetFile, targetFile);
+ } else {
+ // target file must exist
+ File file =
+ TsFileNameGenerator.increaseCrossCompactionCnt(
+ new File(
+ targetFileIdentifiers
+ .get(i)
+ .getFilePath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX)));
+
+ targetFile = getFileFromDataDirs(file.getPath());
+ }
+ if (targetFile == null) {
+ LOGGER.error(
+ "{} [Compaction][Recover] target file of source seq file {} does not exist (<0.13).",
+ fullStorageGroupName,
+ sourceFileIdentifier.getFilePath());
+ return false;
+ }
+
+ // serialize target resource file if not exist
+ TsFileResource targetResource = new TsFileResource(targetFile);
+ if (!targetResource.resourceFileExists()) {
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(targetFile.getAbsolutePath())) {
+ FileLoaderUtils.updateTsFileResource(reader, targetResource);
+ }
+ targetResource.serialize();
+ }
+
+ targetFileResources.add(targetResource);
+
+ // append compaction modifications to target mods file and delete compaction mods file
+ if (compactionModsFileFromOld.exists()) {
+ ModificationFile compactionModsFile =
+ new ModificationFile(compactionModsFileFromOld.getPath());
+ appendCompactionModificationsBefore013(targetResource, compactionModsFile);
+ }
+
+ // delete tmp target file
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
Review comment:
check the return value of all `delete` function, or use `FileUtils.delete` which throws exception if the deletion is failed. Here we can simply new a `TsFileResource` instance, and use the `remove` method to delete file, resourace file and modification.
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverManager.java
##########
@@ -0,0 +1,181 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverManager searches compaction log and call {@link CompactionRecoverTask} to
+ * execute the recover process for all compaction task sequentially, including InnerCompactionTask
+ * in sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverManager {
+ private static final Logger logger =
+ LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final TsFileManager tsFileManager;
+ private final String logicalStorageGroupName;
+ private final String virtualStorageGroupId;
+
+ public CompactionRecoverManager(
+ TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ this.tsFileManager = tsFileManager;
+ this.logicalStorageGroupName = logicalStorageGroupName;
+ this.virtualStorageGroupId = virtualStorageGroupId;
+ }
+
+ public void recoverInnerSpaceCompaction(boolean isSequence) {
+ // search compaction log for SizeTieredCompaction
+ List<String> dirs;
+ if (isSequence) {
+ dirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
+ } else {
+ dirs = DirectoryManager.getInstance().getAllUnSequenceFileFolders();
+ }
+ for (String dir : dirs) {
+ File storageGroupDir =
+ new File(
+ dir
+ + File.separator
+ + logicalStorageGroupName
+ + File.separator
+ + virtualStorageGroupId);
+ if (!storageGroupDir.exists()) {
+ return;
+ }
+ File[] timePartitionDirs = storageGroupDir.listFiles();
+ if (timePartitionDirs == null) {
+ return;
+ }
+ for (File timePartitionDir : timePartitionDirs) {
+ if (!timePartitionDir.isDirectory()
+ || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
+ continue;
+ }
+ File[] compactionLogs =
+ InnerSpaceCompactionUtils.findInnerSpaceCompactionLogs(timePartitionDir.getPath());
+ for (File compactionLog : compactionLogs) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ compactionLog,
+ tsFileManager)
+ .doCompaction();
+ }
+ }
+ }
+
+ // search compaction log for old LevelCompaction
+ File logFile =
+ FSFactoryProducer.getFSFactory()
+ .getFile(
+ tsFileManager.getStorageGroupDir(),
+ logicalStorageGroupName + INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD);
+ if (logFile.exists()) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ logFile,
+ tsFileManager)
+ .doCompaction();
+ }
+ }
+
+ public void recoverCrossSpaceCompaction() {
+ logger.info("recovering cross compaction");
+ recoverCrossCompactionBefore013();
Review comment:
Please unify the styles of inner-space recovery and cross-space recovery
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionExceptionHandler.java
##########
@@ -182,38 +194,40 @@ private static boolean handleWhenAllSourceFilesExist(
}
/**
- * Some source files are lost, check if the compaction has finished. If the compaction has
- * finished, delete the remaining source files and compaction mods files. If the compaction has
- * not finished, set the allowCompaction in tsFileManager to false and print some error logs.
+ * Some source files are lost, check if all target files are complete. If all target files are
+ * complete, delete the remaining source files and compaction mods files. If some target files are
+ * not complete, set the allowCompaction in tsFileManager to false and print some error logs.
*/
- public static boolean handleWhenSomeSourceFilesLost(
- String storageGroup,
- List<TsFileResource> seqFileList,
- List<TsFileResource> unseqFileList,
- List<TsFileResource> targetFileList,
- List<TsFileResource> lostSourceFiles)
+ private static boolean handleWhenSomeSourceFilesLost(
+ List<TsFileResource> targetResourceList,
+ List<TsFileResource> sourceSeqResourceList,
+ List<TsFileResource> sourceUnseqResourceList,
+ List<TsFileResource> lostSourceResourceList,
+ String fullStorageGroupName)
throws IOException {
- if (!checkIsTargetFilesComplete(targetFileList, lostSourceFiles, storageGroup)) {
+ // check whether is all target files complete
+ if (!checkIsTargetFilesComplete(
+ targetResourceList, lostSourceResourceList, fullStorageGroupName)) {
return false;
}
// delete source files
- for (TsFileResource unseqFile : unseqFileList) {
- unseqFile.remove();
- unseqFile.setStatus(TsFileResourceStatus.DELETED);
+ for (TsFileResource resource : sourceSeqResourceList) {
+ resource.setStatus(TsFileResourceStatus.DELETED);
+ resource.remove();
Review comment:
remove the tsfile resource from the `TsFileResourceManager`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r831944438
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
Review comment:
How about rename to `isCrossCompactionLogBefore013` ?
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistFromOld(
Review comment:
The method should specify which versions could be handle.
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistFromOld(
+ List<TsFileIdentifier> targetFileIdentifiers) {
+ // delete tmp target file
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile.merge
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * 1. If target file does not exist, then move .merge file to target file <br>
+ * 2. If target resource file does not exist, then serialize it. <br>
+ * 3. Append merging modification to target mods file and delete merging mods file. <br>
+ * 4. Delete source files and .merge file. <br>
+ */
+ private boolean handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers) {
+ try {
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ List<TsFileResource> targetFileResources = new ArrayList<>();
+ for (int i = 0; i < sourceFileIdentifiers.size(); i++) {
+ TsFileIdentifier sourceFileIdentifier = sourceFileIdentifiers.get(i);
+ if (sourceFileIdentifier.isSequence()) {
+ File tmpTargetFile = targetFileIdentifiers.get(i).getFileFromDataDirs();
+ File targetFile = null;
+
+ // move tmp target file to target file if not exist
+ if (tmpTargetFile != null) {
+ // move tmp target file to target file
+ String sourceFilePath =
+ tmpTargetFile
+ .getPath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX);
+ targetFile = TsFileNameGenerator.increaseCrossCompactionCnt(new File(sourceFilePath));
+ FSFactoryProducer.getFSFactory().moveFile(tmpTargetFile, targetFile);
+ } else {
+ // target file must exist
+ File file =
+ TsFileNameGenerator.increaseCrossCompactionCnt(
+ new File(
+ targetFileIdentifiers
+ .get(i)
+ .getFilePath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX)));
+
+ targetFile = getFileFromDataDirs(file.getPath());
+ }
+ if (targetFile == null) {
+ LOGGER.error(
+ "{} [Compaction][Recover] target file of source seq file {} does not exist (<0.13).",
+ fullStorageGroupName,
+ sourceFileIdentifier.getFilePath());
+ return false;
+ }
+
+ // serialize target resource file if not exist
+ TsFileResource targetResource = new TsFileResource(targetFile);
+ if (!targetResource.resourceFileExists()) {
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(targetFile.getAbsolutePath())) {
+ FileLoaderUtils.updateTsFileResource(reader, targetResource);
+ }
+ targetResource.serialize();
+ }
+
+ targetFileResources.add(targetResource);
+
+ // append compaction modifications to target mods file and delete compaction mods file
+ if (compactionModsFileFromOld.exists()) {
+ ModificationFile compactionModsFile =
+ new ModificationFile(compactionModsFileFromOld.getPath());
+ appendCompactionModificationsFromOld(targetResource, compactionModsFile);
+ }
+
+ // delete tmp target file
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+
+ // delete source tsfile
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ sourceFile.delete();
+ }
+
+ // delete source resource file
+ sourceFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + TsFileResource.RESOURCE_SUFFIX);
+ if (sourceFile != null) {
+ sourceFile.delete();
+ }
+
+ // delete source mods file
+ sourceFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (sourceFile != null) {
+ sourceFile.delete();
+ }
+ }
+
+ // delete compaction mods file
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to handle with some source files lost from old version.",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+
+ return true;
+ }
+
+ private void appendCompactionModificationsFromOld(
Review comment:
Same as above. Old is not specific.
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistFromOld(
+ List<TsFileIdentifier> targetFileIdentifiers) {
+ // delete tmp target file
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile.merge
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * 1. If target file does not exist, then move .merge file to target file <br>
+ * 2. If target resource file does not exist, then serialize it. <br>
+ * 3. Append merging modification to target mods file and delete merging mods file. <br>
+ * 4. Delete source files and .merge file. <br>
+ */
+ private boolean handleCrossCompactionWithSomeSourceFilesLostFromOld(
Review comment:
Same as above.
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,481 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
+import java.util.concurrent.atomic.AtomicInteger;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
-public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
+public class CompactionRecoverTask extends AbstractCompactionTask {
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
- this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ long timePartition,
+ TsFileManager tsFileManager,
+ AtomicInteger currentTaskNum,
+ File logFile,
+ boolean isInnerSpace) {
+ super(
+ logicalStorageGroupName + "-" + virtualStorageGroupName,
+ timePartition,
+ tsFileManager,
+ currentTaskNum);
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ @Override
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isOldInnerCompactionLog()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isOldCrossCompactionLog()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistFromOld(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(
+ targetFileIdentifiers, sourceFileIdentifiers, isInnerSpace);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostFromOld(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess = handleWithSomeSourceFilesLost(sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers,
+ boolean isInnerSpace) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(List<TsFileIdentifier> sourceFileIdentifiers) {
+ // some source files have been deleted, while target file must exist.
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
}
}
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
}
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public boolean equalsOtherTask(AbstractCompactionTask otherTask) {
+ if (otherTask instanceof CompactionRecoverTask) {
+ return compactionLogFile.equals(((CompactionRecoverTask) otherTask).compactionLogFile);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean checkValidAndSetMerging() {
+ return compactionLogFile.exists();
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ // do nothing
+ }
+
+ @Override
+ public void resetCompactionCandidateStatusForAllSourceFiles() {
+ // do nothing
+ }
+
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isOldCrossCompactionLog() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isOldInnerCompactionLog() {
Review comment:
How about rename to `isInnerCompactionLogBefore013`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r834981844
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,483 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.utils.TsFileUtils;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
+ private final String fullStorageGroupName;
+ private final TsFileManager tsFileManager;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ TsFileManager tsFileManager,
+ File logFile,
+ boolean isInnerSpace) {
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
+ this.fullStorageGroupName = logicalStorageGroupName + "-" + virtualStorageGroupName;
this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ public void doCompaction() {
+ boolean handleSuccess = true;
Review comment:
Resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] qiaojialin merged pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
qiaojialin merged pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] THUMarkLau commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
THUMarkLau commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r835031124
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverTask.java
##########
@@ -20,102 +20,483 @@
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
-import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.CompactionUtils;
+import org.apache.iotdb.db.engine.compaction.TsFileIdentifier;
+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;
+import org.apache.iotdb.db.engine.storagegroup.TsFileNameGenerator;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.utils.FileLoaderUtils;
+import org.apache.iotdb.tsfile.common.constant.TsFileConstant;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+import org.apache.iotdb.tsfile.read.TsFileSequenceReader;
+import org.apache.iotdb.tsfile.utils.TsFileUtils;
+import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
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.regex.Pattern;
-/**
- * CompactionRecoverTask execute the recover process for all compaction task sequentially, including
- * InnerCompactionTask in sequence/unsequence space, CrossSpaceCompaction.
- */
+/** CompactionRecoverTask executes the recover process for all compaction tasks. */
public class CompactionRecoverTask {
- private static final Logger logger =
- LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
- private TsFileManager tsFileManager;
- private String logicalStorageGroupName;
- private String virtualStorageGroupId;
+ private final Logger LOGGER = LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private final File compactionLogFile;
+ private final boolean isInnerSpace;
+ private final String fullStorageGroupName;
+ private final TsFileManager tsFileManager;
public CompactionRecoverTask(
- TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ String logicalStorageGroupName,
+ String virtualStorageGroupName,
+ TsFileManager tsFileManager,
+ File logFile,
+ boolean isInnerSpace) {
+ this.compactionLogFile = logFile;
+ this.isInnerSpace = isInnerSpace;
+ this.fullStorageGroupName = logicalStorageGroupName + "-" + virtualStorageGroupName;
this.tsFileManager = tsFileManager;
- this.logicalStorageGroupName = logicalStorageGroupName;
- this.virtualStorageGroupId = virtualStorageGroupId;
}
- public void recoverCrossSpaceCompaction() throws Exception {
- logger.info("recovering cross compaction");
- recoverCrossCompactionFromOldVersion();
- recoverCrossCompaction();
- logger.info("try to synchronize CompactionScheduler");
+ public void doCompaction() {
+ boolean handleSuccess = true;
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log is {}", fullStorageGroupName, compactionLogFile);
+ try {
+ if (compactionLogFile.exists()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] compaction log file {} exists, start to recover it",
+ fullStorageGroupName,
+ compactionLogFile);
+ CompactionLogAnalyzer logAnalyzer = new CompactionLogAnalyzer(compactionLogFile);
+ CompactionRecoverFromOld compactionRecoverFromOld = new CompactionRecoverFromOld();
+ if (isInnerSpace && compactionRecoverFromOld.isInnerCompactionLogBefore013()) {
+ // inner compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldInnerCompactionLog();
+ } else if (!isInnerSpace && compactionRecoverFromOld.isCrossCompactionLogBefore013()) {
+ // cross compaction log from previous version (<0.13)
+ logAnalyzer.analyzeOldCrossCompactionLog();
+ } else {
+ logAnalyzer.analyze();
+ }
+ List<TsFileIdentifier> sourceFileIdentifiers = logAnalyzer.getSourceFileInfos();
+ List<TsFileIdentifier> targetFileIdentifiers = logAnalyzer.getTargetFileInfos();
+
+ // compaction log file is incomplete
+ if (targetFileIdentifiers.isEmpty() || sourceFileIdentifiers.isEmpty()) {
+ LOGGER.info(
+ "{} [Compaction][Recover] incomplete log file, abort recover", fullStorageGroupName);
+ return;
+ }
+
+ // check is all source files existed
+ boolean isAllSourcesFileExisted = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile == null) {
+ isAllSourcesFileExisted = false;
+ break;
+ }
+ }
+
+ if (isAllSourcesFileExisted) {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithAllSourceFilesExistBefore013(
+ targetFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithAllSourceFilesExist(targetFileIdentifiers, sourceFileIdentifiers);
+ }
+ } else {
+ if (!isInnerSpace && logAnalyzer.isLogFromOld()) {
+ handleSuccess =
+ compactionRecoverFromOld.handleCrossCompactionWithSomeSourceFilesLostBefore013(
+ targetFileIdentifiers, sourceFileIdentifiers);
+ } else {
+ handleSuccess =
+ handleWithSomeSourceFilesLost(targetFileIdentifiers, sourceFileIdentifiers);
+ }
+ }
+ }
+ } catch (IOException e) {
+ LOGGER.error("Recover compaction error", e);
+ } finally {
+ if (!handleSuccess) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Failed to recover compaction, set allowCompaction to false",
+ fullStorageGroupName);
+ tsFileManager.setAllowCompaction(false);
+ } else {
+ if (compactionLogFile.exists()) {
+ try {
+ LOGGER.info(
+ "{} [Compaction][Recover] Recover compaction successfully, delete log file {}",
+ fullStorageGroupName,
+ compactionLogFile);
+ FileUtils.delete(compactionLogFile);
+ } catch (IOException e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting log file {}, set allowCompaction to false",
+ fullStorageGroupName,
+ compactionLogFile,
+ e);
+ tsFileManager.setAllowCompaction(false);
+ }
+ }
+ }
+ }
}
- private void recoverCrossCompaction() throws Exception {
- List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
- for (String dir : sequenceDirs) {
- File storageGroupDir =
- new File(
- dir
- + File.separator
- + logicalStorageGroupName
- + File.separator
- + virtualStorageGroupId);
- if (!storageGroupDir.exists()) {
- return;
+ /**
+ * All source files exist: (1) delete all the target files and tmp target files (2) delete
+ * compaction mods files.
+ */
+ private boolean handleWithAllSourceFilesExist(
+ List<TsFileIdentifier> targetFileIdentifiers, List<TsFileIdentifier> sourceFileIdentifiers) {
+ LOGGER.info(
+ "{} [Compaction][Recover] all source files exists, delete all target files.",
+ fullStorageGroupName);
+
+ // remove tmp target files and target files
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.inner or xxx.cross
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ TsFileResource targetResource = null;
+ if (tmpTargetFile != null) {
+ targetResource = new TsFileResource(tmpTargetFile);
+ } else if (targetFile != null) {
+ targetResource = new TsFileResource(targetFile);
}
- File[] timePartitionDirs = storageGroupDir.listFiles();
- if (timePartitionDirs == null) {
- return;
+
+ 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(
+ "{} [Compaction][Recover] failed to remove target file {}",
+ fullStorageGroupName,
+ targetResource);
+ return false;
}
- for (File timePartitionDir : timePartitionDirs) {
- if (!timePartitionDir.isDirectory()
- || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
- continue;
+ }
+
+ // delete compaction mods files
+ List<TsFileResource> sourceTsFileResourceList = new ArrayList<>();
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ sourceTsFileResourceList.add(new TsFileResource(sourceFileIdentifier.getFileFromDataDirs()));
+ }
+ try {
+ CompactionUtils.deleteCompactionModsFile(sourceTsFileResourceList, Collections.emptyList());
+ } catch (Throwable e) {
+ LOGGER.error(
+ "{} [Compaction][Recover] Exception occurs while deleting compaction mods file, set allowCompaction to false",
+ fullStorageGroupName,
+ e);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Some source files lost: delete remaining source files, including: tsfile, resource file, mods
+ * file and compaction mods file.
+ */
+ private boolean handleWithSomeSourceFilesLost(
+ List<TsFileIdentifier> targetFileIdentifiers, List<TsFileIdentifier> sourceFileIdentifiers)
+ throws IOException {
+ // some source files have been deleted, while target file must exist and complete.
+ if (!checkIsTargetFilesComplete(targetFileIdentifiers)) {
+ return false;
+ }
+
+ boolean handleSuccess = true;
+ for (TsFileIdentifier sourceFileIdentifier : sourceFileIdentifiers) {
+ File sourceFile = sourceFileIdentifier.getFileFromDataDirs();
+ if (sourceFile != null) {
+ // delete source tsfile, resource file and mods file
+ if (!new TsFileResource(sourceFile).remove()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete remaining source file {}.",
+ fullStorageGroupName,
+ sourceFile);
+ handleSuccess = false;
}
- File[] compactionLogs =
- CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
- for (File compactionLog : compactionLogs) {
- logger.info("calling cross compaction task");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- Long.parseLong(timePartitionDir.getName()),
- compactionLog,
- tsFileManager)
- .call();
+ } 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 .mods file of source tsfile
+ File modFile =
+ getFileFromDataDirs(sourceFileIdentifier.getFilePath() + ModificationFile.FILE_SUFFIX);
+ if (modFile != null && !modFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ modFile);
+ handleSuccess = false;
+ }
+ }
+ // delete .compaction.mods file of all source files
+ File compactionModFile =
+ getFileFromDataDirs(
+ sourceFileIdentifier.getFilePath() + ModificationFile.COMPACTION_FILE_SUFFIX);
+ if (compactionModFile != null && !compactionModFile.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModFile);
+ handleSuccess = false;
+ }
+ }
+ return handleSuccess;
+ }
+
+ /**
+ * This method find the File object of given filePath by searching it in every data directory. If
+ * the file is not found, it will return null.
+ */
+ private File getFileFromDataDirs(String filePath) {
+ String[] dataDirs = IoTDBDescriptor.getInstance().getConfig().getDataDirs();
+ for (String dataDir : dataDirs) {
+ File f = new File(dataDir, filePath);
+ if (f.exists()) {
+ return f;
+ }
+ }
+ return null;
+ }
+
+ private boolean checkIsTargetFilesComplete(List<TsFileIdentifier> targetFileIdentifiers)
+ throws IOException {
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile
+ File targetFile =
+ getFileFromDataDirs(
+ targetFileIdentifier
+ .getFilePath()
+ .replace(
+ isInnerSpace
+ ? IoTDBConstant.INNER_COMPACTION_TMP_FILE_SUFFIX
+ : IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX,
+ TsFileConstant.TSFILE_SUFFIX));
+ if (targetFile == null
+ || !TsFileUtils.isTsFileComplete(new TsFileResource(targetFile).getTsFile())) {
+ LOGGER.error(
+ "{} [Compaction][ExceptionHandler] target file {} is not complete, and some source files is lost, do nothing. Set allowCompaction to false",
+ fullStorageGroupName,
+ targetFileIdentifier.getFilePath());
+ IoTDBDescriptor.getInstance().getConfig().setReadOnly(true);
+ return false;
}
}
+ return true;
}
- private void recoverCrossCompactionFromOldVersion() throws Exception {
- // check whether there is old compaction log from previous version (<0.13)
- File mergeLogFromOldVersion =
- new File(
- tsFileManager.getStorageGroupDir()
- + File.separator
- + CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
- if (mergeLogFromOldVersion.exists()) {
- logger.info("calling cross compaction task to recover from previous version.");
- IoTDBDescriptor.getInstance()
- .getConfig()
- .getCrossCompactionStrategy()
- .getCompactionRecoverTask(
- logicalStorageGroupName,
- virtualStorageGroupId,
- 0L,
- mergeLogFromOldVersion,
- tsFileManager)
- .call();
+ /**
+ * Used to check whether it is recoverd from last version (<0.13) and perform corresponding
+ * process.
+ */
+ private class CompactionRecoverFromOld {
+
+ /** Return whether cross compaction log file is from previous version (<0.13). */
+ private boolean isCrossCompactionLogBefore013() {
+ return compactionLogFile
+ .getName()
+ .equals(CompactionLogger.CROSS_COMPACTION_LOG_NAME_FROM_OLD);
+ }
+
+ /** Return whether inner compaction log file is from previous version (<0.13). */
+ private boolean isInnerCompactionLogBefore013() {
+ return compactionLogFile.getName().startsWith(tsFileManager.getStorageGroupName());
+ }
+
+ /** Delete tmp target file and compaction mods file. */
+ private boolean handleCrossCompactionWithAllSourceFilesExistBefore013(
+ List<TsFileIdentifier> targetFileIdentifiers) {
+ // delete tmp target file
+ for (TsFileIdentifier targetFileIdentifier : targetFileIdentifiers) {
+ // xxx.tsfile.merge
+ File tmpTargetFile = targetFileIdentifier.getFileFromDataDirs();
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
+ }
+ }
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ if (compactionModsFileFromOld.exists() && !compactionModsFileFromOld.delete()) {
+ LOGGER.error(
+ "{} [Compaction][Recover] fail to delete target file {}, this may cause data incorrectness",
+ fullStorageGroupName,
+ compactionModsFileFromOld);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * 1. If target file does not exist, then move .merge file to target file <br>
+ * 2. If target resource file does not exist, then serialize it. <br>
+ * 3. Append merging modification to target mods file and delete merging mods file. <br>
+ * 4. Delete source files and .merge file. <br>
+ */
+ private boolean handleCrossCompactionWithSomeSourceFilesLostBefore013(
+ List<TsFileIdentifier> targetFileIdentifiers,
+ List<TsFileIdentifier> sourceFileIdentifiers) {
+ try {
+ File compactionModsFileFromOld =
+ new File(
+ tsFileManager.getStorageGroupDir()
+ + File.separator
+ + IoTDBConstant.COMPACTION_MODIFICATION_FILE_NAME_FROM_OLD);
+ List<TsFileResource> targetFileResources = new ArrayList<>();
+ for (int i = 0; i < sourceFileIdentifiers.size(); i++) {
+ TsFileIdentifier sourceFileIdentifier = sourceFileIdentifiers.get(i);
+ if (sourceFileIdentifier.isSequence()) {
+ File tmpTargetFile = targetFileIdentifiers.get(i).getFileFromDataDirs();
+ File targetFile = null;
+
+ // move tmp target file to target file if not exist
+ if (tmpTargetFile != null) {
+ // move tmp target file to target file
+ String sourceFilePath =
+ tmpTargetFile
+ .getPath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX);
+ targetFile = TsFileNameGenerator.increaseCrossCompactionCnt(new File(sourceFilePath));
+ FSFactoryProducer.getFSFactory().moveFile(tmpTargetFile, targetFile);
+ } else {
+ // target file must exist
+ File file =
+ TsFileNameGenerator.increaseCrossCompactionCnt(
+ new File(
+ targetFileIdentifiers
+ .get(i)
+ .getFilePath()
+ .replace(
+ TsFileConstant.TSFILE_SUFFIX
+ + IoTDBConstant.CROSS_COMPACTION_TMP_FILE_SUFFIX_FROM_OLD,
+ TsFileConstant.TSFILE_SUFFIX)));
+
+ targetFile = getFileFromDataDirs(file.getPath());
+ }
+ if (targetFile == null) {
+ LOGGER.error(
+ "{} [Compaction][Recover] target file of source seq file {} does not exist (<0.13).",
+ fullStorageGroupName,
+ sourceFileIdentifier.getFilePath());
+ return false;
+ }
+
+ // serialize target resource file if not exist
+ TsFileResource targetResource = new TsFileResource(targetFile);
+ if (!targetResource.resourceFileExists()) {
+ try (TsFileSequenceReader reader =
+ new TsFileSequenceReader(targetFile.getAbsolutePath())) {
+ FileLoaderUtils.updateTsFileResource(reader, targetResource);
+ }
+ targetResource.serialize();
+ }
+
+ targetFileResources.add(targetResource);
+
+ // append compaction modifications to target mods file and delete compaction mods file
+ if (compactionModsFileFromOld.exists()) {
+ ModificationFile compactionModsFile =
+ new ModificationFile(compactionModsFileFromOld.getPath());
+ appendCompactionModificationsBefore013(targetResource, compactionModsFile);
+ }
+
+ // delete tmp target file
+ if (tmpTargetFile != null) {
+ tmpTargetFile.delete();
Review comment:
Get it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r831930696
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverHelper.java
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverHelper searches compaction log and call {@link CompactionRecoverTask} to execute
+ * the recover process for all compaction task sequentially, including InnerCompactionTask in
+ * sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverHelper {
Review comment:
Name of `CompactionRecoverHelper ` feels more like a until class, how about rename to `CompactionRecoveryManager`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r833376899
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverHelper.java
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverHelper searches compaction log and call {@link CompactionRecoverTask} to execute
+ * the recover process for all compaction task sequentially, including InnerCompactionTask in
+ * sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverHelper {
Review comment:
Resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] THUMarkLau commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
THUMarkLau commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r835031382
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionExceptionHandler.java
##########
@@ -182,38 +194,40 @@ private static boolean handleWhenAllSourceFilesExist(
}
/**
- * Some source files are lost, check if the compaction has finished. If the compaction has
- * finished, delete the remaining source files and compaction mods files. If the compaction has
- * not finished, set the allowCompaction in tsFileManager to false and print some error logs.
+ * Some source files are lost, check if all target files are complete. If all target files are
+ * complete, delete the remaining source files and compaction mods files. If some target files are
+ * not complete, set the allowCompaction in tsFileManager to false and print some error logs.
*/
- public static boolean handleWhenSomeSourceFilesLost(
- String storageGroup,
- List<TsFileResource> seqFileList,
- List<TsFileResource> unseqFileList,
- List<TsFileResource> targetFileList,
- List<TsFileResource> lostSourceFiles)
+ private static boolean handleWhenSomeSourceFilesLost(
+ List<TsFileResource> targetResourceList,
+ List<TsFileResource> sourceSeqResourceList,
+ List<TsFileResource> sourceUnseqResourceList,
+ List<TsFileResource> lostSourceResourceList,
+ String fullStorageGroupName)
throws IOException {
- if (!checkIsTargetFilesComplete(targetFileList, lostSourceFiles, storageGroup)) {
+ // check whether is all target files complete
+ if (!checkIsTargetFilesComplete(
+ targetResourceList, lostSourceResourceList, fullStorageGroupName)) {
return false;
}
// delete source files
- for (TsFileResource unseqFile : unseqFileList) {
- unseqFile.remove();
- unseqFile.setStatus(TsFileResourceStatus.DELETED);
+ for (TsFileResource resource : sourceSeqResourceList) {
+ resource.setStatus(TsFileResourceStatus.DELETED);
+ resource.remove();
Review comment:
Get it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r834994820
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionExceptionHandler.java
##########
@@ -182,38 +194,40 @@ private static boolean handleWhenAllSourceFilesExist(
}
/**
- * Some source files are lost, check if the compaction has finished. If the compaction has
- * finished, delete the remaining source files and compaction mods files. If the compaction has
- * not finished, set the allowCompaction in tsFileManager to false and print some error logs.
+ * Some source files are lost, check if all target files are complete. If all target files are
+ * complete, delete the remaining source files and compaction mods files. If some target files are
+ * not complete, set the allowCompaction in tsFileManager to false and print some error logs.
*/
- public static boolean handleWhenSomeSourceFilesLost(
- String storageGroup,
- List<TsFileResource> seqFileList,
- List<TsFileResource> unseqFileList,
- List<TsFileResource> targetFileList,
- List<TsFileResource> lostSourceFiles)
+ private static boolean handleWhenSomeSourceFilesLost(
+ List<TsFileResource> targetResourceList,
+ List<TsFileResource> sourceSeqResourceList,
+ List<TsFileResource> sourceUnseqResourceList,
+ List<TsFileResource> lostSourceResourceList,
+ String fullStorageGroupName)
throws IOException {
- if (!checkIsTargetFilesComplete(targetFileList, lostSourceFiles, storageGroup)) {
+ // check whether is all target files complete
+ if (!checkIsTargetFilesComplete(
+ targetResourceList, lostSourceResourceList, fullStorageGroupName)) {
return false;
}
// delete source files
- for (TsFileResource unseqFile : unseqFileList) {
- unseqFile.remove();
- unseqFile.setStatus(TsFileResourceStatus.DELETED);
+ for (TsFileResource resource : sourceSeqResourceList) {
+ resource.setStatus(TsFileResourceStatus.DELETED);
+ resource.remove();
Review comment:
If some source files have been deleted, resource memory must be updated before. Therefore, it is unnecessary to remove here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] choubenson commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
choubenson commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r834981041
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/CrossCompactionStrategy.java
##########
@@ -59,22 +59,16 @@ public AbstractCrossSpaceCompactionTask getCompactionTask(
}
}
- public AbstractCrossSpaceCompactionTask getCompactionRecoverTask(
+ public CompactionRecoverTask getCompactionRecoverTask(
String logicalStorageGroupName,
String virtualStorageGroupName,
- long timePartitionId,
File logFile,
TsFileManager tsFileManager) {
switch (this) {
case REWRITE_COMPACTION:
default:
- return new RewriteCrossCompactionRecoverTask(
Review comment:
Resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [iotdb] chengjianyun commented on a change in pull request #5316: [IOTDB-2561]Compaction code refactoring for restart recovery and exception handling
Posted by GitBox <gi...@apache.org>.
chengjianyun commented on a change in pull request #5316:
URL: https://github.com/apache/iotdb/pull/5316#discussion_r831935738
##########
File path: server/src/main/java/org/apache/iotdb/db/engine/compaction/task/CompactionRecoverHelper.java
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.task;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.conf.directories.DirectoryManager;
+import org.apache.iotdb.db.engine.compaction.inner.utils.InnerSpaceCompactionUtils;
+import org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger;
+import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import static org.apache.iotdb.db.engine.compaction.utils.log.CompactionLogger.INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD;
+
+/**
+ * CompactionRecoverHelper searches compaction log and call {@link CompactionRecoverTask} to execute
+ * the recover process for all compaction task sequentially, including InnerCompactionTask in
+ * sequence/unsequence space, CrossSpaceCompaction.
+ */
+public class CompactionRecoverHelper {
+ private static final Logger logger =
+ LoggerFactory.getLogger(IoTDBConstant.COMPACTION_LOGGER_NAME);
+ private TsFileManager tsFileManager;
+ private String logicalStorageGroupName;
+ private String virtualStorageGroupId;
+
+ public CompactionRecoverHelper(
+ TsFileManager tsFileManager, String logicalStorageGroupName, String virtualStorageGroupId) {
+ this.tsFileManager = tsFileManager;
+ this.logicalStorageGroupName = logicalStorageGroupName;
+ this.virtualStorageGroupId = virtualStorageGroupId;
+ }
+
+ public void recoverInnerSpaceCompaction(boolean isSequence) throws Exception {
+ // search compaction log for SizeTieredCompaction
+ List<String> dirs;
+ if (isSequence) {
+ dirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
+ } else {
+ dirs = DirectoryManager.getInstance().getAllUnSequenceFileFolders();
+ }
+ for (String dir : dirs) {
+ File storageGroupDir =
+ new File(
+ dir
+ + File.separator
+ + logicalStorageGroupName
+ + File.separator
+ + virtualStorageGroupId);
+ if (!storageGroupDir.exists()) {
+ return;
+ }
+ File[] timePartitionDirs = storageGroupDir.listFiles();
+ if (timePartitionDirs == null) {
+ return;
+ }
+ for (File timePartitionDir : timePartitionDirs) {
+ if (!timePartitionDir.isDirectory()
+ || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
+ continue;
+ }
+ File[] compactionLogs =
+ InnerSpaceCompactionUtils.findInnerSpaceCompactionLogs(timePartitionDir.getPath());
+ for (File compactionLog : compactionLogs) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ Long.parseLong(
+ timePartitionDir
+ .getPath()
+ .substring(timePartitionDir.getPath().lastIndexOf(File.separator) + 1)),
+ compactionLog,
+ timePartitionDir.getPath(),
+ isSequence,
+ tsFileManager)
+ .call();
+ }
+ }
+ }
+
+ // search compaction log for old LevelCompaction
+ File logFile =
+ FSFactoryProducer.getFSFactory()
+ .getFile(
+ tsFileManager.getStorageGroupDir(),
+ logicalStorageGroupName + INNER_COMPACTION_LOG_NAME_SUFFIX_FROM_OLD);
+ if (logFile.exists()) {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getInnerCompactionStrategy()
+ .getCompactionRecoverTask(
+ tsFileManager.getStorageGroupName(),
+ tsFileManager.getVirtualStorageGroup(),
+ -1,
+ logFile,
+ logFile.getParent(),
+ isSequence,
+ tsFileManager)
+ .call();
+ }
+ }
+
+ public void recoverCrossSpaceCompaction() throws Exception {
+ logger.info("recovering cross compaction");
+ recoverCrossCompactionFromOldVersion();
+ recoverCrossCompaction();
+ logger.info("try to synchronize CompactionScheduler");
+ }
+
+ private void recoverCrossCompaction() throws Exception {
+ List<String> sequenceDirs = DirectoryManager.getInstance().getAllSequenceFileFolders();
+ for (String dir : sequenceDirs) {
+ File storageGroupDir =
+ new File(
+ dir
+ + File.separator
+ + logicalStorageGroupName
+ + File.separator
+ + virtualStorageGroupId);
+ if (!storageGroupDir.exists()) {
+ return;
+ }
+ File[] timePartitionDirs = storageGroupDir.listFiles();
+ if (timePartitionDirs == null) {
+ return;
+ }
+ for (File timePartitionDir : timePartitionDirs) {
+ if (!timePartitionDir.isDirectory()
+ || !Pattern.compile("[0-9]*").matcher(timePartitionDir.getName()).matches()) {
+ continue;
+ }
+ File[] compactionLogs =
+ CompactionLogger.findCrossSpaceCompactionLogs(timePartitionDir.getPath());
+ for (File compactionLog : compactionLogs) {
+ logger.info("calling cross compaction task");
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getCrossCompactionStrategy()
+ .getCompactionRecoverTask(
+ logicalStorageGroupName,
+ virtualStorageGroupId,
+ Long.parseLong(timePartitionDir.getName()),
+ compactionLog,
+ tsFileManager)
+ .call();
+ }
+ }
+ }
+ }
+
+ private void recoverCrossCompactionFromOldVersion() throws Exception {
Review comment:
It's better to point out the specific version number in method name such `recoverCrossCompactionBelow013` and add comments for the method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org