[python] Fix branch creation from tags after snapshot expiration

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(

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):