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 76cc48587b [python] Use Native REST catalog branch operations (#10619)
76cc48587b is described below
commit 76cc48587b29d6294a38a4f86877cb23587c02ef
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 17 11:32:32 2026 +0800
[python] Use Native REST catalog branch operations (#10619)
---
.github/workflows/ci-python.yml | 5 +-
paimon-python/README.md | 22 +
paimon-python/pypaimon/api/api_request.py | 7 -
paimon-python/pypaimon/api/resource_paths.py | 5 -
paimon-python/pypaimon/api/rest_api.py | 41 +-
paimon-python/pypaimon/branch/branch_manager.py | 4 +-
.../pypaimon/branch/catalog_branch_manager.py | 9 +-
.../pypaimon/catalog/rest/native_branch.py | 137 ++++++
.../pypaimon/catalog/rest/rest_catalog.py | 27 +-
.../pypaimon/tests/api/test_branch_dto_serde.py | 19 +-
.../tests/branch/catalog_branch_manager_test.py | 5 +
.../pypaimon/tests/native_branch_write_test.py | 16 +-
.../pypaimon/tests/native_rest_branch_test.py | 543 +++++++++++++++++++++
.../pypaimon/tests/rest/dlf_signer_test.py | 2 +-
.../pypaimon/tests/rest/rest_branch_test.py | 37 +-
paimon-python/pypaimon/tests/rest/rest_server.py | 25 -
16 files changed, 776 insertions(+), 128 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index b6a0fb9f81..00778ca3fc 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -231,8 +231,11 @@ jobs:
normalize_array_for_write,
read_arrow_schema,
)
- from pypaimon_rust.datafusion import ReadBuilder
+ from pypaimon_rust.datafusion import CatalogError, PaimonCatalog,
ReadBuilder
assert callable(getattr(ReadBuilder, 'with_partition_filter', None))
+ assert issubclass(CatalogError, ValueError)
+ for method in ('create_branch', 'drop_branch', 'fast_forward',
'list_branches'):
+ assert callable(getattr(PaimonCatalog, method, None))
PY
- name: Run Python tests with native planning, reading, writing,
updating and committing
diff --git a/paimon-python/README.md b/paimon-python/README.md
index ccb3cb5c18..0c88ad530e 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -139,6 +139,28 @@ table = table.copy({"parquet.filter.columnindex.enabled":
"false"})
Unsupported reads use the normal path. Reading fewer bytes may require more
object-store requests.
+# Native REST branch management
+
+REST catalogs delegate `list_branches` to Rust when the catalog option
+`read.native.enabled=true`, and `create_branch`, `drop_branch` and
`fast_forward`
+when `write.native.enabled=true`. Existing `table.branch_manager()` calls use
+the same route. Install the optional `pypaimon-rust` runtime to enable it.
+
+The bridge reuses the catalog's resolved prefix, endpoint, authentication and
+request headers, and preserves the public Python exception types. Custom
clients,
+authentication providers or error handlers use Python. Once a Native operation
+starts, failures propagate without replaying the request through Python.
+
+OpenAPI V4 (`dlf.signing-algorithm=openapi-v4`) is signed in Rust core,
including
+the action header. Session headers are merged case-insensitively, preserving an
+explicit User-Agent. Sessions with Cookie, custom authentication or TLS
settings
+use Python to retain their transport state. Native listing retries HTTP
429/503;
+mutations are sent once and accept successful responses without a JSON body.
+
+As in Java, creation without a tag produces an empty branch; creation from a
tag
+uses its snapshot. REST branch renaming is unsupported. Dropping branch
metadata
+does not remove data files shared with another branch or tag.
+
# Native write and commit
PyPaimon can write Arrow batches through the optional `pypaimon-rust` runtime.
diff --git a/paimon-python/pypaimon/api/api_request.py
b/paimon-python/pypaimon/api/api_request.py
index 7f06c789a3..c03b75319f 100644
--- a/paimon-python/pypaimon/api/api_request.py
+++ b/paimon-python/pypaimon/api/api_request.py
@@ -212,13 +212,6 @@ class CreateBranchRequest(RESTRequest):
from_tag: Optional[str] = json_field(FIELD_FROM_TAG, default=None)
-@dataclass
-class RenameBranchRequest(RESTRequest):
- FIELD_TO_BRANCH = "toBranch"
-
- to_branch: str = json_field(FIELD_TO_BRANCH)
-
-
@dataclass
class ForwardBranchRequest(RESTRequest):
"""Empty body request; serializes to ``{}`` per Java
ForwardBranchRequest."""
diff --git a/paimon-python/pypaimon/api/resource_paths.py
b/paimon-python/pypaimon/api/resource_paths.py
index 309d829d59..ac5e5dca95 100644
--- a/paimon-python/pypaimon/api/resource_paths.py
+++ b/paimon-python/pypaimon/api/resource_paths.py
@@ -33,7 +33,6 @@ class ResourcePaths:
FUNCTION_DETAILS = "function-details"
TAGS = "tags"
BRANCHES = "branches"
- RENAME = "rename"
FORWARD = "forward"
PERMISSIONS = "permissions"
POLICIES = "policies"
@@ -135,10 +134,6 @@ class ResourcePaths:
self.TABLES, RESTUtil.encode_string(table_name), self.BRANCHES,
RESTUtil.encode_string(branch_name))
- def rename_branch(self, database_name: str, table_name: str, branch_name:
str) -> str:
- return "{}/{}".format(
- self.branch(database_name, table_name, branch_name), self.RENAME)
-
def forward_branch(self, database_name: str, table_name: str, branch_name:
str) -> str:
return "{}/{}".format(
self.branch(database_name, table_name, branch_name), self.FORWARD)
diff --git a/paimon-python/pypaimon/api/rest_api.py
b/paimon-python/pypaimon/api/rest_api.py
index d441ea5cbd..b8148a1dfe 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -27,7 +27,7 @@ from pypaimon.api.api_request import (AlterDatabaseRequest,
AlterFunctionRequest
CreateTableRequest, CreateTagRequest,
DropPolicyRequest, ForwardBranchRequest,
GrantPermissionRequest, PolicyRequest,
- RenameBranchRequest, RenameTableRequest,
+ RenameTableRequest,
RevokePermissionRequest,
RollbackTableRequest)
from pypaimon.api.api_response import (CommitTableResponse, ConfigResponse,
@@ -128,6 +128,10 @@ class RESTApi:
self.options = options
self.resource_paths = ResourcePaths.for_catalog_properties(options)
+ @staticmethod
+ def _is_auth_option(key):
+ return key in ('uri', 'token.provider', 'token') or
key.startswith('dlf.')
+
def __build_paged_query_params(
self,
max_results: Optional[int],
@@ -545,13 +549,24 @@ class RESTApi:
)
# Branch CRUD wrappers — mirror Java RESTApi branch methods.
+ @staticmethod
+ def _validate_branch_path(identifier, branch_name=None):
+ database_name, table_name = RESTApi.__validate_identifier(identifier)
+ components = [(database_name, 'Database name'), (table_name, 'Table
name')]
+ if branch_name is not None:
+ components.append((branch_name, 'Branch name'))
+ for value, label in components:
+ if not value or not value.strip() or value in ('.', '..'):
+ raise ValueError('{} must be a non-empty resource path
component'.format(label))
+ return database_name, table_name
+
def create_branch(
self,
identifier: Identifier,
branch_name: str,
tag_name: Optional[str] = None,
) -> None:
- database_name, table_name = self.__validate_identifier(identifier)
+ database_name, table_name = self._validate_branch_path(identifier)
if not branch_name or not branch_name.strip():
raise ValueError("Branch name cannot be empty")
request = CreateBranchRequest(branch=branch_name, from_tag=tag_name)
@@ -562,30 +577,14 @@ class RESTApi:
)
def drop_branch(self, identifier: Identifier, branch_name: str) -> None:
- database_name, table_name = self.__validate_identifier(identifier)
+ database_name, table_name = self._validate_branch_path(identifier,
branch_name)
self.client.delete(
self.resource_paths.branch(database_name, table_name, branch_name),
self.rest_auth_function,
)
- def rename_branch(
- self,
- identifier: Identifier,
- from_branch: str,
- to_branch: str,
- ) -> None:
- database_name, table_name = self.__validate_identifier(identifier)
- if not to_branch or not to_branch.strip():
- raise ValueError("Target branch name cannot be empty")
- request = RenameBranchRequest(to_branch=to_branch)
- self.client.post(
- self.resource_paths.rename_branch(database_name, table_name,
from_branch),
- request,
- self.rest_auth_function,
- )
-
def fast_forward(self, identifier: Identifier, branch_name: str) -> None:
- database_name, table_name = self.__validate_identifier(identifier)
+ database_name, table_name = self._validate_branch_path(identifier,
branch_name)
self.client.post(
self.resource_paths.forward_branch(database_name, table_name,
branch_name),
ForwardBranchRequest(),
@@ -593,7 +592,7 @@ class RESTApi:
)
def list_branches(self, identifier: Identifier) -> List[str]:
- database_name, table_name = self.__validate_identifier(identifier)
+ database_name, table_name = self._validate_branch_path(identifier)
response = self.client.get(
self.resource_paths.branches(database_name, table_name),
ListBranchesResponse,
diff --git a/paimon-python/pypaimon/branch/branch_manager.py
b/paimon-python/pypaimon/branch/branch_manager.py
index ae0d160f17..8e9e249aeb 100644
--- a/paimon-python/pypaimon/branch/branch_manager.py
+++ b/paimon-python/pypaimon/branch/branch_manager.py
@@ -40,11 +40,11 @@ class BranchManager:
ignore_if_exists: bool = False
) -> None:
"""
- Create a branch from the current state or from a tag.
+ Create an empty branch or initialize it from a tag.
Args:
branch_name: Name of the branch to create
- tag_name: Optional tag name to create branch from, None for
current state
+ tag_name: Optional tag name to create branch from, None for an
empty branch
ignore_if_exists: If true, do nothing when branch already exists;
if false, throw exception
diff --git a/paimon-python/pypaimon/branch/catalog_branch_manager.py
b/paimon-python/pypaimon/branch/catalog_branch_manager.py
index 3c3009c88c..5cad0ab520 100644
--- a/paimon-python/pypaimon/branch/catalog_branch_manager.py
+++ b/paimon-python/pypaimon/branch/catalog_branch_manager.py
@@ -85,11 +85,11 @@ class CatalogBranchManager(BranchManager):
ignore_if_exists: bool = False
) -> None:
"""
- Create a branch from the current state or from a tag.
+ Create an empty branch or initialize it from a tag.
Args:
branch_name: Name of the branch to create
- tag_name: Optional tag name to create branch from, None for
current state
+ tag_name: Optional tag name to create branch from, None for an
empty branch
ignore_if_exists: If true, do nothing when branch already exists
Raises:
@@ -97,10 +97,7 @@ class CatalogBranchManager(BranchManager):
"""
def _create(catalog: Catalog):
BranchManager.validate_branch(branch_name)
- existing_branches = catalog.list_branches(self.identifier)
- if branch_name in existing_branches:
- if not ignore_if_exists:
- raise ValueError(f"Branch name '{branch_name}' already
exists.")
+ if ignore_if_exists and branch_name in
catalog.list_branches(self.identifier):
return
catalog.create_branch(self.identifier, branch_name, tag_name)
diff --git a/paimon-python/pypaimon/catalog/rest/native_branch.py
b/paimon-python/pypaimon/catalog/rest/native_branch.py
new file mode 100644
index 0000000000..4d46b19f18
--- /dev/null
+++ b/paimon-python/pypaimon/catalog/rest/native_branch.py
@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Transport REST branch operations and structured errors to/from Rust core."""
+
+import logging
+import os
+
+import requests
+from requests.structures import CaseInsensitiveDict
+
+from pypaimon.api.api_response import ErrorResponse
+from pypaimon.api.auth import RESTAuthFunction
+from pypaimon.api.auth.bearer import BearTokenAuthProvider
+from pypaimon.api.auth.dlf_provider import DLFAuthProvider
+from pypaimon.api.client import DefaultErrorHandler, HttpClient
+from pypaimon.api.rest_api import RESTApi
+from pypaimon.catalog.catalog_context import CatalogContext
+from pypaimon.common.options.core_options import CoreOptions
+
+logger = logging.getLogger(__name__)
+
+
+def _native_headers(api):
+ session = api.client.session
+ if (type(session) is not requests.Session or session.auth is not None
+ or session.cert is not None or session.verify is not True or
session.proxies or session.params
+ or not session.trust_env or any(session.hooks.values())
+ or set(session.adapters) != {'http://', 'https://'}
+ or any(type(adapter) is not requests.adapters.HTTPAdapter for
adapter in session.adapters.values())
+ or api.client._timeout != (180, 180)
+ or os.environ.get('REQUESTS_CA_BUNDLE') or
os.environ.get('CURL_CA_BUNDLE')):
+ # Keep custom transport credentials and policies in their owning
client.
+ return None
+ headers = CaseInsensitiveDict(session.headers)
+ # Let each HTTP transport advertise only the codecs it can decode and
manage
+ # its own connections. An explicitly customized transport header stays
Python.
+ for name in ('Accept-Encoding', 'Connection'):
+ if (headers.get(name) != requests.utils.default_headers().get(name)
+ or any(key.lower() == name.lower() for key in
api.rest_auth_function.init_header)):
+ return None
+ headers.pop(name, None)
+ headers.update(api.rest_auth_function.init_header)
+ return {key: value for key, value in headers.items() if value is not None}
+
+
+def _native_catalog(catalog, operation):
+ from pypaimon.catalog.rest.rest_catalog import RESTCatalog
+
+ if (type(catalog) is not RESTCatalog or type(catalog.context) is not
CatalogContext
+ or type(catalog.rest_api) is not RESTApi
+ or any(getattr(catalog.context, attr, None) is not None for attr
in (
+ 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader'))):
+ return None
+ options = CoreOptions(catalog.context.options)
+ enabled = options.native_read_enabled() if operation == 'list_branches'
else options.native_write_enabled()
+ if not enabled:
+ return None
+ api = catalog.rest_api
+ if (type(api.client) is not HttpClient
+ or type(api.client.error_handler) is not DefaultErrorHandler
+ or type(api.rest_auth_function) is not RESTAuthFunction
+ or type(api.rest_auth_function.auth_provider) not in
(BearTokenAuthProvider, DLFAuthProvider)):
+ return None
+ headers = _native_headers(api)
+ if headers is None:
+ return None
+ try:
+ from pypaimon_rust.datafusion import CatalogError, PaimonCatalog
+ except ImportError:
+ return None
+ from pypaimon.read.native_plan import _option_value_to_string,
_raise_if_native_fork_safety_error
+ resolved = {key: value for key, value in api.options.to_map().items()
+ if not RESTApi._is_auth_option(key) and not
key.startswith(RESTApi.HEADER_PREFIX)}
+ resolved['uri'] = api.client.uri
+ resolved.update({RESTApi.HEADER_PREFIX + key: value for key, value in
headers.items()})
+ resolved = {str(key): _option_value_to_string(value)
+ for key, value in resolved.items() if value is not None}
+ resolved['metastore'] = 'rest'
+ try:
+ provider = api.rest_auth_function.auth_provider
+ if type(provider) is DLFAuthProvider:
+ # The owning provider retains Java's STS cache/refresh policy. Pass
+ # its current credentials for this call; Rust still signs the
request.
+ token = provider.get_token()
+ resolved.update({
+ 'token.provider': 'dlf', 'dlf.region': provider.region,
+ 'dlf.signing-algorithm': provider.signing_algorithm,
+ 'dlf.access-key-id': token.access_key_id,
+ 'dlf.access-key-secret': token.access_key_secret,
+ })
+ if token.security_token is not None:
+ resolved['dlf.security-token'] = token.security_token
+ else:
+ resolved.update({'token.provider': 'bear', 'token':
provider.token})
+ # Reuse the resolved configuration, as RESTCatalogLoader does. Keep no
+ # opaque native state on the Python catalog across pickle or process
fork.
+ return PaimonCatalog(resolved, config_required=False), CatalogError
+ except Exception as error:
+ _raise_if_native_fork_safety_error(error)
+ logger.debug('Native branch preparation failed: %s', error)
+ return None
+
+
+def call_branch(catalog, operation, identifier, *args):
+ # Both backends must reject a path that HTTP clients normalize to another
+ # resource. CREATE's branch is in JSON; only DROP/FORWARD put it in the
URL.
+ path_branch = args[0] if operation in ('drop_branch', 'fast_forward') else
None
+ RESTApi._validate_branch_path(identifier, path_branch)
+ if operation == 'create_branch' and (not args[0] or not args[0].strip()):
+ raise ValueError('Branch name cannot be empty')
+ prepared = _native_catalog(catalog, operation)
+ if prepared is None:
+ return getattr(catalog.rest_api, operation)(identifier, *args)
+ native, error_type = prepared
+ try:
+ # Once an operation starts, never execute it again through Python:
+ # the REST server may have published it before returning an error.
+ return getattr(native, operation)((identifier.get_database_name(),
identifier.get_object_name()), *args)
+ except error_type as error:
+ DefaultErrorHandler.get_instance().accept(ErrorResponse(
+ resource_type=error.resource_type,
resource_name=error.resource_name,
+ message=error.message, code=error.code),
HttpClient.DEFAULT_REQUEST_ID)
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 2e0d5a046e..6ae9e6fd24 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -648,14 +648,17 @@ class RESTCatalog(Catalog):
# Branch CRUD: mirrors Java RESTCatalog branch handlers.
def create_branch(self, identifier: Union[str, Identifier], branch_name:
str,
tag_name: Optional[str] = None) -> None:
+ from pypaimon.catalog.rest.native_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- self.rest_api.create_branch(identifier, branch_name, tag_name)
+ call_branch(self, 'create_branch', identifier, branch_name,
tag_name)
except NoSuchResourceException as e:
if e.resource_type == ErrorResponse.RESOURCE_TYPE_TAG:
raise TagNotExistException(tag_name) from e
- raise TableNotExistException(identifier) from e
+ if e.resource_type == ErrorResponse.RESOURCE_TYPE_TABLE:
+ raise TableNotExistException(identifier) from e
+ raise
except AlreadyExistsException as e:
raise BranchAlreadyExistException(branch_name) from e
except ForbiddenException as e:
@@ -664,10 +667,11 @@ class RESTCatalog(Catalog):
raise IllegalArgumentError(str(e)) from e
def drop_branch(self, identifier: Union[str, Identifier], branch_name:
str) -> None:
+ from pypaimon.catalog.rest.native_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- self.rest_api.drop_branch(identifier, branch_name)
+ call_branch(self, 'drop_branch', identifier, branch_name)
except NoSuchResourceException as e:
raise BranchNotExistException(branch_name) from e
except ForbiddenException as e:
@@ -675,32 +679,25 @@ class RESTCatalog(Catalog):
def rename_branch(self, identifier: Union[str, Identifier], from_branch:
str,
to_branch: str) -> None:
- if not isinstance(identifier, Identifier):
- identifier = Identifier.from_string(identifier)
- try:
- self.rest_api.rename_branch(identifier, from_branch, to_branch)
- except NoSuchResourceException as e:
- raise BranchNotExistException(from_branch) from e
- except AlreadyExistsException as e:
- raise BranchAlreadyExistException(to_branch) from e
- except ForbiddenException as e:
- raise TableNoPermissionException(identifier) from e
+ raise NotImplementedError('REST branch rename is not supported,
matching Java RESTCatalog.')
def fast_forward(self, identifier: Union[str, Identifier], branch_name:
str) -> None:
+ from pypaimon.catalog.rest.native_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- self.rest_api.fast_forward(identifier, branch_name)
+ call_branch(self, 'fast_forward', identifier, branch_name)
except NoSuchResourceException as e:
raise BranchNotExistException(branch_name) from e
except ForbiddenException as e:
raise TableNoPermissionException(identifier) from e
def list_branches(self, identifier: Union[str, Identifier]) -> List[str]:
+ from pypaimon.catalog.rest.native_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- return self.rest_api.list_branches(identifier)
+ return call_branch(self, 'list_branches', identifier)
except NoSuchResourceException as e:
raise TableNotExistException(identifier) from e
except ForbiddenException as e:
diff --git a/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
b/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
index a211c7af69..2b32d0a393 100644
--- a/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
+++ b/paimon-python/pypaimon/tests/api/test_branch_dto_serde.py
@@ -19,7 +19,7 @@
These tests pin the Python wire format to the Java contract so review
against ``paimon-api/.../rest/requests/CreateBranchRequest.java``,
-``RenameBranchRequest.java``, ``ForwardBranchRequest.java``,
+``ForwardBranchRequest.java``,
``responses/ListBranchesResponse.java`` and ``ResourcePaths.java`` does
not need to inspect a running mock server. They explicitly assert the
*exact* JSON field names and URL templates Java uses, and that
@@ -30,8 +30,7 @@ returns plain ``List<String>`` — pypaimon mirrors this).
import json
import unittest
-from pypaimon.api.api_request import (CreateBranchRequest,
ForwardBranchRequest,
- RenameBranchRequest)
+from pypaimon.api.api_request import CreateBranchRequest, ForwardBranchRequest
from pypaimon.api.api_response import ListBranchesResponse
from pypaimon.api.resource_paths import ResourcePaths
from pypaimon.common.json_util import JSON
@@ -56,14 +55,6 @@ class CreateBranchRequestSerdeTest(unittest.TestCase):
self.assertIsNone(parsed.get("fromTag"))
-class RenameBranchRequestSerdeTest(unittest.TestCase):
-
- def test_uses_to_branch_field(self):
- request = RenameBranchRequest(to_branch="b2")
- parsed = json.loads(JSON.to_json(request))
- self.assertEqual(parsed, {"toBranch": "b2"})
-
-
class ForwardBranchRequestSerdeTest(unittest.TestCase):
def test_serializes_empty_object(self):
@@ -110,12 +101,6 @@ class ResourcePathsBranchesTest(unittest.TestCase):
"/v1/mock/databases/db/tables/tbl/branches/b1",
)
- def test_rename_branch_url(self):
- self.assertEqual(
- self.paths.rename_branch("db", "tbl", "b1"),
- "/v1/mock/databases/db/tables/tbl/branches/b1/rename",
- )
-
def test_forward_branch_url(self):
self.assertEqual(
self.paths.forward_branch("db", "tbl", "b1"),
diff --git a/paimon-python/pypaimon/tests/branch/catalog_branch_manager_test.py
b/paimon-python/pypaimon/tests/branch/catalog_branch_manager_test.py
index dda8a85fc4..51338d3a63 100644
--- a/paimon-python/pypaimon/tests/branch/catalog_branch_manager_test.py
+++ b/paimon-python/pypaimon/tests/branch/catalog_branch_manager_test.py
@@ -54,6 +54,11 @@ class TestCatalogBranchManager(unittest.TestCase):
self.identifier, "test-branch", "tag-1"
)
+ def test_strict_create_does_not_require_list_permission(self):
+ self.catalog.list_branches.side_effect = AssertionError('Strict create
must not list branches')
+ self.branch_manager.create_branch('new-branch')
+ self.catalog.create_branch.assert_called_once_with(self.identifier,
'new-branch', None)
+
def test_create_branch_ignore_if_exists(self):
"""Test creating branch with ignore_if_exists flag."""
self.catalog.list_branches.return_value = ["existing-branch"]
diff --git a/paimon-python/pypaimon/tests/native_branch_write_test.py
b/paimon-python/pypaimon/tests/native_branch_write_test.py
index 38de93fd4e..da0945ab13 100644
--- a/paimon-python/pypaimon/tests/native_branch_write_test.py
+++ b/paimon-python/pypaimon/tests/native_branch_write_test.py
@@ -410,24 +410,26 @@ def
test_native_branch_lifecycle_preserves_histories(tmp_path, native_rest_catal
_write(main, [seed])
branch = _branch(catalog, main)
_write(branch, [added])
- catalog.rename_branch(main.identifier, 'dev', 'renamed')
+ name = 'dev' if rest else 'renamed'
+ if not rest:
+ catalog.rename_branch(main.identifier, 'dev', name)
catalog.rename_table(main.identifier, Identifier('default', 'renamed'))
main = catalog.get_table('default.renamed')
- branch = catalog.get_table(Identifier('default', 'renamed',
branch='renamed'))
+ branch = catalog.get_table(Identifier('default', 'renamed', branch=name))
_assert_rows(main, [seed])
_assert_rows(branch, [seed, added])
- assert catalog.list_branches(main.identifier) == ['renamed']
+ assert catalog.list_branches(main.identifier) == [name]
assert catalog.get_tag(main.identifier, 'base').snapshot.id == 1
with pytest.raises(TableNotExistException):
catalog.get_table('default.t')
third = {'id': 3, 'value': 30, 'p': 'b'}
_write(branch, [third])
_assert_rows(branch, [seed, added, third])
- catalog.drop_branch(main.identifier, 'renamed')
+ catalog.drop_branch(main.identifier, name)
assert catalog.list_branches(main.identifier) == []
_assert_rows(main, [seed])
- catalog.create_branch(main.identifier, 'renamed')
- empty = catalog.get_table(Identifier('default', 'renamed',
branch='renamed'))
+ catalog.create_branch(main.identifier, name)
+ empty = catalog.get_table(Identifier('default', 'renamed', branch=name))
_assert_rows(empty, [])
_write(empty, [added])
_assert_rows(empty, [added])
@@ -449,7 +451,7 @@ def
test_native_rest_branch_invalid_names_preserve_main(tmp_path, native_rest_ca
operation()
_assert_rows(main, [seed])
branch = _branch(catalog, main)
- with pytest.raises((BadRequestException, IllegalArgumentError,
ValueError)):
+ with pytest.raises(NotImplementedError):
catalog.rename_branch(main.identifier, 'dev', name)
_assert_rows(branch, [seed])
_assert_rows(main, [seed])
diff --git a/paimon-python/pypaimon/tests/native_rest_branch_test.py
b/paimon-python/pypaimon/tests/native_rest_branch_test.py
new file mode 100644
index 0000000000..8782030d93
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_rest_branch_test.py
@@ -0,0 +1,543 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""REST branch operations retain PyPaimon's contracts when Rust executes
them."""
+
+from contextlib import ExitStack, contextmanager
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import json
+import threading
+from unittest.mock import patch
+from urllib.parse import parse_qsl, urlsplit
+
+import pyarrow as pa
+import pytest
+import requests
+
+from pypaimon import Schema
+from pypaimon.catalog.catalog_context import CatalogContext
+from pypaimon.catalog.rest.rest_catalog import RESTCatalog
+from pypaimon.api.api_response import ErrorResponse
+from pypaimon.api.auth.dlf_signer import DLFOpenApiV4Signer
+from pypaimon.api.token_loader import DLFToken
+from pypaimon.api.typedef import RESTAuthParameter
+from pypaimon.api.rest_exception import (
+ NoSuchResourceException, NotImplementedException, RESTException,
ServiceFailureException,
+)
+from pypaimon.catalog.catalog_exception import (
+ BranchAlreadyExistException, BranchNotExistException, IllegalArgumentError,
+ TableNotExistException, TableNoPermissionException, TagNotExistException,
+)
+from pypaimon.common.identifier import Identifier
+from pypaimon.tests.native_branch_write_test import _assert_rows, _branch,
_table, _write
+from pypaimon.tests.rest.rest_server import RESTCatalogServer
+
+pytestmark = pytest.mark.native_plan
+_OPERATIONS = ('create_branch', 'drop_branch', 'fast_forward', 'list_branches')
+
+
[email protected]
+def transport_server():
+ class Handler(BaseHTTPRequestHandler):
+ def operation(self):
+ size = int(self.headers.get('Content-Length', 0))
+ data = self.rfile.read(size).decode() if size else None
+ self.server.requests.append((self.command, self.path,
self.headers))
+ status = 200
+ if self.path.startswith('/v1/config'):
+ body = json.dumps({'defaults': {'prefix': 'test'}})
+ elif self.server.require_v4 and not
self.headers.get('Authorization', '').startswith('ACS4-HMAC-SHA256 '):
+ status, body = 403, json.dumps({'message': 'wrong signature
algorithm'})
+ elif self.server.temporary_status and self.server.branch_requests
== 0:
+ status = self.server.temporary_status
+ body = json.dumps({'message': 'temporary failure'})
+ self.server.branch_requests += 1
+ else:
+ body = json.dumps({'branches': ['dev']}) if self.command ==
'GET' else 'OK'
+ if self.server.require_v4:
+ signed_headers = {key.lower(): value for key, value in
self.headers.items()
+ if key.lower().startswith('x-acs-') or
key.lower() in ('host', 'content-type')}
+ url = urlsplit(self.path)
+ expected = DLFOpenApiV4Signer('cn-hangzhou').authorization(
+ RESTAuthParameter(self.command, url.path, data,
dict(parse_qsl(url.query))),
+ self.server.expected_token,
+ self.headers['Host'], signed_headers)
+ if self.headers.get('Authorization') != expected:
+ status, body = 403, json.dumps({'message': 'signature
mismatch'})
+ output = body.encode()
+ self.send_response(status)
+ self.send_header('Content-Length', str(len(output)))
+ self.end_headers()
+ self.wfile.write(output)
+
+ do_GET = do_POST = do_DELETE = operation
+
+ def log_message(self, *args):
+ pass
+
+ server = HTTPServer(('localhost', 0), Handler)
+ server.requests = []
+ server.require_v4 = False
+ server.temporary_status = server.branch_requests = 0
+ server.expected_token = DLFToken('test-key', 'test-secret', 'test-sts',
None)
+ thread = threading.Thread(target=server.serve_forever)
+ thread.start()
+ try:
+ yield server
+ finally:
+ server.shutdown()
+ server.server_close()
+ thread.join()
+
+
+def transport_catalog(server, **extra):
+ options = {
+ 'metastore': 'rest', 'uri': 'http://localhost:%d' %
server.server_port, 'warehouse': 'w',
+ 'token.provider': 'bear', 'token': 'test-token',
+ 'read.native.enabled': 'true', 'write.native.enabled': 'true',
+ }
+ options.update(extra)
+ return RESTCatalog(CatalogContext.create_from_options(options))
+
+
[email protected]('operation', _OPERATIONS)
+def test_openapi_v4_uses_native_signer(transport_server, operation):
+ server = transport_server
+ server.require_v4 = True
+ catalog = transport_catalog(server, **{
+ 'token.provider': 'dlf', 'dlf.region': 'cn-hangzhou',
'dlf.signing-algorithm': 'openapi-v4',
+ 'dlf.access-key-id': 'test-key', 'dlf.access-key-secret':
'test-secret',
+ 'dlf.security-token': 'test-sts',
+ })
+ catalog.rest_api.client.session.headers.update({
+ 'HOST': 'stale.example', 'X-ACS-DATE': 'stale', 'authorization':
'stale',
+ })
+ if operation in ('create_branch', 'fast_forward'):
+ catalog.rest_api.client.session.headers['Content-Type'] = 'text/plain'
+ server.requests.clear()
+ args = () if operation == 'list_branches' else ('dev 一/?#',)
+ if operation == 'create_branch':
+ args += ('源 一',)
+ with native_only(catalog):
+ result = getattr(catalog, operation)(Identifier('db 一/space', 'table
?$branch_dev'), *args)
+ assert result == (['dev'] if operation == 'list_branches' else None)
+ assert len(server.requests) == 1
+ headers = server.requests[0][2]
+ assert headers['Authorization'].startswith('ACS4-HMAC-SHA256 ')
+ for name in ('Authorization', 'Host', 'x-acs-date'):
+ assert len(headers.get_all(name)) == 1
+ if operation in ('create_branch', 'fast_forward'):
+ assert headers.get_all('Content-Type') == ['application/json']
+ assert headers['x-acs-security-token'] == 'test-sts'
+ assert headers['x-acs-action'] == {
+ 'create_branch': 'CreateBranch', 'drop_branch': 'DropBranch',
+ 'fast_forward': 'FastForwardBranch', 'list_branches': 'ListBranches',
+ }[operation]
+
+
[email protected]('operation', _OPERATIONS)
+def test_native_dlf_reuses_cached_credentials(transport_server, tmp_path,
operation):
+ server = transport_server
+ server.require_v4 = True
+ token_file = tmp_path / 'credentials.json'
+ token_file.write_text(json.dumps({
+ 'AccessKeyId': 'test-key', 'AccessKeySecret': 'test-secret',
+ 'SecurityToken': 'test-sts', 'Expiration': '2099-01-01T00:00:00Z',
+ }))
+ catalog = transport_catalog(server, **{
+ 'token.provider': 'dlf', 'dlf.region': 'cn-hangzhou',
'dlf.signing-algorithm': 'openapi-v4',
+ 'dlf.token-loader': 'local_file', 'dlf.token-path': str(token_file),
+ })
+ provider = catalog.rest_api.rest_auth_function.auth_provider
+ assert provider.token is not None # /config has loaded a valid cached STS.
+ token_file.unlink()
+ server.requests.clear()
+ args = () if operation == 'list_branches' else ('dev',)
+ with patch.object(provider.token_loader, 'load_token',
side_effect=AssertionError('unnecessary refresh')), \
+ native_only(catalog):
+ result = getattr(catalog, operation)('db.t', *args)
+ assert result == (['dev'] if operation == 'list_branches' else None)
+ assert len(server.requests) == 1
+ assert server.requests[0][2]['x-acs-security-token'] == 'test-sts'
+
+
+def test_native_dlf_refreshes_through_owning_provider(transport_server,
tmp_path):
+ server = transport_server
+ server.require_v4 = True
+ token_file = tmp_path / 'credentials.json'
+
+ def write_token(sts):
+ token_file.write_text(json.dumps({
+ 'AccessKeyId': 'test-key', 'AccessKeySecret': 'test-secret',
+ 'SecurityToken': sts, 'Expiration': '2099-01-01T00:00:00Z',
+ }))
+
+ write_token('test-sts')
+ catalog = transport_catalog(server, **{
+ 'token.provider': 'dlf', 'dlf.region': 'cn-hangzhou',
'dlf.signing-algorithm': 'openapi-v4',
+ 'dlf.token-loader': 'local_file', 'dlf.token-path': str(token_file),
+ })
+ provider = catalog.rest_api.rest_auth_function.auth_provider
+ provider.token.expiration_at_millis = 0
+ write_token('refreshed-sts')
+ server.expected_token = DLFToken('test-key', 'test-secret',
'refreshed-sts', None)
+ server.requests.clear()
+ with patch.object(provider.signer, 'authorization',
side_effect=AssertionError('Python signing')), \
+ patch.object(provider.token_loader, 'load_token',
wraps=provider.token_loader.load_token) as load, \
+ native_only(catalog):
+ assert catalog.list_branches('db.t') == ['dev']
+ token_file.unlink()
+ assert catalog.list_branches('db.t') == ['dev']
+ assert load.call_count == 1
+ assert len(server.requests) == 2
+ assert all(request[2]['x-acs-security-token'] == 'refreshed-sts' for
request in server.requests)
+
+
[email protected]('customization', ['hook', 'adapter', 'trust_env',
'timeout', 'ca_bundle'])
+def test_custom_session_keeps_owning_transport(transport_server, monkeypatch,
customization):
+ from pypaimon_rust.datafusion import PaimonCatalog
+
+ catalog = transport_catalog(transport_server)
+ client = catalog.rest_api.client
+ called = []
+
+ def hook(response, **kwargs):
+ called.append('hook')
+
+ class Adapter(requests.adapters.HTTPAdapter):
+ def send(self, request, **kwargs):
+ called.append('adapter')
+ return super().send(request, **kwargs)
+
+ if customization == 'hook':
+ client.session.hooks['response'].append(hook)
+ elif customization == 'adapter':
+ client.session.mount('http://', Adapter())
+ elif customization == 'trust_env':
+ client.session.trust_env = False
+ elif customization == 'timeout':
+ client._timeout = (12, 34)
+ else:
+ monkeypatch.setenv('REQUESTS_CA_BUNDLE', '/custom/ca.pem')
+ with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog)
as construct:
+ assert catalog.list_branches('db.t') == ['dev']
+ construct.assert_not_called()
+ assert called == ([customization] if customization in ('hook', 'adapter')
else [])
+
+
[email protected]('agent_header', ['User-Agent', 'user-agent',
'USER-AGENT'])
+def
test_native_preserves_session_headers_and_one_explicit_user_agent(transport_server,
agent_header):
+ catalog = transport_catalog(transport_server, **{'header.' + agent_header:
'explicit-agent'})
+ catalog.rest_api.client.session.headers['X-Tenant'] = 'tenant-a'
+ catalog.rest_api.client.session.headers['authorization'] = 'Bearer stale'
+ catalog.rest_api.client.session.headers['x-priority'] = 'session'
+ catalog.rest_api.rest_auth_function.init_header['X-Priority'] = 'request'
+ transport_server.requests.clear()
+ with native_only(catalog):
+ assert catalog.list_branches('db.t') == ['dev']
+ assert len(transport_server.requests) == 1
+ headers = transport_server.requests[0][2]
+ assert headers['X-Tenant'] == 'tenant-a'
+ assert headers.get_all('X-Priority') == ['request']
+ assert headers.get_all('User-Agent') == ['explicit-agent']
+ assert headers.get_all('Authorization') == ['Bearer test-token']
+
+
[email protected]('operation', ['create_branch', 'drop_branch',
'fast_forward'])
+def test_native_mutations_accept_plain_text_success(transport_server,
operation):
+ catalog = transport_catalog(transport_server)
+ transport_server.requests.clear()
+ with native_only(catalog):
+ assert getattr(catalog, operation)('db.t', 'dev') is None
+ assert len(transport_server.requests) == 1
+
+
[email protected]('status', [429, 503])
+def test_native_branch_listing_retries_temporary_status(transport_server,
status):
+ catalog = transport_catalog(transport_server)
+ transport_server.requests.clear()
+ transport_server.temporary_status = status
+ with native_only(catalog):
+ assert catalog.list_branches('db.t') == ['dev']
+ assert len(transport_server.requests) == 2
+ assert all(request[0] == 'GET' for request in transport_server.requests)
+
+
[email protected]
+def catalog(native_rest_catalog):
+ native_rest_catalog.context.options.to_map().update({
+ 'read.native.enabled': 'true', 'write.native.enabled': 'true',
+ })
+ return native_rest_catalog
+
+
+@contextmanager
+def native_only(catalog):
+ with ExitStack() as guards:
+ for name in _OPERATIONS:
+ guards.enter_context(patch.object(
+ catalog.rest_api, name, side_effect=AssertionError('Python
branch fallback: ' + name)))
+ yield
+
+
+def test_native_rest_branch_lifecycle_with_shared_data(tmp_path, catalog):
+ catalog, main = _table(tmp_path, catalog, True)
+ seed = dict(id=1, value=10, p='a')
+ added = dict(id=2, value=20, p='b')
+ _write(main, [seed])
+ with native_only(catalog):
+ empty = _branch(catalog, main, from_tag=False, name='empty')
+ _assert_rows(empty, [])
+ branch = _branch(catalog, main)
+ _write(branch, [added])
+ _assert_rows(main, [seed])
+ _assert_rows(branch, [seed, added])
+ assert catalog.list_branches(main.identifier) == ['dev', 'empty']
+ paths = {path for path in tmp_path.rglob('*.parquet')}
+ assert paths
+ catalog.fast_forward(main.identifier, 'dev')
+ catalog.drop_branch(main.identifier, 'dev')
+ catalog.drop_branch(main.identifier, 'empty')
+ assert catalog.list_branches(main.identifier) == []
+ assert all(path.exists() for path in paths)
+ catalog.create_branch(main.identifier, 'restored', 'base')
+ restored = catalog.get_table(Identifier('default', 't',
branch='restored'))
+ _assert_rows(restored, [seed])
+
+
+def test_native_branch_errors_keep_public_python_types(tmp_path, catalog):
+ catalog, table = _table(tmp_path, catalog, True)
+ with native_only(catalog):
+ catalog.create_branch(table.identifier, 'dev')
+ cases = [
+ (TableNotExistException, lambda:
catalog.create_branch('default.missing', 'dev')),
+ (TableNotExistException, lambda:
catalog.list_branches('default.missing')),
+ (TagNotExistException, lambda:
catalog.create_branch(table.identifier, 'tagged', 'missing')),
+ (BranchAlreadyExistException, lambda:
catalog.create_branch(table.identifier, 'dev')),
+ (BranchNotExistException, lambda:
catalog.drop_branch(table.identifier, 'missing')),
+ (BranchNotExistException, lambda:
catalog.fast_forward(table.identifier, 'missing')),
+ (IllegalArgumentError, lambda:
catalog.create_branch(table.identifier, 'main')),
+ ]
+ for expected, invoke in cases:
+ with pytest.raises(expected):
+ invoke()
+ manager = table.branch_manager()
+ manager.create_branch('dev', ignore_if_exists=True)
+ with pytest.raises(ValueError, match='already exists'):
+ manager.create_branch('dev')
+ with pytest.raises(ValueError, match="doesn't exist"):
+ manager.drop_branch('missing')
+
+
[email protected]('operation', _OPERATIONS)
[email protected]('name', ['.', '..', '', ' '])
+def test_invalid_resource_paths_send_no_requests(catalog, operation, name):
+ args = () if operation == 'list_branches' else ('dev',)
+ with patch('pypaimon.catalog.rest.native_branch._native_catalog',
+ side_effect=AssertionError('Invalid paths must be rejected
before preparing a backend')):
+ for identifier in (Identifier(name, 't'), Identifier('default', name)):
+ with pytest.raises(ValueError):
+ getattr(catalog, operation)(identifier, *args)
+ if operation in ('drop_branch', 'fast_forward'):
+ with pytest.raises(ValueError):
+ getattr(catalog, operation)('default.t', name)
+
+
+def test_native_catalog_constructor_uses_resolved_configuration(catalog):
+ from pypaimon_rust.datafusion import PaimonCatalog
+
+ with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog)
as construct:
+ with native_only(catalog):
+ # A valid missing table reaches the catalog, rather than failing
construction.
+ with pytest.raises(TableNotExistException):
+ catalog.list_branches('default.missing')
+ options = construct.call_args.args[0]
+ assert options['prefix'] == 'native-test'
+ assert options['metastore'] == 'rest'
+ assert construct.call_args.kwargs == {'config_required': False}
+
+
+def test_preparation_failure_can_use_python_before_mutation(tmp_path, catalog):
+ catalog, table = _table(tmp_path, catalog, True)
+ with patch('pypaimon_rust.datafusion.PaimonCatalog',
side_effect=ValueError('preparation failed')):
+ catalog.create_branch(table.identifier, 'dev')
+ with native_only(catalog):
+ assert catalog.list_branches(table.identifier) == ['dev']
+
+
+def test_fork_safety_failure_never_uses_python(catalog):
+ from pypaimon_rust import ForkSafetyError
+
+ error = ForkSafetyError('inherited native state')
+ with patch('pypaimon_rust.datafusion.PaimonCatalog', side_effect=error),
native_only(catalog):
+ with pytest.raises(ForkSafetyError) as raised:
+ catalog.list_branches('default.t')
+ assert raised.value is error
+
+
[email protected]('status,body,exception', [
+ (403, '{"message":"denied"}', TableNoPermissionException),
+ (429, 'too many requests', RESTException),
+ (502, 'bad gateway', RESTException),
+ (504, 'gateway timeout', RESTException),
+])
+def test_real_http_errors_preserve_python_exceptions(catalog, status, body,
exception):
+ # POST is deliberately used so neither HTTP client retries status
responses.
+ with patch.object(RESTCatalogServer, '_branches_handle',
return_value=(body, status)) as handle, \
+ native_only(catalog):
+ with pytest.raises(exception):
+ catalog.create_branch('default.t', 'dev')
+ assert handle.call_count == 1
+
+
+def test_native_response_parse_failure_is_a_rest_exception(catalog):
+ with patch.object(RESTCatalogServer, '_branches_handle',
return_value=('not JSON', 200)) as handle, \
+ native_only(catalog):
+ with pytest.raises(RESTException):
+ catalog.list_branches('default.t')
+ assert handle.call_count == 1
+
+
+def test_native_connection_failure_is_a_rest_exception(catalog):
+ import socket
+
+ with socket.socket() as listener:
+ listener.bind(('localhost', 0))
+ # A bound, non-listening port rejects connections without racing port
allocation.
+ catalog.rest_api.client.uri = 'http://localhost:%d' %
listener.getsockname()[1]
+ with native_only(catalog), pytest.raises(RESTException):
+ catalog.drop_branch('default.t', 'dev')
+
+
+def test_published_branch_survives_uncertain_response(tmp_path, catalog):
+ catalog, table = _table(tmp_path, catalog, True)
+ original = RESTCatalogServer._branches_handle
+ published = []
+
+ def publish_then_fail(server, method, data, identifier):
+ response = original(server, method, data, identifier)
+ if method == 'POST':
+ published.append(identifier)
+ return server._mock_response(ErrorResponse(None, None, 'uncertain
result', 500), 500)
+ return response
+
+ with patch.object(RESTCatalogServer, '_branches_handle',
publish_then_fail), native_only(catalog):
+ with pytest.raises(ServiceFailureException):
+ catalog.create_branch(table.identifier, 'dev')
+ assert catalog.list_branches(table.identifier) == ['dev']
+ assert published == [table.identifier]
+
+
+def test_custom_error_handler_uses_python_before_request(catalog):
+ from pypaimon.api.client import ErrorHandler
+
+ class CustomFailure(RESTException):
+ pass
+
+ class Handler(ErrorHandler):
+ def accept(self, error, request_id):
+ raise CustomFailure('custom error handler')
+
+ catalog.rest_api.client.set_error_handler(Handler())
+ with patch('pypaimon_rust.datafusion.PaimonCatalog',
side_effect=AssertionError('custom handler bypassed')):
+ with pytest.raises(CustomFailure, match='custom error handler'):
+ catalog.list_branches('default.missing')
+
+
+def test_native_keeps_live_endpoint_and_original_authentication(catalog):
+ # Config merging changes options without rebuilding Python's existing
client
+ # and provider. Delegation must keep their endpoint, credentials and
headers.
+ from pypaimon_rust.datafusion import PaimonCatalog
+
+ original_token = catalog.rest_api.rest_auth_function.auth_provider.token
+ catalog.rest_api.options.to_map().update({
+ 'token.provider': 'dlf', 'token': 'server-token', 'dlf.token-loader':
'ecs',
+ 'uri': 'http://invalid.example', 'header.X-Test': 'stale',
+ })
+ catalog.rest_api.rest_auth_function.init_header['X-Test'] = 'current'
+ catalog.rest_api.client.set_user_agent('branch-client')
+ with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog)
as construct, native_only(catalog):
+ with pytest.raises(TableNotExistException):
+ catalog.list_branches('default.missing')
+ options = construct.call_args.args[0]
+ assert options['token.provider'] == 'bear'
+ assert options['token'] == original_token
+ assert options['uri'] == catalog.rest_api.client.uri
+ assert 'dlf.token-loader' not in options
+ assert options['header.X-Test'] == 'current'
+ assert options['header.User-Agent'] == 'branch-client'
+
+
[email protected]('code,exception', [
+ (403, TableNoPermissionException), (500, ServiceFailureException), (501,
NotImplementedException),
+ (404, NoSuchResourceException),
+])
+def test_native_errors_are_translated_without_python_retry(catalog, code,
exception):
+ from pypaimon_rust.datafusion import CatalogError
+
+ error = CatalogError('native error')
+ error.code, error.resource_type, error.resource_name, error.message =
code, 'OTHER', 'unknown', 'native error'
+
+ class Native:
+ calls = 0
+
+ def create_branch(self, *args):
+ self.calls += 1
+ raise error
+
+ native = Native()
+ with patch('pypaimon.catalog.rest.native_branch._native_catalog',
return_value=(native, CatalogError)), \
+ native_only(catalog):
+ with pytest.raises(exception):
+ catalog.create_branch('default.t', 'dev')
+ assert native.calls == 1
+
+
+def test_native_branch_blob_files_survive_drop(tmp_path, catalog):
+ schema = pa.schema([('id', pa.int32()), ('payload', pa.large_binary())])
+ catalog.create_table('default.blobs', Schema.from_pyarrow_schema(schema,
options={
+ 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
'blob-field': 'payload',
+ 'write.native.enabled': 'true', 'commit.native.enabled': 'true',
+ }), False)
+ main = catalog.get_table('default.blobs')
+
+ def write(table, value):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [value], 'payload': [b'payload'
* value]}, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ write(main, 1)
+ catalog.create_tag(main.identifier, 'base')
+ with native_only(catalog):
+ catalog.create_branch(main.identifier, 'dev', 'base')
+ branch = catalog.get_table(Identifier('default', 'blobs',
branch='dev'))
+ write(branch, 2)
+ files = set(tmp_path.rglob('*.blob'))
+ assert files
+ catalog.drop_branch(main.identifier, 'dev')
+ assert all(path.exists() for path in files)
+ catalog.create_branch(main.identifier, 'restored', 'base')
+ restored = catalog.get_table(Identifier('default', 'blobs',
branch='restored'))
+ builder = restored.new_read_builder()
+ assert
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pydict() == {
+ 'id': [1], 'payload': [b'payload'],
+ }
diff --git a/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
b/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
index fcbb1347ad..17e380c7ca 100644
--- a/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
+++ b/paimon-python/pypaimon/tests/rest/dlf_signer_test.py
@@ -548,7 +548,7 @@ class DLFSignerTest(unittest.TestCase):
def test_openapi_v4_unregistered_requests_have_no_action(self):
paths = ResourcePaths("clg-paimon-1")
self.assertIsNone(resolve_action("GET", paths.tables()))
- self.assertIsNone(resolve_action("POST", paths.rename_branch("db",
"t", "b")))
+ self.assertIsNone(resolve_action("POST", paths.branch("db", "t", "b")
+ "/rename"))
self.assertIsNone(resolve_action("PUT", paths.table("db", "t")))
self.assertIsNone(resolve_action("GET", ResourcePaths.config() + "/"))
self.assertIsNone(resolve_action("GET", "/v1//databases"))
diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
index 5b13d6f195..ea99508568 100644
--- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
@@ -17,6 +17,7 @@
import json
import unittest
+from unittest.mock import patch
import pyarrow as pa
@@ -79,6 +80,16 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
self.rest_catalog.create_branch(
self._identifier(), "b1", tag_name="missing")
+ def test_create_branch_preserves_unknown_missing_resource(self):
+ from pypaimon.api.rest_exception import NoSuchResourceException
+
+ error = NoSuchResourceException('OTHER', 'resource', 'unknown resource
missing')
+ with patch('pypaimon.catalog.rest.native_branch._native_catalog',
return_value=None), \
+ patch.object(self.rest_catalog.rest_api, 'create_branch',
side_effect=error):
+ with self.assertRaises(NoSuchResourceException) as raised:
+ self.rest_catalog.create_branch(self._identifier(), 'new')
+ self.assertIs(raised.exception, error)
+
def test_branch_table_uses_branch_schema_manager(self):
identifier = self._identifier()
self.rest_catalog.create_branch(identifier, "b1")
@@ -157,27 +168,11 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
self.assertEqual(
sorted(self.rest_catalog.list_branches(identifier)), ["b1", "b2"])
- def test_rename_branch_happy(self):
- identifier = self._identifier()
- self.rest_catalog.create_branch(identifier, "b1")
- self.rest_catalog.rename_branch(identifier, "b1", "b2")
- result = self.rest_catalog.list_branches(identifier)
- self.assertNotIn("b1", result)
- self.assertIn("b2", result)
-
- def test_rename_branch_to_existing_raises(self):
- identifier = self._identifier()
- self.rest_catalog.create_branch(identifier, "b1")
- self.rest_catalog.create_branch(identifier, "b2")
- with self.assertRaises(BranchAlreadyExistException) as cm:
- self.rest_catalog.rename_branch(identifier, "b1", "b2")
- self.assertEqual(cm.exception.branch, "b2")
-
- def test_rename_branch_from_missing_raises(self):
- identifier = self._identifier()
- with self.assertRaises(BranchNotExistException) as cm:
- self.rest_catalog.rename_branch(identifier, "absent", "b2")
- self.assertEqual(cm.exception.branch, "absent")
+ def test_rename_branch_is_unsupported_without_a_request(self):
+ with patch.object(self.rest_catalog.rest_api.client, 'post',
+ side_effect=AssertionError('REST rename must not
send a request')):
+ with self.assertRaises(NotImplementedError):
+ self.rest_catalog.rename_branch(self._identifier(), 'old',
'new')
def test_drop_branch_happy(self):
identifier = self._identifier()
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py
b/paimon-python/pypaimon/tests/rest/rest_server.py
index db5a88c0a2..e61ea4388b 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -35,7 +35,6 @@ from pypaimon.api.api_request import (AlterDatabaseRequest,
AlterTableRequest,
CreatePartitionsRequest,
CreateTableRequest, CreateTagRequest,
GrantPermissionRequest,
- RenameBranchRequest,
RenameTableRequest,
RevokePermissionRequest)
from pypaimon.api.api_response import (ConfigResponse,
CreatePartitionsResponse,
@@ -662,8 +661,6 @@ class RESTCatalogServer:
elif len(path_parts) == 6 and path_parts[3] == ResourcePaths.BRANCHES:
branch_name = RESTUtil.decode_string(path_parts[4])
sub = path_parts[5]
- if sub == ResourcePaths.RENAME:
- return self._branch_rename_handle(method, data,
lookup_identifier, branch_name)
if sub == ResourcePaths.FORWARD:
return self._branch_forward_handle(method, lookup_identifier,
branch_name)
return self._mock_response(ErrorResponse(None, None, "Not Found",
404), 404)
@@ -1163,28 +1160,6 @@ class RESTCatalogServer:
return self._mock_response("", 200)
return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
- def _branch_rename_handle(self, method: str, data: str, identifier:
Identifier,
- from_branch: str) -> Tuple[str, int]:
- if method != "POST":
- return self._mock_response(ErrorResponse(None, None, "Method Not
Allowed", 405), 405)
- if identifier.get_full_name() not in self.table_metadata_store:
- raise TableNotExistException(identifier)
-
- store = self.branch_store.setdefault(identifier.get_full_name(), set())
- if from_branch not in store:
- raise BranchNotExistException(from_branch)
- request = JSON.from_json(data, RenameBranchRequest)
- self._validate_branch_name(from_branch)
- self._validate_branch_name(request.to_branch)
- if request.to_branch in store:
- raise BranchAlreadyExistException(request.to_branch)
- source = self._schema_manager(identifier, from_branch)
- target = self._schema_manager(identifier, request.to_branch)
- Path(source.branch_path).rename(target.branch_path)
- store.discard(from_branch)
- store.add(request.to_branch)
- return self._mock_response("", 200)
-
def _branch_forward_handle(self, method: str, identifier: Identifier,
branch_name: str) -> Tuple[str, int]:
if method != "POST":
