[core] Cover snapshot sequence bucket rescale check (#8286)

This PR makes write-only snapshot sequence APPEND commits participate in
the same bucket-number consistency check as unordered write-only append
commits. It prevents INSERT INTO from writing new files with a rescaled
bucket count before existing data layout is rewritten, while INSERT
OVERWRITE can still be used to rescale the layout.

Author: Aitozi

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -782,8 +782,17 @@ private void checkSameBucketFromSnapshot(
     private boolean shouldCheckSameBucket(CommitKind commitKind) {
         return commitKind == CommitKind.APPEND
                 && bucketMode == BucketMode.HASH_FIXED
-                && options.writeOnly()
-                && !options.bucketAppendOrdered();
+                && (isUnorderedWriteOnlyAppend() || isWriteOnlySnapshotSequenceAppend());
+    }
+
+    private boolean isUnorderedWriteOnlyAppend() {
+        return options.writeOnly() && !options.bucketAppendOrdered();
+    }
+
+    private boolean isWriteOnlySnapshotSequenceAppend() {
+        return options.writeOnly()
+                && options.writeSequenceNumberInitMode()
+                        == CoreOptions.SequenceNumberInitMode.SNAPSHOT;
     }
 
     private OptionalLong maxSequenceNumber(List<ManifestFileMeta> manifests) {

paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java

@@ -35,6 +35,7 @@
 import org.apache.paimon.index.IndexFileHandler;
 import org.apache.paimon.index.IndexFileMeta;
 import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataIncrement;
 import org.apache.paimon.manifest.FileKind;
 import org.apache.paimon.manifest.IndexManifestEntry;
@@ -188,6 +189,63 @@ public void testLatestHint() throws Exception {
         assertThat(snapshotManager.latestSnapshotId()).isEqualTo(latestId);
     }
 
+    @Test
+    public void testWriteOnlySnapshotSequenceCommitChecksRescaledBucketNumber() throws Exception {
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.WRITE_ONLY.key(), "true");
+        options.put(CoreOptions.WRITE_SEQUENCE_NUMBER_INIT_MODE.key(), "snapshot");
+        options.put(CoreOptions.BUCKET.key(), "2");
+        options.put(CoreOptions.BUCKET_KEY.key(), "orderId");
+        TestAppendFileStore store = TestAppendFileStore.createAppendStore(tempDir, options);
+        BinaryRow partition =
+                gen.getPartition(gen.nextInsert("20201110", 10, 1L, new int[] {1, 1}, "first"));
+
+        try (FileStoreCommitImpl commit = store.newCommit()) {
+            assertThat(
+                            commit.tryCommitOnce(
+                                            null,
+                                            Collections.singletonList(addFile(partition, 1, 2, 0)),
+                                            Collections.emptyList(),
+                                            Collections.emptyList(),
+                                            0,
+                                            null,
+                                            new HashMap<>(),
+                                            Snapshot.CommitKind.APPEND,
+                                            false,
+                                            null,
+                                            false,
+                                            null)
+                                    .isSuccess())
+                    .isTrue();
+        }
+        assertThat(store.snapshotManager().latestSnapshot().properties())
+                .containsKey(SequenceSnapshotProperties.MAX_SEQUENCE_NUMBER);
+
+        Map<String, String> rescaledOptions = new HashMap<>(options);
+        rescaledOptions.put(CoreOptions.BUCKET.key(), "4");
+        TestAppendFileStore rescaledStore =
+                TestAppendFileStore.createAppendStore(tempDir, rescaledOptions);
+        try (FileStoreCommitImpl commit = rescaledStore.newCommit()) {
+            assertThatThrownBy(
+                            () ->
+                                    commit.tryCommitOnce(
+                                            null,
+                                            Collections.singletonList(addFile(partition, 1, 4, 1)),
+                                            Collections.emptyList(),
+                                            Collections.emptyList(),
+                                            1,
+                                            null,
+                                            new HashMap<>(),
+                                            Snapshot.CommitKind.APPEND,
+                                            false,
+                                            rescaledStore.snapshotManager().latestSnapshot(),
+                                            false,
+                                            null))
+                    .hasMessageContaining("new bucket num 4")
+                    .hasMessageContaining("previous bucket num is 2");
+        }
+    }
+
     @Test
     public void testFilterCommittedAfterExpire() throws Exception {
         testRandomConcurrentNoConflict(1, false, CoreOptions.ChangelogProducer.NONE);
@@ -1135,6 +1193,30 @@ private FileStoreCommitImpl newCommitWithSnapshotCommit(
                 store.options().dataEvolutionEnabled());
     }
 
+    private ManifestEntry addFile(
+            BinaryRow partition, int bucket, int totalBuckets, long maxSequenceNumber) {
+        return ManifestEntry.create(
+                FileKind.ADD,
+                partition,
+                bucket,
+                totalBuckets,
+                DataFileMeta.forAppend(
+                        String.format("test-%d.orc", maxSequenceNumber),
+                        1,
+                        1,
+                        EMPTY_STATS,
+                        maxSequenceNumber,
+                        maxSequenceNumber,
+                        0,
+                        Collections.emptyList(),
+                        null,
+                        null,
+                        null,
+                        null,
+                        null,
+                        null));
+    }
+
     private FileStoreCommitImpl newCommitWithSnapshotCommit(
             TestFileStore store,
             String commitUser,

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

@@ -201,6 +201,46 @@ public void testWriteOnlySnapshotSequenceNumberInitOverwritePreviousValue() thro
         assertThat(actual).containsExactly(Row.of(1, "new"), Row.of(2, "keep"));
     }
 
+    @Test
+    @Timeout(TIMEOUT)
+    public void testSnapshotSequenceInsertIntoCheckSameBucketAndInsertOverwriteRescale()
+            throws Exception {
+        TableEnvironment bEnv = tableEnvironmentBuilder().batchMode().parallelism(1).build();
+        bEnv.executeSql(createCatalogSql("testCatalog", path));
+        bEnv.executeSql("USE CATALOG testCatalog");
+        bEnv.executeSql(
+                "CREATE TABLE T ("
+                        + "  k INT,"
+                        + "  v STRING,"
+                        + "  PRIMARY KEY (k) NOT ENFORCED"
+                        + ") WITH ("
+                        + "  'bucket' = '1',"
+                        + "  'write-only' = 'true',"
+                        + "  'write.sequence-number-init-mode' = 'snapshot'"
+                        + ")");
+
+        bEnv.executeSql("INSERT INTO T VALUES (1, 'AAA'), (2, 'BBB')").await();
+        bEnv.executeSql("ALTER TABLE T SET ('bucket' = '2')");
+
+        assertThatCode(() -> bEnv.executeSql("INSERT INTO T VALUES (3, 'CCC')").await())
+                .rootCause()
+                .isInstanceOf(RuntimeException.class)
+                .hasMessage(
+                        "Try to write table with a new bucket num 2, but the previous bucket num is 1. "
+                                + "Please switch to batch mode, and perform INSERT OVERWRITE to rescale current data layout first.");
+
+        bEnv.executeSql("INSERT OVERWRITE T VALUES (3, 'CCC'), (4, 'DDD')").await();
+
+        List<Row> actual = new ArrayList<>();
+        try (CloseableIterator<Row> it = bEnv.executeSql("SELECT * FROM T ORDER BY k").collect()) {
+            while (it.hasNext()) {
+                actual.add(it.next());
+            }
+        }
+
+        assertThat(actual).containsExactly(Row.of(3, "CCC"), Row.of(4, "DDD"));
+    }
+
     @Test
     @Timeout(TIMEOUT)
     public void testFullCompactionTriggerInterval() throws Exception {