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 c6292e1ee3 [python] Add regression coverage for automatic REST
rollback (#10555)
c6292e1ee3 is described below
commit c6292e1ee346613ae9e2edbba99a12960dc0dd8d
Author: QuakeWang <[email protected]>
AuthorDate: Fri Oct 16 13:02:29 2026 +0800
[python] Add regression coverage for automatic REST rollback (#10555)
---
.../pypaimon/tests/file_store_commit_test.py | 32 ++++++++
.../pypaimon/tests/rest/rollback_protocol_test.py | 95 ++++++++++++++++++----
2 files changed, 111 insertions(+), 16 deletions(-)
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index e4c3d1d568..c9ac5da573 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -447,6 +447,38 @@ class TestFileStoreCommit(unittest.TestCase):
file_store_commit.commit_scanner.read_incremental_changes.assert_not_called()
read_all.assert_called_once()
+ def test_conflict_rollback_failure_preserves_original_conflict(
+ self, mock_manifest_list_manager, mock_manifest_file_manager):
+ file_store_commit = self._create_file_store_commit()
+ conflict = RuntimeError("conflicting compaction")
+ latest_snapshot = Mock(id=3)
+
file_store_commit.commit_scanner.read_all_entries_from_changed_partitions =
Mock(
+ return_value=[]
+ )
+ 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
+
+ with self.assertRaises(RuntimeError) as failure:
+ file_store_commit._try_commit_once(
+ retry_result=None,
+ commit_kind="APPEND",
+ commit_entries=[Mock()],
+ changelog_entries=[],
+ 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)
+ 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()
+
def test_commit_fail_retry_missing_snapshot_still_fails_closed(
self, mock_manifest_list_manager, mock_manifest_file_manager):
file_store_commit = self._create_file_store_commit()
diff --git a/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
b/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
index 085d251515..56ac97c37b 100644
--- a/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
+++ b/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
@@ -17,6 +17,7 @@
import json
from types import SimpleNamespace
+from unittest.mock import Mock
from urllib.parse import unquote_plus, urlsplit
import pyarrow as pa
@@ -34,8 +35,9 @@ from pypaimon.common.identifier import Identifier
from pypaimon.common.json_util import JSON
from pypaimon.common.options import Options
from pypaimon.read.native_plan import _resolved_rest_table_response
+from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.table.instant import Instant
-from pypaimon.write.commit.commit_rollback import CommitRollback
+from pypaimon.write.file_store_commit import FileStoreCommit
from pypaimon.write.native_commit import _rest_catalog_supported
@@ -83,6 +85,15 @@ def table(catalog, transport):
return loaded
[email protected]
+def compact_head():
+ return Snapshot(
+ version=3, id=3, schema_id=0, base_manifest_list='base-list',
+ delta_manifest_list='delta-list', total_record_count=0,
delta_record_count=0,
+ commit_user='snapshot-author', commit_identifier=1,
commit_kind='COMPACT',
+ time_millis=0, uuid='observed-head-uuid')
+
+
def request_body(request):
return json.loads(request.body)
@@ -117,16 +128,41 @@ def test_table_rollback_transmits_loaded_uuid(table,
transport, target, instant)
}
-def test_compact_rollback_transmits_uuid_and_expected_head(table, transport):
- rollback =
CommitRollback(table.catalog_environment.catalog_table_rollback())
[email protected]('snapshot_uuid', ['observed-head-uuid', 'missing',
None])
[email protected]('branch', [None, 'dev / blue'])
+def test_compact_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':
+ payload.pop('uuid')
+ else:
+ payload['uuid'] = snapshot_uuid
+ head = JSON.from_json(json.dumps(payload), Snapshot)
+ assert head.uuid == (None if snapshot_uuid == 'missing' else snapshot_uuid)
+ identifier = Identifier('db', 'table', branch=branch)
+ table = catalog.get_table(identifier)
+ transport.requests.clear()
+ commit = FileStoreCommit(Mock(), table, 'current-writer')
- assert rollback.try_to_rollback(SimpleNamespace(commit_kind='COMPACT',
id=3))
+ assert commit.rollback.try_to_rollback(head)
assert len(transport.requests) == 1
+ assert transport.requests[0].method == 'POST'
assert request_body(transport.requests[0]) == {
'instant': {'type': 'snapshot', 'snapshotId': 2},
'fromSnapshot': 3, 'tableId': 'loaded-table-uuid',
}
+ object_name =
unquote_plus(urlsplit(transport.requests[0].url).path.split('/')[-2])
+ 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'])
@@ -200,15 +236,23 @@ def
test_replaced_table_does_not_refresh_retry_or_clean_locally(table, transport
assert request_body(transport.requests[0])['tableId'] ==
'loaded-table-uuid'
-def test_compact_rollback_rejection_preserves_identity(table, transport):
- transport.status = 400
- rollback =
CommitRollback(table.catalog_environment.catalog_table_rollback())
[email protected]('status', [400, 403, 404, 409])
+def test_compact_rollback_rejection_preserves_identity(table, transport,
compact_head, monkeypatch, status):
+ transport.status = status
+ commit = FileStoreCommit(Mock(), table, 'current-writer')
- assert not rollback.try_to_rollback(SimpleNamespace(commit_kind='COMPACT',
id=3))
+ 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)
assert len(transport.requests) == 1
- assert request_body(transport.requests[0])['tableId'] ==
'loaded-table-uuid'
- assert request_body(transport.requests[0])['fromSnapshot'] == 3
+ assert transport.requests[0].method == 'POST'
+ assert request_body(transport.requests[0]) == {
+ 'instant': {'type': 'snapshot', 'snapshotId': 2},
+ 'fromSnapshot': 3, 'tableId': 'loaded-table-uuid',
+ }
@pytest.mark.parametrize('table_uuid', [None, 'known-uuid'])
@@ -232,22 +276,31 @@ def
test_legacy_catalog_override_never_discards_known_uuid(table_uuid):
@pytest.mark.parametrize('source,target', [(None, 'dev'), ('dev', None),
('dev', 'release')])
-def test_copied_branch_rollback_omits_source_identity(catalog, transport,
source, target):
[email protected]('automatic', [False, True])
+def test_copied_branch_rollback_omits_source_identity(catalog, transport,
compact_head, source, target, automatic):
table = catalog.get_table(Identifier('db', 'table', branch=source))
copied = table.copy({'branch': target})
transport.requests.clear()
- copied.rollback_to(1)
+ if automatic:
+ commit = FileStoreCommit(Mock(), copied, 'current-writer')
+ assert commit.rollback.try_to_rollback(compact_head)
+ else:
+ copied.rollback_to(1)
assert table.catalog_environment.uuid == 'loaded-table-uuid'
assert copied.catalog_environment.uuid is None
assert len(transport.requests) == 1
- assert 'tableId' not in request_body(transport.requests[0])
+ assert request_body(transport.requests[0]) == {
+ 'instant': {'type': 'snapshot', 'snapshotId': 2 if automatic else 1},
+ 'fromSnapshot': 3 if automatic else None,
+ }
object_name =
unquote_plus(urlsplit(transport.requests[0].url).path.split('/')[-2])
assert object_name == Identifier('db', 'table',
branch=target).get_object_name()
-def test_same_branch_copy_and_independent_load_preserve_identity(table,
catalog, transport):
[email protected]('automatic', [False, True])
+def test_same_branch_copy_and_independent_load_preserve_identity(table,
catalog, transport, compact_head, automatic):
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)
@@ -262,10 +315,20 @@ def
test_same_branch_copy_and_independent_load_preserve_identity(table, catalog,
assert _rest_catalog_supported(branch)
transport.requests.clear()
- branch.rollback_to(1)
+ if automatic:
+ copied = branch.copy({'branch': 'dev'})
+ commit = FileStoreCommit(Mock(), copied, 'current-writer')
+ assert commit.rollback.try_to_rollback(compact_head)
+ else:
+ branch.rollback_to(1)
assert len(transport.requests) == 1
- assert request_body(transport.requests[0])['tableId'] == 'loaded-dev-uuid'
+ assert request_body(transport.requests[0]) == {
+ 'instant': {'type': 'snapshot', 'snapshotId': 2 if automatic else 1},
+ 'fromSnapshot': 3 if automatic else None, '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()
def test_copied_branch_rejects_source_read_metadata(table, monkeypatch):
