This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.12 by this push:
new 82da977 [IOTDB-1461][To rel/0.12] Fix compaction conflicts with ttl
(#3509)
82da977 is described below
commit 82da9773fec13d44483ad39fd26e2264458e0fe9
Author: zhanglingzhe0820 <[email protected]>
AuthorDate: Mon Jul 5 17:13:08 2021 +0800
[IOTDB-1461][To rel/0.12] Fix compaction conflicts with ttl (#3509)
Co-authored-by: zhanglingzhe <[email protected]>
---
.../level/LevelCompactionTsFileManagement.java | 62 ++++++++++++++--------
.../engine/storagegroup/StorageGroupProcessor.java | 9 +---
2 files changed, 41 insertions(+), 30 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
index 5a96761..f740c6b 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/level/LevelCompactionTsFileManagement.java
@@ -476,7 +476,11 @@ public class LevelCompactionTsFileManagement extends
TsFileManagement {
List<TsFileResource> sourceTsFileResources = new ArrayList<>();
for (String file : sourceFileList) {
// get tsfile resource from list, as they have been recovered in
StorageGroupProcessor
- sourceTsFileResources.add(getTsFileResource(file, isSeq));
+ TsFileResource sourceTsFileResource = getTsFileResource(file,
isSeq);
+ if (sourceTsFileResource == null) {
+ throw new IOException();
+ }
+ sourceTsFileResources.add(sourceTsFileResource);
}
int level = TsFileResource.getMergeLevel(new
File(sourceFileList.get(0)).getName());
RestorableTsFileIOWriter writer = new
RestorableTsFileIOWriter(target);
@@ -526,6 +530,7 @@ public class LevelCompactionTsFileManagement extends
TsFileManagement {
}
} catch (IOException | IllegalPathException | InterruptedException e) {
logger.error("recover level tsfile management error ", e);
+ restoreCompaction();
} finally {
if (logFile.exists()) {
try {
@@ -801,34 +806,45 @@ public class LevelCompactionTsFileManagement extends
TsFileManagement {
throw new IOException();
}
- private TsFileResource getTsFileResource(String filePath, boolean isSeq)
throws IOException {
- if (isSeq) {
- for (List<SortedSet<TsFileResource>> tsFileResourcesWithLevel :
- sequenceTsFileResources.values()) {
- for (SortedSet<TsFileResource> tsFileResources :
tsFileResourcesWithLevel) {
- for (TsFileResource tsFileResource : tsFileResources) {
- if (Files.isSameFile(
- tsFileResource.getTsFile().toPath(), new
File(filePath).toPath())) {
- return tsFileResource;
+ private TsFileResource getTsFileResource(String filePath, boolean isSeq) {
+ readLock();
+ try {
+ File file = new File(filePath);
+ if (!file.exists()) {
+ return null;
+ }
+ if (isSeq) {
+ for (List<SortedSet<TsFileResource>> tsFileResourcesWithLevel :
+ sequenceTsFileResources.values()) {
+ for (SortedSet<TsFileResource> tsFileResources :
tsFileResourcesWithLevel) {
+ for (TsFileResource tsFileResource : tsFileResources) {
+ if (Files.isSameFile(
+ tsFileResource.getTsFile().toPath(), new
File(filePath).toPath())) {
+ return tsFileResource;
+ }
}
}
}
- }
- } else {
- for (List<List<TsFileResource>> tsFileResourcesWithLevel :
- unSequenceTsFileResources.values()) {
- for (List<TsFileResource> tsFileResources : tsFileResourcesWithLevel) {
- for (TsFileResource tsFileResource : tsFileResources) {
- if (Files.isSameFile(
- tsFileResource.getTsFile().toPath(), new
File(filePath).toPath())) {
- return tsFileResource;
+ } else {
+ for (List<List<TsFileResource>> tsFileResourcesWithLevel :
+ unSequenceTsFileResources.values()) {
+ for (List<TsFileResource> tsFileResources :
tsFileResourcesWithLevel) {
+ for (TsFileResource tsFileResource : tsFileResources) {
+ if (Files.isSameFile(
+ tsFileResource.getTsFile().toPath(), new
File(filePath).toPath())) {
+ return tsFileResource;
+ }
}
}
}
}
+ } catch (Exception e) {
+ logger.error("cannot get tsfile resource path: {}", filePath, e);
+ return null;
+ } finally {
+ readUnLock();
}
- logger.error("cannot get tsfile resource path: {}", filePath);
- throw new IOException();
+ return null;
}
/** restore the files back to the status before the compaction task is
submitted */
@@ -845,7 +861,9 @@ public class LevelCompactionTsFileManagement extends
TsFileManagement {
boolean isSeq = logAnalyzer.isSeq();
for (String file : sourceFileList) {
TsFileResource fileResource = getTsFileResource(file, isSeq);
- fileResource.setMerging(false);
+ if (fileResource != null) {
+ fileResource.setMerging(false);
+ }
}
if (targetFilePath != null) {
File targetFile = new File(targetFilePath);
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index f3a0a3d..9461bac 100755
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -1431,9 +1431,7 @@ public class StorageGroupProcessor {
}
private void checkFileTTL(TsFileResource resource, long timeLowerBound,
boolean isSeq) {
- if (resource.isMerging()
- || !resource.isClosed()
- || !resource.isDeleted() && resource.stillLives(timeLowerBound)) {
+ if (!resource.isClosed() || !resource.isDeleted() &&
resource.stillLives(timeLowerBound)) {
return;
}
@@ -1441,11 +1439,6 @@ public class StorageGroupProcessor {
try {
// prevent new merges and queries from choosing this file
resource.setDeleted(true);
- // the file may be chosen for merge after the last check and before
writeLock()
- // double check to ensure the file is not used by a merge
- if (resource.isMerging()) {
- return;
- }
// ensure that the file is not used by any queries
if (resource.tryWriteLock()) {
