[core] Append commit should check bucket number if latest commit user is different (#6723)

yuzelin · web-flow · commit 7a1187e9354d · 2025-12-02T19:23:37.000+08:00
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -36,6 +36,8 @@ public interface FileStoreCommit extends AutoCloseable {
 
     FileStoreCommit withPartitionExpire(PartitionExpire partitionExpire);
 
+    FileStoreCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);
+
     /** Find out which committables need to be retried when recovering from the failure. */
     List<ManifestCommittable> filterCommitted(List<ManifestCommittable> committables);
 
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
@@ -154,6 +154,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
 
     private boolean ignoreEmptyCommit;
     private CommitMetrics commitMetrics;
+    private boolean appendCommitCheckConflict = false;
 
     public FileStoreCommitImpl(
             SnapshotCommit snapshotCommit,
@@ -246,6 +247,12 @@ public FileStoreCommit withPartitionExpire(PartitionExpire partitionExpire) {
         return this;
     }
 
+    @Override
+    public FileStoreCommit appendCommitCheckConflict(boolean appendCommitCheckConflict) {
+        this.appendCommitCheckConflict = appendCommitCheckConflict;
+        return this;
+    }
+
     @Override
     public List<ManifestCommittable> filterCommitted(List<ManifestCommittable> committables) {
         // nothing to filter, fast exit
@@ -327,6 +334,8 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                 if (containsFileDeletionOrDeletionVectors(appendSimpleEntries, appendIndexFiles)) {
                     commitKind = CommitKind.OVERWRITE;
                     conflictCheck = mustConflictCheck();
+                } else if (latestSnapshot != null && appendCommitCheckConflict) {
+                    conflictCheck = mustConflictCheck();
                 }
 
                 boolean discardDuplicate = discardDuplicateFiles && commitKind == CommitKind.APPEND;
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchWriteBuilderImpl.java
@@ -40,6 +40,7 @@ public class BatchWriteBuilderImpl implements BatchWriteBuilder {
     private final String commitUser;
 
     private Map<String, String> staticPartition;
+    private boolean appendCommitCheckConflict = false;
 
     public BatchWriteBuilderImpl(InnerTable table) {
         this.table = table;
@@ -81,15 +82,23 @@ public BatchTableWrite newWrite() {
 
     @Override
     public BatchTableCommit newCommit() {
-        InnerTableCommit commit = table.newCommit(commitUser).withOverwrite(staticPartition);
+        InnerTableCommit commit =
+                table.newCommit(commitUser)
+                        .withOverwrite(staticPartition)
+                        .appendCommitCheckConflict(appendCommitCheckConflict);
         commit.ignoreEmptyCommit(
                 Options.fromMap(table.options())
                         .getOptional(CoreOptions.SNAPSHOT_IGNORE_EMPTY_COMMIT)
                         .orElse(true));
         return commit;
     }
 
-    public BatchWriteBuilder copyWithNewTable(Table newTable) {
+    public BatchWriteBuilderImpl copyWithNewTable(Table newTable) {
         return new BatchWriteBuilderImpl((InnerTable) newTable, commitUser, staticPartition);
     }
+
+    public BatchWriteBuilderImpl appendCommitCheckConflict(boolean appendCommitCheckConflict) {
+        this.appendCommitCheckConflict = appendCommitCheckConflict;
+        return this;
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/InnerTableCommit.java
@@ -46,6 +46,8 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {
 
     InnerTableCommit expireForEmptyCommit(boolean expireForEmptyCommit);
 
+    InnerTableCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);
+
     @Override
     InnerTableCommit withMetricRegistry(MetricRegistry registry);
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -158,6 +158,12 @@ public TableCommitImpl expireForEmptyCommit(boolean expireForEmptyCommit) {
         return this;
     }
 
+    @Override
+    public TableCommitImpl appendCommitCheckConflict(boolean appendCommitCheckConflict) {
+        commit.appendCommitCheckConflict(appendCommitCheckConflict);
+        return this;
+    }
+
     @Override
     public InnerTableCommit withMetricRegistry(MetricRegistry registry) {
         commit.withMetrics(new CommitMetrics(registry, tableName));
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableCommitTest.java
@@ -50,6 +50,8 @@
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -67,6 +69,7 @@
 import static java.util.Collections.singletonMap;
 import static org.apache.paimon.utils.FileStorePathFactoryTest.createNonPartFactory;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Tests for {@link TableCommit}. */
@@ -322,6 +325,71 @@ public void testGiveUpCommitWhenTotalBucketsChanged() throws Exception {
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testGiveUpCommitWhenAppendFoundTotalBucketsChanged(boolean checkAppend)
+            throws Exception {
+        String path = tempDir.toString();
+        RowType rowType =
+                RowType.of(
+                        new DataType[] {DataTypes.INT(), DataTypes.BIGINT()},
+                        new String[] {"k", "v"});
+
+        Options options = new Options();
+        options.set(CoreOptions.PATH, path);
+        options.set(CoreOptions.BUCKET, 1);
+        TableSchema tableSchema =
+                SchemaUtils.forceCommit(
+                        new SchemaManager(LocalFileIO.create(), new Path(path)),
+                        new Schema(
+                                rowType.getFields(),
+                                Collections.emptyList(),
+                                Collections.singletonList("k"),
+                                options.toMap(),
+                                ""));
+        FileStoreTable table =
+                FileStoreTableFactory.create(
+                        LocalFileIO.create(),
+                        new Path(path),
+                        tableSchema,
+                        CatalogEnvironment.empty());
+
+        String commitUser1 = UUID.randomUUID().toString();
+        TableWriteImpl<?> write1 = table.newWrite(commitUser1);
+        TableCommitImpl commit1 = table.newCommit(commitUser1);
+        for (int i = 1; i < 10; i++) {
+            write1.write(GenericRow.of(i, (long) i));
+        }
+
+        // mock rescale
+        String commitUser2 = UUID.randomUUID().toString();
+        options = new Options(table.options());
+        options.set(CoreOptions.BUCKET, 2);
+        FileStoreTable rescaleTable = table.copy(tableSchema.copy(options.toMap()));
+        try (TableWriteImpl<?> write = rescaleTable.newWrite(commitUser2);
+                TableCommitImpl commit =
+                        rescaleTable.newCommit(commitUser2).withOverwrite(Collections.emptyMap())) {
+            for (int i = 1; i < 10; i++) {
+                write.write(GenericRow.of(i, (long) i));
+            }
+            commit.commit(1, write.prepareCommit(false, 1));
+        }
+
+        if (checkAppend) {
+            commit1.appendCommitCheckConflict(true);
+            assertThatThrownBy(() -> commit1.commit(1, write1.prepareCommit(false, 1)))
+                    .isInstanceOf(RuntimeException.class)
+                    .hasMessageContaining("changed from 2 to 1 without overwrite");
+        } else {
+            // the commit result is error, but here verify that no check if
+            // appendCommitCheckConflict was not set
+            assertThatCode(() -> commit1.commit(1, write1.prepareCommit(false, 1)))
+                    .doesNotThrowAnyException();
+        }
+        write1.close();
+        commit1.close();
+    }
+
     @Test
     public void testStrictModeForCompact() throws Exception {
         String path = tempDir.toString();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/PostponeFixedBucketSink.java
@@ -68,8 +68,16 @@ protected CommittableStateManager<ManifestCommittable> createCommittableStateMan
     @Override
     protected Committer.Factory<Committable, ManifestCommittable> createCommitterFactory() {
         if (overwritePartition == null) {
-            // The table has copied bucket option outside, no need to change anything
-            return super.createCommitterFactory();
+            // The table has copied bucket option outside, no need to change.
+            return context ->
+                    new StoreCommitter(
+                            table,
+                            table.newCommit(context.commitUser())
+                                    .withOverwrite(overwritePartition)
+                                    .ignoreEmptyCommit(!context.streamingCheckpointEnabled())
+                                    // Need to check conflict
+                                    .appendCommitCheckConflict(true),
+                            context);
         } else {
             // When overwriting, the postpone bucket files need to be deleted, so using a postpone
             // bucket table commit here
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -410,6 +410,8 @@ case class PaimonSparkWriter(
       writeBuilder
         .asInstanceOf[BatchWriteBuilderImpl]
         .copyWithNewTable(PostponeUtils.tableForCommit(table))
+        // Need to check conflict
+        .appendCommitCheckConflict(true)
     } else {
       writeBuilder
     }

Original file line number	Diff line number	Diff line change
`@@ -46,6 +46,8 @@ public interface InnerTableCommit extends StreamTableCommit, BatchTableCommit {`
`46`	`46`
`47`	`47`	`InnerTableCommit expireForEmptyCommit(boolean expireForEmptyCommit);`
`48`	`48`
	`49`	`+ InnerTableCommit appendCommitCheckConflict(boolean appendCommitCheckConflict);`
	`50`	`+`
`49`	`51`	`@Override`
`50`	`52`	`InnerTableCommit withMetricRegistry(MetricRegistry registry);`
`51`	`53`	`}`
Original file line number	Diff line number	Diff line change
`@@ -410,6 +410,8 @@ case class PaimonSparkWriter(`
`410`	`410`	`writeBuilder`
`411`	`411`	`.asInstanceOf[BatchWriteBuilderImpl]`
`412`	`412`	`.copyWithNewTable(PostponeUtils.tableForCommit(table))`
	`413`	`+ // Need to check conflict`
	`414`	`+ .appendCommitCheckConflict(true)`
`413`	`415`	`} else {`
`414`	`416`	`writeBuilder`
`415`	`417`	`}`