[FLINK-27705] Prevent num-sorted-run.compaction-trigger from interfering num-levels

flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/ForceCompactionITCase.java

@@ -20,7 +20,7 @@
 
 import org.junit.Test;
 
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -30,12 +30,17 @@ public class ForceCompactionITCase extends FileStoreTableITCase {
 
     @Override
     protected List<String> ddl() {
-        return Collections.singletonList(
+        return Arrays.asList(
                 "CREATE TABLE IF NOT EXISTS T (\n"
                         + "  f0 INT\n, "
                         + "  f1 STRING\n, "
                         + "  f2 STRING\n"
-                        + ") PARTITIONED BY (f1)");
+                        + ") PARTITIONED BY (f1)",
+                "CREATE TABLE IF NOT EXISTS T1 (\n"
+                        + "  f0 INT\n, "
+                        + "  f1 STRING\n, "
+                        + "  f2 STRING\n"
+                        + ")");
     }
 
     @Test
@@ -74,4 +79,37 @@ public void testDynamicPartition() {
 
         assertThat(batchSql("SELECT * FROM T")).hasSize(21);
     }
+
+    @Test
+    public void testNoDefaultNumOfLevels() throws Exception {
+        bEnv.executeSql("ALTER TABLE T1 SET ('commit.force-compact' = 'true')");
+        bEnv.executeSql(
+                        "INSERT INTO T1 VALUES(1, 'Winter', 'Winter is Coming'),"
+                                + "(2, 'Winter', 'The First Snowflake'), "
+                                + "(2, 'Spring', 'The First Rose in Spring'), "
+                                + "(7, 'Summer', 'Summertime Sadness')")
+                .await();
+        bEnv.executeSql("INSERT INTO T1 VALUES(12, 'Winter', 'Last Christmas')").await();
+        bEnv.executeSql("INSERT INTO T1 VALUES(11, 'Winter', 'Winter is Coming')").await();
+        bEnv.executeSql("INSERT INTO T1 VALUES(10, 'Autumn', 'Refrain')").await();
+        bEnv.executeSql(
+                        "INSERT INTO T1 VALUES(6, 'Summer', 'Watermelon Sugar'), "
+                                + "(4, 'Spring', 'Spring Water')")
+                .await();
+        bEnv.executeSql(
+                        "INSERT INTO T1 VALUES(66, 'Summer', 'Summer Vibe'), "
+                                + "(9, 'Autumn', 'Wake Me Up When September Ends')")
+                .await();
+        bEnv.executeSql(
+                        "INSERT INTO T1 VALUES(666, 'Summer', 'Summer Vibe'), "
+                                + "(9, 'Autumn', 'Wake Me Up When September Ends')")
+                .await();
+        bEnv.executeSql("ALTER TABLE T1 SET ('num-sorted-run.compaction-trigger' = '2')");
+        bEnv.executeSql(
+                        "INSERT INTO T1 VALUES(666, 'Summer', 'Summer Vibe'), "
+                                + "(9, 'Autumn', 'Wake Me Up When September Ends')")
+                .await();
+
+        assertThat(batchSql("SELECT * FROM T1")).hasSize(15);
+    }
 }

flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/Levels.java

@@ -43,6 +43,16 @@ public class Levels {
 
     public Levels(Comparator<RowData> keyComparator, List<DataFileMeta> inputFiles, int numLevels) {
         this.keyComparator = keyComparator;
+
+        // in case the num of levels is not specified explicitly
+        numLevels =
+                Math.max(
+                        numLevels,
+                        inputFiles.stream()
+                                        .map(DataFileMeta::level)
+                                        .max(Comparator.naturalOrder())
+                                        .orElse(-1)
+                                + 1);
         checkArgument(numLevels > 1, "levels must be at least 2.");
         this.level0 =
                 new TreeSet<>(Comparator.comparing(DataFileMeta::maxSequenceNumber).reversed());