1

Zouxxyy · Jan 19, 2024 · a99234b · a99234b
1 parent 97d1314
commit a99234b
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 68 deletions.
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -275,7 +275,8 @@ public class CoreOptions implements Serializable {
                     .noDefaultValue()
                     .withDescription(
                             "If the maximum number of sort readers exceeds this value, a spill will be attempted. "
-                                    + "This prevents too many readers from consuming too much memory and causing OOM.");
+                                    + "This prevents too many readers from consuming too much memory and causing OOM,"
+                                    + "the default value is 'num-sorted-run.compaction-trigger' + 1.");
 
     public static final ConfigOption<MemorySize> SORT_SPILL_BUFFER_SIZE =
             key("sort-spill-buffer-size")
@@ -383,7 +384,8 @@ public class CoreOptions implements Serializable {
                     .intType()
                     .noDefaultValue()
                     .withDescription(
-                            "Total level number, for example, there are 3 levels, including 0,1,2 levels.");
+                            "Total level number, for example, there are 3 levels, including 0,1,2 levels,"
+                                    + " the default value is 'num-sorted-run.compaction-trigger' + 1.");
 
     public static final ConfigOption<Boolean> COMMIT_FORCE_COMPACT =
             key("commit.force-compact")
@@ -1200,8 +1202,9 @@ public SortEngine sortEngine() {
     public int sortSpillThreshold() {
         Integer maxSortedRunNum = options.get(SORT_SPILL_THRESHOLD);
         if (maxSortedRunNum == null) {
-            int stopNum = numSortedRunStopTrigger();
-            maxSortedRunNum = Math.max(stopNum, stopNum + 1);
+            maxSortedRunNum = numSortedRunCompactionTrigger() + 1;
+        } else {
+            maxSortedRunNum = Math.max(numSortedRunStopTrigger(), maxSortedRunNum);
         }
         checkArgument(maxSortedRunNum > 1, "The sort spill threshold cannot be smaller than 2.");
         return maxSortedRunNum;

diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java
@@ -41,8 +41,8 @@ public interface CompactStrategy {
     /** Pick a compaction unit consisting of all existing files. */
     static Optional<CompactUnit> pickFullCompaction(int numLevels, List<LevelSortedRun> runs) {
         int maxLevel = numLevels - 1;
-        if (runs.size() == 1 && runs.get(0).level() == maxLevel) {
-            // only 1 sorted run on the max level, nothing to compact
+        if (runs.isEmpty() || (runs.size() == 1 && runs.get(0).level() == maxLevel)) {
+            // no sorted run or only 1 sorted run on the max level, no need to compact
             return Optional.empty();
         } else {
             return Optional.of(CompactUnit.fromLevelRuns(maxLevel, runs));

diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupCompaction.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupCompaction.java
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -43,8 +43,8 @@
 import org.apache.paimon.mergetree.compact.CompactRewriter;
 import org.apache.paimon.mergetree.compact.CompactStrategy;
 import org.apache.paimon.mergetree.compact.FirstRowMergeTreeCompactRewriter;
+import org.apache.paimon.mergetree.compact.ForceUpLevel0Compaction;
 import org.apache.paimon.mergetree.compact.FullChangelogMergeTreeCompactRewriter;
-import org.apache.paimon.mergetree.compact.LookupCompaction;
 import org.apache.paimon.mergetree.compact.LookupMergeTreeCompactRewriter;
 import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
 import org.apache.paimon.mergetree.compact.MergeTreeCompactManager;
@@ -160,7 +160,7 @@ protected MergeTreeWriter createWriter(
                         options.optimizedCompactionInterval());
         CompactStrategy compactStrategy =
                 options.changelogProducer() == ChangelogProducer.LOOKUP
-                        ? new LookupCompaction(universalCompaction)
+                        ? new ForceUpLevel0Compaction(universalCompaction)
                         : universalCompaction;
         CompactManager compactManager =
                 createCompactManager(partition, bucket, compactStrategy, compactExecutor, levels);

diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/UniversalCompactionTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/UniversalCompactionTest.java
@@ -266,7 +266,8 @@ public void testSizeRatioThreshold() {
 
     @Test
     public void testLookup() {
-        LookupCompaction compaction = new LookupCompaction(new UniversalCompaction(25, 1, 3));
+        ForceUpLevel0Compaction compaction =
+                new ForceUpLevel0Compaction(new UniversalCompaction(25, 1, 3));
 
         // level 0 to max level
         Optional<CompactUnit> pick = compaction.pick(3, level0(1, 2, 2, 2));