You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by er...@apache.org on 2022/03/08 08:04:47 UTC
[iotdb] branch master updated: [IOTDB-2651] Fix write performance deteriorates severely (#5170)
This is an automated email from the ASF dual-hosted git repository.
ericpai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new b2d060c [IOTDB-2651] Fix write performance deteriorates severely (#5170)
b2d060c is described below
commit b2d060cf52c5746f37502b9e887ec05ce516e576
Author: Liu Xuxin <37...@users.noreply.github.com>
AuthorDate: Tue Mar 8 16:00:22 2022 +0800
[IOTDB-2651] Fix write performance deteriorates severely (#5170)
---
.../engine/compaction/CompactionTaskManager.java | 1 +
.../cross/AbstractCrossSpaceCompactionTask.java | 14 ++-
.../RewriteCrossSpaceCompactionSelector.java | 2 -
.../task/RewriteCrossSpaceCompactionTask.java | 3 +-
.../inner/AbstractInnerSpaceCompactionTask.java | 5 +
.../sizetiered/SizeTieredCompactionSelector.java | 1 -
.../inner/sizetiered/SizeTieredCompactionTask.java | 2 +-
.../compaction/task/AbstractCompactionTask.java | 2 +
.../db/engine/storagegroup/TsFileResource.java | 4 +-
.../compaction/CompactionTaskManagerTest.java | 104 ++++++++++++++++++++-
10 files changed, 126 insertions(+), 12 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java
index 3284379..b608b14 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManager.java
@@ -218,6 +218,7 @@ public class CompactionTaskManager implements IService {
throws InterruptedException {
if (!candidateCompactionTaskQueue.contains(compactionTask)
&& !runningCompactionTaskList.contains(compactionTask)) {
+ compactionTask.setSourceFilesToCompactionCandidate();
candidateCompactionTaskQueue.put(compactionTask);
// add metrics
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/AbstractCrossSpaceCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/AbstractCrossSpaceCompactionTask.java
index d13d3fa..7fe754c 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/AbstractCrossSpaceCompactionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/AbstractCrossSpaceCompactionTask.java
@@ -47,6 +47,12 @@ public abstract class AbstractCrossSpaceCompactionTask extends AbstractCompactio
this.selectedUnsequenceFiles = null;
}
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ this.selectedSequenceFiles.forEach(x -> x.setCompactionCandidate(true));
+ this.selectedUnsequenceFiles.forEach(x -> x.setCompactionCandidate(true));
+ }
+
public List<TsFileResource> getSelectedSequenceFiles() {
return selectedSequenceFiles;
}
@@ -88,10 +94,10 @@ public abstract class AbstractCrossSpaceCompactionTask extends AbstractCompactio
.append(fullStorageGroupName)
.append("-")
.append(timePartition)
- .append(" task seq file num is ")
- .append(selectedSequenceFiles.size())
- .append(" , unseq file num is ")
- .append(selectedUnsequenceFiles.size())
+ .append(" task seq files are ")
+ .append(selectedSequenceFiles.toString())
+ .append(" , unseq files are ")
+ .append(selectedUnsequenceFiles.toString())
.toString();
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/RewriteCrossSpaceCompactionSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/RewriteCrossSpaceCompactionSelector.java
index 7afc400..452e4d0 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/RewriteCrossSpaceCompactionSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/RewriteCrossSpaceCompactionSelector.java
@@ -115,8 +115,6 @@ public class RewriteCrossSpaceCompactionSelector extends AbstractCrossSpaceCompa
mergeFiles[1]);
if (mergeFiles[0].size() > 0 && mergeFiles[1].size() > 0) {
- mergeFiles[0].forEach(x -> ((TsFileResource) x).setCompactionCandidate(true));
- mergeFiles[1].forEach(x -> ((TsFileResource) x).setCompactionCandidate(true));
AbstractCompactionTask compactionTask =
taskFactory.createTask(
logicalStorageGroupName,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java
index f06011f..dffd329 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/cross/rewrite/task/RewriteCrossSpaceCompactionTask.java
@@ -187,7 +187,6 @@ public class RewriteCrossSpaceCompactionTask extends AbstractCrossSpaceCompactio
return false;
}
tsFileResource.setCompacting(true);
- tsFileResource.setCompactionCandidate(false);
}
return true;
}
@@ -208,6 +207,8 @@ public class RewriteCrossSpaceCompactionTask extends AbstractCrossSpaceCompactio
}
private void releaseAllLock() {
+ selectedSeqTsFileResourceList.forEach(x -> x.setCompactionCandidate(false));
+ selectedUnSeqTsFileResourceList.forEach(x -> x.setCompactionCandidate(false));
for (TsFileResource tsFileResource : holdReadLockList) {
tsFileResource.readUnlock();
tsFileResource.setCompacting(false);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionTask.java
index 7183114..ca00952 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/AbstractInnerSpaceCompactionTask.java
@@ -54,6 +54,11 @@ public abstract class AbstractInnerSpaceCompactionTask extends AbstractCompactio
collectSelectedFilesInfo();
}
+ @Override
+ public void setSourceFilesToCompactionCandidate() {
+ this.selectedTsFileResourceList.forEach(x -> x.setCompactionCandidate(true));
+ }
+
private void collectSelectedFilesInfo() {
selectedFileSize = 0L;
sumOfCompactionCount = 0;
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java
index e2c0f52..7a8ea62 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionSelector.java
@@ -167,7 +167,6 @@ public class SizeTieredCompactionSelector extends AbstractInnerSpaceCompactionSe
private boolean createAndSubmitTask(List<TsFileResource> selectedFileList)
throws InterruptedException {
- selectedFileList.forEach(x -> x.setCompactionCandidate(true));
AbstractCompactionTask compactionTask =
taskFactory.createTask(
logicalStorageGroupName,
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
index fb6bfbe..a77e625 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/inner/sizetiered/SizeTieredCompactionTask.java
@@ -227,6 +227,7 @@ public class SizeTieredCompactionTask extends AbstractInnerSpaceCompactionTask {
@Override
public boolean checkValidAndSetMerging() {
+ selectedTsFileResourceList.forEach(x -> x.setCompactionCandidate(false));
for (int i = 0; i < selectedTsFileResourceList.size(); ++i) {
TsFileResource resource = selectedTsFileResourceList.get(i);
resource.readLock();
@@ -244,7 +245,6 @@ public class SizeTieredCompactionTask extends AbstractInnerSpaceCompactionTask {
for (TsFileResource resource : selectedTsFileResourceList) {
resource.setCompacting(true);
- resource.setCompactionCandidate(false);
}
return true;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/AbstractCompactionTask.java b/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/AbstractCompactionTask.java
index 07135ab..ce80aff 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/AbstractCompactionTask.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/compaction/task/AbstractCompactionTask.java
@@ -56,6 +56,8 @@ public abstract class AbstractCompactionTask implements Callable<Void> {
this.currentTaskNum = currentTaskNum;
}
+ public abstract void setSourceFilesToCompactionCandidate();
+
protected abstract void doCompaction() throws Exception;
@Override
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index 0d4c1b0..e067233 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -555,7 +555,9 @@ public class TsFileResource {
@Override
public String toString() {
- return file.toString();
+ return String.format(
+ "file is %s, compactionCandidate: %s, compacting: %s",
+ file.toString(), compactionCandidate, isCompacting);
}
@Override
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManagerTest.java b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManagerTest.java
index 09603cc..67e3109 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManagerTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskManagerTest.java
@@ -19,12 +19,16 @@
package org.apache.iotdb.db.engine.compaction;
import org.apache.iotdb.db.constant.TestConstant;
+import org.apache.iotdb.db.engine.compaction.cross.rewrite.task.RewriteCrossSpaceCompactionTask;
import org.apache.iotdb.db.engine.compaction.inner.InnerCompactionTest;
import org.apache.iotdb.db.engine.compaction.inner.sizetiered.SizeTieredCompactionTask;
import org.apache.iotdb.db.engine.compaction.task.AbstractCompactionTask;
import org.apache.iotdb.db.engine.storagegroup.TsFileManager;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.commons.io.FileUtils;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -32,6 +36,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
@@ -51,6 +56,12 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
super.setUp();
}
+ @After
+ public void tearDown() throws StorageEngineException, IOException {
+ CompactionTaskManager.getInstance().waitAllCompactionFinish();
+ super.tearDown();
+ }
+
@Test
public void testRepeatedSubmitBeforeExecution() throws Exception {
logger.warn("testRepeatedSubmitBeforeExecution");
@@ -66,21 +77,33 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
seqResources.get(0).readLock();
CompactionTaskManager manager = CompactionTaskManager.getInstance();
try {
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
Assert.assertTrue(manager.addTaskToWaitingQueue(task1));
Assert.assertEquals(manager.getTotalTaskCount(), 1);
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.isCompactionCandidate());
+ }
// a same task should not be submitted compaction task manager
Assert.assertFalse(manager.addTaskToWaitingQueue(task2));
Assert.assertEquals(manager.getTotalTaskCount(), 1);
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.isCompactionCandidate());
+ }
manager.submitTaskFromTaskQueue();
} finally {
seqResources.get(0).readUnlock();
}
Thread.sleep(5000);
Assert.assertEquals(0, manager.getTotalTaskCount());
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
long waitingTime = 0;
while (manager.getRunningCompactionTaskList().size() > 0) {
- Thread.sleep(100);
- waitingTime += 100;
+ Thread.sleep(50);
+ waitingTime += 50;
if (waitingTime % 10000 == 0) {
logger.warn("{}", manager.getRunningCompactionTaskList());
}
@@ -105,12 +128,24 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
seqResources.get(0).readLock();
try {
CompactionTaskManager manager = CompactionTaskManager.getInstance();
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
manager.addTaskToWaitingQueue(task1);
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.isCompactionCandidate());
+ }
manager.submitTaskFromTaskQueue();
Thread.sleep(2000);
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
// When a same compaction task is executing, the compaction task should not be submitted!
Assert.assertEquals(manager.getExecutingTaskCount(), 1);
Assert.assertFalse(manager.addTaskToWaitingQueue(task2));
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
} finally {
seqResources.get(0).readUnlock();
}
@@ -125,6 +160,9 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
Assert.fail();
}
}
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
}
@Test
@@ -203,4 +241,66 @@ public class CompactionTaskManagerTest extends InnerCompactionTest {
}
}
}
+
+ @Test
+ public void testSizeTieredCompactionStatus() throws Exception {
+ TsFileManager tsFileManager =
+ new TsFileManager("root.compactionTest", "0", tempSGDir.getAbsolutePath());
+ tsFileManager.addAll(seqResources, true);
+ SizeTieredCompactionTask task =
+ new SizeTieredCompactionTask(
+ "root.compactionTest", "0", 0, tsFileManager, seqResources, true, new AtomicInteger(0));
+ CompactionTaskManager.getInstance().addTaskToWaitingQueue(task);
+
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.isCompactionCandidate());
+ }
+
+ CompactionTaskManager.getInstance().submitTaskFromTaskQueue();
+ Thread.sleep(50);
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
+ }
+
+ @Test
+ public void testRewriteCrossCompactionFileStatus() throws Exception {
+ TsFileManager tsFileManager =
+ new TsFileManager("root.compactionTest", "0", tempSGDir.getAbsolutePath());
+ tsFileManager.addAll(seqResources, true);
+ tsFileManager.addAll(unseqResources, false);
+ RewriteCrossSpaceCompactionTask task =
+ new RewriteCrossSpaceCompactionTask(
+ COMPACTION_TEST_SG,
+ "0",
+ 0,
+ tsFileManager,
+ seqResources,
+ unseqResources,
+ new AtomicInteger(0));
+
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
+
+ CompactionTaskManager.getInstance().addTaskToWaitingQueue(task);
+ for (TsFileResource resource : seqResources) {
+ Assert.assertTrue(resource.isCompactionCandidate());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertTrue(resource.isCompactionCandidate());
+ }
+
+ CompactionTaskManager.getInstance().submitTaskFromTaskQueue();
+ CompactionTaskManager.getInstance().waitAllCompactionFinish();
+ for (TsFileResource resource : seqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
+ for (TsFileResource resource : unseqResources) {
+ Assert.assertFalse(resource.isCompactionCandidate());
+ }
+ }
}