[python] Add tests for Data Evolution BLOB writes on branches

paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py

@@ -73,6 +73,21 @@ def setUp(self):
     def tearDown(self):
         shutil.rmtree(self.temp_dir, ignore_errors=True)
 
+    @staticmethod
+    def _write(table, data):
+        builder = table.new_batch_write_builder()
+        writer = builder.new_write()
+        try:
+            writer.write_arrow(data)
+            builder.new_commit().commit(writer.prepare_commit())
+        finally:
+            writer.close()
+
+    @staticmethod
+    def _read(table):
+        builder = table.new_read_builder()
+        return builder.new_read().to_arrow(builder.new_scan().plan().splits())
+
     # -- create + list --------------------------------------------------------
 
     def test_create_branch_without_from_tag(self):
@@ -97,6 +112,49 @@ def test_alter_table_isolated_to_branch(self):
         self.assertIn(
             "branch_col", self.catalog.get_table(branch_identifier).field_names)
 
+    def test_write_blob_to_data_evolution_branch(self):
+        schema = pa.schema([
+            ("id", pa.int64()),
+            ("payload", pa.large_binary()),
+        ])
+        identifier = Identifier.from_string("default.test_de_blob_branch")
+        self.catalog.create_table(
+            identifier,
+            Schema.from_pyarrow_schema(
+                schema,
+                options={
+                    "data-evolution.enabled": "true",
+                    "row-tracking.enabled": "true",
+                    "blob-field": "payload",
+                },
+            ),
+            False,
+        )
+        main = self.catalog.get_table(identifier)
+        self._write(main, pa.table({
+            "id": [1],
+            "payload": pa.array([b"main"], pa.large_binary()),
+        }, schema=schema))
+        main.create_tag("base")
+        self.catalog.create_branch(identifier, "b1", tag_name="base")
+
+        branch_identifier = Identifier(
+            identifier.get_database_name(), identifier.get_table_name(), branch="b1")
+        branch = self.catalog.get_table(branch_identifier)
+        self._write(branch, pa.table({
+            "id": [2],
+            "payload": pa.array([b"branch"], pa.large_binary()),
+        }, schema=schema))
+
+        self.assertEqual(
+            self._read(main).to_pydict(),
+            {"id": [1], "payload": [b"main"]},
+        )
+        self.assertEqual(
+            self._read(branch).to_pydict(),
+            {"id": [1, 2], "payload": [b"main", b"branch"]},
+        )
+
     def test_create_branch_duplicate_raises(self):
         self.catalog.create_branch(self.identifier, "b1")
         with self.assertRaises(BranchAlreadyExistException) as cm:

paimon-python/pypaimon/tests/rest/rest_branch_test.py

@@ -17,9 +17,13 @@
 
 import unittest
 
+import pyarrow as pa
+
+from pypaimon import Schema
 from pypaimon.catalog.catalog_exception import (BranchAlreadyExistException,
                                                 BranchNotExistException,
-                                                TableNotExistException)
+                                                TableNotExistException,
+                                                TagNotExistException)
 from pypaimon.common.identifier import Identifier
 from pypaimon.tests.rest.rest_base_test import RESTBaseTest
 
@@ -36,6 +40,21 @@ def _identifier(self):
         # snapshot in setUp.
         return Identifier.from_string("default.test_reader_iterator")
 
+    @staticmethod
+    def _write(table, data):
+        builder = table.new_batch_write_builder()
+        writer = builder.new_write()
+        try:
+            writer.write_arrow(data)
+            builder.new_commit().commit(writer.prepare_commit())
+        finally:
+            writer.close()
+
+    @staticmethod
+    def _read(table):
+        builder = table.new_read_builder()
+        return builder.new_read().to_arrow(builder.new_scan().plan().splits())
+
     def test_create_branch_table_not_exist(self):
         with self.assertRaises(TableNotExistException):
             self.rest_catalog.create_branch(
@@ -50,6 +69,14 @@ def test_create_branch_without_from_tag(self):
         identifier = self._identifier()
         self.rest_catalog.create_branch(identifier, "b1")
         self.assertEqual(self.rest_catalog.list_branches(identifier), ["b1"])
+        branch = self.rest_catalog.get_table(Identifier(
+            identifier.get_database_name(), identifier.get_table_name(), branch="b1"))
+        self.assertEqual(self._read(branch).num_rows, 0)
+
+    def test_create_branch_from_missing_tag_raises(self):
+        with self.assertRaises(TagNotExistException):
+            self.rest_catalog.create_branch(
+                self._identifier(), "b1", tag_name="missing")
 
     def test_branch_table_uses_branch_schema_manager(self):
         identifier = self._identifier()
@@ -65,6 +92,53 @@ def test_branch_table_uses_branch_schema_manager(self):
         self.assertEqual(table.current_branch(), "b1")
         self.assertEqual(table.schema_manager.branch, "b1")
 
+    def test_write_blob_to_data_evolution_branch(self):
+        schema = pa.schema([
+            ("id", pa.int64()),
+            ("payload", pa.large_binary()),
+        ])
+        identifier = Identifier.from_string("default.test_de_blob_branch")
+        self.rest_catalog.create_table(
+            identifier,
+            Schema.from_pyarrow_schema(
+                schema,
+                options={
+                    "data-evolution.enabled": "true",
+                    "row-tracking.enabled": "true",
+                    "blob-field": "payload",
+                },
+            ),
+            False,
+        )
+        main = self.rest_catalog.get_table(identifier)
+        self._write(main, pa.table({
+            "id": [1],
+            "payload": pa.array([b"main"], pa.large_binary()),
+        }, schema=schema))
+        self.assertEqual(
+            self._read(main).to_pydict(),
+            {"id": [1], "payload": [b"main"]},
+        )
+        self.rest_catalog.create_tag(identifier, "base")
+        self.rest_catalog.create_branch(identifier, "b1", tag_name="base")
+
+        branch_identifier = Identifier(
+            identifier.get_database_name(), identifier.get_table_name(), branch="b1")
+        branch = self.rest_catalog.get_table(branch_identifier)
+        self._write(branch, pa.table({
+            "id": [2],
+            "payload": pa.array([b"branch"], pa.large_binary()),
+        }, schema=schema))
+
+        self.assertEqual(
+            self._read(main).to_pydict(),
+            {"id": [1], "payload": [b"main"]},
+        )
+        self.assertEqual(
+            self._read(branch).to_pydict(),
+            {"id": [1, 2], "payload": [b"main", b"branch"]},
+        )
+
     def test_create_branch_duplicate_raises(self):
         identifier = self._identifier()
         self.rest_catalog.create_branch(identifier, "b1")

paimon-python/pypaimon/tests/rest/rest_server.py

@@ -548,7 +548,8 @@ def _handle_table_resource(self, method: str, path_parts: List[str],
             elif operation == "rollback":
                 return self._table_rollback_handle(method, data, lookup_identifier)
             elif operation == "snapshot":
-                return self._table_snapshot_handle(method, lookup_identifier)
+                return self._table_snapshot_handle(
+                    method, lookup_identifier, branch_part)
             elif operation == ResourcePaths.PARTITIONS:
                 return self._table_partitions_handle(method, data, lookup_identifier, parameters)
             elif operation == ResourcePaths.TAGS:
@@ -882,14 +883,18 @@ def _branches_handle(self, method: str, data: str,
 
         if method == "POST":
             request = JSON.from_json(data, CreateBranchRequest)
-            # Mock simplification: ``from_tag`` existence is NOT validated here.
-            # The real Java REST server checks against TagManager and returns
-            # 404+TAG when the tag is missing. pypaimon's mock doesn't track
-            # tag-to-branch dependencies; a TODO for full validation lives
-            # with the Tag CRUD work in #7746.
             store = self.branch_store.setdefault(identifier.get_full_name(), set())
             if request.branch in store:
                 raise BranchAlreadyExistException(request.branch)
+
+            if request.from_tag is not None:
+                tags = self.tag_store.get(identifier.get_full_name(), {})
+                if request.from_tag not in tags:
+                    raise TagNotExistException(request.from_tag)
+                snapshot = tags[request.from_tag].snapshot
+                if snapshot is not None:
+                    self._write_snapshot_files(
+                        identifier, snapshot, None, request.branch)
             store.add(request.branch)
             return self._mock_response("", 200)
 
@@ -1115,7 +1120,7 @@ def _table_commit_handle(self, method: str, data: str, identifier: Identifier,
                     ErrorResponse("SNAPSHOT", None, "Snapshot is required for commit operation", 400), 400
                 )
 
-            table = self._get_file_table(identifier)
+            table = self._get_file_table(identifier, branch)
             current_snapshot = table.snapshot_manager().get_latest_snapshot()
             current_snapshot_uuid = (
                 current_snapshot.uuid if current_snapshot else None
@@ -1126,7 +1131,9 @@ def _table_commit_handle(self, method: str, data: str, identifier: Identifier,
                 )
 
             # Write snapshot to file system
-            self._write_snapshot_files(identifier, commit_request.snapshot, commit_request.statistics)
+            self._write_snapshot_files(
+                identifier, commit_request.snapshot,
+                commit_request.statistics, branch)
 
             self.logger.info(f"Successfully committed snapshot for table {identifier.get_full_name()}, "
                              f"branch: {branch or 'main'}")
@@ -1220,7 +1227,8 @@ def _rollback_table_by_tag(self, identifier: Identifier,
         table.rollback_to(tag_name)
         return self._mock_response("", 200)
 
-    def _table_snapshot_handle(self, method: str, identifier: Identifier) -> Tuple[str, int]:
+    def _table_snapshot_handle(self, method: str, identifier: Identifier,
+                               branch: str = None) -> Tuple[str, int]:
         """Handle table snapshot operations.
 
         Args:
@@ -1246,7 +1254,7 @@ def _table_snapshot_handle(self, method: str, identifier: Identifier) -> Tuple[s
             return self._mock_response(response, 404)
 
         # Get the table and snapshot manager to retrieve snapshot
-        table = self._get_file_table(identifier)
+        table = self._get_file_table(identifier, branch)
         snapshot_manager = table.snapshot_manager()
 
         # Get latest snapshot
@@ -1273,7 +1281,7 @@ def _table_snapshot_handle(self, method: str, identifier: Identifier) -> Tuple[s
         response = GetTableSnapshotResponse(table_snapshot)
         return self._mock_response(response, 200)
 
-    def _get_file_table(self, identifier: Identifier):
+    def _get_file_table(self, identifier: Identifier, branch: str = None):
         """Construct a FileStoreTable from the metadata store.
 
         loads the schema from the metadata store, builds a CatalogEnvironment
@@ -1294,23 +1302,32 @@ def _get_file_table(self, identifier: Identifier):
             f'file://{self.data_path}/{self.warehouse}/'
             f'{identifier.get_database_name()}/{identifier.get_object_name()}')
 
+        table_identifier = Identifier.create(
+            identifier.get_database_name(), identifier.get_table_name(), branch=branch)
         catalog_env = CatalogEnvironment(
-            identifier=identifier,
+            identifier=table_identifier,
             uuid=table_metadata.uuid,
             catalog_loader=None,
             supports_version_management=False
         )
 
         file_io = FileIO.get(table_path, Options({}))
-        return FileStoreTable(file_io, identifier, table_path, table_schema, catalog_env)
+        return FileStoreTable(
+            file_io, table_identifier, table_path, table_schema, catalog_env)
 
-    def _write_snapshot_files(self, identifier: Identifier, snapshot, statistics):
+    def _write_snapshot_files(self, identifier: Identifier, snapshot, statistics,
+                              branch: str = None):
         """Write snapshot and related files to the file system"""
         import os
 
         # Construct table path: {warehouse}/{database}/{table}
-        table_path = os.path.join(self.data_path, self.warehouse, identifier.get_database_name(),
-                                  identifier.get_object_name())
+        from pypaimon.branch.branch_manager import BranchManager
+
+        table_path = os.path.join(
+            self.data_path, self.warehouse, identifier.get_database_name(),
+            identifier.get_object_name())
+        table_path = BranchManager.branch_path(
+            table_path, BranchManager.normalize_branch(branch))
 
         # Create directory structure
         snapshot_dir = os.path.join(table_path, "snapshot")