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 4790703c21 [core][python] Disable automatic conflict rollback by
default (#10768)
4790703c21 is described below
commit 4790703c21d09627a27b5f336faeb5afa6a7c690
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Oct 24 11:07:17 2026 +0800
[core][python] Disable automatic conflict rollback by default (#10768)
---
docs/generated/core_configuration.html | 6 ++
.../main/java/org/apache/paimon/CoreOptions.java | 9 +++
.../paimon/operation/FileStoreCommitImpl.java | 5 +-
.../org/apache/paimon/rest/RESTCatalogTest.java | 2 +-
.../paimon/rest/RESTCommitLockIntegrationTest.java | 1 +
.../paimon/table/DataEvolutionTableTest.java | 85 ++++++++++++++++++++++
.../python/write/commit/commit_rollback.py | 61 ----------------
.../execution/python/write/file_store_commit.py | 28 +------
.../pypaimon/tests/file_store_commit_test.py | 65 +----------------
.../tests/ray_data_evolution_merge_into_test.py | 23 ++----
.../pypaimon/tests/rest/rollback_protocol_test.py | 53 ++++++--------
.../pypaimon/tests/shard_table_updator_test.py | 9 +--
12 files changed, 142 insertions(+), 205 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 7572026db7..d519e80197 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -296,6 +296,12 @@ under the License.
<td>String</td>
<td>Specifies the comparison algorithm used for range
partitioning, including 'zorder', 'hilbert', and 'order', corresponding to the
z-order curve algorithm, hilbert curve algorithm, and basic type comparison
algorithm, respectively. When not configured, it will automatically determine
the algorithm based on the number of columns in 'clustering.columns'. 'order'
is used for 1 column, 'zorder' for less than 5 columns, and 'hilbert' for 5 or
more columns.</td>
</tr>
+ <tr>
+ <td><h5>commit.auto-rollback.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Allow Java writers to roll back compaction on write conflicts.
May delete snapshots and tags. Does not affect explicit rollback or Python
writers.</td>
+ </tr>
<tr>
<td><h5>commit.callback.#.param</h5></td>
<td style="word-wrap: break-word;">(none)</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 25cf6876ad..de729d86bb 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1014,6 +1014,15 @@ public class CoreOptions implements Serializable {
.noDefaultValue()
.withDescription("Timeout duration of retry when commit
failed.");
+ public static final ConfigOption<Boolean> COMMIT_AUTO_ROLLBACK_ENABLED =
+ key("commit.auto-rollback.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Allow Java writers to roll back compaction on
write conflicts. "
+ + "May delete snapshots and tags. Does not
affect explicit rollback "
+ + "or Python writers.");
+
public static final ConfigOption<Integer> COMMIT_MAX_RETRIES =
key("commit.max-retries")
.intType()
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 ffe79cf4bb..a460a5234a 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
@@ -1240,7 +1240,10 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
rowIdConflictChecker,
commitKind);
if (exception.isPresent()) {
- if (allowRollback && rollback != null) {
+ if (allowRollback
+ && rollback != null
+ && options.toConfiguration()
+
.get(CoreOptions.COMMIT_AUTO_ROLLBACK_ENABLED)) {
snapshotCommit.ensureValid();
if (rollback.tryToRollback(latestSnapshot)) {
return RetryCommitResult.forRollback(exception.get());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index ad6f315261..780100cd27 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -6203,7 +6203,7 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
Lists.newArrayList(new DataField(0, "col1",
DataTypes.INT())),
emptyList(),
emptyList(),
- new HashMap<>(),
+
Collections.singletonMap("commit.auto-rollback.enabled", "true"),
""),
true);
Table table = catalog.getTable(identifier);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCommitLockIntegrationTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCommitLockIntegrationTest.java
index 21a37fbf2e..89b8123334 100644
---
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCommitLockIntegrationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCommitLockIntegrationTest.java
@@ -267,6 +267,7 @@ class RESTCommitLockIntegrationTest {
Schema schema =
Schema.newBuilder()
.column("id", DataTypes.INT())
+
.option(CoreOptions.COMMIT_AUTO_ROLLBACK_ENABLED.key(), "true")
.option(CoreOptions.REST_COMMIT_LOCK_ENABLED.key(),
"true")
.option(CoreOptions.REST_COMMIT_ACQUIRE_LOCK.key(),
String.valueOf(locking))
.option(CoreOptions.COMMIT_MIN_RETRY_WAIT.key(), "1
ms")
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
index 38dc6599cb..a6ee4254eb 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import org.apache.paimon.catalog.TableRollback;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.BlobData;
import org.apache.paimon.data.GenericRow;
@@ -59,6 +60,8 @@ import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
import java.util.Arrays;
@@ -1274,6 +1277,88 @@ public class DataEvolutionTableTest extends
DataEvolutionTestBase {
assertThat(readF0AndF2(table)).isEqualTo(Arrays.asList("10|updated-100",
"11|updated-101"));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"default", "false", "true"})
+ public void testConflictingWriterAutoRollback(String autoRollback) throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ long firstRowId = writeFullRows(table, 10, 11);
+ long readSnapshotId = table.latestSnapshot().get().id();
+
+ RowType writeType =
table.rowType().project(Collections.singletonList("f2"));
+ BatchWriteBuilder staleBuilder = table.newBatchWriteBuilder();
+ List<CommitMessage> staleMessages;
+ try (BatchTableWrite write =
staleBuilder.newWrite().withWriteType(writeType)) {
+ write.write(GenericRow.of(BinaryString.fromString("stale-10")));
+ write.write(GenericRow.of(BinaryString.fromString("stale-11")));
+ List<CommitMessage> prepared = write.prepareCommit();
+ setFirstRowId(prepared, firstRowId);
+ staleMessages =
+ prepared.stream()
+ .map(
+ message ->
+ ((CommitMessageImpl) message)
+
.withCheckFromSnapshot(readSnapshotId))
+ .collect(Collectors.toList());
+ }
+
+ updateF2(table, firstRowId, 100, 101);
+ FileStoreTable compactTable = withCompactOptions(table, "1 MB");
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(
+ compactTable, false, false,
compactTable.latestSnapshot().get());
+ List<DataEvolutionCompactTask> tasks = coordinator.plan();
+ assertThat(tasks.size()).isEqualTo(1);
+ CommitMessage compact = tasks.get(0).doCompact(compactTable,
"tagged-compaction");
+ try (BatchTableCommit commit =
compactTable.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(compact));
+ }
+ long concurrentSnapshotId = table.latestSnapshot().get().id();
+ table.createTag("published", concurrentSnapshotId);
+ AtomicInteger rollbackCalls = new AtomicInteger();
+ CatalogEnvironment environment =
+ new CatalogEnvironment(null, null, null, null, null, null,
false, false) {
+ @Override
+ public TableRollback catalogTableRollback(String branch) {
+ return (instant, fromSnapshot) -> {
+ rollbackCalls.incrementAndGet();
+ table.rollbackTo(((Instant.SnapshotInstant)
instant).getSnapshotId());
+ };
+ }
+ };
+ FileStoreTable writerTable =
+ new AppendOnlyFileStoreTable(
+ table.fileIO(), table.location(), table.schema(),
environment);
+ if (!autoRollback.equals("default")) {
+ writerTable =
+ writerTable.copy(
+
Collections.singletonMap("commit.auto-rollback.enabled", autoRollback));
+ }
+ FileStoreTable configuredWriter = writerTable;
+
+ assertThatThrownBy(
+ () -> {
+ try (BatchTableCommit commit =
+
configuredWriter.newBatchWriteBuilder().newCommit()) {
+ commit.commit(staleMessages);
+ }
+ })
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(DATA_EVOLUTION_ROW_ID_CONFLICT_MESSAGE);
+ if (autoRollback.equals("true")) {
+ assertThat(rollbackCalls.get()).isEqualTo(1);
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(concurrentSnapshotId -
1);
+ assertThat(table.tagManager().tagExists("published")).isFalse();
+ return;
+ }
+
assertThat(table.tagManager().getOrThrow("published").id()).isEqualTo(concurrentSnapshotId);
+ assertThat(rollbackCalls.get()).isZero();
+
assertThat(readF0AndF2(table.copy(Collections.singletonMap("scan.tag-name",
"published"))))
+ .isEqualTo(Arrays.asList("10|updated-100", "11|updated-101"));
+
assertThat(table.latestSnapshot().get().id()).isEqualTo(concurrentSnapshotId);
+
assertThat(readF0AndF2(table)).isEqualTo(Arrays.asList("10|updated-100",
"11|updated-101"));
+ }
+
@Test
public void testRejectDifferentRowIdCheckSnapshotsInOneCommit() throws
Exception {
createTableDefault();
diff --git
a/paimon-python/pypaimon/execution/python/write/commit/commit_rollback.py
b/paimon-python/pypaimon/execution/python/write/commit/commit_rollback.py
deleted file mode 100644
index ac59051925..0000000000
--- a/paimon-python/pypaimon/execution/python/write/commit/commit_rollback.py
+++ /dev/null
@@ -1,61 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing,
-# software distributed under the License is distributed on an
-# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-
-"""
-Commit rollback to rollback 'COMPACT' commits for resolving conflicts.
-"""
-
-from pypaimon.table.instant import Instant
-
-
-class CommitRollback:
- """Rollback COMPACT commits to resolve conflicts.
-
- When a conflict is detected during commit, if the latest snapshot is a
- COMPACT commit, it can be rolled back via TableRollback.
- """
-
- def __init__(self, table_rollback):
- """Initialize CommitRollback.
-
- Args:
- table_rollback: A TableRollback instance used to perform the
rollback.
- """
- self._table_rollback = table_rollback
-
- def try_to_rollback(self, latest_snapshot):
- """Try to rollback a COMPACT commit to resolve conflicts.
-
- Only rolls back COMPACT type commits. Delegates to TableRollback
- to rollback to the previous snapshot (latest - 1), passing the
- latest snapshot ID as from_snapshot.
-
- Args:
- latest_snapshot: The latest snapshot that may need to be rolled
back.
-
- Returns:
- True if rollback succeeded, False otherwise.
- """
- if latest_snapshot.commit_kind == "COMPACT":
- latest_id = latest_snapshot.id
- try:
- self._table_rollback.rollback_to(
- Instant.snapshot(latest_id - 1), latest_id)
- return True
- except Exception:
- pass
- return False
diff --git a/paimon-python/pypaimon/execution/python/write/file_store_commit.py
b/paimon-python/pypaimon/execution/python/write/file_store_commit.py
index 2451488e28..de76eff182 100644
--- a/paimon-python/pypaimon/execution/python/write/file_store_commit.py
+++ b/paimon-python/pypaimon/execution/python/write/file_store_commit.py
@@ -36,7 +36,6 @@ from pypaimon.snapshot.snapshot_commit import
(PartitionStatistics,
from pypaimon.table.row.generic_row import GenericRow, GenericRowSerializer
from pypaimon.table.row.offset_row import OffsetRow
from pypaimon.utils.file_store_path_factory import canonical_data_file_path
-from pypaimon.execution.python.write.commit.commit_rollback import
CommitRollback
from pypaimon.execution.python.write.commit.commit_scanner import CommitScanner
from pypaimon.execution.python.write.commit.conflict_detection import (
ConflictDetection,
@@ -122,13 +121,6 @@ class CommitFailRetryResult(RetryResult):
"""Retry after an atomic snapshot commit failed, matching Java."""
-class RollbackRetryResult(RetryResult):
- """Retry after a conflicting compaction was rolled back, matching Java."""
-
- def __init__(self, exception: Optional[Exception] = None):
- super().__init__(None, exception)
-
-
class RewriteResult(CommitResult):
def __init__(self, rewrite: RowIdRewriteResult):
@@ -181,9 +173,6 @@ class FileStoreCommit:
commit_scanner=self.commit_scanner
)
- table_rollback = table.catalog_environment.catalog_table_rollback()
- self.rollback = CommitRollback(table_rollback) if table_rollback is
not None else None
-
def _set_fixed_bucket_commit_check(self, messages):
from pypaimon.execution.python.write.commit.fixed_bucket_commit_check
import FixedBucketCommitCheck
@@ -238,15 +227,12 @@ class FileStoreCommit:
commit_kind = "APPEND"
detect_conflicts = False
- allow_rollback = False
if self.conflict_detection.should_be_overwrite_commit(
commit_entries, index_adds + index_deletes):
commit_kind = "OVERWRITE"
detect_conflicts = True
- allow_rollback = True
if self.conflict_detection.has_row_id_check_from_snapshot():
detect_conflicts = True
- allow_rollback = True
if self.conflict_detection.has_global_index_additions(index_adds):
detect_conflicts = True
if self.conflict_detection.has_hash_index_changes(
@@ -262,7 +248,6 @@ class FileStoreCommit:
commit_entries_plan=lambda snapshot: commit_entries,
changelog_entries=changelog_entries,
detect_conflicts=detect_conflicts,
- allow_rollback=allow_rollback,
index_deletes=index_deletes,
index_adds=index_adds,
snapshot_properties=snapshot_properties,
@@ -308,7 +293,6 @@ class FileStoreCommit:
commit_entries_plan=provider.provide,
changelog_entries=[],
detect_conflicts=True,
- allow_rollback=False,
snapshot_properties=snapshot_properties,
# Java records static/table overwrite even when no files match.
allow_empty_commit=True,
@@ -341,7 +325,6 @@ class FileStoreCommit:
commit_identifier=commit_identifier,
commit_entries_plan=provider.provide,
detect_conflicts=True,
- allow_rollback=False,
# Java records a partition overwrite even when no files match.
allow_empty_commit=True,
)
@@ -354,12 +337,11 @@ class FileStoreCommit:
commit_identifier=commit_identifier,
commit_entries_plan=provider.provide,
detect_conflicts=True,
- allow_rollback=False,
allow_empty_commit=True,
)
def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan,
- detect_conflicts=False, allow_rollback=False,
index_deletes=None,
+ detect_conflicts=False, index_deletes=None,
index_adds=None, changelog_entries=None,
snapshot_properties: Optional[Dict[str, str]] = None,
allow_empty_commit=False):
@@ -406,7 +388,6 @@ class FileStoreCommit:
commit_identifier=commit_identifier,
latest_snapshot=latest_snapshot,
detect_conflicts=detect_conflicts,
- allow_rollback=allow_rollback,
index_deletes=attempt_index_deletes,
index_adds=attempt_index_adds,
commit_result_may_be_uncertain=commit_result_may_be_uncertain,
@@ -485,7 +466,6 @@ class FileStoreCommit:
commit_identifier: int,
latest_snapshot: Optional[Snapshot],
detect_conflicts: bool = False,
- allow_rollback: bool = False,
index_deletes=None,
index_adds=None,
commit_result_may_be_uncertain: bool = False,
@@ -564,11 +544,7 @@ class FileStoreCommit:
)
if rewrite_result is not None:
return RewriteResult(rewrite_result)
- if allow_rollback and self.rollback is not None:
- if self.rollback.try_to_rollback(latest_snapshot):
- # Rolled back: base/snapshot no longer valid; next
attempt
- # re-scans from scratch (matches Java
RollbackRetryResult).
- return RollbackRetryResult(conflict_exception)
+ # Automatic rollback can delete snapshots and tags already
used by readers.
raise conflict_exception
# Apply row tracking logic after conflict detection (matches Java
ordering)
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 3c37d1d8fd..9a4bdb47ac 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -39,7 +39,6 @@ from pypaimon.write.commit_message import CommitMessage
from pypaimon.execution.python.write.file_store_commit import (
CommitFailRetryResult,
FileStoreCommit,
- RollbackRetryResult,
RewriteResult,
_reject_compact_increment,
_row_id_check_from_messages,
@@ -392,62 +391,7 @@ class TestFileStoreCommit(unittest.TestCase):
schema_id=0,
)
- def test_conflict_rollback_retry_skips_history_and_rescans_base(
- self, mock_manifest_list_manager, mock_manifest_file_manager):
- file_store_commit = self._create_file_store_commit()
- conflict = RuntimeError("conflicting compaction")
- read_all = Mock(return_value=[])
-
file_store_commit.commit_scanner.read_all_entries_from_changed_partitions =
read_all
- file_store_commit.conflict_detection.check_conflicts = Mock(
- return_value=conflict
- )
- file_store_commit.rollback = Mock()
- file_store_commit.rollback.try_to_rollback.return_value = True
-
- result = file_store_commit._try_commit_once(
- retry_result=None,
- commit_kind="APPEND",
- commit_entries=[Mock()],
- changelog_entries=[],
- commit_identifier=11,
- latest_snapshot=Mock(id=3),
- detect_conflicts=True,
- allow_rollback=True,
- )
-
- self.assertIsInstance(result, RollbackRetryResult)
- self.assertIs(result.exception, conflict)
-
- file_store_commit.snapshot_manager.get_snapshot_by_id.side_effect = (
- AssertionError("rollback retry must not scan snapshot history")
- )
- file_store_commit.commit_scanner.read_incremental_changes = Mock(
- side_effect=AssertionError(
- "rollback retry must not reuse the previous conflict base"
- )
- )
- read_all.reset_mock()
- file_store_commit.conflict_detection.check_conflicts.return_value = (
- RuntimeError("conflict after rollback retry")
- )
-
- with self.assertRaisesRegex(
- RuntimeError, "conflict after rollback retry"):
- file_store_commit._try_commit_once(
- retry_result=result,
- commit_kind="APPEND",
- commit_entries=[Mock()],
- changelog_entries=[],
- commit_identifier=11,
- latest_snapshot=Mock(id=2),
- detect_conflicts=True,
- )
-
-
file_store_commit.snapshot_manager.get_snapshot_by_id.assert_not_called()
-
file_store_commit.commit_scanner.read_incremental_changes.assert_not_called()
- read_all.assert_called_once()
-
- def test_conflict_rollback_failure_preserves_original_conflict(
+ def test_conflict_preserves_committed_snapshot_without_rollback(
self, mock_manifest_list_manager, mock_manifest_file_manager):
file_store_commit = self._create_file_store_commit()
conflict = RuntimeError("conflicting compaction")
@@ -458,8 +402,8 @@ class TestFileStoreCommit(unittest.TestCase):
file_store_commit.conflict_detection.check_conflicts = Mock(
return_value=conflict
)
- file_store_commit.rollback = Mock()
- file_store_commit.rollback.try_to_rollback.return_value = False
+ rollback =
file_store_commit.table.catalog_environment.catalog_table_rollback
+ rollback.reset_mock()
with self.assertRaises(RuntimeError) as failure:
file_store_commit._try_commit_once(
@@ -470,11 +414,10 @@ class TestFileStoreCommit(unittest.TestCase):
commit_identifier=11,
latest_snapshot=latest_snapshot,
detect_conflicts=True,
- allow_rollback=True,
)
self.assertIs(failure.exception, conflict)
-
file_store_commit.rollback.try_to_rollback.assert_called_once_with(latest_snapshot)
+ rollback.assert_not_called()
self.mock_snapshot_commit.commit.assert_not_called()
mock_manifest_file_manager.return_value.write.assert_not_called()
mock_manifest_list_manager.return_value.write.assert_not_called()
diff --git a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
index 37f51f9e74..6022dba61c 100644
--- a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
@@ -2689,7 +2689,7 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
# Match Spark: replaced staging files are left for orphan cleanup.
self.assertTrue(all(os.path.exists(path) for path in stale_paths))
- def test_self_merge_compaction_retry_checks_core_rollback(self):
+ def test_self_merge_compaction_retry_preserves_committed_snapshot(self):
from pypaimon.ray import data_evolution_merge_into as merge_module
from pypaimon.ray import row_id_conflict_rewriter as rewriter_module
from pypaimon.execution.python.write.file_store_commit import
FileStoreCommit
@@ -2700,12 +2700,14 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
table = self.catalog.get_table(target)
real_apply = merge_module.distributed_self_merge_update_apply
real_rewrite = rewriter_module._rewrite_updates
- real_commit_init = FileStoreCommit.__init__
- rollbacks = []
+ tagged_versions = []
def stage_then_compact(*args, **kwargs):
result = real_apply(*args, **kwargs)
self._compact_all_data_files(table)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ table.create_tag("published", snapshot_id=snapshot.id)
+ tagged_versions.append(snapshot.id)
return result
rewrite_calls = [0]
@@ -2716,12 +2718,6 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
return None
return real_rewrite(*args, **kwargs)
- def init_with_rollback(commit, *args, **kwargs):
- real_commit_init(commit, *args, **kwargs)
- commit.rollback = Mock()
- commit.rollback.try_to_rollback.return_value = False
- rollbacks.append(commit.rollback)
-
with patch.object(
merge_module,
'distributed_self_merge_update_apply',
@@ -2730,10 +2726,6 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
rewriter_module,
'_rewrite_updates',
side_effect=miss_precommit_race,
- ), patch.object(
- FileStoreCommit,
- '__init__',
- new=init_with_rollback,
), patch.object(
FileStoreCommit,
'_commit_retry_wait',
@@ -2750,12 +2742,11 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
num_partitions=_TEST_NUM_PARTITIONS,
)
+ self.assertEqual(table.tag_manager().get_or_throw("published").id,
tagged_versions[0])
+
self.assertIsNotNone(table.snapshot_manager().get_snapshot_by_id(tagged_versions[0]))
self.assertEqual(result['num_matched'], 4)
self.assertEqual(self._read_sorted(target)['age'], [99, 99, 99, 99])
commit_retry_wait.assert_not_called()
- self.assertTrue(any(
- rollback.try_to_rollback.called for rollback in rollbacks
- ))
ray_retry_wait.assert_called_once()
self.assertEqual(ray_retry_wait.call_args[0][1], 0)
self.assertEqual(rewrite_calls[0], 2)
diff --git a/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
b/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
index 6c28457d6f..e9dc8b7651 100644
--- a/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
+++ b/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
@@ -17,7 +17,6 @@
import json
from types import SimpleNamespace
-from unittest.mock import Mock
from urllib.parse import unquote_plus, urlsplit
import pyarrow as pa
@@ -37,7 +36,6 @@ from pypaimon.common.options import Options
from pypaimon.execution.native.table import (_resolved_rest_table_response)
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.table.instant import Instant
-from pypaimon.execution.python.write.file_store_commit import FileStoreCommit
from pypaimon.execution.native.write.native_commit import
_rest_catalog_supported
@@ -133,7 +131,7 @@ def test_table_rollback_transmits_loaded_uuid(table,
transport, target, instant)
@pytest.mark.parametrize('snapshot_uuid', ['observed-head-uuid', 'missing',
None])
@pytest.mark.parametrize('branch', [None, 'dev / blue'])
-def test_compact_rollback_transmits_uuid_and_expected_head(
+def test_conditional_rollback_transmits_uuid_and_expected_head(
catalog, transport, compact_head, snapshot_uuid, branch):
payload = json.loads(JSON.to_json(compact_head))
if snapshot_uuid == 'missing':
@@ -145,9 +143,8 @@ def test_compact_rollback_transmits_uuid_and_expected_head(
identifier = Identifier('db', 'table', branch=branch)
table = catalog.get_table(identifier)
transport.requests.clear()
- commit = FileStoreCommit(Mock(), table, 'current-writer')
-
- assert commit.rollback.try_to_rollback(head)
+ table.catalog_environment.catalog_table_rollback().rollback_to(
+ Instant.snapshot(head.id - 1), head.id)
assert len(transport.requests) == 1
assert transport.requests[0].method == 'POST'
@@ -159,15 +156,6 @@ def test_compact_rollback_transmits_uuid_and_expected_head(
assert object_name == identifier.get_object_name()
[email protected]('commit_kind', ['APPEND', 'OVERWRITE'])
-def test_non_compact_head_is_not_rolled_back(table, transport, compact_head,
commit_kind):
- compact_head.commit_kind = commit_kind
- commit = FileStoreCommit(Mock(), table, 'current-writer')
-
- assert not commit.rollback.try_to_rollback(compact_head)
- assert transport.requests == []
-
-
@pytest.mark.parametrize('entrypoint', ['catalog', 'api'])
def test_direct_rollback_accepts_optional_uuid(catalog, transport, entrypoint):
delegate = catalog if entrypoint == 'catalog' else catalog.rest_api
@@ -241,15 +229,16 @@ def
test_replaced_table_does_not_refresh_retry_or_clean_locally(table, transport
@pytest.mark.parametrize('status', [400, 403, 404, 409])
-def test_compact_rollback_rejection_preserves_identity(table, transport,
compact_head, monkeypatch, status):
+def test_conditional_rollback_rejection_preserves_identity(table, transport,
compact_head, monkeypatch, status):
transport.status = status
- commit = FileStoreCommit(Mock(), table, 'current-writer')
def unexpected_cleanup():
pytest.fail('Catalog rejection must not trigger local cleanup')
monkeypatch.setattr(table, 'rollback_helper', unexpected_cleanup)
- assert not commit.rollback.try_to_rollback(compact_head)
+ with pytest.raises(RuntimeError, match="Failed to rollback table"):
+ table.catalog_environment.catalog_table_rollback().rollback_to(
+ Instant.snapshot(compact_head.id - 1), compact_head.id)
assert len(transport.requests) == 1
assert transport.requests[0].method == 'POST'
@@ -280,15 +269,15 @@ def
test_legacy_catalog_override_never_discards_known_uuid(table_uuid):
@pytest.mark.parametrize('source,target', [(None, 'dev'), ('dev', None),
('dev', 'release')])
[email protected]('automatic', [False, True])
-def test_copied_branch_rollback_omits_source_identity(catalog, transport,
compact_head, source, target, automatic):
[email protected]('conditional', [False, True])
+def test_copied_branch_rollback_omits_source_identity(catalog, transport,
compact_head, source, target, conditional):
table = catalog.get_table(Identifier('db', 'table', branch=source))
copied = table.copy({'branch': target})
transport.requests.clear()
- if automatic:
- commit = FileStoreCommit(Mock(), copied, 'current-writer')
- assert commit.rollback.try_to_rollback(compact_head)
+ if conditional:
+ copied.catalog_environment.catalog_table_rollback().rollback_to(
+ Instant.snapshot(compact_head.id - 1), compact_head.id)
else:
copied.rollback_to(1)
@@ -296,15 +285,15 @@ def
test_copied_branch_rollback_omits_source_identity(catalog, transport, compac
assert copied.catalog_environment.uuid is None
assert len(transport.requests) == 1
assert request_body(transport.requests[0]) == {
- 'instant': {'type': 'snapshot', 'snapshotId': 2 if automatic else 1},
- **({'fromSnapshot': 3} if automatic else {}),
+ 'instant': {'type': 'snapshot', 'snapshotId': 2 if conditional else 1},
+ **({'fromSnapshot': 3} if conditional else {}),
}
object_name =
unquote_plus(urlsplit(transport.requests[0].url).path.split('/')[-2])
assert object_name == Identifier('db', 'table',
branch=target).get_object_name()
[email protected]('automatic', [False, True])
-def test_same_branch_copy_and_independent_load_preserve_identity(table,
catalog, transport, compact_head, automatic):
[email protected]('conditional', [False, True])
+def test_same_branch_copy_and_independent_load_preserve_identity(table,
catalog, transport, compact_head, conditional):
assert table.copy({'scan.parallelism': '2'}).catalog_environment.uuid ==
'loaded-table-uuid'
assert table.copy({'branch': 'main'}).catalog_environment.uuid ==
'loaded-table-uuid'
assert _rest_catalog_supported(table)
@@ -319,17 +308,17 @@ def
test_same_branch_copy_and_independent_load_preserve_identity(table, catalog,
assert _rest_catalog_supported(branch)
transport.requests.clear()
- if automatic:
+ if conditional:
copied = branch.copy({'branch': 'dev'})
- commit = FileStoreCommit(Mock(), copied, 'current-writer')
- assert commit.rollback.try_to_rollback(compact_head)
+ copied.catalog_environment.catalog_table_rollback().rollback_to(
+ Instant.snapshot(compact_head.id - 1), compact_head.id)
else:
branch.rollback_to(1)
assert len(transport.requests) == 1
assert request_body(transport.requests[0]) == {
- 'instant': {'type': 'snapshot', 'snapshotId': 2 if automatic else 1},
- **({'fromSnapshot': 3} if automatic else {}), 'tableId':
'loaded-dev-uuid',
+ 'instant': {'type': 'snapshot', 'snapshotId': 2 if conditional else 1},
+ **({'fromSnapshot': 3} if conditional else {}), 'tableId':
'loaded-dev-uuid',
}
object_name =
unquote_plus(urlsplit(transport.requests[0].url).path.split('/')[-2])
assert object_name == Identifier('db', 'table',
branch='dev').get_object_name()
diff --git a/paimon-python/pypaimon/tests/shard_table_updator_test.py
b/paimon-python/pypaimon/tests/shard_table_updator_test.py
index fb3968f81d..1bcc57eba3 100644
--- a/paimon-python/pypaimon/tests/shard_table_updator_test.py
+++ b/paimon-python/pypaimon/tests/shard_table_updator_test.py
@@ -615,7 +615,7 @@ class ShardTableUpdatorTest(unittest.TestCase):
% actual_columns
)
- def test_shard_update_passes_allow_rollback_true(self):
+ def test_shard_update_enables_conflict_detection(self):
table_schema = pa.schema([
('a', pa.int32()),
('b', pa.int32()),
@@ -624,7 +624,7 @@ class ShardTableUpdatorTest(unittest.TestCase):
table_schema,
options={'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true'}
)
- name = self._create_unique_table_name('rollback')
+ name = self._create_unique_table_name('conflict_detection')
self.catalog.create_table(name, schema, False)
table = self.catalog.get_table(name)
@@ -664,11 +664,6 @@ class ShardTableUpdatorTest(unittest.TestCase):
tc2.commit(commit_messages)
tc2.close()
- self.assertTrue(
- captured_args.get('allow_rollback', False),
- "Row-id-check commits must pass allow_rollback=True so that "
- "concurrent COMPACT snapshots can be rolled back on conflict."
- )
self.assertTrue(
captured_args.get('detect_conflicts', False),
"Row-id-check commits must enable conflict detection."
