[core] add more tests for clustering bucketed append table (#6981)

Author: LsomeYeah

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/AppendOnlyTableITCase.java

@@ -472,6 +472,37 @@ public void testAutoCompaction() {
                         Row.of(12, "MMM"));
     }
 
+    @Test
+    public void testAutoCluster() {
+        batchSql("ALTER TABLE append_table SET ('num-sorted-run.compaction-trigger' = '3')");
+        batchSql("ALTER TABLE append_table SET ('num-levels' = '6')");
+        batchSql("ALTER TABLE append_table SET ('bucket-append-ordered' = 'false')");
+        batchSql("ALTER TABLE append_table SET ('clustering.columns' = 'data')");
+        batchSql("ALTER TABLE append_table SET ('clustering.strategy' = 'order')");
+        batchSql("ALTER TABLE append_table SET ('clustering.incremental' = 'true')");
+
+        assertAutoCompaction(
+                "INSERT INTO append_table VALUES (1, '9')", 1L, Snapshot.CommitKind.APPEND);
+        assertAutoCompaction(
+                "INSERT INTO append_table VALUES (2, '8')", 2L, Snapshot.CommitKind.APPEND);
+        assertAutoCompaction(
+                "INSERT INTO append_table VALUES (3, '7')", 4L, Snapshot.CommitKind.COMPACT);
+        assertAutoCompaction(
+                "INSERT INTO append_table VALUES (4, '6')", 5L, Snapshot.CommitKind.APPEND);
+        assertAutoCompaction(
+                "INSERT INTO append_table VALUES (5, '5')", 7L, Snapshot.CommitKind.COMPACT);
+
+        List<Row> rows = batchSql("SELECT * FROM append_table");
+        assertThat(rows.size()).isEqualTo(5);
+        assertThat(rows)
+                .containsExactly(
+                        Row.of(5, "5"),
+                        Row.of(4, "6"),
+                        Row.of(3, "7"),
+                        Row.of(2, "8"),
+                        Row.of(1, "9"));
+    }
+
     @Test
     public void testRejectDelete() {
         testRejectChanges(RowKind.DELETE);

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/IncrementalClusterActionITCase.java

@@ -40,10 +40,12 @@
 import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.table.sink.CommitMessage;
 import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.ReadBuilder;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.StringUtils;
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
@@ -56,6 +58,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.Collectors;
 
@@ -686,6 +689,7 @@ public void testClusterWithDeletionVector() throws Exception {
     @Test
     public void testClusterWithBucket() throws Exception {
         Map<String, String> dynamicOptions = commonOptions();
+        dynamicOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
         dynamicOptions.put(CoreOptions.BUCKET.key(), "2");
         dynamicOptions.put(CoreOptions.BUCKET_KEY.key(), "pt");
         dynamicOptions.put(CoreOptions.BUCKET_APPEND_ORDERED.key(), "false");
@@ -848,6 +852,35 @@ public void testClusterWithBucket() throws Exception {
         assertThat(result5).containsExactlyElementsOf(expected5);
     }
 
+    @Test
+    public void testStreamingClusterWithBucket() throws Exception {
+        Map<String, String> dynamicOptions = commonOptions();
+        dynamicOptions.put(CoreOptions.WRITE_ONLY.key(), "true");
+        dynamicOptions.put(CoreOptions.BUCKET.key(), "1");
+        dynamicOptions.put(CoreOptions.BUCKET_KEY.key(), "pt");
+        dynamicOptions.put(CoreOptions.BUCKET_APPEND_ORDERED.key(), "false");
+        dynamicOptions.put(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL.key(), "1s");
+        FileStoreTable table = createTable(null, dynamicOptions);
+        StreamWriteBuilder streamWriteBuilder =
+                table.newStreamWriteBuilder().withCommitUser(commitUser);
+        write = streamWriteBuilder.newWrite();
+        commit = streamWriteBuilder.newCommit();
+
+        // base records
+        writeData(GenericRow.of(2, 2, BinaryString.fromString("test"), 0));
+        writeData(GenericRow.of(2, 1, BinaryString.fromString("test"), 0));
+        writeData(GenericRow.of(2, 0, BinaryString.fromString("test"), 0));
+
+        checkSnapshot(table, Snapshot.CommitKind.APPEND);
+        runAction(true, Collections.emptyList());
+        checkSnapshot(table, 4, Snapshot.CommitKind.COMPACT, 60_000);
+
+        // incremental records
+        writeData(GenericRow.of(1, 2, BinaryString.fromString("test"), 0));
+        writeData(GenericRow.of(1, 1, BinaryString.fromString("test"), 0));
+        checkSnapshot(table, 7, Snapshot.CommitKind.COMPACT, 60_000);
+    }
+
     protected FileStoreTable createTable(String partitionKeys) throws Exception {
         return createTable(partitionKeys, commonOptions());
     }
@@ -923,8 +956,28 @@ private static String randomString(int length) {
     }
 
     private void checkSnapshot(FileStoreTable table) {
-        assertThat(table.latestSnapshot().get().commitKind())
-                .isEqualTo(Snapshot.CommitKind.COMPACT);
+        checkSnapshot(table, Snapshot.CommitKind.COMPACT);
+    }
+
+    private void checkSnapshot(FileStoreTable table, Snapshot.CommitKind commitKind) {
+        assertThat(table.latestSnapshot().get().commitKind()).isEqualTo(commitKind);
+    }
+
+    protected void checkSnapshot(
+            FileStoreTable table, long snapshotId, Snapshot.CommitKind commitKind, long timeout)
+            throws Exception {
+        SnapshotManager snapshotManager = table.snapshotManager();
+        long start = System.currentTimeMillis();
+        while (!Objects.equals(snapshotManager.latestSnapshotId(), snapshotId)) {
+            Thread.sleep(500);
+            if (System.currentTimeMillis() - start > timeout) {
+                throw new RuntimeException("can't wait for a compaction.");
+            }
+        }
+
+        Snapshot snapshot = snapshotManager.snapshot(snapshotManager.latestSnapshotId());
+        assertThat(snapshot.id()).isEqualTo(snapshotId);
+        assertThat(snapshot.commitKind()).isEqualTo(commitKind);
     }
 
     private List<CommitMessage> produceDvIndexMessages(
@@ -954,7 +1007,16 @@ private List<CommitMessage> produceDvIndexMessages(
     }
 
     private void runAction(List<String> extra) throws Exception {
-        StreamExecutionEnvironment env = streamExecutionEnvironmentBuilder().batchMode().build();
+        runAction(false, extra);
+    }
+
+    private void runAction(boolean isStreaming, List<String> extra) throws Exception {
+        StreamExecutionEnvironment env;
+        if (isStreaming) {
+            env = streamExecutionEnvironmentBuilder().streamingMode().build();
+        } else {
+            env = streamExecutionEnvironmentBuilder().batchMode().build();
+        }
         ArrayList<String> baseArgs =
                 Lists.newArrayList("compact", "--database", database, "--table", tableName);
         ThreadLocalRandom random = ThreadLocalRandom.current();
@@ -966,7 +1028,12 @@ private void runAction(List<String> extra) throws Exception {
         baseArgs.addAll(extra);
 
         CompactAction action = createAction(CompactAction.class, baseArgs.toArray(new String[0]));
-        action.withStreamExecutionEnvironment(env);
-        action.run();
+        action.withStreamExecutionEnvironment(env).build();
+        if (isStreaming) {
+            env.executeAsync();
+        } else {
+            env.execute();
+        }
+        //        action.run();
     }
 }