You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by zy...@apache.org on 2023/03/21 15:20:32 UTC
[iotdb] branch rel/1.1 updated: [To rel/1.1][IOTDB-5639][compaction]Fix file not found exception in cross space compaction selector (#9255)
This is an automated email from the ASF dual-hosted git repository.
zyk pushed a commit to branch rel/1.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.1 by this push:
new d8a34fbf07 [To rel/1.1][IOTDB-5639][compaction]Fix file not found exception in cross space compaction selector (#9255)
d8a34fbf07 is described below
commit d8a34fbf070bb6e5410663470e8c3cf0566237e3
Author: 周沛辰 <45...@users.noreply.github.com>
AuthorDate: Tue Mar 21 23:20:24 2023 +0800
[To rel/1.1][IOTDB-5639][compaction]Fix file not found exception in cross space compaction selector (#9255)
---
.../utils/CrossSpaceCompactionCandidate.java | 44 ++++++++++++++++----
.../db/engine/compaction/cross/MergeTest.java | 1 +
.../cross/RewriteCompactionFileSelectorTest.java | 47 +++++++++++++++++++---
3 files changed, 79 insertions(+), 13 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossSpaceCompactionCandidate.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossSpaceCompactionCandidate.java
index 2b21345957..22da3695e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossSpaceCompactionCandidate.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/selector/utils/CrossSpaceCompactionCandidate.java
@@ -77,20 +77,26 @@ public class CrossSpaceCompactionCandidate {
List<TsFileResourceCandidate> ret = new ArrayList<>();
// The startTime and endTime of each device are different in one TsFile. So we need to do the
- // check
- // one by one. And we cannot skip any device in the unseq file because it may lead to omission
- // of
- // target seq file
+ // check one by one. And we cannot skip any device in the unseq file because it may lead to
+ // omission of target seq file
+ if (!unseqFile.hasDetailedDeviceInfo()) {
+ // It means unseq file resource has been deleted due to TTL and cannot upgrade to
+ // DEVICE_TIME_INDEX
+ return false;
+ }
for (DeviceInfo unseqDeviceInfo : unseqFile.getDevices()) {
for (TsFileResourceCandidate seqFile : seqFiles) {
+ // If the seqFile may need to be selected but its invalid, the selection should be
+ // terminated.
+ if ((!seqFile.isValidCandidate || !seqFile.hasDetailedDeviceInfo())
+ && seqFile.mayHasOverlapWithUnseqFile(unseqDeviceInfo)) {
+ return false;
+ }
if (!seqFile.containsDevice(unseqDeviceInfo.deviceId)) {
continue;
}
DeviceInfo seqDeviceInfo = seqFile.getDeviceInfoById(unseqDeviceInfo.deviceId);
- // If the seqFile should be selected but its invalid, the selection should be terminated.
- if (!seqFile.isValidCandidate && unseqDeviceInfo.startTime <= seqDeviceInfo.endTime) {
- return false;
- }
+
// If the unsealed file is unclosed, the file should not be selected only when its startTime
// is larger than endTime of unseqFile. Or, the selection should be terminated.
if (seqFile.unsealed() && unseqDeviceInfo.endTime >= seqDeviceInfo.startTime) {
@@ -181,6 +187,8 @@ public class CrossSpaceCompactionCandidate {
public boolean isValidCandidate;
private Map<String, DeviceInfo> deviceInfoMap;
+ private boolean hasDetailedDeviceInfo;
+
protected TsFileResourceCandidate(TsFileResource tsFileResource) {
this.resource = tsFileResource;
this.selected = false;
@@ -206,6 +214,10 @@ public class CrossSpaceCompactionCandidate {
}
deviceInfoMap = new LinkedHashMap<>();
if (resource.getTimeIndexType() == ITimeIndex.FILE_TIME_INDEX_TYPE) {
+ if (!resource.resourceFileExists()) {
+ hasDetailedDeviceInfo = false;
+ return;
+ }
DeviceTimeIndex timeIndex = resource.buildDeviceTimeIndex();
for (String deviceId : timeIndex.getDevices()) {
deviceInfoMap.put(
@@ -221,6 +233,7 @@ public class CrossSpaceCompactionCandidate {
deviceId, resource.getStartTime(deviceId), resource.getEndTime(deviceId)));
}
}
+ hasDetailedDeviceInfo = true;
}
protected void markAsSelected() {
@@ -241,6 +254,21 @@ public class CrossSpaceCompactionCandidate {
prepareDeviceInfos();
return deviceInfoMap.containsKey(deviceId);
}
+
+ protected boolean hasDetailedDeviceInfo() throws IOException {
+ prepareDeviceInfos();
+ return hasDetailedDeviceInfo;
+ }
+
+ protected boolean mayHasOverlapWithUnseqFile(DeviceInfo unseqFileDeviceInfo)
+ throws IOException {
+ prepareDeviceInfos();
+ long endTime =
+ containsDevice(unseqFileDeviceInfo.deviceId)
+ ? getDeviceInfoById(unseqFileDeviceInfo.deviceId).endTime
+ : resource.getFileEndTime();
+ return unseqFileDeviceInfo.startTime <= endTime;
+ }
}
protected static class DeviceInfo {
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/MergeTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/MergeTest.java
index 9d2c4e4194..be70513008 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/MergeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/MergeTest.java
@@ -176,6 +176,7 @@ abstract class MergeTest {
}
}
fileWriter.close();
+ tsFileResource.serialize();
}
void mkdirs(File file) {
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCompactionFileSelectorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCompactionFileSelectorTest.java
index 7df9d4ad23..11bb18a7de 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCompactionFileSelectorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/cross/RewriteCompactionFileSelectorTest.java
@@ -39,7 +39,6 @@ import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-import org.apache.commons.io.FileUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -1001,6 +1000,7 @@ public class RewriteCompactionFileSelectorTest extends MergeTest {
new Thread(
() -> {
try {
+ Thread.sleep(1000);
List<CrossCompactionTaskResource> selected =
selector.selectCrossSpaceTask(seqResources, unseqResources);
} catch (Exception e) {
@@ -1009,12 +1009,49 @@ public class RewriteCompactionFileSelectorTest extends MergeTest {
}
});
Thread thread2 =
+ new Thread(
+ () -> {
+ if (!seqResources.get(0).remove()) {
+ fail.set(true);
+ }
+ if (!unseqResources.get(0).remove()) {
+ fail.set(true);
+ }
+ });
+ thread1.start();
+ thread2.start();
+ thread1.join();
+ thread2.join();
+ if (fail.get()) {
+ Assert.fail();
+ }
+ }
+
+ @Test
+ public void testDeleteAndDegradeInSelection() throws Exception {
+ RewriteCrossSpaceCompactionSelector selector =
+ new RewriteCrossSpaceCompactionSelector("", "", 0, null);
+ AtomicBoolean fail = new AtomicBoolean(false);
+ Thread thread1 =
new Thread(
() -> {
try {
- FileUtils.delete(seqResources.get(0).getTsFile());
- FileUtils.delete(unseqResources.get(0).getTsFile());
- } catch (IOException e) {
+ Thread.sleep(1000);
+ List<CrossCompactionTaskResource> selected =
+ selector.selectCrossSpaceTask(seqResources, unseqResources);
+ Assert.assertEquals(1, selected.get(0).getSeqFiles().size());
+ Assert.assertEquals(1, selected.get(0).getUnseqFiles().size());
+ } catch (Exception e) {
+ logger.error("Exception occurs", e);
+ fail.set(true);
+ }
+ });
+ Thread thread2 =
+ new Thread(
+ () -> {
+ seqResources.get(1).degradeTimeIndex();
+ if (!seqResources.get(1).remove()) {
+ fail.set(true);
}
});
thread1.start();
@@ -1022,7 +1059,7 @@ public class RewriteCompactionFileSelectorTest extends MergeTest {
thread1.join();
thread2.join();
if (fail.get()) {
- // fail();
+ Assert.fail();
}
}
}