From 712e6ec1f06b17557980963d7a1cbbcd22876291 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 09:51:59 +0800 Subject: [PATCH 1/4] [python] Use Native REST catalog branch operations --- .github/workflows/ci-python.yml | 5 +- paimon-python/README.md | 16 + paimon-python/pypaimon/api/api_request.py | 7 - paimon-python/pypaimon/api/resource_paths.py | 5 - paimon-python/pypaimon/api/rest_api.py | 45 +-- .../pypaimon/branch/branch_manager.py | 4 +- .../pypaimon/branch/catalog_branch_manager.py | 9 +- .../pypaimon/catalog/rest/native_branch.py | 95 ++++++ .../pypaimon/catalog/rest/rest_catalog.py | 27 +- .../tests/api/test_branch_dto_serde.py | 19 +- .../branch/catalog_branch_manager_test.py | 5 + .../tests/native_branch_write_test.py | 16 +- .../pypaimon/tests/native_rest_branch_test.py | 306 ++++++++++++++++++ .../pypaimon/tests/rest/dlf_signer_test.py | 2 +- .../pypaimon/tests/rest/rest_branch_test.py | 37 +-- .../pypaimon/tests/rest/rest_server.py | 25 -- 16 files changed, 495 insertions(+), 128 deletions(-) create mode 100644 paimon-python/pypaimon/catalog/rest/native_branch.py create mode 100644 paimon-python/pypaimon/tests/native_rest_branch_test.py diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index b6a0fb9f81dc..00778ca3fc73 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 ccb3cb5c1873..5481b52f355c 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -139,6 +139,22 @@ 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. + +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 7f06c789a3ed..c03b75319fca 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 309d829d5903..ac5e5dca9584 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 @@ def branch(self, database_name: str, table_name: str, branch_name: str) -> str: 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 d441ea5cbdcd..eb630c268689 100755 --- a/paimon-python/pypaimon/api/rest_api.py +++ b/paimon-python/pypaimon/api/rest_api.py @@ -27,7 +27,7 @@ CreateTableRequest, CreateTagRequest, DropPolicyRequest, ForwardBranchRequest, GrantPermissionRequest, PolicyRequest, - RenameBranchRequest, RenameTableRequest, + RenameTableRequest, RevokePermissionRequest, RollbackTableRequest) from pypaimon.api.api_response import (CommitTableResponse, ConfigResponse, @@ -101,6 +101,10 @@ def __init__(self, options: Union[Options, Dict[str, str]], config_required: boo self.logger = logging.getLogger(self.__class__.__name__) self.client = HttpClient(uri, user_agent.rest_user_agent(options)) auth_provider = AuthProviderFactory.create_auth_provider(options) + # Java retains the provider created before config merging. Preserve its + # input when delegating an operation to a separately constructed client. + self._auth_options = {key: value for key, value in options.to_map().items() + if self._is_auth_option(key)} base_headers = RESTUtil.extract_prefix_map(options, self.HEADER_PREFIX) if config_required: @@ -128,6 +132,10 @@ def __init__(self, options: Union[Options, Dict[str, str]], config_required: boo 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 +553,24 @@ def delete_tag(self, identifier: Identifier, tag_name: str) -> None: ) # 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 +581,14 @@ def create_branch( ) 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 +596,7 @@ def fast_forward(self, identifier: Identifier, branch_name: str) -> None: ) 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 ae0d160f1782..8e9e249aeb52 100644 --- a/paimon-python/pypaimon/branch/branch_manager.py +++ b/paimon-python/pypaimon/branch/branch_manager.py @@ -40,11 +40,11 @@ def create_branch( 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 3c3009c88c2d..5cad0ab5205a 100644 --- a/paimon-python/pypaimon/branch/catalog_branch_manager.py +++ b/paimon-python/pypaimon/branch/catalog_branch_manager.py @@ -85,11 +85,11 @@ def create_branch( 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 @@ def create_branch( """ 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 000000000000..f104218fca96 --- /dev/null +++ b/paimon-python/pypaimon/catalog/rest/native_branch.py @@ -0,0 +1,95 @@ +# 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 + +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_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 + 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.update(api._auth_options) + resolved['uri'] = api.client.uri + resolved['header.User-Agent'] = api.client.session.headers['User-Agent'] + resolved.update({RESTApi.HEADER_PREFIX + key: value + for key, value in api.rest_auth_function.init_header.items()}) + resolved = {str(key): _option_value_to_string(value) + for key, value in resolved.items() if value is not None} + resolved['metastore'] = 'rest' + try: + # 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 2e0d5a046ed1..6ae9e6fd2401 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py +++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py @@ -648,14 +648,17 @@ def replace_tag(self, identifier: Union[str, Identifier], tag_name: str, # 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 @@ def create_branch(self, identifier: Union[str, Identifier], branch_name: str, 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 @@ def drop_branch(self, identifier: Union[str, Identifier], branch_name: str) -> N 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 a211c7af697d..2b32d0a393bd 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 @@ 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 @@ def test_without_tag_serializes_null_from_tag(self): 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 @@ def test_single_branch_url(self): "/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 dda8a85fc46b..51338d3a637e 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 @@ def test_create_branch_from_tag(self): 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 38de93fd4e03..da0945ab1332 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 000000000000..f179fc32b1ea --- /dev/null +++ b/paimon-python/pypaimon/tests/native_rest_branch_test.py @@ -0,0 +1,306 @@ +# 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 unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon import Schema +from pypaimon.api.api_response import ErrorResponse +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') + + +@pytest.fixture +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') + + +@pytest.mark.parametrize('operation', _OPERATIONS) +@pytest.mark.parametrize('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 + + +@pytest.mark.parametrize('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' + + +@pytest.mark.parametrize('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 fcbb1347ada7..17e380c7cac3 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 @@ def test_openapi_v4_names_spelled_like_literals(self): 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 5b13d6f195d0..ea99508568ed 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 @@ def test_create_branch_from_missing_tag_raises(self): 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 @@ def test_list_branches_returns_created(self): 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 db5a88c0a2f4..e61ea4388b5a 100755 --- a/paimon-python/pypaimon/tests/rest/rest_server.py +++ b/paimon-python/pypaimon/tests/rest/rest_server.py @@ -35,7 +35,6 @@ CreatePartitionsRequest, CreateTableRequest, CreateTagRequest, GrantPermissionRequest, - RenameBranchRequest, RenameTableRequest, RevokePermissionRequest) from pypaimon.api.api_response import (ConfigResponse, CreatePartitionsResponse, @@ -662,8 +661,6 @@ def _handle_table_resource(self, method: str, path_parts: List[str], 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 @@ def _branch_handle(self, method: str, identifier: Identifier, 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": From 84b6277f60d5678ab1c8cc76fc1c1159c59e450a Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 10:31:44 +0800 Subject: [PATCH 2/4] [python] Preserve REST branch session and header semantics --- paimon-python/README.md | 6 + .../pypaimon/catalog/rest/native_branch.py | 29 ++- .../pypaimon/tests/native_rest_branch_test.py | 166 ++++++++++++++++++ 3 files changed, 198 insertions(+), 3 deletions(-) diff --git a/paimon-python/README.md b/paimon-python/README.md index 5481b52f355c..0c88ad530efb 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -151,6 +151,12 @@ 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. diff --git a/paimon-python/pypaimon/catalog/rest/native_branch.py b/paimon-python/pypaimon/catalog/rest/native_branch.py index f104218fca96..010b3124ce4c 100644 --- a/paimon-python/pypaimon/catalog/rest/native_branch.py +++ b/paimon-python/pypaimon/catalog/rest/native_branch.py @@ -19,6 +19,9 @@ import logging +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 @@ -31,6 +34,25 @@ logger = logging.getLogger(__name__) +def _native_headers(api): + session = api.client.session + if (type(session) is not requests.Session or session.cookies or session.auth is not None + or session.cert is not None or session.verify is not True or session.proxies or session.params): + # Keep cookie selection/expiry and custom transport credentials in their + # owning Session. Rebuilding a client cannot preserve those semantics. + 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 @@ -49,6 +71,9 @@ def _native_catalog(catalog, operation): 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: @@ -58,9 +83,7 @@ def _native_catalog(catalog, operation): if not RESTApi._is_auth_option(key) and not key.startswith(RESTApi.HEADER_PREFIX)} resolved.update(api._auth_options) resolved['uri'] = api.client.uri - resolved['header.User-Agent'] = api.client.session.headers['User-Agent'] - resolved.update({RESTApi.HEADER_PREFIX + key: value - for key, value in api.rest_auth_function.init_header.items()}) + 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' diff --git a/paimon-python/pypaimon/tests/native_rest_branch_test.py b/paimon-python/pypaimon/tests/native_rest_branch_test.py index f179fc32b1ea..cb1b1538d716 100644 --- a/paimon-python/pypaimon/tests/native_rest_branch_test.py +++ b/paimon-python/pypaimon/tests/native_rest_branch_test.py @@ -18,13 +18,22 @@ """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 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, ) @@ -40,6 +49,163 @@ _OPERATIONS = ('create_branch', 'drop_branch', 'fast_forward', 'list_branches') +@pytest.fixture +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_cookie and self.headers.get('Cookie') != 'sessionid=accepted': + status, body = 403, json.dumps({'message': 'missing cookie'}) + 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))), + DLFToken('test-key', 'test-secret', 'test-sts', None), + 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))) + if self.path.startswith('/v1/config') and self.server.require_cookie: + self.send_header('Set-Cookie', 'sessionid=accepted; Path=/') + 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_cookie = server.require_v4 = False + server.temporary_status = server.branch_requests = 0 + 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)) + + +@pytest.mark.parametrize('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] + + +@pytest.mark.parametrize('operation', _OPERATIONS) +def test_config_cookie_keeps_existing_session(transport_server, operation): + from pypaimon_rust.datafusion import PaimonCatalog + + server = transport_server + server.require_cookie = True + catalog = transport_catalog(server) + server.requests.clear() + args = () if operation == 'list_branches' else ('dev',) + with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog) as construct: + result = getattr(catalog, operation)('db.t', *args) + construct.assert_not_called() + assert result == (['dev'] if operation == 'list_branches' else None) + assert len(server.requests) == 1 + assert server.requests[0][2]['Cookie'] == 'sessionid=accepted' + + +@pytest.mark.parametrize('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'] + + +@pytest.mark.parametrize('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 + + +@pytest.mark.parametrize('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) + + @pytest.fixture def catalog(native_rest_catalog): native_rest_catalog.context.options.to_map().update({ From ab474f17a6e5352cda3fd99014ffc353a03f44c7 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 11:02:10 +0800 Subject: [PATCH 3/4] [python] Preserve Native branch credential and session lifecycles --- paimon-python/pypaimon/api/rest_api.py | 4 - .../pypaimon/catalog/rest/native_branch.py | 55 ++++- .../pypaimon/tests/native_rest_branch_test.py | 192 +++++++++++++++++- 3 files changed, 242 insertions(+), 9 deletions(-) diff --git a/paimon-python/pypaimon/api/rest_api.py b/paimon-python/pypaimon/api/rest_api.py index eb630c268689..b8148a1dfe55 100755 --- a/paimon-python/pypaimon/api/rest_api.py +++ b/paimon-python/pypaimon/api/rest_api.py @@ -101,10 +101,6 @@ def __init__(self, options: Union[Options, Dict[str, str]], config_required: boo self.logger = logging.getLogger(self.__class__.__name__) self.client = HttpClient(uri, user_agent.rest_user_agent(options)) auth_provider = AuthProviderFactory.create_auth_provider(options) - # Java retains the provider created before config merging. Preserve its - # input when delegating an operation to a separately constructed client. - self._auth_options = {key: value for key, value in options.to_map().items() - if self._is_auth_option(key)} base_headers = RESTUtil.extract_prefix_map(options, self.HEADER_PREFIX) if config_required: diff --git a/paimon-python/pypaimon/catalog/rest/native_branch.py b/paimon-python/pypaimon/catalog/rest/native_branch.py index 010b3124ce4c..5b58cca91dc2 100644 --- a/paimon-python/pypaimon/catalog/rest/native_branch.py +++ b/paimon-python/pypaimon/catalog/rest/native_branch.py @@ -18,6 +18,10 @@ """Transport REST branch operations and structured errors to/from Rust core.""" import logging +import os +import re +import time +from email.message import Message import requests from requests.structures import CaseInsensitiveDict @@ -37,7 +41,12 @@ def _native_headers(api): session = api.client.session if (type(session) is not requests.Session or session.cookies 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 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 cookie selection/expiry and custom transport credentials in their # owning Session. Rebuilding a client cannot preserve those semantics. return None @@ -81,22 +90,58 @@ def _native_catalog(catalog, operation): 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.update(api._auth_options) 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 + return PaimonCatalog(resolved, config_required=False, _capture_response_cookies=True), CatalogError except Exception as error: _raise_if_native_fork_safety_error(error) logger.debug('Native branch preparation failed: %s', error) return None +def _restore_response_cookies(session, updates): + for url, cookies, received_at in updates: + request = requests.cookies.MockRequest(requests.Request('GET', url).prepare()) + for value in cookies: + # Max-Age is relative to receipt, not the end of GET retries or + # response-body reads. Keep invalid attributes for CookieJar to + # reject using its usual parser. + elapsed = int(time.time()) - received_at + + def adjust_max_age(match): + try: + age = int(match[2]) + return match[1] + str(age - elapsed if age > 0 else age) + except ValueError: + return match[0] + + headers = Message() + headers.add_header('Set-Cookie', re.sub( + r'(;\s*max-age\s*=\s*)([^;]*)', adjust_max_age, value, flags=re.IGNORECASE)) + session.cookies.extract_cookies(requests.cookies.MockResponse(headers), request) + + 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. @@ -116,3 +161,7 @@ def call_branch(catalog, operation, identifier, *args): 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) + finally: + # A response may establish or revoke a resource-path session even when + # its status/body fails. Retain it without replaying the operation. + _restore_response_cookies(catalog.rest_api.client.session, native._take_response_cookies()) diff --git a/paimon-python/pypaimon/tests/native_rest_branch_test.py b/paimon-python/pypaimon/tests/native_rest_branch_test.py index cb1b1538d716..c1a9b451fd8b 100644 --- a/paimon-python/pypaimon/tests/native_rest_branch_test.py +++ b/paimon-python/pypaimon/tests/native_rest_branch_test.py @@ -26,6 +26,7 @@ import pyarrow as pa import pytest +import requests from pypaimon import Schema from pypaimon.catalog.catalog_context import CatalogContext @@ -63,19 +64,24 @@ def operation(self): status, body = 403, json.dumps({'message': 'missing cookie'}) 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.cookie_on_retry and self.server.branch_requests + and self.headers.get('Cookie') != 'sessionid=accepted'): + status, body = 403, json.dumps({'message': 'missing retry cookie'}) 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.branch_status: + status, body = self.server.branch_status, json.dumps({'message': 'branch failed'}) 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))), - DLFToken('test-key', 'test-secret', 'test-sts', None), + self.server.expected_token, self.headers['Host'], signed_headers) if self.headers.get('Authorization') != expected: status, body = 403, json.dumps({'message': 'signature mismatch'}) @@ -84,6 +90,11 @@ def operation(self): self.send_header('Content-Length', str(len(output))) if self.path.startswith('/v1/config') and self.server.require_cookie: self.send_header('Set-Cookie', 'sessionid=accepted; Path=/') + elif (not self.path.startswith('/v1/config') and self.server.branch_response_cookie + and (not self.server.cookie_only_on_temporary or status in (429, 503))): + self.send_header('Set-Cookie', self.server.branch_response_cookie) + if status in (429, 503) and self.server.retry_after: + self.send_header('Retry-After', self.server.retry_after) self.end_headers() self.wfile.write(output) @@ -96,6 +107,12 @@ def log_message(self, *args): server.requests = [] server.require_cookie = server.require_v4 = False server.temporary_status = server.branch_requests = 0 + server.branch_response_cookie = None + server.branch_status = 0 + server.cookie_on_retry = False + server.cookie_only_on_temporary = False + server.retry_after = None + server.expected_token = DLFToken('test-key', 'test-secret', 'test-sts', None) thread = threading.Thread(target=server.serve_forever) thread.start() try: @@ -168,6 +185,174 @@ def test_config_cookie_keeps_existing_session(transport_server, operation): assert server.requests[0][2]['Cookie'] == 'sessionid=accepted' +@pytest.mark.parametrize('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) + + +@pytest.mark.parametrize('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 []) + + +@pytest.mark.parametrize('operation', _OPERATIONS) +@pytest.mark.parametrize('status', [200, 500]) +def test_native_response_cookies_rejoin_owning_session(transport_server, operation, status): + from pypaimon_rust.datafusion import PaimonCatalog + + server = transport_server + catalog = transport_catalog(server) + server.requests.clear() + server.branch_response_cookie = 'sessionid=accepted; Path=/v1/test/databases/' + server.branch_status = status + args = () if operation == 'list_branches' else ('dev',) + with native_only(catalog): + if status == 500: + with pytest.raises(ServiceFailureException): + getattr(catalog, operation)('db.t', *args) + else: + getattr(catalog, operation)('db.t', *args) + assert len(server.requests) == 1 + assert catalog.rest_api.client.session.cookies.get('sessionid') == 'accepted' + server.branch_status = 0 + server.require_cookie = True + # The response's resource-path cookie stays in the original Session for + # subsequent branch calls and other REST APIs, even after a failed mutation. + with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog) as construct: + assert catalog.list_branches('db.t') == ['dev'] + construct.assert_not_called() + assert server.requests[-1][2]['Cookie'] == 'sessionid=accepted' + + +def test_native_retry_uses_response_cookie_and_returns_it_to_session(transport_server): + server = transport_server + catalog = transport_catalog(server) + server.requests.clear() + server.temporary_status = 503 + server.branch_response_cookie = 'sessionid=accepted; Path=/v1/test/databases/' + server.cookie_on_retry = True + with native_only(catalog): + assert catalog.list_branches('db.t') == ['dev'] + assert [request[2].get('Cookie') for request in server.requests] == [None, 'sessionid=accepted'] + assert catalog.rest_api.client.session.cookies.get('sessionid') == 'accepted' + + +def test_cookie_expiring_during_native_retry_is_not_reactivated(transport_server): + server = transport_server + catalog = transport_catalog(server) + server.requests.clear() + server.temporary_status, server.retry_after = 503, '2' + server.branch_response_cookie = 'sessionid=accepted; Max-Age=1; Path=/v1/test/databases/' + server.cookie_only_on_temporary = True + with native_only(catalog): + assert catalog.list_branches('db.t') == ['dev'] + assert not catalog.rest_api.client.session.cookies + assert catalog.list_branches('db.t') == ['dev'] + assert [request[2].get('Cookie') for request in server.requests] == [None, None, None] + + +@pytest.mark.parametrize('attribute,now,expected', [ + ('Max-Age=120', 1700000030, 1700000120), + ('mAx-AgE=120', 1700000030, 1700000120), + ('Max-Age=20; Expires=Thu, 01 Jan 2099 00:00:00 GMT', 1700000030, None), + ('Max-Age=0', 1699999970, None), + ('Max-Age=invalid', 1700000030, None), + ('Expires=Tue, 14 Nov 2023 23:13:20 GMT', 1700000030, 1700003600), +]) +def test_cookie_handoff_preserves_expiry_and_parser_rules(attribute, now, expected): + from pypaimon.catalog.rest.native_branch import _restore_response_cookies + + session = requests.Session() + updates = [('https://catalog.example/v1/branches', ['sid=accepted; Path=/v1/; ' + attribute], 1700000000)] + with patch('pypaimon.catalog.rest.native_branch.time.time', return_value=now): + _restore_response_cookies(session, updates) + if expected is None: + assert not session.cookies + else: + cookie, = session.cookies + assert cookie.expires == expected + + @pytest.mark.parametrize('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'}) @@ -297,7 +482,7 @@ def test_native_catalog_constructor_uses_resolved_configuration(catalog): options = construct.call_args.args[0] assert options['prefix'] == 'native-test' assert options['metastore'] == 'rest' - assert construct.call_args.kwargs == {'config_required': False} + assert construct.call_args.kwargs == {'config_required': False, '_capture_response_cookies': True} def test_preparation_failure_can_use_python_before_mutation(tmp_path, catalog): @@ -424,6 +609,9 @@ def test_native_errors_are_translated_without_python_retry(catalog, code, except class Native: calls = 0 + def _take_response_cookies(self): + return [] + def create_branch(self, *args): self.calls += 1 raise error From 2817f36133873a4c6362b59663efd369d87668eb Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 11:27:22 +0800 Subject: [PATCH 4/4] [python] Remove Native REST cookie session synchronization --- .../pypaimon/catalog/rest/native_branch.py | 36 +----- .../pypaimon/tests/native_rest_branch_test.py | 121 +----------------- 2 files changed, 5 insertions(+), 152 deletions(-) diff --git a/paimon-python/pypaimon/catalog/rest/native_branch.py b/paimon-python/pypaimon/catalog/rest/native_branch.py index 5b58cca91dc2..4d46b19f180c 100644 --- a/paimon-python/pypaimon/catalog/rest/native_branch.py +++ b/paimon-python/pypaimon/catalog/rest/native_branch.py @@ -19,9 +19,6 @@ import logging import os -import re -import time -from email.message import Message import requests from requests.structures import CaseInsensitiveDict @@ -40,15 +37,14 @@ def _native_headers(api): session = api.client.session - if (type(session) is not requests.Session or session.cookies or session.auth is not None + 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 cookie selection/expiry and custom transport credentials in their - # owning Session. Rebuilding a client cannot preserve those semantics. + # 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 @@ -113,35 +109,13 @@ def _native_catalog(catalog, operation): 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, _capture_response_cookies=True), CatalogError + 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 _restore_response_cookies(session, updates): - for url, cookies, received_at in updates: - request = requests.cookies.MockRequest(requests.Request('GET', url).prepare()) - for value in cookies: - # Max-Age is relative to receipt, not the end of GET retries or - # response-body reads. Keep invalid attributes for CookieJar to - # reject using its usual parser. - elapsed = int(time.time()) - received_at - - def adjust_max_age(match): - try: - age = int(match[2]) - return match[1] + str(age - elapsed if age > 0 else age) - except ValueError: - return match[0] - - headers = Message() - headers.add_header('Set-Cookie', re.sub( - r'(;\s*max-age\s*=\s*)([^;]*)', adjust_max_age, value, flags=re.IGNORECASE)) - session.cookies.extract_cookies(requests.cookies.MockResponse(headers), request) - - 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. @@ -161,7 +135,3 @@ def call_branch(catalog, operation, identifier, *args): 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) - finally: - # A response may establish or revoke a resource-path session even when - # its status/body fails. Retain it without replaying the operation. - _restore_response_cookies(catalog.rest_api.client.session, native._take_response_cookies()) diff --git a/paimon-python/pypaimon/tests/native_rest_branch_test.py b/paimon-python/pypaimon/tests/native_rest_branch_test.py index c1a9b451fd8b..8782030d93de 100644 --- a/paimon-python/pypaimon/tests/native_rest_branch_test.py +++ b/paimon-python/pypaimon/tests/native_rest_branch_test.py @@ -60,21 +60,14 @@ def operation(self): status = 200 if self.path.startswith('/v1/config'): body = json.dumps({'defaults': {'prefix': 'test'}}) - elif self.server.require_cookie and self.headers.get('Cookie') != 'sessionid=accepted': - status, body = 403, json.dumps({'message': 'missing cookie'}) 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.cookie_on_retry and self.server.branch_requests - and self.headers.get('Cookie') != 'sessionid=accepted'): - status, body = 403, json.dumps({'message': 'missing retry cookie'}) 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.branch_status: - status, body = self.server.branch_status, json.dumps({'message': 'branch failed'}) 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')} @@ -88,13 +81,6 @@ def operation(self): output = body.encode() self.send_response(status) self.send_header('Content-Length', str(len(output))) - if self.path.startswith('/v1/config') and self.server.require_cookie: - self.send_header('Set-Cookie', 'sessionid=accepted; Path=/') - elif (not self.path.startswith('/v1/config') and self.server.branch_response_cookie - and (not self.server.cookie_only_on_temporary or status in (429, 503))): - self.send_header('Set-Cookie', self.server.branch_response_cookie) - if status in (429, 503) and self.server.retry_after: - self.send_header('Retry-After', self.server.retry_after) self.end_headers() self.wfile.write(output) @@ -105,13 +91,8 @@ def log_message(self, *args): server = HTTPServer(('localhost', 0), Handler) server.requests = [] - server.require_cookie = server.require_v4 = False + server.require_v4 = False server.temporary_status = server.branch_requests = 0 - server.branch_response_cookie = None - server.branch_status = 0 - server.cookie_on_retry = False - server.cookie_only_on_temporary = False - server.retry_after = None server.expected_token = DLFToken('test-key', 'test-secret', 'test-sts', None) thread = threading.Thread(target=server.serve_forever) thread.start() @@ -168,23 +149,6 @@ def test_openapi_v4_uses_native_signer(transport_server, operation): }[operation] -@pytest.mark.parametrize('operation', _OPERATIONS) -def test_config_cookie_keeps_existing_session(transport_server, operation): - from pypaimon_rust.datafusion import PaimonCatalog - - server = transport_server - server.require_cookie = True - catalog = transport_catalog(server) - server.requests.clear() - args = () if operation == 'list_branches' else ('dev',) - with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog) as construct: - result = getattr(catalog, operation)('db.t', *args) - construct.assert_not_called() - assert result == (['dev'] if operation == 'list_branches' else None) - assert len(server.requests) == 1 - assert server.requests[0][2]['Cookie'] == 'sessionid=accepted' - - @pytest.mark.parametrize('operation', _OPERATIONS) def test_native_dlf_reuses_cached_credentials(transport_server, tmp_path, operation): server = transport_server @@ -275,84 +239,6 @@ def send(self, request, **kwargs): assert called == ([customization] if customization in ('hook', 'adapter') else []) -@pytest.mark.parametrize('operation', _OPERATIONS) -@pytest.mark.parametrize('status', [200, 500]) -def test_native_response_cookies_rejoin_owning_session(transport_server, operation, status): - from pypaimon_rust.datafusion import PaimonCatalog - - server = transport_server - catalog = transport_catalog(server) - server.requests.clear() - server.branch_response_cookie = 'sessionid=accepted; Path=/v1/test/databases/' - server.branch_status = status - args = () if operation == 'list_branches' else ('dev',) - with native_only(catalog): - if status == 500: - with pytest.raises(ServiceFailureException): - getattr(catalog, operation)('db.t', *args) - else: - getattr(catalog, operation)('db.t', *args) - assert len(server.requests) == 1 - assert catalog.rest_api.client.session.cookies.get('sessionid') == 'accepted' - server.branch_status = 0 - server.require_cookie = True - # The response's resource-path cookie stays in the original Session for - # subsequent branch calls and other REST APIs, even after a failed mutation. - with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog) as construct: - assert catalog.list_branches('db.t') == ['dev'] - construct.assert_not_called() - assert server.requests[-1][2]['Cookie'] == 'sessionid=accepted' - - -def test_native_retry_uses_response_cookie_and_returns_it_to_session(transport_server): - server = transport_server - catalog = transport_catalog(server) - server.requests.clear() - server.temporary_status = 503 - server.branch_response_cookie = 'sessionid=accepted; Path=/v1/test/databases/' - server.cookie_on_retry = True - with native_only(catalog): - assert catalog.list_branches('db.t') == ['dev'] - assert [request[2].get('Cookie') for request in server.requests] == [None, 'sessionid=accepted'] - assert catalog.rest_api.client.session.cookies.get('sessionid') == 'accepted' - - -def test_cookie_expiring_during_native_retry_is_not_reactivated(transport_server): - server = transport_server - catalog = transport_catalog(server) - server.requests.clear() - server.temporary_status, server.retry_after = 503, '2' - server.branch_response_cookie = 'sessionid=accepted; Max-Age=1; Path=/v1/test/databases/' - server.cookie_only_on_temporary = True - with native_only(catalog): - assert catalog.list_branches('db.t') == ['dev'] - assert not catalog.rest_api.client.session.cookies - assert catalog.list_branches('db.t') == ['dev'] - assert [request[2].get('Cookie') for request in server.requests] == [None, None, None] - - -@pytest.mark.parametrize('attribute,now,expected', [ - ('Max-Age=120', 1700000030, 1700000120), - ('mAx-AgE=120', 1700000030, 1700000120), - ('Max-Age=20; Expires=Thu, 01 Jan 2099 00:00:00 GMT', 1700000030, None), - ('Max-Age=0', 1699999970, None), - ('Max-Age=invalid', 1700000030, None), - ('Expires=Tue, 14 Nov 2023 23:13:20 GMT', 1700000030, 1700003600), -]) -def test_cookie_handoff_preserves_expiry_and_parser_rules(attribute, now, expected): - from pypaimon.catalog.rest.native_branch import _restore_response_cookies - - session = requests.Session() - updates = [('https://catalog.example/v1/branches', ['sid=accepted; Path=/v1/; ' + attribute], 1700000000)] - with patch('pypaimon.catalog.rest.native_branch.time.time', return_value=now): - _restore_response_cookies(session, updates) - if expected is None: - assert not session.cookies - else: - cookie, = session.cookies - assert cookie.expires == expected - - @pytest.mark.parametrize('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'}) @@ -482,7 +368,7 @@ def test_native_catalog_constructor_uses_resolved_configuration(catalog): options = construct.call_args.args[0] assert options['prefix'] == 'native-test' assert options['metastore'] == 'rest' - assert construct.call_args.kwargs == {'config_required': False, '_capture_response_cookies': True} + assert construct.call_args.kwargs == {'config_required': False} def test_preparation_failure_can_use_python_before_mutation(tmp_path, catalog): @@ -609,9 +495,6 @@ def test_native_errors_are_translated_without_python_retry(catalog, code, except class Native: calls = 0 - def _take_response_cookies(self): - return [] - def create_branch(self, *args): self.calls += 1 raise error