This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 93b899b6b21966e22c343363a82521546ad6151d Author: YeJunHao <[email protected]> AuthorDate: Fri Sep 18 19:05:45 2026 +0800 [core] Separate commit history bounds from strict mode (#9960) --- docs/generated/core_configuration.html | 14 +++-- .../main/java/org/apache/paimon/CoreOptions.java | 41 +++++++++++-- .../paimon/operation/FileStoreCommitImpl.java | 9 +-- .../paimon/operation/commit/StrictModeChecker.java | 18 +++--- .../org/apache/paimon/table/PostponeUtils.java | 4 +- .../java/org/apache/paimon/CoreOptionsTest.java | 32 +++++++++++ .../paimon/operation/FileStoreCommitTest.java | 39 +++++++++++++ .../org/apache/paimon/table/PostponeUtilsTest.java | 2 +- .../apache/paimon/table/sink/TableCommitTest.java | 67 ++++++++++++++-------- .../flink/action/DataEvolutionMergeIntoAction.java | 2 +- .../apache/paimon/flink/action/RescaleAction.java | 3 +- .../dataevolution/DataEvolutionDeleteSink.java | 3 +- .../action/DeleteActionDataEvolutionITCase.java | 2 +- .../paimon/spark/procedure/RescaleProcedure.java | 2 +- .../spark/SparkPostponeStagedCommitter.scala | 4 +- .../paimon/spark/commands/PaimonSparkWriter.scala | 4 +- 16 files changed, 183 insertions(+), 63 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 2fb5acc403..4c8538b4f1 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -326,6 +326,12 @@ under the License. <td>Boolean</td> <td>In streaming job, whether to force creating snapshot when there is no data in this write-commit phase.</td> </tr> + <tr> + <td><h5>commit.last-safe-snapshot</h5></td> + <td style="word-wrap: break-word;">(none)</td> + <td>Long</td> + <td>Snapshot preceding the earliest snapshot to inspect when committing. Only later snapshots are searched for this commit user's previous commits. This also provides the starting point for strict-mode checks when enabled. Keep this bound unchanged across retries and recovery.</td> + </tr> <tr> <td><h5>commit.max-retries</h5></td> <td style="word-wrap: break-word;">10</td> @@ -345,10 +351,10 @@ under the License. <td>Min retry wait time when commit failed.</td> </tr> <tr> - <td><h5>commit.strict-mode.last-safe-snapshot</h5></td> - <td style="word-wrap: break-word;">(none)</td> - <td>Long</td> - <td>If set, committer will check if there are other commit user's snapshot starting from the snapshot after this one. If found a COMPACT / OVERWRITE snapshot, or found a APPEND snapshot which committed files to fixed bucket, commit will be aborted.If the value of this option is -1, committer will not check for its first commit.</td> + <td><h5>commit.strict-mode.enabled</h5></td> + <td style="word-wrap: break-word;">true</td> + <td>Boolean</td> + <td>Whether to check concurrent snapshot changes after commit.last-safe-snapshot, when that bound is configured. Rejects COMPACT or OVERWRITE changes in the same partition, and fixed-bucket APPEND changes when committing OVERWRITE. Disabling this does not disable regular conflict detection or the history search bound.</td> </tr> <tr> <td><h5>commit.timeout</h5></td> diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 60605bc74c..3842623860 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2461,15 +2461,34 @@ public class CoreOptions implements Serializable { + "instead of at the end of the schema. " + "This only takes effect for partitioned tables."); + public static final ConfigOption<Long> COMMIT_LAST_SAFE_SNAPSHOT = + ConfigOptions.key("commit.last-safe-snapshot") + .longType() + .noDefaultValue() + .withFallbackKeys("commit.strict-mode.last-safe-snapshot") + .withDescription( + "Snapshot preceding the earliest snapshot to inspect when committing. " + + "Only later snapshots are searched for this commit user's previous commits. " + + "This also provides the starting point for strict-mode checks when enabled. " + + "Keep this bound unchanged across retries and recovery."); + + public static final ConfigOption<Boolean> COMMIT_STRICT_MODE_ENABLED = + ConfigOptions.key("commit.strict-mode.enabled") + .booleanType() + .defaultValue(true) + .withDescription( + "Whether to check concurrent snapshot changes after commit.last-safe-snapshot, " + + "when that bound is configured. Rejects COMPACT or OVERWRITE changes " + + "in the same partition, and fixed-bucket APPEND changes when committing OVERWRITE. " + + "Disabling this does not disable regular conflict detection or the history search bound."); + + /** @deprecated Use {@link #COMMIT_LAST_SAFE_SNAPSHOT}. */ + @Deprecated public static final ConfigOption<Long> COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT = ConfigOptions.key("commit.strict-mode.last-safe-snapshot") .longType() .noDefaultValue() - .withDescription( - "If set, committer will check if there are other commit user's snapshot starting from the " - + "snapshot after this one. If found a COMPACT / OVERWRITE snapshot, or found a " - + "APPEND snapshot which committed files to fixed bucket, commit will be aborted." - + "If the value of this option is -1, committer will not check for its first commit."); + .withDescription("Deprecated alias for commit.last-safe-snapshot."); public static final ConfigOption<String> CLUSTERING_COLUMNS = key("clustering.columns") @@ -4612,8 +4631,18 @@ public class CoreOptions implements Serializable { return options.get(AGGREGATION_REMOVE_RECORD_ON_DELETE); } + public Optional<Long> commitLastSafeSnapshot() { + return options.getOptional(COMMIT_LAST_SAFE_SNAPSHOT); + } + + public boolean commitStrictModeEnabled() { + return options.get(COMMIT_STRICT_MODE_ENABLED); + } + + /** @deprecated Use {@link #commitLastSafeSnapshot()}. */ + @Deprecated public Optional<Long> commitStrictModeLastSafeSnapshot() { - return options.getOptional(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT); + return commitLastSafeSnapshot(); } public List<String> clusteringColumns() { 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 0f60d6a3cb..e867a1cc85 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 @@ -222,7 +222,8 @@ public class FileStoreCommitImpl implements FileStoreCommit { this.statsFileHandler = statsFileHandler; this.bucketMode = bucketMode; this.strictModeChecker = - options.commitStrictModeLastSafeSnapshot() + options.commitLastSafeSnapshot() + .filter(ignored -> options.commitStrictModeEnabled()) .map( id -> new StrictModeChecker( @@ -294,14 +295,14 @@ public class FileStoreCommitImpl implements FileStoreCommit { "Committables must be sorted according to identifiers before filtering. This is unexpected."); } - Optional<Long> optionalStrictSnapshot = options.commitStrictModeLastSafeSnapshot(); + Optional<Long> lastSafeSnapshot = options.commitLastSafeSnapshot(); Optional<Snapshot> latestSnapshot; - if (optionalStrictSnapshot.isPresent()) { + if (lastSafeSnapshot.isPresent()) { latestSnapshot = snapshotManager.latestSnapshotOfUser( commitUser, snapshotManager.latestSnapshotId(), - optionalStrictSnapshot.get() + 1); + lastSafeSnapshot.get() + 1); } else { latestSnapshot = snapshotManager.latestSnapshotOfUser(commitUser); } diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java index 00d94845ea..dc93bf8dc9 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/commit/StrictModeChecker.java @@ -44,7 +44,7 @@ public class StrictModeChecker { private final IndexManifestFile indexManifestFile; private final boolean dataEvolutionEnabled; - private long strictModeLastSafeSnapshot; + private long lastSafeSnapshot; public StrictModeChecker( SnapshotManager snapshotManager, @@ -52,19 +52,19 @@ public class StrictModeChecker { Supplier<FileStoreScan> scanSupplier, IndexManifestFile indexManifestFile, boolean dataEvolutionEnabled, - long strictModeLastSafeSnapshot) { + long lastSafeSnapshot) { this.snapshotManager = snapshotManager; this.commitUser = commitUser; this.scanSupplier = scanSupplier; this.indexManifestFile = indexManifestFile; this.dataEvolutionEnabled = dataEvolutionEnabled; - this.strictModeLastSafeSnapshot = strictModeLastSafeSnapshot; + this.lastSafeSnapshot = lastSafeSnapshot; } public void check( long newSnapshotId, CommitKind newCommitKind, List<BinaryRow> newChangedPartitions) { Set<BinaryRow> newPartitions = new HashSet<>(newChangedPartitions); - for (long id = strictModeLastSafeSnapshot + 1; id < newSnapshotId; id++) { + for (long id = lastSafeSnapshot + 1; id < newSnapshotId; id++) { Snapshot snapshot = snapshotManager.snapshot(id); if (snapshot.commitUser().equals(commitUser)) { continue; @@ -84,13 +84,13 @@ public class StrictModeChecker { String.format( "When trying to commit snapshot %d, " + "commit user %s has found a %s snapshot (id: %d) by another user %s " - + "which modified the same partition. Giving up committing as %s is set.", + + "which modified the same partition. Giving up committing as %s is true.", newSnapshotId, commitUser, snapshot.commitKind().name(), id, snapshot.commitUser(), - CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key())); + CoreOptions.COMMIT_STRICT_MODE_ENABLED.key())); } } if (snapshot.commitKind() == CommitKind.APPEND @@ -109,12 +109,12 @@ public class StrictModeChecker { "When trying to commit snapshot %d, " + "commit user %s has found a APPEND snapshot (id: %d) by another user %s " + "which committed files to fixed bucket on the same partition. " - + "Giving up committing as %s is set.", + + "Giving up committing as %s is true.", newSnapshotId, commitUser, id, snapshot.commitUser(), - CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key())); + CoreOptions.COMMIT_STRICT_MODE_ENABLED.key())); } } } @@ -173,6 +173,6 @@ public class StrictModeChecker { } public void update(long newSafeSnapshot) { - strictModeLastSafeSnapshot = newSafeSnapshot; + lastSafeSnapshot = newSafeSnapshot; } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java index 7ea59f78a9..1aeaf94096 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/PostponeUtils.java @@ -54,7 +54,7 @@ import java.util.Optional; import java.util.Set; import static org.apache.paimon.CoreOptions.BUCKET; -import static org.apache.paimon.CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT; +import static org.apache.paimon.CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT; import static org.apache.paimon.CoreOptions.WRITE_ONLY; import static org.apache.paimon.utils.Preconditions.checkArgument; @@ -511,7 +511,7 @@ public class PostponeUtils { Map<String, String> rewriteOptions = new HashMap<>(); rewriteOptions.put(BUCKET.key(), String.valueOf(numBuckets)); rewriteOptions.put(WRITE_ONLY.key(), "false"); - rewriteOptions.put(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), String.valueOf(snapshotId)); + rewriteOptions.put(COMMIT_LAST_SAFE_SNAPSHOT.key(), String.valueOf(snapshotId)); return table.copy(rewriteOptions); } diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index e83ca0d4ac..585dd718dd 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -29,6 +29,38 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link org.apache.paimon.CoreOptions}. */ public class CoreOptionsTest { + @Test + void testCommitLastSafeSnapshotAndStrictMode() { + Options options = new Options(); + CoreOptions core = new CoreOptions(options); + assertThat(core.commitLastSafeSnapshot()).isEmpty(); + assertThat(core.commitStrictModeEnabled()).isTrue(); + + options.setString("commit.strict-mode.last-safe-snapshot", "7"); + assertThat(core.commitLastSafeSnapshot()).contains(7L); + assertThat(core.commitStrictModeEnabled()).isTrue(); + + options.set(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT, 11L); + assertThat(core.commitLastSafeSnapshot()).contains(11L); + + options.set(CoreOptions.COMMIT_STRICT_MODE_ENABLED, false); + assertThat(core.commitStrictModeEnabled()).isFalse(); + assertThat(core.commitLastSafeSnapshot()).contains(11L); + } + + @Test + @SuppressWarnings("deprecation") + void testLegacyCommitLastSafeSnapshotApi() { + Options options = new Options(); + options.set(CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT, 7L); + CoreOptions core = new CoreOptions(options); + assertThat(core.commitLastSafeSnapshot()).contains(7L); + assertThat(core.commitStrictModeLastSafeSnapshot()).contains(7L); + + options.set(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT, 11L); + assertThat(core.commitStrictModeLastSafeSnapshot()).contains(11L); + } + @Test void testManifestSidecarDefaultsToManifestSort() { assertThat(CoreOptions.MANIFEST_SIDECAR_ENABLED.defaultValue()).isNull(); 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 437944fae5..7a0d26b160 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 @@ -367,6 +367,45 @@ public class FileStoreCommitTest { } } + @ParameterizedTest + @ValueSource(strings = {"commit.last-safe-snapshot", "commit.strict-mode.last-safe-snapshot"}) + public void testFilterCommittedWithStrictModeDisabled(String lastSafeKey) throws Exception { + Map<String, String> options = new HashMap<>(); + options.put(lastSafeKey, "2"); + options.put(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), "false"); + TestFileStore store = createStore(false, options); + try (FileStoreCommit commit = store.newCommit("older-user", null)) { + commit.ignoreEmptyCommit(false); + commit.commit(new ManifestCommittable(1), false); + commit.commit(new ManifestCommittable(2), false); + } + + // A disabled strict checker must still honor the search bound. Fail if the lookup + // reaches old history, rather than relying on timing to detect a full history scan. + Path oldSnapshot = store.snapshotManager().snapshotPath(1); + store.fileIO().deleteQuietly(oldSnapshot); + store.fileIO().writeFile(oldSnapshot, "not a snapshot", false); + store.snapshotManager().invalidateCache(); + ManifestCommittable pending = new ManifestCommittable(10); + try (FileStoreCommit commit = store.newCommit("new-user", null)) { + assertThat(commit.filterCommitted(Collections.singletonList(pending))) + .containsExactly(pending); + commit.ignoreEmptyCommit(false); + commit.commit(pending, false); + } + try (FileStoreCommit commit = store.newCommit("other-user", null)) { + commit.ignoreEmptyCommit(false); + commit.commit(new ManifestCommittable(1), false); + } + + // Recovery after a successful commit with a lost response still deduplicates it. + try (FileStoreCommit recovered = store.newCommit("new-user", null)) { + ManifestCommittable next = new ManifestCommittable(11); + assertThat(recovered.filterCommitted(Arrays.asList(pending, next))) + .containsExactly(next); + } + } + protected void testRandomConcurrentNoConflict( int numThreads, boolean failing, CoreOptions.ChangelogProducer changelogProducer) throws Exception { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java index d6c6ddc5d3..b117fa1bda 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/PostponeUtilsTest.java @@ -253,7 +253,7 @@ public class PostponeUtilsTest { assertThat(options.getValue()) .containsEntry("bucket", "4") .containsEntry("write-only", "false") - .containsEntry("commit.strict-mode.last-safe-snapshot", "5"); + .containsEntry("commit.last-safe-snapshot", "5"); } @Test 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 index 4c1a3a94e9..00eb30d2f9 100644 --- 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 @@ -52,6 +52,7 @@ import org.apache.paimon.utils.SnapshotManager; 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.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import java.util.Arrays; @@ -68,7 +69,7 @@ import java.util.stream.Collectors; import java.util.stream.LongStream; import static java.util.Collections.singletonMap; -import static org.apache.paimon.CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT; +import static org.apache.paimon.CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT; import static org.apache.paimon.utils.FileStorePathFactoryTest.createNonPartFactory; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatCode; @@ -388,8 +389,15 @@ public class TableCommitTest { commit1.close(); } - @Test - public void testStrictModeForCompact() throws Exception { + @ParameterizedTest + @CsvSource({ + "commit.last-safe-snapshot, true", + "commit.last-safe-snapshot, false", + "commit.strict-mode.last-safe-snapshot, true", + "commit.strict-mode.last-safe-snapshot, false" + }) + public void testStrictModeForCompact(String lastSafeKey, boolean strictEnabled) + throws Exception { String path = tempDir.toString(); RowType rowType = RowType.of( @@ -428,7 +436,11 @@ public class TableCommitTest { // test skip this commit check String user2 = UUID.randomUUID().toString(); - table = table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2")); + Map<String, String> commitOptions = new HashMap<>(); + commitOptions.put(lastSafeKey, "2"); + commitOptions.put( + CoreOptions.COMMIT_STRICT_MODE_ENABLED.key(), String.valueOf(strictEnabled)); + table = table.copy(commitOptions); TableWriteImpl<?> write2 = table.newWrite(user2); TableCommitImpl commit2 = table.newCommit(user2); @@ -444,16 +456,21 @@ public class TableCommitTest { assertThatCode(() -> commit2.commit(2, write2.prepareCommit(false, 2))) .doesNotThrowAnyException(); - // COMPACT on the same partition should be checked and fail + // COMPACT on the same partition is rejected only when strict mode is enabled. write1.write(GenericRow.of(2, 6, 6L)); write1.compact(pt2, 0, true); commit1.commit(3, write1.prepareCommit(true, 3)); write2.write(GenericRow.of(2, 7, 7L)); - assertThatThrownBy(() -> commit2.commit(3, write2.prepareCommit(false, 3))) - .isInstanceOf(RuntimeException.class) - .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + if (strictEnabled) { + assertThatThrownBy(() -> commit2.commit(3, write2.prepareCommit(false, 3))) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining( + "Giving up committing as commit.strict-mode.enabled is true."); + } else { + commit2.commit(3, write2.prepareCommit(false, 3)); + assertThat(table.snapshotManager().latestSnapshot().commitUser()).isEqualTo(user2); + } write1.close(); commit1.close(); @@ -502,7 +519,7 @@ public class TableCommitTest { // test skip this commit check String user2 = UUID.randomUUID().toString(); FileStoreTable tableWithStrict = - table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2")); + table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "2")); TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2); TableCommitImpl commit2 = tableWithStrict.newCommit(user2); @@ -532,7 +549,7 @@ public class TableCommitTest { assertThatThrownBy(() -> commit2.commit(3, write2.prepareCommit(true, 3))) .isInstanceOf(RuntimeException.class) .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + "Giving up committing as commit.strict-mode.enabled is true."); write1.close(); commit1.close(); @@ -585,7 +602,7 @@ public class TableCommitTest { String user2 = UUID.randomUUID().toString(); FileStoreTable tableWithStrict = - table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "1")); + table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "1")); TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2); TableCommitImpl commit2 = tableWithStrict.newCommit(user2); write2.write(GenericRow.of(1, 1, 1L)); @@ -620,7 +637,7 @@ public class TableCommitTest { assertThatThrownBy(() -> commit2.commit(1, write2.prepareCommit(true, 1))) .isInstanceOf(RuntimeException.class) .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + "Giving up committing as commit.strict-mode.enabled is true."); } write1.close(); @@ -675,7 +692,7 @@ public class TableCommitTest { // user2 with strict mode, last-safe-snapshot=2 (skip its own snapshot 2) String user2 = UUID.randomUUID().toString(); FileStoreTable tableWithStrict = - table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2")); + table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "2")); TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2); TableCommitImpl commit2 = tableWithStrict.newCommit(user2); write2.write(GenericRow.of(1, 1, 1L)); @@ -699,7 +716,7 @@ public class TableCommitTest { assertThatThrownBy(() -> commit2.commit(3, write2.prepareCommit(true, 3))) .isInstanceOf(RuntimeException.class) .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + "Giving up committing as commit.strict-mode.enabled is true."); write1.close(); commit1.close(); @@ -764,7 +781,7 @@ public class TableCommitTest { // and avoid throwing. String user2 = UUID.randomUUID().toString(); FileStoreTable tableWithStrict = - table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2")); + table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "2")); TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2); TableCommitImpl commit2 = tableWithStrict.newCommit(user2); write2.write(GenericRow.of(3, 1, 1L)); @@ -852,7 +869,7 @@ public class TableCommitTest { // test skip this commit check String user2 = UUID.randomUUID().toString(); - table = table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2")); + table = table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "2")); TableWriteImpl<?> write2 = table.newWrite(user2); TableCommitImpl commit2 = table.newCommit(user2).withOverwrite(singletonMap("pt", "1")); @@ -898,7 +915,7 @@ public class TableCommitTest { assertThatThrownBy(() -> commit2.commit(4, write2.prepareCommit(false, 4))) .isInstanceOf(RuntimeException.class) .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + "Giving up committing as commit.strict-mode.enabled is true."); write1.close(); commit1.close(); @@ -983,7 +1000,7 @@ public class TableCommitTest { // get filtered out and the conflict is silently missed. String user2 = UUID.randomUUID().toString(); FileStoreTable tableWithStrict = - fixedTable.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "0")); + fixedTable.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "0")); TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2); TableCommitImpl commit2 = tableWithStrict.newCommit(user2).withOverwrite(singletonMap("pt", "1")); @@ -991,7 +1008,7 @@ public class TableCommitTest { assertThatThrownBy(() -> commit2.commit(3, write2.prepareCommit(false, 3))) .isInstanceOf(RuntimeException.class) .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + "Giving up committing as commit.strict-mode.enabled is true."); write2.close(); commit2.close(); @@ -1036,7 +1053,7 @@ public class TableCommitTest { // test skip this commit check String user2 = UUID.randomUUID().toString(); - table = table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2")); + table = table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "2")); TableWriteImpl<?> write2 = table.newWrite(user2); TableCommitImpl commit2 = table.newCommit(user2); @@ -1091,7 +1108,7 @@ public class TableCommitTest { // test skip this commit check String user2 = UUID.randomUUID().toString(); FileStoreTable tableWithStrict = - table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "2")); + table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "2")); TableWriteImpl<?> write2 = tableWithStrict.newWrite(user2); TableCommitImpl commit2 = tableWithStrict.newCommit(user2); @@ -1108,7 +1125,7 @@ public class TableCommitTest { assertThatThrownBy(() -> commit2.commit(2, write2.prepareCommit(false, 2))) .isInstanceOf(RuntimeException.class) .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + "Giving up committing as commit.strict-mode.enabled is true."); // APPEND with fixed bucket files should be caught when user2 commits OVERWRITE, // since non-partitioned table entries share BinaryRow.EMPTY_ROW so they always @@ -1119,7 +1136,7 @@ public class TableCommitTest { commit1Append.commit(3, write1Append.prepareCommit(false, 3)); FileStoreTable tableWithStrict2 = - table.copy(singletonMap(COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), "3")); + table.copy(singletonMap(COMMIT_LAST_SAFE_SNAPSHOT.key(), "3")); TableWriteImpl<?> write2Ow = tableWithStrict2.newWrite(user2); TableCommitImpl commit2Ow = tableWithStrict2.newCommit(user2).withOverwrite(Collections.emptyMap()); @@ -1127,7 +1144,7 @@ public class TableCommitTest { assertThatThrownBy(() -> commit2Ow.commit(4, write2Ow.prepareCommit(false, 4))) .isInstanceOf(RuntimeException.class) .hasMessageContaining( - "Giving up committing as commit.strict-mode.last-safe-snapshot is set."); + "Giving up committing as commit.strict-mode.enabled is true."); write1Append.close(); commit1Append.close(); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java index 4407ca3a85..5bda307641 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java @@ -162,7 +162,7 @@ public class DataEvolutionMergeIntoAction extends TableActionBase { table = table.copy( Collections.singletonMap( - CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), + CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), latestSnapshotId.toString())); this.coreOptions = ((FileStoreTable) table).coreOptions(); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RescaleAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RescaleAction.java index 3855a5115d..ed60db636f 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RescaleAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RescaleAction.java @@ -91,8 +91,7 @@ public class RescaleAction extends TableActionBase { // So we use strict mode to make sure nothing is lost. Map<String, String> dynamicOptions = new HashMap<>(); dynamicOptions.put( - CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), - String.valueOf(snapshot.id())); + CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), String.valueOf(snapshot.id())); fileStoreTable = fileStoreTable.copy(dynamicOptions); PartitionPredicate partitionPredicate = diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java index 55d6814c73..eddcf3bf91 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java @@ -81,8 +81,7 @@ public class DataEvolutionDeleteSink implements Serializable { : (FileStoreTable) table.copy( Collections.singletonMap( - CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT - .key(), + CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), String.valueOf(baseSnapshotId))); this.baseSnapshotId = baseSnapshotId; this.sinkParallelism = sinkParallelism; diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java index f220f84808..eedc8bfadf 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java @@ -326,7 +326,7 @@ public class DeleteActionDataEvolutionITCase extends ActionITCaseBase { first.run(); assertThatThrownBy(stale::run) - .hasStackTraceContaining(CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key()); + .hasStackTraceContaining(CoreOptions.COMMIT_STRICT_MODE_ENABLED.key()); testBatchRead( "SELECT id, name, dt FROM T ORDER BY id", Arrays.asList( diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java index 232f4370b5..1b549c6197 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RescaleProcedure.java @@ -125,7 +125,7 @@ public class RescaleProcedure extends BaseProcedure { // So we use strict mode to make sure nothing is lost. Map<String, String> dynamicOptions = new HashMap<>(); dynamicOptions.put( - CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), + CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), String.valueOf(snapshot.id())); fileStoreTable = fileStoreTable.copy(dynamicOptions); diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala index a4c5babc0d..da28f43d05 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/SparkPostponeStagedCommitter.scala @@ -60,9 +60,7 @@ private[spark] class SparkPostponeStagedCommitter( private val fixedWriteCommitUser = coreOptions.createCommitUser() private val fixedWriteTable = { val options = new java.util.HashMap[String, String]() - options.put( - CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), - baseSnapshotId.getOrElse(0L).toString) + options.put(CoreOptions.COMMIT_LAST_SAFE_SNAPSHOT.key(), baseSnapshotId.getOrElse(0L).toString) // Every overwrite in this coordinator supplies its exact BinaryRow partitions. Disable the // table-level dynamic rewrite so an empty rewritten partition is still removed, and implement // INSERT OVERWRITE dynamic-partition semantics explicitly in commitCurrentBatch. 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 index 91efc3d541..45f408fb78 100644 --- 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 @@ -19,7 +19,7 @@ package org.apache.paimon.spark.commands import org.apache.paimon.{CoreOptions, Snapshot} -import org.apache.paimon.CoreOptions.{COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT, PartitionSinkStrategy, WRITE_ONLY} +import org.apache.paimon.CoreOptions.{COMMIT_LAST_SAFE_SNAPSHOT, PartitionSinkStrategy, WRITE_ONLY} import org.apache.paimon.codegen.CodeGenUtils import org.apache.paimon.crosspartition.{IndexBootstrap, KeyPartOrRow} import org.apache.paimon.data.BinaryRow @@ -141,7 +141,7 @@ case class PaimonSparkWriter( case Some(_) => val directWriteOptions = new java.util.HashMap[String, String]() directWriteOptions.put( - COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(), + COMMIT_LAST_SAFE_SNAPSHOT.key(), postponeBaseSnapshotId.getOrElse(0L).toString) val builder = table.copy(directWriteOptions).newPostponeFixedBucketWriteBuilder() overwritePartitionSpec.foreach(spec => builder.withOverwrite(spec.asJava))
