From 28c529f445dac5445e0bae80f4f6cd2856dc558c Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 14 Aug 2026 03:21:58 -0700 Subject: [PATCH 1/3] [python] Test Data Evolution BLOB writes on branches --- .../tests/filesystem_catalog_branch_test.py | 58 +++++++++++++++++ .../pypaimon/tests/rest/rest_branch_test.py | 64 +++++++++++++++++++ .../pypaimon/tests/rest/rest_server.py | 35 ++++++---- 3 files changed, 146 insertions(+), 11 deletions(-) diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py index 4bf22b85aac2..b1fe5cb6813b 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py +++ b/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: diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py b/paimon-python/pypaimon/tests/rest/rest_branch_test.py index 3b9eb80ff229..b01f37f13cce 100644 --- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py @@ -17,6 +17,9 @@ import unittest +import pyarrow as pa + +from pypaimon import Schema from pypaimon.catalog.catalog_exception import (BranchAlreadyExistException, BranchNotExistException, TableNotExistException) @@ -36,6 +39,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( @@ -65,6 +83,52 @@ 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_branch(identifier, "b1") + + 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": [2], "payload": [b"branch"]}, + ) + def test_create_branch_duplicate_raises(self): identifier = self._identifier() self.rest_catalog.create_branch(identifier, "b1") diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py b/paimon-python/pypaimon/tests/rest/rest_server.py index 92d089b3029a..3ccb09d34615 100755 --- a/paimon-python/pypaimon/tests/rest/rest_server.py +++ b/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: @@ -1115,7 +1116,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 +1127,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 +1223,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 +1250,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 +1277,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 +1298,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") From 5ea649a6df5ebba8eb7c0af685ce53e2d6d2149a Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 14 Aug 2026 19:19:53 -0700 Subject: [PATCH 2/3] [python] Make mock REST branches inherit main snapshot --- .../pypaimon/tests/rest/rest_branch_test.py | 2 +- paimon-python/pypaimon/tests/rest/rest_server.py | 15 ++++++++++----- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py b/paimon-python/pypaimon/tests/rest/rest_branch_test.py index b01f37f13cce..a296abf70b6a 100644 --- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py @@ -126,7 +126,7 @@ def test_write_blob_to_data_evolution_branch(self): ) self.assertEqual( self._read(branch).to_pydict(), - {"id": [2], "payload": [b"branch"]}, + {"id": [1, 2], "payload": [b"main", b"branch"]}, ) def test_create_branch_duplicate_raises(self): diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py b/paimon-python/pypaimon/tests/rest/rest_server.py index 3ccb09d34615..d6941c38b853 100755 --- a/paimon-python/pypaimon/tests/rest/rest_server.py +++ b/paimon-python/pypaimon/tests/rest/rest_server.py @@ -883,14 +883,19 @@ 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 None: + snapshot = ( + self._get_file_table(identifier) + .snapshot_manager() + .get_latest_snapshot() + ) + if snapshot is not None: + self._write_snapshot_files( + identifier, snapshot, None, request.branch) store.add(request.branch) return self._mock_response("", 200) From 0cfd0339665b669ec91d5aea50d3868904ee8f42 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 14 Aug 2026 20:02:59 -0700 Subject: [PATCH 3/3] [python] Fix mock REST branch inheritance semantics --- .../pypaimon/tests/rest/rest_branch_test.py | 14 ++++++++++++-- paimon-python/pypaimon/tests/rest/rest_server.py | 11 +++++------ 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py b/paimon-python/pypaimon/tests/rest/rest_branch_test.py index a296abf70b6a..e3c8951fa97d 100644 --- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py @@ -22,7 +22,8 @@ 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 @@ -68,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() @@ -110,7 +119,8 @@ def test_write_blob_to_data_evolution_branch(self): self._read(main).to_pydict(), {"id": [1], "payload": [b"main"]}, ) - self.rest_catalog.create_branch(identifier, "b1") + 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") diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py b/paimon-python/pypaimon/tests/rest/rest_server.py index d6941c38b853..612bf7393b57 100755 --- a/paimon-python/pypaimon/tests/rest/rest_server.py +++ b/paimon-python/pypaimon/tests/rest/rest_server.py @@ -887,12 +887,11 @@ def _branches_handle(self, method: str, data: str, if request.branch in store: raise BranchAlreadyExistException(request.branch) - if request.from_tag is None: - snapshot = ( - self._get_file_table(identifier) - .snapshot_manager() - .get_latest_snapshot() - ) + 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)