This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch rc/1.3.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rc/1.3.3 by this push:
new 06a3831a603 [to rc/1.3.3] Fix compaction device info cache (#13454)
06a3831a603 is described below
commit 06a3831a603a6244c4770f0dc8788a21899d53d6
Author: shuwenwei <[email protected]>
AuthorDate: Tue Sep 10 10:18:57 2024 +0800
[to rc/1.3.3] Fix compaction device info cache (#13454)
* fix device info cache
* simplify test
---
.../selector/utils/TsFileResourceCandidate.java | 3 +-
.../InsertionCrossSpaceCompactionSelectorTest.java | 76 ++++++++++++++++++++++
2 files changed, 78 insertions(+), 1 deletion(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/utils/TsFileResourceCandidate.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/utils/TsFileResourceCandidate.java
index 0c82b4726eb..f46ca053a79 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/utils/TsFileResourceCandidate.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/utils/TsFileResourceCandidate.java
@@ -72,6 +72,7 @@ public class TsFileResourceCandidate {
}
private void prepareDeviceInfos() throws IOException {
+ boolean canCacheDeviceInfo = resource.getStatus() !=
TsFileResourceStatus.UNCLOSED;
if (deviceInfoMap == null && compactionScheduleContext != null) {
// get device info from cache
deviceInfoMap =
compactionScheduleContext.getResourceDeviceInfo(this.resource);
@@ -107,7 +108,7 @@ public class TsFileResourceCandidate {
}
}
hasDetailedDeviceInfo = true;
- if (compactionScheduleContext != null) {
+ if (compactionScheduleContext != null && canCacheDeviceInfo) {
compactionScheduleContext.addResourceDeviceTimeIndex(this.resource,
deviceInfoMap);
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionSelectorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionSelectorTest.java
index 2fb43073e1b..3f36a2baeb4 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionSelectorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionSelectorTest.java
@@ -25,12 +25,15 @@ import org.apache.iotdb.db.exception.MergeException;
import org.apache.iotdb.db.exception.StorageEngineException;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InsertionCrossSpaceCompactionTask;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionScheduleContext;
+import
org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionScheduler;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.impl.RewriteCrossSpaceCompactionSelector;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.utils.CrossSpaceCompactionCandidate;
import
org.apache.iotdb.db.storageengine.dataregion.compaction.selector.utils.InsertionCrossCompactionTaskResource;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus;
import
org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.DeviceTimeIndex;
+import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.exception.write.WriteProcessException;
@@ -99,6 +102,79 @@ public class InsertionCrossSpaceCompactionSelectorTest
extends AbstractCompactio
Assert.assertEquals(seqResource2, result.nextSeqFile);
}
+ @Test
+ public void testInsertionCompactionWithCachedDeviceInfoAndUnclosedResource()
+ throws InterruptedException, IOException {
+ CompactionScheduleContext context = new CompactionScheduleContext();
+
+ IDeviceID d1 = new PlainDeviceID("root.testsg.d1");
+ IDeviceID d2 = new PlainDeviceID("root.testsg.d2");
+ IDeviceID d3 = new PlainDeviceID("root.testsg.d3");
+
+ TsFileResource seqResource1 = createTsFileResource("1-1-0-0.tsfile", true);
+ seqResource1.getTsFile().createNewFile();
+ seqResource1.updateStartTime(d1, 10);
+ seqResource1.updateEndTime(d1, 20);
+ seqResource1.serialize();
+
+ TsFileResource seqResource2 = createTsFileResource("3-3-0-0.tsfile", true);
+ seqResource2.getTsFile().createNewFile();
+ seqResource2.updateStartTime(d1, 40);
+ seqResource2.updateEndTime(d1, 50);
+ seqResource2.serialize();
+
+ // unclosed
+ TsFileResource seqResource3 = createTsFileResource("6-6-0-0.tsfile", true);
+ seqResource3.getTsFile().createNewFile();
+ seqResource3.setStatusForTest(TsFileResourceStatus.UNCLOSED);
+ seqResource3.updateStartTime(d1, 70);
+
+ seqResources.add(seqResource1);
+ seqResources.add(seqResource2);
+ seqResources.add(seqResource3);
+ TsFileResource unseqResource1 = createTsFileResource("5-5-1-0.tsfile",
false);
+ unseqResource1.getTsFile().createNewFile();
+ unseqResource1.updateStartTime(d1, 30);
+ unseqResource1.updateEndTime(d1, 35);
+ unseqResource1.updateStartTime(d3, 10);
+ unseqResource1.updateEndTime(d3, 20);
+ unseqResource1.serialize();
+ unseqResources.add(unseqResource1);
+
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+
+ Phaser phaser = new Phaser(1);
+ int submitTaskNum =
+ CompactionScheduler.scheduleInsertionCompaction(tsFileManager, 0,
phaser, context);
+ Assert.assertEquals(1, submitTaskNum);
+ // perform insertion compaction
+ phaser.awaitAdvanceInterruptibly(phaser.arrive());
+
+ // unclosed file has sealed
+ seqResource3.updateEndTime(d1, 80);
+ seqResource3.updateStartTime(d2, 10);
+ seqResource3.updateEndTime(d2, 20);
+ seqResource3.setStatusForTest(TsFileResourceStatus.NORMAL);
+
+ TsFileResource unseqResource2 = createTsFileResource("7-7-1-0.tsfile",
false);
+ unseqResource2.getTsFile().createNewFile();
+ unseqResource2.updateStartTime(d2, 10);
+ unseqResource2.updateEndTime(d2, 20);
+ unseqResource2.serialize();
+ tsFileManager.keepOrderInsert(unseqResource2, false);
+ // Should not select unseq resource2
+ // The unclosed resource should not be cached. Otherwise, the results here
will be incorrect.
+ // seq resource3: d1[70, 80] d2[10, 20]
+ // unseq resource2 d2[10, 20]
+
+ submitTaskNum =
+ CompactionScheduler.scheduleInsertionCompaction(tsFileManager, 0,
phaser, context);
+ Assert.assertEquals(0, submitTaskNum);
+ Assert.assertTrue(
+
TsFileResourceUtils.validateTsFileResourcesHasNoOverlap(tsFileManager.getTsFileList(true)));
+ }
+
@Test
public void testSimpleInsertionCompactionWithMultiUnseqFiles()
throws IOException, MergeException {
