From 78cc2a3abfe5fcdeab7ac44dabc6ee1b4dfe5803 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 12:08:19 +0800 Subject: [PATCH 1/2] [python] Use Native REST catalog version management --- .github/workflows/ci-python.yml | 7 +- paimon-python/pypaimon/api/api_request.py | 2 +- paimon-python/pypaimon/api/rest_api.py | 22 +- .../{native_branch.py => native_catalog.py} | 46 ++- .../pypaimon/catalog/rest/rest_catalog.py | 37 ++- paimon-python/pypaimon/common/json_util.py | 2 + .../pypaimon/tests/native_rest_branch_test.py | 11 +- .../tests/native_rest_version_test.py | 304 ++++++++++++++++++ .../pypaimon/tests/rest/rest_branch_test.py | 2 +- .../tests/rest/rollback_protocol_test.py | 20 +- 10 files changed, 402 insertions(+), 51 deletions(-) rename paimon-python/pypaimon/catalog/rest/{native_branch.py => native_catalog.py} (74%) create mode 100644 paimon-python/pypaimon/tests/native_rest_version_test.py diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 00778ca3fc73..014460867ab3 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -231,11 +231,14 @@ jobs: normalize_array_for_write, read_arrow_schema, ) - from pypaimon_rust.datafusion import CatalogError, PaimonCatalog, ReadBuilder + from pypaimon_rust.datafusion import CatalogError, Instant, 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'): + for method in ('create_branch', 'drop_branch', 'fast_forward', 'list_branches', + 'create_tag', 'get_tag', 'list_tags_paged', 'delete_tag', + 'load_snapshot', 'rollback_to'): assert callable(getattr(PaimonCatalog, method, None)) + assert callable(Instant.snapshot) and callable(Instant.tag) PY - name: Run Python tests with native planning, reading, writing, updating and committing diff --git a/paimon-python/pypaimon/api/api_request.py b/paimon-python/pypaimon/api/api_request.py index c03b75319fca..6ff6e08934a1 100644 --- a/paimon-python/pypaimon/api/api_request.py +++ b/paimon-python/pypaimon/api/api_request.py @@ -109,7 +109,7 @@ class RollbackTableRequest(RESTRequest): FIELD_TABLE_ID = "tableId" instant: Instant = json_field(FIELD_INSTANT) - from_snapshot: Optional[int] = json_field(FIELD_FROM_SNAPSHOT) + from_snapshot: Optional[int] = optional_json_field(FIELD_FROM_SNAPSHOT, "non_null") table_id: Optional[str] = optional_json_field(FIELD_TABLE_ID, "non_null") diff --git a/paimon-python/pypaimon/api/rest_api.py b/paimon-python/pypaimon/api/rest_api.py index b8148a1dfe55..de3adf18646a 100755 --- a/paimon-python/pypaimon/api/rest_api.py +++ b/paimon-python/pypaimon/api/rest_api.py @@ -142,11 +142,11 @@ def __build_paged_query_params( if max_results is not None and max_results > 0: query_params[RESTApi.MAX_RESULTS] = str(max_results) - if page_token is not None and page_token.strip(): + if page_token is not None: query_params[RESTApi.PAGE_TOKEN] = page_token for key, value in name_patterns.items(): - if key and value and key.strip() and value.strip(): + if key and value: query_params[key] = value return query_params @@ -538,8 +538,8 @@ def list_tags_paged( ListTagsResponse, self.rest_auth_function, ) - tags = response.data() or [] - return PagedList(tags, response.get_next_page_token()) + tags = response.data() + return PagedList([], None) if tags is None else PagedList(tags, response.get_next_page_token()) def delete_tag(self, identifier: Identifier, tag_name: str) -> None: database_name, table_name = self.__validate_identifier(identifier) @@ -550,11 +550,11 @@ 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): + def _validate_resource_path(identifier, resource_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')) + if resource_name is not None: + components.append((resource_name, 'Resource 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)) @@ -566,7 +566,7 @@ def create_branch( branch_name: str, tag_name: Optional[str] = None, ) -> None: - database_name, table_name = self._validate_branch_path(identifier) + database_name, table_name = self._validate_resource_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) @@ -577,14 +577,14 @@ def create_branch( ) def drop_branch(self, identifier: Identifier, branch_name: str) -> None: - database_name, table_name = self._validate_branch_path(identifier, branch_name) + database_name, table_name = self._validate_resource_path(identifier, branch_name) self.client.delete( self.resource_paths.branch(database_name, table_name, branch_name), self.rest_auth_function, ) def fast_forward(self, identifier: Identifier, branch_name: str) -> None: - database_name, table_name = self._validate_branch_path(identifier, branch_name) + database_name, table_name = self._validate_resource_path(identifier, branch_name) self.client.post( self.resource_paths.forward_branch(database_name, table_name, branch_name), ForwardBranchRequest(), @@ -592,7 +592,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_branch_path(identifier) + database_name, table_name = self._validate_resource_path(identifier) response = self.client.get( self.resource_paths.branches(database_name, table_name), ListBranchesResponse, diff --git a/paimon-python/pypaimon/catalog/rest/native_branch.py b/paimon-python/pypaimon/catalog/rest/native_catalog.py similarity index 74% rename from paimon-python/pypaimon/catalog/rest/native_branch.py rename to paimon-python/pypaimon/catalog/rest/native_catalog.py index 4d46b19f180c..a3faebf8423f 100644 --- a/paimon-python/pypaimon/catalog/rest/native_branch.py +++ b/paimon-python/pypaimon/catalog/rest/native_catalog.py @@ -15,7 +15,7 @@ # specific language governing permissions and limitations # under the License. -"""Transport REST branch operations and structured errors to/from Rust core.""" +"""Delegate REST catalog version management and structured errors to Rust core.""" import logging import os @@ -67,7 +67,8 @@ def _native_catalog(catalog, operation): '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() + read = operation in ('list_branches', 'get_tag', 'list_tags_paged', 'load_snapshot') + enabled = options.native_read_enabled() if read else options.native_write_enabled() if not enabled: return None api = catalog.rest_api @@ -81,6 +82,12 @@ def _native_catalog(catalog, operation): return None try: from pypaimon_rust.datafusion import CatalogError, PaimonCatalog + if not callable(getattr(PaimonCatalog, operation, None)): + return None + if operation == 'rollback_to': + from pypaimon_rust.datafusion import Instant + if not all(callable(getattr(Instant, name, None)) for name in ('snapshot', 'tag')): + return None except ImportError: return None from pypaimon.read.native_plan import _option_value_to_string, _raise_if_native_fork_safety_error @@ -112,26 +119,47 @@ def _native_catalog(catalog, operation): 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) + logger.debug('Native catalog preparation failed: %s', error) return None -def call_branch(catalog, operation, identifier, *args): +def call_catalog(catalog, operation, identifier, *args, **kwargs): # 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) + # resource. Names in CREATE/ROLLBACK bodies are not URL components. + path_name = args[0] if operation in ('drop_branch', 'fast_forward', 'get_tag', 'delete_tag') else None + RESTApi._validate_resource_path(identifier, path_name) 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) + return getattr(catalog.rest_api, operation)(identifier, *args, **kwargs) native, error_type = prepared + if operation == 'rollback_to': + from pypaimon.table.instant import SnapshotInstant, TagInstant + from pypaimon_rust.datafusion import Instant + instant = args[0] + if isinstance(instant, SnapshotInstant): + instant = Instant.snapshot(instant.snapshot_id) + elif isinstance(instant, TagInstant): + instant = Instant.tag(instant.tag_name) + else: + raise TypeError('Rollback requires a snapshot or tag instant') + args = (instant,) + args[1:] 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) + result = getattr(native, operation)( + (identifier.get_database_name(), identifier.get_object_name()), *args, **kwargs) 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) + if operation in ('get_tag', 'load_snapshot') and result is not None: + from pypaimon.api.api_response import GetTagResponse + from pypaimon.snapshot.table_snapshot import TableSnapshot + from pypaimon.common.json_util import JSON + result = JSON.from_json(result.to_json(), GetTagResponse if operation == 'get_tag' else TableSnapshot) + elif operation == 'list_tags_paged': + from pypaimon.api.api_response import PagedList + result = PagedList(*result) + return result diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py b/paimon-python/pypaimon/catalog/rest/rest_catalog.py index 6ae9e6fd2401..ec6c501b8d7a 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py +++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py @@ -430,8 +430,8 @@ def rollback_to(self, identifier: Union[str, Identifier], instant: Instant, if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - self.rest_api.rollback_to( - identifier, instant, from_snapshot, table_uuid=table_uuid) + from pypaimon.catalog.rest.native_catalog import call_catalog + call_catalog(self, 'rollback_to', identifier, instant, from_snapshot, table_uuid=table_uuid) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_SNAPSHOT: raise ValueError( @@ -459,7 +459,8 @@ def load_snapshot(self, identifier: Union[str, Identifier]) -> Optional[TableSna if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - return self.rest_api.load_snapshot(identifier) + from pypaimon.catalog.rest.native_catalog import call_catalog + return call_catalog(self, 'load_snapshot', identifier) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_SNAPSHOT: return None @@ -574,7 +575,9 @@ def create_tag(self, identifier: Union[str, Identifier], tag_name: str, if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - self.rest_api.create_tag(identifier, tag_name, snapshot_id, time_retained) + from pypaimon.catalog.rest.native_catalog import call_catalog + # Ignore-if-exists remains in the public catalog error handling below. + call_catalog(self, 'create_tag', identifier, tag_name, snapshot_id, time_retained) except AlreadyExistsException as e: if not ignore_if_exists: raise TagAlreadyExistException(tag_name) from e @@ -593,7 +596,8 @@ def get_tag(self, identifier: Union[str, Identifier], tag_name: str) -> GetTagRe if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - return self.rest_api.get_tag(identifier, tag_name) + from pypaimon.catalog.rest.native_catalog import call_catalog + return call_catalog(self, 'get_tag', identifier, tag_name) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_TAG: raise TagNotExistException(tag_name) from e @@ -608,8 +612,8 @@ def list_tags_paged(self, identifier: Union[str, Identifier], if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - return self.rest_api.list_tags_paged( - identifier, max_results, page_token, tag_name_prefix) + from pypaimon.catalog.rest.native_catalog import call_catalog + return call_catalog(self, 'list_tags_paged', identifier, max_results, page_token, tag_name_prefix) except NoSuchResourceException as e: raise TableNotExistException(identifier) from e except ForbiddenException as e: @@ -619,7 +623,8 @@ def delete_tag(self, identifier: Union[str, Identifier], tag_name: str) -> None: if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - self.rest_api.delete_tag(identifier, tag_name) + from pypaimon.catalog.rest.native_catalog import call_catalog + call_catalog(self, 'delete_tag', identifier, tag_name) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_TAG: raise TagNotExistException(tag_name) from e @@ -648,11 +653,11 @@ 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 + from pypaimon.catalog.rest.native_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - call_branch(self, 'create_branch', identifier, branch_name, tag_name) + call_catalog(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 @@ -667,11 +672,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 + from pypaimon.catalog.rest.native_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - call_branch(self, 'drop_branch', identifier, branch_name) + call_catalog(self, 'drop_branch', identifier, branch_name) except NoSuchResourceException as e: raise BranchNotExistException(branch_name) from e except ForbiddenException as e: @@ -682,22 +687,22 @@ def rename_branch(self, identifier: Union[str, Identifier], from_branch: str, 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 + from pypaimon.catalog.rest.native_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - call_branch(self, 'fast_forward', identifier, branch_name) + call_catalog(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 + from pypaimon.catalog.rest.native_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - return call_branch(self, 'list_branches', identifier) + return call_catalog(self, 'list_branches', identifier) except NoSuchResourceException as e: raise TableNotExistException(identifier) from e except ForbiddenException as e: diff --git a/paimon-python/pypaimon/common/json_util.py b/paimon-python/pypaimon/common/json_util.py index df9c95427ce9..15e8d616321a 100644 --- a/paimon-python/pypaimon/common/json_util.py +++ b/paimon-python/pypaimon/common/json_util.py @@ -70,6 +70,8 @@ def to_json(obj: Any, **kwargs) -> str: def from_json(json_str: str, target_class: Type[T]) -> T: """Create instance from JSON string""" data = json.loads(json_str) + if data is None: + return None return JSON.__from_dict(data, target_class) @staticmethod diff --git a/paimon-python/pypaimon/tests/native_rest_branch_test.py b/paimon-python/pypaimon/tests/native_rest_branch_test.py index 8782030d93de..57943c0e0594 100644 --- a/paimon-python/pypaimon/tests/native_rest_branch_test.py +++ b/paimon-python/pypaimon/tests/native_rest_branch_test.py @@ -57,6 +57,7 @@ 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)) + self.server.bodies.append(json.loads(data) if data else None) status = 200 if self.path.startswith('/v1/config'): body = json.dumps({'defaults': {'prefix': 'test'}}) @@ -66,6 +67,8 @@ def operation(self): status = self.server.temporary_status body = json.dumps({'message': 'temporary failure'}) self.server.branch_requests += 1 + elif self.server.respond is not None: + status, body = self.server.respond(self.command, self.path, data) else: body = json.dumps({'branches': ['dev']}) if self.command == 'GET' else 'OK' if self.server.require_v4: @@ -73,7 +76,7 @@ def operation(self): 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))), + RESTAuthParameter(self.command, url.path, data, dict(parse_qsl(url.query, keep_blank_values=True))), self.server.expected_token, self.headers['Host'], signed_headers) if self.headers.get('Authorization') != expected: @@ -91,6 +94,8 @@ def log_message(self, *args): server = HTTPServer(('localhost', 0), Handler) server.requests = [] + server.bodies = [] + server.respond = None server.require_v4 = False server.temporary_status = server.branch_requests = 0 server.expected_token = DLFToken('test-key', 'test-secret', 'test-sts', None) @@ -347,7 +352,7 @@ def test_native_branch_errors_keep_public_python_types(tmp_path, catalog): @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', + with patch('pypaimon.catalog.rest.native_catalog._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): @@ -500,7 +505,7 @@ def create_branch(self, *args): raise error native = Native() - with patch('pypaimon.catalog.rest.native_branch._native_catalog', return_value=(native, CatalogError)), \ + with patch('pypaimon.catalog.rest.native_catalog._native_catalog', return_value=(native, CatalogError)), \ native_only(catalog): with pytest.raises(exception): catalog.create_branch('default.t', 'dev') diff --git a/paimon-python/pypaimon/tests/native_rest_version_test.py b/paimon-python/pypaimon/tests/native_rest_version_test.py new file mode 100644 index 000000000000..0d004a08e3e1 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_rest_version_test.py @@ -0,0 +1,304 @@ +# 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. + +"""Existing REST version-management behavior through Rust and Python transports.""" + +import json +from unittest.mock import patch +from urllib.parse import parse_qs, unquote_plus, urlsplit + +import pyarrow as pa +import pytest + +from pypaimon import Schema +from pypaimon.api.rest_exception import BadRequestException, NotImplementedException, ServiceFailureException +from pypaimon.catalog.catalog_exception import ( + TableNotExistException, TableNoPermissionException, TagAlreadyExistException, TagNotExistException, +) +from pypaimon.common.identifier import Identifier +from pypaimon.common.json_util import JSON +from pypaimon.table.instant import Instant +from pypaimon.tests import native_rest_branch_test as branch_fixtures + +pytestmark = pytest.mark.native_plan +transport_server = branch_fixtures.transport_server +_OPERATIONS = ('create_tag', 'get_tag', 'list_tags_paged', 'delete_tag', 'load_snapshot', 'rollback_to') +_SNAPSHOT = dict(version=3, id=7, schemaId=0, baseManifestList='base', deltaManifestList='delta', + commitUser='writer', commitIdentifier=1, commitKind='APPEND', timeMillis=123, + uuid='snapshot-uuid', totalRecordCount=8, deltaRecordCount=1, properties={'source': 'test'}, + baseManifestListSize=321, deltaManifestListSize=0) + + +@pytest.fixture +def version_server(transport_server, tmp_path): + server = transport_server + server.require_v4 = True + server.uuid, server.head, server.status = 'incarnation-A', 9, None + server.tags, server.next_token = ['z', 'a', 'z'], 'next +/%一' + server.error_resource, server.error_name = None, None + server.snapshot_response = dict(snapshot=dict( + snapshot=_SNAPSHOT, recordCount=0, fileCount=2, fileSizeInBytes=123, lastFileCreationTime=456)) + schema = Schema.from_pyarrow_schema(pa.schema([('id', pa.int64())])) + + def respond(method, path, data): + if server.status is not None: + return server.status, json.dumps(dict( + resourceType=server.error_resource, resourceName=server.error_name, message='rejected')) + parsed = urlsplit(path) + if parsed.path.endswith('/rollback'): + body = json.loads(data) + if 'tableId' in body and body['tableId'] != server.uuid: + return 400, json.dumps({'message': 'table UUID mismatch'}) + if 'fromSnapshot' in body and body['fromSnapshot'] != server.head: + return 400, json.dumps({'message': 'snapshot head mismatch'}) + instant = body['instant'] + server.head = instant['snapshotId'] if instant['type'] == 'snapshot' else 7 + return 200, 'rolled back' + if parsed.path.endswith('/snapshot'): + return 200, json.dumps(server.snapshot_response) + if parsed.path.endswith('/tags'): + return (200, json.dumps(dict(tags=server.tags, nextPageToken=server.next_token))) \ + if method == 'GET' else (200, 'created') + if '/tags/' in parsed.path: + return (200, json.dumps(dict( + tagName=unquote_plus(parsed.path.rsplit('/', 1)[1]), snapshot=_SNAPSHOT, + tagCreateTime=123, tagTimeRetained='2 d'))) if method == 'GET' else (200, 'deleted') + return 200, json.dumps(dict( + id=server.uuid, name=unquote_plus(parsed.path.rsplit('/', 1)[1]), + schema=json.loads(JSON.to_json(schema)), schemaId=0, path=tmp_path.as_uri(), isExternal=False)) + + server.respond = respond + return server + + +def make_catalog(server, native=True): + return branch_fixtures.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', + 'read.native.enabled': str(native).lower(), 'write.native.enabled': str(native).lower(), + }) + + +def invoke(catalog, operation, identifier=Identifier('db /一', 'table', branch='dev /一')): + if operation == 'create_tag': + return catalog.create_tag(identifier, 'release /一', 7, '2 d') + if operation in ('get_tag', 'delete_tag'): + return getattr(catalog, operation)(identifier, 'release /一') + if operation == 'list_tags_paged': + return catalog.list_tags_paged(identifier, 3, 'next +/%一', 'prefix +/%一') + if operation == 'rollback_to': + return catalog.rollback_to(identifier, Instant.snapshot(7), 9, table_uuid='incarnation-A') + return catalog.load_snapshot(identifier) + + +@pytest.mark.parametrize('operation', _OPERATIONS) +def test_version_operations_use_rust_with_v4_and_public_python_results(version_server, operation): + server = version_server + catalog = make_catalog(server) + server.requests.clear() + server.bodies.clear() + with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python fallback')): + result = invoke(catalog, operation) + assert len(server.requests) == 1 + method, path, headers = server.requests[0] + assert headers['Authorization'].startswith('ACS4-HMAC-SHA256 ') + assert headers['x-acs-action'] == { + 'create_tag': 'CreateTag', 'get_tag': 'GetTag', 'list_tags_paged': 'ListTags', + 'delete_tag': 'DropTag', 'load_snapshot': 'GetTableSnapshot', 'rollback_to': 'RollbackToSnapshot', + }[operation] + assert '/tables/table%24branch_dev+%2F%E4%B8%80/' in path + if operation == 'create_tag': + assert server.bodies == [dict(tagName='release /一', snapshotId=7, timeRetained='2 d')] + elif operation == 'rollback_to': + assert server.bodies == [dict( + instant=dict(type='snapshot', snapshotId=7), fromSnapshot=9, tableId='incarnation-A')] + assert server.head == 7 + elif operation == 'get_tag': + assert result.tag_name == 'release /一' + assert result.snapshot.id == 7 + assert result.snapshot.uuid == 'snapshot-uuid' + assert (result.snapshot.base_manifest_list_size, result.snapshot.delta_manifest_list_size) == (321, 0) + assert result.tag_create_time == 123 and result.tag_time_retained == '2 d' + elif operation == 'list_tags_paged': + assert result.elements == ['z', 'a', 'z'] and result.next_page_token == 'next +/%一' + assert parse_qs(urlsplit(path).query) == { + 'maxResults': ['3'], 'pageToken': ['next +/%一'], 'tagNamePrefix': ['prefix +/%一']} + elif operation == 'load_snapshot': + assert result.snapshot.id == 7 and result.snapshot.uuid == 'snapshot-uuid' + assert (result.snapshot.base_manifest_list_size, result.snapshot.delta_manifest_list_size) == (321, 0) + assert (result.record_count, result.file_count, result.file_size_in_bytes, + result.last_file_creation_time) == (0, 2, 123, 456) + else: + assert result is None and method == 'DELETE' + + +@pytest.mark.parametrize('native', [False, True]) +@pytest.mark.parametrize('tags,expected_token', [(None, None), ([], 'next +/%一')]) +def test_null_tag_page_ends_paging_but_empty_page_preserves_token(version_server, native, tags, expected_token): + server = version_server + catalog = make_catalog(server, native) + server.tags = tags + server.requests.clear() + page = catalog.list_tags_paged('db.table', 0, '', ' ') + assert page.elements == [] and page.next_page_token == expected_token + assert parse_qs(urlsplit(server.requests[0][1]).query, keep_blank_values=True) == { + 'pageToken': [''], 'tagNamePrefix': [' ']} + + +@pytest.mark.parametrize('native', [False, True]) +@pytest.mark.parametrize('instant', [Instant.snapshot(7), Instant.tag('release /一')]) +def test_optional_rollback_guards_are_omitted_on_both_paths(version_server, native, instant): + server = version_server + catalog = make_catalog(server, native) + server.bodies.clear() + catalog.rollback_to('db.table', instant) + assert server.bodies == [{'instant': instant.to_dict()}] + + +@pytest.mark.parametrize('operation', _OPERATIONS) +@pytest.mark.parametrize('status', [403, 404]) +def test_version_errors_keep_public_exception_types(version_server, operation, status): + server = version_server + catalog = make_catalog(server) + server.requests.clear() + server.status, server.error_resource, server.error_name = status, 'TABLE', 'db.table' + expected = TableNoPermissionException if status == 403 else TableNotExistException + with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python replay')): + with pytest.raises(expected): + invoke(catalog, operation) + assert len(server.requests) == 1 + + +@pytest.mark.parametrize('operation,resource,expected', [ + ('create_tag', 'TAG', TagAlreadyExistException), + ('create_tag', 'SNAPSHOT', ValueError), + ('get_tag', 'TAG', TagNotExistException), + ('delete_tag', 'TAG', TagNotExistException), + ('rollback_to', 'TAG', ValueError), + ('rollback_to', 'SNAPSHOT', ValueError), +]) +def test_missing_tag_and_snapshot_errors_remain_distinct(version_server, operation, resource, expected): + server = version_server + catalog = make_catalog(server) + server.status = 409 if expected is TagAlreadyExistException else 404 + server.error_resource, server.error_name = resource, '7' if resource == 'SNAPSHOT' else 'release /一' + with pytest.raises(expected): + invoke(catalog, operation) + + +@pytest.mark.parametrize('operation', ['create_tag', 'delete_tag', 'rollback_to']) +def test_failure_after_publication_has_no_backend_replay_or_local_cleanup(version_server, operation, tmp_path): + server = version_server + catalog = make_catalog(server) + sentinel = tmp_path / 'shared-data.parquet' + sentinel.write_bytes(b'existing data') + server.status = 500 + server.requests.clear() + with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python replay')): + with pytest.raises(ServiceFailureException): + invoke(catalog, operation) + assert len(server.requests) == 1 + assert sentinel.read_bytes() == b'existing data' + + +@pytest.mark.parametrize('rejection', ['identity', 'head']) +def test_loaded_table_rollback_preserves_identity_and_never_reloads_on_rejection(version_server, rejection): + server = version_server + catalog = make_catalog(server) + table = catalog.get_table('db.table') + server.requests.clear() + server.bodies.clear() + if rejection == 'identity': + server.uuid = 'incarnation-B' + with patch.object(table, 'rollback_helper', side_effect=AssertionError('local cleanup')): + if rejection == 'identity': + with pytest.raises(RuntimeError, match='table UUID mismatch'): + table.rollback_to(7) + else: + with pytest.raises(BadRequestException, match='snapshot head mismatch'): + catalog.rollback_to('db.table', Instant.snapshot(7), 8, table_uuid='incarnation-A') + assert len(server.requests) == 1 + assert server.bodies[0]['tableId'] == 'incarnation-A' + assert server.head == 9 + + +@pytest.mark.parametrize('native', [False, True]) +@pytest.mark.parametrize('response', [None, {}, {'snapshot': None}]) +def test_empty_snapshot_responses_are_empty_views(version_server, native, response): + server = version_server + catalog = make_catalog(server, native) + server.snapshot_response = response + assert catalog.load_snapshot('db.table') is None + + +def test_empty_snapshot_error_is_not_a_missing_table(version_server): + server = version_server + catalog = make_catalog(server) + server.status, server.error_resource = 404, 'SNAPSHOT' + assert catalog.load_snapshot('db.table') is None + + +@pytest.mark.parametrize('operation', _OPERATIONS) +def test_unimplemented_version_endpoints_keep_rest_exception_and_never_replay(version_server, operation): + server = version_server + catalog = make_catalog(server) + server.status = 501 + server.requests.clear() + with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python replay')): + with pytest.raises(NotImplementedException): + invoke(catalog, operation) + assert len(server.requests) == 1 + + +@pytest.mark.parametrize('operation', _OPERATIONS) +def test_missing_native_operation_is_detected_before_execution(version_server, operation): + server = version_server + catalog = make_catalog(server) + + class UnavailableCatalog: + def __init__(self, *args, **kwargs): + raise AssertionError('Unavailable operation must be detected before constructing the catalog') + + server.requests.clear() + with patch('pypaimon_rust.datafusion.PaimonCatalog', UnavailableCatalog), \ + patch.object(catalog.rest_api, operation, wraps=getattr(catalog.rest_api, operation)) as execute: + invoke(catalog, operation) + execute.assert_called_once() + assert len(server.requests) == 1 + + +@pytest.mark.parametrize('native', [False, True]) +@pytest.mark.parametrize('snapshot_name', ['null', None]) +def test_create_tag_on_empty_table_preserves_missing_snapshot(version_server, native, snapshot_name): + server = version_server + catalog = make_catalog(server, native) + server.status, server.error_resource, server.error_name = 404, 'SNAPSHOT', snapshot_name + server.requests.clear() + with pytest.raises(ValueError, match='Snapshot'): + catalog.create_tag('db.table', 'release') + assert len(server.requests) == 1 + + +@pytest.mark.parametrize('native', [False, True]) +@pytest.mark.parametrize('operation', ['create_tag', 'get_tag', 'delete_tag']) +@pytest.mark.parametrize('resource', ['tag', 'snapshot']) +def test_resource_types_use_java_case_sensitive_names(version_server, native, operation, resource): + server = version_server + catalog = make_catalog(server, native) + server.status, server.error_resource, server.error_name = 404, resource, '7' + with pytest.raises(TableNotExistException): + invoke(catalog, operation) diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py b/paimon-python/pypaimon/tests/rest/rest_branch_test.py index ea99508568ed..73f7d3212e9d 100644 --- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py @@ -84,7 +84,7 @@ 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), \ + with patch('pypaimon.catalog.rest.native_catalog._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') diff --git a/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py b/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py index 56ac97c37bc1..2790806373d9 100644 --- a/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py +++ b/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py @@ -75,6 +75,9 @@ def catalog(transport): 'metastore': 'rest', 'uri': 'http://127.0.0.1:1', 'warehouse': 'warehouse', 'prefix': 'rollback-test', 'token.provider': 'bear', 'token': 'test-token', 'data-token.enabled': 'false', + # This suite spies on requests.Session.send. Native wire behavior is + # covered by the real HTTP Native version-management tests. + 'read.native.enabled': 'false', 'write.native.enabled': 'false', })), config_required=False) @@ -108,9 +111,9 @@ def test_request_serialization(table_uuid): assert json.loads(JSON.to_json(request)) == expected -def test_legacy_request_constructor(): +def test_unguarded_request_omits_absent_guard(): assert json.loads(JSON.to_json(RollbackTableRequest(Instant.snapshot(1), None))) == { - 'instant': {'type': 'snapshot', 'snapshotId': 1}, 'fromSnapshot': None, + 'instant': {'type': 'snapshot', 'snapshotId': 1}, } @@ -124,7 +127,7 @@ def test_table_rollback_transmits_loaded_uuid(table, transport, target, instant) assert len(transport.requests) == 1 assert transport.requests[0].method == 'POST' assert request_body(transport.requests[0]) == { - 'instant': instant, 'fromSnapshot': None, 'tableId': 'loaded-table-uuid', + 'instant': instant, 'tableId': 'loaded-table-uuid', } @@ -185,8 +188,9 @@ def test_existing_calls_and_encoded_branch_url(catalog, transport): catalog.rollback_to(identifier, Instant.snapshot(1), 2) catalog.rollback_to('db.table', Instant.tag('v1')) - assert [request_body(request)['fromSnapshot'] for request in transport.requests] == [None, 2, None] - assert all(set(request_body(request)) == {'instant', 'fromSnapshot'} for request in transport.requests) + assert [request_body(request).get('fromSnapshot') for request in transport.requests] == [None, 2, None] + assert [set(request_body(request)) for request in transport.requests] == [ + {'instant'}, {'instant', 'fromSnapshot'}, {'instant'}] path = urlsplit(transport.requests[0].url).path.split('/') assert len(path) == 8 assert unquote_plus(path[4]) == identifier.get_database_name() @@ -203,7 +207,7 @@ def test_table_rollback_without_uuid(catalog, transport): assert len(transport.requests) == 1 assert request_body(transport.requests[0]) == { - 'instant': {'type': 'snapshot', 'snapshotId': 1}, 'fromSnapshot': None, + 'instant': {'type': 'snapshot', 'snapshotId': 1}, } @@ -293,7 +297,7 @@ def test_copied_branch_rollback_omits_source_identity(catalog, transport, compac assert len(transport.requests) == 1 assert request_body(transport.requests[0]) == { 'instant': {'type': 'snapshot', 'snapshotId': 2 if automatic else 1}, - 'fromSnapshot': 3 if automatic else None, + **({'fromSnapshot': 3} if automatic else {}), } object_name = unquote_plus(urlsplit(transport.requests[0].url).path.split('/')[-2]) assert object_name == Identifier('db', 'table', branch=target).get_object_name() @@ -325,7 +329,7 @@ def test_same_branch_copy_and_independent_load_preserve_identity(table, catalog, assert len(transport.requests) == 1 assert request_body(transport.requests[0]) == { 'instant': {'type': 'snapshot', 'snapshotId': 2 if automatic else 1}, - 'fromSnapshot': 3 if automatic else None, 'tableId': 'loaded-dev-uuid', + **({'fromSnapshot': 3} if automatic else {}), 'tableId': 'loaded-dev-uuid', } object_name = unquote_plus(urlsplit(transport.requests[0].url).path.split('/')[-2]) assert object_name == Identifier('db', 'table', branch='dev').get_object_name() From da99d2f19ec5218fcde05e36c61e094b900bfb6a Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Sat, 17 Oct 2026 13:21:32 +0800 Subject: [PATCH 2/2] [python] Keep REST catalog operations on Python --- .github/workflows/ci-python.yml | 8 +- paimon-python/README.md | 27 +- paimon-python/pypaimon/api/rest_api.py | 16 +- .../pypaimon/catalog/rest/native_catalog.py | 165 ------ .../pypaimon/catalog/rest/rest_catalog.py | 33 +- .../pypaimon/tests/native_rest_branch_test.py | 548 ------------------ .../pypaimon/tests/rest/rest_branch_test.py | 3 +- .../tests/rest/rest_branch_transport_test.py | 266 +++++++++ .../rest_version_test.py} | 51 +- 9 files changed, 317 insertions(+), 800 deletions(-) delete mode 100644 paimon-python/pypaimon/catalog/rest/native_catalog.py delete mode 100644 paimon-python/pypaimon/tests/native_rest_branch_test.py create mode 100644 paimon-python/pypaimon/tests/rest/rest_branch_transport_test.py rename paimon-python/pypaimon/tests/{native_rest_version_test.py => rest/rest_version_test.py} (91%) diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 014460867ab3..b6a0fb9f81dc 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -231,14 +231,8 @@ jobs: normalize_array_for_write, read_arrow_schema, ) - from pypaimon_rust.datafusion import CatalogError, Instant, PaimonCatalog, ReadBuilder + from pypaimon_rust.datafusion import ReadBuilder assert callable(getattr(ReadBuilder, 'with_partition_filter', None)) - assert issubclass(CatalogError, ValueError) - for method in ('create_branch', 'drop_branch', 'fast_forward', 'list_branches', - 'create_tag', 'get_tag', 'list_tags_paged', 'delete_tag', - 'load_snapshot', 'rollback_to'): - assert callable(getattr(PaimonCatalog, method, None)) - assert callable(Instant.snapshot) and callable(Instant.tag) 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 0c88ad530efb..7ea5a3c127c8 100644 --- a/paimon-python/README.md +++ b/paimon-python/README.md @@ -139,27 +139,12 @@ 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. +# REST catalog management + +PyPaimon executes REST catalog operations through its Python REST client, +including branch and tag management, snapshot loading and rollback. The client +retains its authentication provider, request headers and transport settings. +Native planning, reading, writing and committing remain available for table data. # Native write and commit diff --git a/paimon-python/pypaimon/api/rest_api.py b/paimon-python/pypaimon/api/rest_api.py index de3adf18646a..8564795490ee 100755 --- a/paimon-python/pypaimon/api/rest_api.py +++ b/paimon-python/pypaimon/api/rest_api.py @@ -128,10 +128,6 @@ 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], @@ -423,7 +419,7 @@ def rollback_to(self, identifier, instant, from_snapshot=None, *, table_uuid=Non NoSuchResourceException: If the table, snapshot or tag does not exist. ForbiddenException: If no permission to access this table. """ - database_name, table_name = self.__validate_identifier(identifier) + database_name, table_name = self._validate_resource_path(identifier) request = RollbackTableRequest( instant=instant, from_snapshot=from_snapshot, table_id=table_uuid) self.client.post( @@ -441,7 +437,7 @@ def load_snapshot(self, identifier: Identifier) -> Optional['TableSnapshot']: Returns: TableSnapshot instance or None if snapshot not found. """ - database_name, table_name = self.__validate_identifier(identifier) + database_name, table_name = self._validate_resource_path(identifier) response = self.client.get( self.resource_paths.table_snapshot(database_name, table_name), GetTableSnapshotResponse, @@ -500,7 +496,7 @@ def create_tag( snapshot_id: Optional[int] = None, time_retained: Optional[str] = None, ) -> None: - database_name, table_name = self.__validate_identifier(identifier) + database_name, table_name = self._validate_resource_path(identifier) request = CreateTagRequest( tag_name=tag_name, snapshot_id=snapshot_id, @@ -513,7 +509,7 @@ def create_tag( ) def get_tag(self, identifier: Identifier, tag_name: str) -> GetTagResponse: - database_name, table_name = self.__validate_identifier(identifier) + database_name, table_name = self._validate_resource_path(identifier, tag_name) return self.client.get( self.resource_paths.tag(database_name, table_name, tag_name), GetTagResponse, @@ -527,7 +523,7 @@ def list_tags_paged( page_token: Optional[str] = None, tag_name_prefix: Optional[str] = None, ) -> PagedList[str]: - database_name, table_name = self.__validate_identifier(identifier) + database_name, table_name = self._validate_resource_path(identifier) response = self.client.get_with_params( self.resource_paths.tags(database_name, table_name), self.__build_paged_query_params( @@ -542,7 +538,7 @@ def list_tags_paged( return PagedList([], None) if tags is None else PagedList(tags, response.get_next_page_token()) def delete_tag(self, identifier: Identifier, tag_name: str) -> None: - database_name, table_name = self.__validate_identifier(identifier) + database_name, table_name = self._validate_resource_path(identifier, tag_name) self.client.delete( self.resource_paths.tag(database_name, table_name, tag_name), self.rest_auth_function, diff --git a/paimon-python/pypaimon/catalog/rest/native_catalog.py b/paimon-python/pypaimon/catalog/rest/native_catalog.py deleted file mode 100644 index a3faebf8423f..000000000000 --- a/paimon-python/pypaimon/catalog/rest/native_catalog.py +++ /dev/null @@ -1,165 +0,0 @@ -# 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. - -"""Delegate REST catalog version management and structured errors to 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) - read = operation in ('list_branches', 'get_tag', 'list_tags_paged', 'load_snapshot') - enabled = options.native_read_enabled() if read 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 - if not callable(getattr(PaimonCatalog, operation, None)): - return None - if operation == 'rollback_to': - from pypaimon_rust.datafusion import Instant - if not all(callable(getattr(Instant, name, None)) for name in ('snapshot', 'tag')): - return None - 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 catalog preparation failed: %s', error) - return None - - -def call_catalog(catalog, operation, identifier, *args, **kwargs): - # Both backends must reject a path that HTTP clients normalize to another - # resource. Names in CREATE/ROLLBACK bodies are not URL components. - path_name = args[0] if operation in ('drop_branch', 'fast_forward', 'get_tag', 'delete_tag') else None - RESTApi._validate_resource_path(identifier, path_name) - 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, **kwargs) - native, error_type = prepared - if operation == 'rollback_to': - from pypaimon.table.instant import SnapshotInstant, TagInstant - from pypaimon_rust.datafusion import Instant - instant = args[0] - if isinstance(instant, SnapshotInstant): - instant = Instant.snapshot(instant.snapshot_id) - elif isinstance(instant, TagInstant): - instant = Instant.tag(instant.tag_name) - else: - raise TypeError('Rollback requires a snapshot or tag instant') - args = (instant,) + args[1:] - try: - # Once an operation starts, never execute it again through Python: - # the REST server may have published it before returning an error. - result = getattr(native, operation)( - (identifier.get_database_name(), identifier.get_object_name()), *args, **kwargs) - 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) - if operation in ('get_tag', 'load_snapshot') and result is not None: - from pypaimon.api.api_response import GetTagResponse - from pypaimon.snapshot.table_snapshot import TableSnapshot - from pypaimon.common.json_util import JSON - result = JSON.from_json(result.to_json(), GetTagResponse if operation == 'get_tag' else TableSnapshot) - elif operation == 'list_tags_paged': - from pypaimon.api.api_response import PagedList - result = PagedList(*result) - return result diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py b/paimon-python/pypaimon/catalog/rest/rest_catalog.py index ec6c501b8d7a..a7069d534540 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py +++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py @@ -430,8 +430,8 @@ def rollback_to(self, identifier: Union[str, Identifier], instant: Instant, if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - from pypaimon.catalog.rest.native_catalog import call_catalog - call_catalog(self, 'rollback_to', identifier, instant, from_snapshot, table_uuid=table_uuid) + self.rest_api.rollback_to( + identifier, instant, from_snapshot, table_uuid=table_uuid) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_SNAPSHOT: raise ValueError( @@ -459,8 +459,7 @@ def load_snapshot(self, identifier: Union[str, Identifier]) -> Optional[TableSna if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - from pypaimon.catalog.rest.native_catalog import call_catalog - return call_catalog(self, 'load_snapshot', identifier) + return self.rest_api.load_snapshot(identifier) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_SNAPSHOT: return None @@ -575,9 +574,7 @@ def create_tag(self, identifier: Union[str, Identifier], tag_name: str, if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - from pypaimon.catalog.rest.native_catalog import call_catalog - # Ignore-if-exists remains in the public catalog error handling below. - call_catalog(self, 'create_tag', identifier, tag_name, snapshot_id, time_retained) + self.rest_api.create_tag(identifier, tag_name, snapshot_id, time_retained) except AlreadyExistsException as e: if not ignore_if_exists: raise TagAlreadyExistException(tag_name) from e @@ -596,8 +593,7 @@ def get_tag(self, identifier: Union[str, Identifier], tag_name: str) -> GetTagRe if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - from pypaimon.catalog.rest.native_catalog import call_catalog - return call_catalog(self, 'get_tag', identifier, tag_name) + return self.rest_api.get_tag(identifier, tag_name) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_TAG: raise TagNotExistException(tag_name) from e @@ -612,8 +608,8 @@ def list_tags_paged(self, identifier: Union[str, Identifier], if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - from pypaimon.catalog.rest.native_catalog import call_catalog - return call_catalog(self, 'list_tags_paged', identifier, max_results, page_token, tag_name_prefix) + return self.rest_api.list_tags_paged( + identifier, max_results, page_token, tag_name_prefix) except NoSuchResourceException as e: raise TableNotExistException(identifier) from e except ForbiddenException as e: @@ -623,8 +619,7 @@ def delete_tag(self, identifier: Union[str, Identifier], tag_name: str) -> None: if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - from pypaimon.catalog.rest.native_catalog import call_catalog - call_catalog(self, 'delete_tag', identifier, tag_name) + self.rest_api.delete_tag(identifier, tag_name) except NoSuchResourceException as e: if e.resource_type == ErrorResponse.RESOURCE_TYPE_TAG: raise TagNotExistException(tag_name) from e @@ -653,11 +648,10 @@ 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_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - call_catalog(self, 'create_branch', identifier, branch_name, tag_name) + self.rest_api.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 @@ -672,11 +666,10 @@ 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_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - call_catalog(self, 'drop_branch', identifier, branch_name) + self.rest_api.drop_branch(identifier, branch_name) except NoSuchResourceException as e: raise BranchNotExistException(branch_name) from e except ForbiddenException as e: @@ -687,22 +680,20 @@ def rename_branch(self, identifier: Union[str, Identifier], from_branch: str, 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_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - call_catalog(self, 'fast_forward', identifier, branch_name) + self.rest_api.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_catalog import call_catalog if not isinstance(identifier, Identifier): identifier = Identifier.from_string(identifier) try: - return call_catalog(self, 'list_branches', identifier) + return self.rest_api.list_branches(identifier) except NoSuchResourceException as e: raise TableNotExistException(identifier) from e except ForbiddenException as e: diff --git a/paimon-python/pypaimon/tests/native_rest_branch_test.py b/paimon-python/pypaimon/tests/native_rest_branch_test.py deleted file mode 100644 index 57943c0e0594..000000000000 --- a/paimon-python/pypaimon/tests/native_rest_branch_test.py +++ /dev/null @@ -1,548 +0,0 @@ -# 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') - - -@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)) - self.server.bodies.append(json.loads(data) if data else None) - 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 - elif self.server.respond is not None: - status, body = self.server.respond(self.command, self.path, data) - 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, keep_blank_values=True))), - 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.bodies = [] - server.respond = None - 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)) - - -@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_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('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({ - '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_catalog._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_catalog._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/rest_branch_test.py b/paimon-python/pypaimon/tests/rest/rest_branch_test.py index 73f7d3212e9d..cebfcd49b5a7 100644 --- a/paimon-python/pypaimon/tests/rest/rest_branch_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_branch_test.py @@ -84,8 +84,7 @@ 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_catalog._native_catalog', return_value=None), \ - patch.object(self.rest_catalog.rest_api, 'create_branch', side_effect=error): + with 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) diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_transport_test.py b/paimon-python/pypaimon/tests/rest/rest_branch_transport_test.py new file mode 100644 index 000000000000..304979fd5373 --- /dev/null +++ b/paimon-python/pypaimon/tests/rest/rest_branch_transport_test.py @@ -0,0 +1,266 @@ +# 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 catalog metadata stays on Python even when Native table execution is enabled.""" + +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, HTTPServer +import builtins +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.api.api_response import ErrorResponse +from pypaimon.api.auth.dlf_signer import DLFOpenApiV4Signer +from pypaimon.api.rest_exception import ServiceFailureException +from pypaimon.api.token_loader import DLFToken +from pypaimon.api.typedef import RESTAuthParameter +from pypaimon.catalog.catalog_context import CatalogContext +from pypaimon.catalog.rest.rest_catalog import RESTCatalog +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 + + +@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)) + self.server.bodies.append(json.loads(data) if data else None) + 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 + elif self.server.respond is not None: + status, body = self.server.respond(self.command, self.path, data) + 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, keep_blank_values=True))), + 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.bodies = [] + server.respond = None + 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)) + + +@contextmanager +def python_rest_only(): + original_import = builtins.__import__ + + def checked_import(name, *args, **kwargs): + if name == 'pypaimon_rust' or name.startswith('pypaimon_rust.'): + raise AssertionError('REST catalog operations must use the Python client') + return original_import(name, *args, **kwargs) + + with patch('builtins.__import__', side_effect=checked_import): + yield + + +@pytest.mark.parametrize('operation', ['create_branch', 'drop_branch', 'fast_forward', 'list_branches']) +def test_branch_operations_use_python_with_native_options_enabled(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', + }) + args = () if operation == 'list_branches' else ('dev 一/?#',) + server.requests.clear() + with python_rest_only(), \ + patch.object(catalog.rest_api, operation, wraps=getattr(catalog.rest_api, operation)) as execute: + result = getattr(catalog, operation)(Identifier('db 一/space', 'table'), *args) + execute.assert_called_once() + assert result == (['dev'] if operation == 'list_branches' else None) + assert len(server.requests) == 1 + assert server.requests[0][2]['x-acs-action'] == { + 'create_branch': 'CreateBranch', 'drop_branch': 'DropBranch', + 'fast_forward': 'FastForwardBranch', 'list_branches': 'ListBranches', + }[operation] + + +@pytest.mark.parametrize('operation', ['create_branch', 'drop_branch', 'fast_forward']) +def test_failed_branch_mutations_are_not_replayed(transport_server, operation): + server = transport_server + catalog = transport_catalog(server) + server.respond = lambda *args: (500, '{"message":"uncertain publication"}') + server.requests.clear() + with python_rest_only(), pytest.raises(ServiceFailureException): + getattr(catalog, operation)('db.table', 'dev') + assert len(server.requests) == 1 + + +@pytest.mark.parametrize('status', [429, 503]) +def test_branch_listing_retries_temporary_status(transport_server, status): + catalog = transport_catalog(transport_server) + transport_server.requests.clear() + transport_server.temporary_status = status + with python_rest_only(): + assert catalog.list_branches('db.table') == ['dev'] + assert len(transport_server.requests) == 2 + assert all(request[0] == 'GET' for request in transport_server.requests) + + +@pytest.mark.parametrize('operation', ['create_branch', 'drop_branch', 'fast_forward', 'list_branches']) +@pytest.mark.parametrize('name', ['.', '..', '', ' ']) +def test_invalid_branch_resource_paths_send_no_requests(transport_server, operation, name): + catalog = transport_catalog(transport_server) + transport_server.requests.clear() + args = () if operation == 'list_branches' else ('dev',) + for identifier in (Identifier(name, 'table'), Identifier('db', name)): + with pytest.raises(ValueError): + getattr(catalog, operation)(identifier, *args) + if operation in ('drop_branch', 'fast_forward'): + with pytest.raises(ValueError): + getattr(catalog, operation)('db.table', name) + assert transport_server.requests == [] + + +@pytest.mark.native_plan +def test_python_rest_branches_preserve_native_shared_data(tmp_path, native_rest_catalog): + catalog = native_rest_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]) + 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]) + + +@pytest.mark.native_plan +def test_published_branch_survives_uncertain_response(tmp_path, native_rest_catalog): + catalog = native_rest_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): + with pytest.raises(ServiceFailureException): + catalog.create_branch(table.identifier, 'dev') + assert catalog.list_branches(table.identifier) == ['dev'] + assert published == [table.identifier] + + +@pytest.mark.native_plan +def test_python_rest_branch_drop_preserves_native_blob_files(tmp_path, native_rest_catalog): + catalog = native_rest_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') + 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/native_rest_version_test.py b/paimon-python/pypaimon/tests/rest/rest_version_test.py similarity index 91% rename from paimon-python/pypaimon/tests/native_rest_version_test.py rename to paimon-python/pypaimon/tests/rest/rest_version_test.py index 0d004a08e3e1..19bff6433f54 100644 --- a/paimon-python/pypaimon/tests/native_rest_version_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_version_test.py @@ -15,7 +15,7 @@ # specific language governing permissions and limitations # under the License. -"""Existing REST version-management behavior through Rust and Python transports.""" +"""REST version management uses the Python client and Java wire semantics.""" import json from unittest.mock import patch @@ -32,9 +32,8 @@ from pypaimon.common.identifier import Identifier from pypaimon.common.json_util import JSON from pypaimon.table.instant import Instant -from pypaimon.tests import native_rest_branch_test as branch_fixtures +from pypaimon.tests.rest import rest_branch_transport_test as branch_fixtures -pytestmark = pytest.mark.native_plan transport_server = branch_fixtures.transport_server _OPERATIONS = ('create_tag', 'get_tag', 'list_tags_paged', 'delete_tag', 'load_snapshot', 'rollback_to') _SNAPSHOT = dict(version=3, id=7, schemaId=0, baseManifestList='base', deltaManifestList='delta', @@ -106,13 +105,15 @@ def invoke(catalog, operation, identifier=Identifier('db /一', 'table', branch= @pytest.mark.parametrize('operation', _OPERATIONS) -def test_version_operations_use_rust_with_v4_and_public_python_results(version_server, operation): +def test_version_operations_use_python_with_v4_and_native_options(version_server, operation): server = version_server catalog = make_catalog(server) server.requests.clear() server.bodies.clear() - with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python fallback')): + with branch_fixtures.python_rest_only(), \ + patch.object(catalog.rest_api, operation, wraps=getattr(catalog.rest_api, operation)) as execute: result = invoke(catalog, operation) + execute.assert_called_once() assert len(server.requests) == 1 method, path, headers = server.requests[0] assert headers['Authorization'].startswith('ACS4-HMAC-SHA256 ') @@ -177,7 +178,7 @@ def test_version_errors_keep_public_exception_types(version_server, operation, s server.requests.clear() server.status, server.error_resource, server.error_name = status, 'TABLE', 'db.table' expected = TableNoPermissionException if status == 403 else TableNotExistException - with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python replay')): + with branch_fixtures.python_rest_only(): with pytest.raises(expected): invoke(catalog, operation) assert len(server.requests) == 1 @@ -201,14 +202,14 @@ def test_missing_tag_and_snapshot_errors_remain_distinct(version_server, operati @pytest.mark.parametrize('operation', ['create_tag', 'delete_tag', 'rollback_to']) -def test_failure_after_publication_has_no_backend_replay_or_local_cleanup(version_server, operation, tmp_path): +def test_failure_after_publication_has_no_request_replay_or_local_cleanup(version_server, operation, tmp_path): server = version_server catalog = make_catalog(server) sentinel = tmp_path / 'shared-data.parquet' sentinel.write_bytes(b'existing data') server.status = 500 server.requests.clear() - with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python replay')): + with branch_fixtures.python_rest_only(): with pytest.raises(ServiceFailureException): invoke(catalog, operation) assert len(server.requests) == 1 @@ -258,29 +259,12 @@ def test_unimplemented_version_endpoints_keep_rest_exception_and_never_replay(ve catalog = make_catalog(server) server.status = 501 server.requests.clear() - with patch.object(catalog.rest_api, operation, side_effect=AssertionError('Python replay')): + with branch_fixtures.python_rest_only(): with pytest.raises(NotImplementedException): invoke(catalog, operation) assert len(server.requests) == 1 -@pytest.mark.parametrize('operation', _OPERATIONS) -def test_missing_native_operation_is_detected_before_execution(version_server, operation): - server = version_server - catalog = make_catalog(server) - - class UnavailableCatalog: - def __init__(self, *args, **kwargs): - raise AssertionError('Unavailable operation must be detected before constructing the catalog') - - server.requests.clear() - with patch('pypaimon_rust.datafusion.PaimonCatalog', UnavailableCatalog), \ - patch.object(catalog.rest_api, operation, wraps=getattr(catalog.rest_api, operation)) as execute: - invoke(catalog, operation) - execute.assert_called_once() - assert len(server.requests) == 1 - - @pytest.mark.parametrize('native', [False, True]) @pytest.mark.parametrize('snapshot_name', ['null', None]) def test_create_tag_on_empty_table_preserves_missing_snapshot(version_server, native, snapshot_name): @@ -302,3 +286,18 @@ def test_resource_types_use_java_case_sensitive_names(version_server, native, op server.status, server.error_resource, server.error_name = 404, resource, '7' with pytest.raises(TableNotExistException): invoke(catalog, operation) + + +@pytest.mark.parametrize('operation', _OPERATIONS) +def test_invalid_version_resource_paths_send_no_requests(version_server, operation): + catalog = make_catalog(version_server) + version_server.requests.clear() + with branch_fixtures.python_rest_only(): + for identifier in (Identifier('..', 'table'), Identifier('db', '.')): + with pytest.raises(ValueError): + invoke(catalog, operation, identifier) + if operation in ('get_tag', 'delete_tag'): + for name in ('.', '..'): + with pytest.raises(ValueError): + getattr(catalog, operation)('db.table', name) + assert version_server.requests == []