[iotdb] branch master updated: add time partition in compaction comparator (#10662)

[xi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xingtanzjr 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 3c6ab756d56 add time partition in compaction comparator (#10662)
3c6ab756d56 is described below

commit 3c6ab756d56331a9f8ba42d7f1c71eeef9b96dce
Author: shuwenwei <55...@users.noreply.github.com>
AuthorDate: Thu Jul 27 16:13:42 2023 +0800

    add time partition in compaction comparator (#10662)
---
 .../DefaultCompactionTaskComparatorImpl.java       | 14 ++++++
 .../compaction/CompactionTaskComparatorTest.java   | 58 ++++++++++++++++++++++
 2 files changed, 72 insertions(+)

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
index a3b743e444e..b6cecd1f026 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/schedule/comparator/DefaultCompactionTaskComparatorImpl.java
@@ -79,6 +79,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
@@ -113,6 +120,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/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java
index 427f533fc85..9582b15309b 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionTaskComparatorTest.java
@@ -355,6 +355,64 @@ public class CompactionTaskComparatorTest {
     fakedTsFileResource.getModFile().remove();
   }
 
+  @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(