This is an automated email from the ASF dual-hosted git repository.
xingtanzjr 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 9c0c0e5b259 add time partition in compaction comparator (#10664)
9c0c0e5b259 is described below
commit 9c0c0e5b25904c578400fdd68874e1a93d48769a
Author: shuwenwei <[email protected]>
AuthorDate: Thu Jul 27 16:13:18 2023 +0800
add time partition in compaction comparator (#10664)
---
.../DefaultCompactionTaskComparatorImpl.java | 14 +++++
.../compaction/CompactionTaskComparatorTest.java | 59 ++++++++++++++++++++++
2 files changed, 73 insertions(+)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
index d2113a97966..6ef8d7f2b24 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
@@ -71,6 +71,13 @@ public class DefaultCompactionTaskComparatorImpl implements
ICompactionTaskCompa
- o2.getSumOfCompactionCount() /
o2.getSelectedTsFileResourceList().size();
}
+ // if the time partition of o1 and o2 are different
+ // we prefer to execute task with greater time partition
+ // because we want to compact files with new data
+ if (o1.getTimePartition() != o2.getTimePartition()) {
+ return o2.getTimePartition() > o1.getTimePartition() ? 1 : -1;
+ }
+
// if the max file version of o1 and o2 are different
// we prefer to execute task with greater file version
// because we want to compact newly written files
@@ -105,6 +112,13 @@ public class DefaultCompactionTaskComparatorImpl
implements ICompactionTaskCompa
public int compareCrossSpaceCompactionTask(
CrossSpaceCompactionTask o1, CrossSpaceCompactionTask o2) {
+ // if the time partition of o1 and o2 are different
+ // we prefer to execute task with greater time partition
+ // because we want to compact files with new data
+ if (o1.getTimePartition() != o2.getTimePartition()) {
+ return o2.getTimePartition() > o1.getTimePartition() ? 1 : -1;
+ }
+
if (o1.getSelectedSequenceFiles().size() !=
o2.getSelectedSequenceFiles().size()) {
// we prefer the task with fewer sequence files
// because this type of tasks consume fewer memory during execution
diff --git
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
index b503ee8e928..8dd67723c13 100644
---
a/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/engine/compaction/CompactionTaskComparatorTest.java
@@ -34,6 +34,7 @@ import
org.apache.iotdb.db.utils.datastructure.FixedPriorityBlockingQueue;
import com.google.common.collect.MinMaxPriorityQueue;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
@@ -315,6 +316,64 @@ public class CompactionTaskComparatorTest {
}
}
+ @Test
+ public void testCompareByTimePartitionWithInnerSpaceCompaction() throws
InterruptedException {
+ List<TsFileResource> resources1 = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ resources1.add(
+ new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile",
i, i)), 10));
+ }
+ FixedPriorityBlockingQueue<AbstractCompactionTask>
candidateCompactionTaskQueue =
+ new FixedPriorityBlockingQueue<>(
+
IoTDBDescriptor.getInstance().getConfig().getCandidateCompactionTaskQueueSize(),
+ new DefaultCompactionTaskComparatorImpl());
+ for (int i = 0; i < 10; i++) {
+ FakedInnerSpaceCompactionTask task =
+ new FakedInnerSpaceCompactionTask(
+ "fakeSg", i, tsFileManager, taskNum, true, resources1, 0);
+ candidateCompactionTaskQueue.put(task);
+ }
+
+ for (int i = 9; i >= 0; i--) {
+
Assert.assertEquals(candidateCompactionTaskQueue.take().getTimePartition(), i);
+ }
+ }
+
+ @Test
+ public void testCompareByTimePartitionWithCrossSpaceCompaction() throws
InterruptedException {
+ List<TsFileResource> seqResources = new ArrayList<>();
+ List<TsFileResource> unseqResources = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ seqResources.add(
+ new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile",
i, i)), 10));
+ }
+ for (int i = 10; i < 20; i++) {
+ unseqResources.add(
+ new FakedTsFileResource(new File(String.format("%d-%d-0-0.tsfile",
i, i)), 10));
+ }
+ FixedPriorityBlockingQueue<AbstractCompactionTask>
candidateCompactionTaskQueue =
+ new FixedPriorityBlockingQueue<>(
+
IoTDBDescriptor.getInstance().getConfig().getCandidateCompactionTaskQueueSize(),
+ new DefaultCompactionTaskComparatorImpl());
+ for (int i = 0; i < 10; i++) {
+ CrossSpaceCompactionTask task =
+ new CrossSpaceCompactionTask(
+ i,
+ tsFileManager,
+ seqResources,
+ unseqResources,
+ new FastCompactionPerformer(true),
+ taskNum,
+ 0,
+ 0);
+ candidateCompactionTaskQueue.put(task);
+ }
+
+ for (int i = 9; i >= 0; i--) {
+
Assert.assertEquals(candidateCompactionTaskQueue.take().getTimePartition(), i);
+ }
+ }
+
private static class FakedInnerSpaceCompactionTask extends
InnerSpaceCompactionTask {
public FakedInnerSpaceCompactionTask(
