This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch rel/0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.11 by this push:
new 4daa1a1 [IOTDB-1461][To rel/0.11] Fix compaction conflicts with ttl
(#3486)
4daa1a1 is described below
commit 4daa1a18958223e70d5e38140e3529f25f2b1e8f
Author: zhanglingzhe0820 <[email protected]>
AuthorDate: Mon Jul 5 21:06:17 2021 +0800
[IOTDB-1461][To rel/0.11] Fix compaction conflicts with ttl (#3486)
Co-authored-by: zhanglingzhe <[email protected]>
---
.../level/LevelCompactionTsFileManagement.java | 66 ++++++++++++++--------
.../engine/storagegroup/StorageGroupProcessor.java | 8 +--
2 files changed, 44 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 427a14c..b5ddd05 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
@@ -452,7 +452,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 = getMergeLevel(new File(sourceFileList.get(0)));
RestorableTsFileIOWriter writer = new
RestorableTsFileIOWriter(target);
@@ -498,6 +502,7 @@ public class LevelCompactionTsFileManagement extends
TsFileManagement {
}
} catch (IOException | IllegalPathException e) {
logger.error("recover level tsfile management error ", e);
+ restoreCompaction();
} finally {
if (logFile.exists()) {
try {
@@ -733,7 +738,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);
@@ -820,34 +827,47 @@ 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;
+ }
+ try {
+ 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;
}
+ return null;
+ } finally {
+ readUnLock();
}
- logger.error("cannot get tsfile resource path: {}", filePath);
- throw new IOException();
}
@TestOnly
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 f2b2e20..d7b83fc 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
@@ -1318,8 +1318,7 @@ public class StorageGroupProcessor {
}
private void checkFileTTL(TsFileResource resource, long timeLowerBound,
boolean isSeq) {
- if (resource.isMerging()
- || !resource.isClosed()
+ if (!resource.isClosed()
|| !resource.isDeleted() && resource.stillLives(timeLowerBound)) {
return;
}
@@ -1328,11 +1327,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()) {
