This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 926459dd77 [core] Cover snapshot sequence bucket rescale check (#8286)
926459dd77 is described below
commit 926459dd77e9b6d7aaab42005e0a4949fe47670f
Author: WenjunMin <[email protected]>
AuthorDate: Fri Jun 19 15:38:37 2026 +0800
[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.
---
.../paimon/operation/FileStoreCommitImpl.java | 13 +++-
.../paimon/operation/FileStoreCommitTest.java | 82 ++++++++++++++++++++++
.../flink/PrimaryKeyFileStoreTableITCase.java | 40 +++++++++++
3 files changed, 133 insertions(+), 2 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index a316c25711..a5cc109614 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -782,8 +782,17 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
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) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index ef40a0fa2c..bed2a3863c 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -35,6 +35,7 @@ import org.apache.paimon.index.GlobalIndexMeta;
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 class FileStoreCommitTest {
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 @@ public class FileStoreCommitTest {
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,
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
index a39f7bc725..c381d99f83 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PrimaryKeyFileStoreTableITCase.java
@@ -201,6 +201,46 @@ public class PrimaryKeyFileStoreTableITCase extends
AbstractTestBase {
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 {
