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 f5bb59de3b [python] Add tests for Data Evolution BLOB writes on
branches (#9221)
f5bb59de3b is described below
commit f5bb59de3bb1aa1b5c9a6e7c4def4ab75657b642
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Aug 15 20:17:26 2026 +0800
[python] Add tests for Data Evolution BLOB writes on branches (#9221)
---
.../tests/filesystem_catalog_branch_test.py | 58 +++++++++++++++++
.../pypaimon/tests/rest/rest_branch_test.py | 76 +++++++++++++++++++++-
paimon-python/pypaimon/tests/rest/rest_server.py | 49 +++++++++-----
3 files changed, 166 insertions(+), 17 deletions(-)
diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
b/paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py
index 4bf22b85aa..b1fe5cb681 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 @@ class FileSystemCatalogBranchCRUDTest(unittest.TestCase):
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 @@ class FileSystemCatalogBranchCRUDTest(unittest.TestCase):
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 3b9eb80ff2..e3c8951fa9 100644
--- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
+++ b/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 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
# 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 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
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 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
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")
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py
b/paimon-python/pypaimon/tests/rest/rest_server.py
index 92d089b302..612bf7393b 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -548,7 +548,8 @@ class RESTCatalogServer:
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 @@ class RESTCatalogServer:
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 @@ class RESTCatalogServer:
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 @@ class RESTCatalogServer:
)
# 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 @@ class RESTCatalogServer:
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 @@ class RESTCatalogServer:
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 @@ class RESTCatalogServer:
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 @@ class RESTCatalogServer:
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")
