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 4602768004 [core] Separate commit history bounds from strict mode
(#9960)
4602768004 is described below
commit 4602768004b3ea98658bc7eb0c9351192ebfae7a
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))
