From 4b3c37b6d0bab9f0ea5520d2079b09910624f011 Mon Sep 17 00:00:00 2001 From: chenjian2664 Date: Thu, 29 Oct 2026 11:28:46 +0800 Subject: [PATCH] [python] Fix branch creation from tags after snapshot expiration --- .../branch/filesystem_branch_manager.py | 15 ++++++---- .../tests/filesystem_catalog_branch_test.py | 28 +++++++++++++++++++ 2 files changed, 38 insertions(+), 5 deletions(-) diff --git a/paimon-python/pypaimon/branch/filesystem_branch_manager.py b/paimon-python/pypaimon/branch/filesystem_branch_manager.py index ddd83a0f65a1..1f3b84e803db 100644 --- a/paimon-python/pypaimon/branch/filesystem_branch_manager.py +++ b/paimon-python/pypaimon/branch/filesystem_branch_manager.py @@ -21,6 +21,7 @@ from pypaimon.branch.branch_manager import BranchManager, BRANCH_PREFIX from pypaimon.common.file_io import FileIO +from pypaimon.common.json_util import JSON from pypaimon.schema.schema_manager import SchemaManager from pypaimon.snapshot.snapshot_manager import SnapshotManager from pypaimon.tag.tag_manager import TagManager @@ -120,6 +121,8 @@ def create_branch( """ Create a branch from the current state or from a tag. + A retained tag can be used even after its snapshot file expires. + Args: branch_name: Name of the branch to create tag_name: Optional tag name to create branch from, None for current state @@ -160,11 +163,13 @@ def create_branch( branch_tag_manager.tag_path(tag_name), overwrite=True ) - self.file_io.copy_file( - self.snapshot_manager.get_snapshot_path(snapshot.id), - branch_snapshot_manager.get_snapshot_path(snapshot.id), - overwrite=True - ) + snapshot_path = self.snapshot_manager.get_snapshot_path(snapshot.id) + branch_snapshot_path = branch_snapshot_manager.get_snapshot_path(snapshot.id) + if self.file_io.exists(snapshot_path): + self.file_io.copy_file(snapshot_path, branch_snapshot_path, overwrite=True) + else: + # The tag retains the snapshot metadata after the original file expires. + self.file_io.write_file(branch_snapshot_path, JSON.to_json(snapshot), overwrite=True) self._copy_schemas_to_branch(branch_name, snapshot.schema_id) except Exception as e: raise RuntimeError( diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py index e8b21523f9e5..3f4f9fbbf625 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py @@ -185,6 +185,34 @@ def test_create_branch_from_existing_tag(self): self.assertTrue(os.path.isdir(branch_root)) self.assertTrue(os.path.isfile("{}/tag/tag-t1".format(branch_root))) + def test_create_branch_from_tag_after_snapshot_expired(self): + table = self.catalog.get_table(self.identifier) + snapshot = table.snapshot_manager().get_latest_snapshot() + table.create_tag("t1", time_retained="1d") + tag = table.tag_manager().get_or_throw("t1") + self._write(table, pa.table({"id": [4], "value": ["d"]})) + + # Simulate snapshot expiration while the tag retains its metadata and data. + snapshot_path = table.snapshot_manager().get_snapshot_path(snapshot.id) + table.file_io.delete(snapshot_path) + expected = {"id": [1, 2, 3], "value": ["a", "b", "c"]} + self.assertEqual( + self._read(table.copy({"scan.tag-name": "t1"})).to_pydict(), expected) + + self.catalog.create_branch(self.identifier, "b1", tag_name="t1") + + branch_identifier = Identifier( + self.identifier.get_database_name(), self.identifier.get_table_name(), branch="b1") + branch = self.catalog.get_table(branch_identifier) + self.assertEqual(branch.snapshot_manager().get_latest_snapshot(), snapshot) + self.assertEqual(branch.tag_manager().get_or_throw("t1"), tag) + self.assertEqual(self._read(branch).to_pydict(), expected) + self.assertFalse(table.file_io.exists(snapshot_path)) + self.assertEqual(table.snapshot_manager().get_latest_snapshot().id, snapshot.id + 1) + self.assertEqual( + self._read(table).sort_by("id").to_pydict(), + {"id": [1, 2, 3, 4], "value": ["a", "b", "c", "d"]}) + # -- list ----------------------------------------------------------------- def test_list_branches_returns_created(self):