This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 220f0fe5fc [python] Keep REST catalog operations on Python (#10622)
220f0fe5fc is described below
commit 220f0fe5fc8fde017adf2e3f23e6075707b6c3c5
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 17 13:32:36 2026 +0800
[python] Keep REST catalog operations on Python (#10622)
---
.github/workflows/ci-python.yml | 5 +-
paimon-python/README.md | 27 +-
paimon-python/pypaimon/api/api_request.py | 2 +-
paimon-python/pypaimon/api/rest_api.py | 38 +-
.../pypaimon/catalog/rest/native_branch.py | 137 ------
.../pypaimon/catalog/rest/rest_catalog.py | 12 +-
paimon-python/pypaimon/common/json_util.py | 2 +
.../pypaimon/tests/native_rest_branch_test.py | 543 ---------------------
.../pypaimon/tests/rest/rest_branch_test.py | 3 +-
.../tests/rest/rest_branch_transport_test.py | 266 ++++++++++
.../pypaimon/tests/rest/rest_version_test.py | 303 ++++++++++++
.../pypaimon/tests/rest/rollback_protocol_test.py | 20 +-
12 files changed, 613 insertions(+), 745 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 00778ca3fc..b6a0fb9f81 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -231,11 +231,8 @@ jobs:
normalize_array_for_write,
read_arrow_schema,
)
- from pypaimon_rust.datafusion import CatalogError, 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'):
- assert callable(getattr(PaimonCatalog, method, None))
PY
- name: Run Python tests with native planning, reading, writing,
updating and committing
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 0c88ad530e..7ea5a3c127 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/api_request.py
b/paimon-python/pypaimon/api/api_request.py
index c03b75319f..6ff6e08934 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 b8148a1dfe..8564795490 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -128,10 +128,6 @@ class RESTApi:
self.options = options
self.resource_paths = ResourcePaths.for_catalog_properties(options)
- @staticmethod
- def _is_auth_option(key):
- return key in ('uri', 'token.provider', 'token') or
key.startswith('dlf.')
-
def __build_paged_query_params(
self,
max_results: Optional[int],
@@ -142,11 +138,11 @@ class RESTApi:
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
@@ -423,7 +419,7 @@ class RESTApi:
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 @@ class RESTApi:
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 @@ class RESTApi:
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 @@ class RESTApi:
)
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 @@ class RESTApi:
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(
@@ -538,11 +534,11 @@ class RESTApi:
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)
+ 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,
@@ -550,11 +546,11 @@ class RESTApi:
# 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 +562,7 @@ class RESTApi:
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 +573,14 @@ class RESTApi:
)
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 +588,7 @@ class RESTApi:
)
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_branch.py
deleted file mode 100644
index 4d46b19f18..0000000000
--- a/paimon-python/pypaimon/catalog/rest/native_branch.py
+++ /dev/null
@@ -1,137 +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.
-
-"""Transport REST branch operations and structured errors to/from Rust core."""
-
-import logging
-import os
-
-import requests
-from requests.structures import CaseInsensitiveDict
-
-from pypaimon.api.api_response import ErrorResponse
-from pypaimon.api.auth import RESTAuthFunction
-from pypaimon.api.auth.bearer import BearTokenAuthProvider
-from pypaimon.api.auth.dlf_provider import DLFAuthProvider
-from pypaimon.api.client import DefaultErrorHandler, HttpClient
-from pypaimon.api.rest_api import RESTApi
-from pypaimon.catalog.catalog_context import CatalogContext
-from pypaimon.common.options.core_options import CoreOptions
-
-logger = logging.getLogger(__name__)
-
-
-def _native_headers(api):
- session = api.client.session
- if (type(session) is not requests.Session or session.auth is not None
- or session.cert is not None or session.verify is not True or
session.proxies or session.params
- or not session.trust_env or any(session.hooks.values())
- or set(session.adapters) != {'http://', 'https://'}
- or any(type(adapter) is not requests.adapters.HTTPAdapter for
adapter in session.adapters.values())
- or api.client._timeout != (180, 180)
- or os.environ.get('REQUESTS_CA_BUNDLE') or
os.environ.get('CURL_CA_BUNDLE')):
- # Keep custom transport credentials and policies in their owning
client.
- return None
- headers = CaseInsensitiveDict(session.headers)
- # Let each HTTP transport advertise only the codecs it can decode and
manage
- # its own connections. An explicitly customized transport header stays
Python.
- for name in ('Accept-Encoding', 'Connection'):
- if (headers.get(name) != requests.utils.default_headers().get(name)
- or any(key.lower() == name.lower() for key in
api.rest_auth_function.init_header)):
- return None
- headers.pop(name, None)
- headers.update(api.rest_auth_function.init_header)
- return {key: value for key, value in headers.items() if value is not None}
-
-
-def _native_catalog(catalog, operation):
- from pypaimon.catalog.rest.rest_catalog import RESTCatalog
-
- if (type(catalog) is not RESTCatalog or type(catalog.context) is not
CatalogContext
- or type(catalog.rest_api) is not RESTApi
- or any(getattr(catalog.context, attr, None) is not None for attr
in (
- 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader'))):
- return None
- options = CoreOptions(catalog.context.options)
- enabled = options.native_read_enabled() if operation == 'list_branches'
else options.native_write_enabled()
- if not enabled:
- return None
- api = catalog.rest_api
- if (type(api.client) is not HttpClient
- or type(api.client.error_handler) is not DefaultErrorHandler
- or type(api.rest_auth_function) is not RESTAuthFunction
- or type(api.rest_auth_function.auth_provider) not in
(BearTokenAuthProvider, DLFAuthProvider)):
- return None
- headers = _native_headers(api)
- if headers is None:
- return None
- try:
- from pypaimon_rust.datafusion import CatalogError, PaimonCatalog
- except ImportError:
- return None
- from pypaimon.read.native_plan import _option_value_to_string,
_raise_if_native_fork_safety_error
- resolved = {key: value for key, value in api.options.to_map().items()
- if not RESTApi._is_auth_option(key) and not
key.startswith(RESTApi.HEADER_PREFIX)}
- resolved['uri'] = api.client.uri
- resolved.update({RESTApi.HEADER_PREFIX + key: value for key, value in
headers.items()})
- resolved = {str(key): _option_value_to_string(value)
- for key, value in resolved.items() if value is not None}
- resolved['metastore'] = 'rest'
- try:
- provider = api.rest_auth_function.auth_provider
- if type(provider) is DLFAuthProvider:
- # The owning provider retains Java's STS cache/refresh policy. Pass
- # its current credentials for this call; Rust still signs the
request.
- token = provider.get_token()
- resolved.update({
- 'token.provider': 'dlf', 'dlf.region': provider.region,
- 'dlf.signing-algorithm': provider.signing_algorithm,
- 'dlf.access-key-id': token.access_key_id,
- 'dlf.access-key-secret': token.access_key_secret,
- })
- if token.security_token is not None:
- resolved['dlf.security-token'] = token.security_token
- else:
- resolved.update({'token.provider': 'bear', 'token':
provider.token})
- # Reuse the resolved configuration, as RESTCatalogLoader does. Keep no
- # opaque native state on the Python catalog across pickle or process
fork.
- return PaimonCatalog(resolved, config_required=False), CatalogError
- except Exception as error:
- _raise_if_native_fork_safety_error(error)
- logger.debug('Native branch preparation failed: %s', error)
- return None
-
-
-def call_branch(catalog, operation, identifier, *args):
- # Both backends must reject a path that HTTP clients normalize to another
- # resource. CREATE's branch is in JSON; only DROP/FORWARD put it in the
URL.
- path_branch = args[0] if operation in ('drop_branch', 'fast_forward') else
None
- RESTApi._validate_branch_path(identifier, path_branch)
- if operation == 'create_branch' and (not args[0] or not args[0].strip()):
- raise ValueError('Branch name cannot be empty')
- prepared = _native_catalog(catalog, operation)
- if prepared is None:
- return getattr(catalog.rest_api, operation)(identifier, *args)
- native, error_type = prepared
- try:
- # Once an operation starts, never execute it again through Python:
- # the REST server may have published it before returning an error.
- return getattr(native, operation)((identifier.get_database_name(),
identifier.get_object_name()), *args)
- except error_type as error:
- DefaultErrorHandler.get_instance().accept(ErrorResponse(
- resource_type=error.resource_type,
resource_name=error.resource_name,
- message=error.message, code=error.code),
HttpClient.DEFAULT_REQUEST_ID)
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 6ae9e6fd24..a7069d5345 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -648,11 +648,10 @@ class RESTCatalog(Catalog):
# Branch CRUD: mirrors Java RESTCatalog branch handlers.
def create_branch(self, identifier: Union[str, Identifier], branch_name:
str,
tag_name: Optional[str] = None) -> None:
- from pypaimon.catalog.rest.native_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- call_branch(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
@@ -667,11 +666,10 @@ class RESTCatalog(Catalog):
raise IllegalArgumentError(str(e)) from e
def drop_branch(self, identifier: Union[str, Identifier], branch_name:
str) -> None:
- from pypaimon.catalog.rest.native_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- call_branch(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:
@@ -682,22 +680,20 @@ class RESTCatalog(Catalog):
raise NotImplementedError('REST branch rename is not supported,
matching Java RESTCatalog.')
def fast_forward(self, identifier: Union[str, Identifier], branch_name:
str) -> None:
- from pypaimon.catalog.rest.native_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- call_branch(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_branch import call_branch
if not isinstance(identifier, Identifier):
identifier = Identifier.from_string(identifier)
try:
- return call_branch(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/common/json_util.py
b/paimon-python/pypaimon/common/json_util.py
index df9c95427c..15e8d61632 100644
--- a/paimon-python/pypaimon/common/json_util.py
+++ b/paimon-python/pypaimon/common/json_util.py
@@ -70,6 +70,8 @@ class JSON:
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
deleted file mode 100644
index 8782030d93..0000000000
--- a/paimon-python/pypaimon/tests/native_rest_branch_test.py
+++ /dev/null
@@ -1,543 +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')
-
-
[email protected]
-def transport_server():
- class Handler(BaseHTTPRequestHandler):
- def operation(self):
- size = int(self.headers.get('Content-Length', 0))
- data = self.rfile.read(size).decode() if size else None
- self.server.requests.append((self.command, self.path,
self.headers))
- status = 200
- if self.path.startswith('/v1/config'):
- body = json.dumps({'defaults': {'prefix': 'test'}})
- elif self.server.require_v4 and not
self.headers.get('Authorization', '').startswith('ACS4-HMAC-SHA256 '):
- status, body = 403, json.dumps({'message': 'wrong signature
algorithm'})
- elif self.server.temporary_status and self.server.branch_requests
== 0:
- status = self.server.temporary_status
- body = json.dumps({'message': 'temporary failure'})
- self.server.branch_requests += 1
- else:
- body = json.dumps({'branches': ['dev']}) if self.command ==
'GET' else 'OK'
- if self.server.require_v4:
- signed_headers = {key.lower(): value for key, value in
self.headers.items()
- if key.lower().startswith('x-acs-') or
key.lower() in ('host', 'content-type')}
- url = urlsplit(self.path)
- expected = DLFOpenApiV4Signer('cn-hangzhou').authorization(
- RESTAuthParameter(self.command, url.path, data,
dict(parse_qsl(url.query))),
- self.server.expected_token,
- self.headers['Host'], signed_headers)
- if self.headers.get('Authorization') != expected:
- status, body = 403, json.dumps({'message': 'signature
mismatch'})
- output = body.encode()
- self.send_response(status)
- self.send_header('Content-Length', str(len(output)))
- self.end_headers()
- self.wfile.write(output)
-
- do_GET = do_POST = do_DELETE = operation
-
- def log_message(self, *args):
- pass
-
- server = HTTPServer(('localhost', 0), Handler)
- server.requests = []
- server.require_v4 = False
- server.temporary_status = server.branch_requests = 0
- server.expected_token = DLFToken('test-key', 'test-secret', 'test-sts',
None)
- thread = threading.Thread(target=server.serve_forever)
- thread.start()
- try:
- yield server
- finally:
- server.shutdown()
- server.server_close()
- thread.join()
-
-
-def transport_catalog(server, **extra):
- options = {
- 'metastore': 'rest', 'uri': 'http://localhost:%d' %
server.server_port, 'warehouse': 'w',
- 'token.provider': 'bear', 'token': 'test-token',
- 'read.native.enabled': 'true', 'write.native.enabled': 'true',
- }
- options.update(extra)
- return RESTCatalog(CatalogContext.create_from_options(options))
-
-
[email protected]('operation', _OPERATIONS)
-def test_openapi_v4_uses_native_signer(transport_server, operation):
- server = transport_server
- server.require_v4 = True
- catalog = transport_catalog(server, **{
- 'token.provider': 'dlf', 'dlf.region': 'cn-hangzhou',
'dlf.signing-algorithm': 'openapi-v4',
- 'dlf.access-key-id': 'test-key', 'dlf.access-key-secret':
'test-secret',
- 'dlf.security-token': 'test-sts',
- })
- catalog.rest_api.client.session.headers.update({
- 'HOST': 'stale.example', 'X-ACS-DATE': 'stale', 'authorization':
'stale',
- })
- if operation in ('create_branch', 'fast_forward'):
- catalog.rest_api.client.session.headers['Content-Type'] = 'text/plain'
- server.requests.clear()
- args = () if operation == 'list_branches' else ('dev 一/?#',)
- if operation == 'create_branch':
- args += ('源 一',)
- with native_only(catalog):
- result = getattr(catalog, operation)(Identifier('db 一/space', 'table
?$branch_dev'), *args)
- assert result == (['dev'] if operation == 'list_branches' else None)
- assert len(server.requests) == 1
- headers = server.requests[0][2]
- assert headers['Authorization'].startswith('ACS4-HMAC-SHA256 ')
- for name in ('Authorization', 'Host', 'x-acs-date'):
- assert len(headers.get_all(name)) == 1
- if operation in ('create_branch', 'fast_forward'):
- assert headers.get_all('Content-Type') == ['application/json']
- assert headers['x-acs-security-token'] == 'test-sts'
- assert headers['x-acs-action'] == {
- 'create_branch': 'CreateBranch', 'drop_branch': 'DropBranch',
- 'fast_forward': 'FastForwardBranch', 'list_branches': 'ListBranches',
- }[operation]
-
-
[email protected]('operation', _OPERATIONS)
-def test_native_dlf_reuses_cached_credentials(transport_server, tmp_path,
operation):
- server = transport_server
- server.require_v4 = True
- token_file = tmp_path / 'credentials.json'
- token_file.write_text(json.dumps({
- 'AccessKeyId': 'test-key', 'AccessKeySecret': 'test-secret',
- 'SecurityToken': 'test-sts', 'Expiration': '2099-01-01T00:00:00Z',
- }))
- catalog = transport_catalog(server, **{
- 'token.provider': 'dlf', 'dlf.region': 'cn-hangzhou',
'dlf.signing-algorithm': 'openapi-v4',
- 'dlf.token-loader': 'local_file', 'dlf.token-path': str(token_file),
- })
- provider = catalog.rest_api.rest_auth_function.auth_provider
- assert provider.token is not None # /config has loaded a valid cached STS.
- token_file.unlink()
- server.requests.clear()
- args = () if operation == 'list_branches' else ('dev',)
- with patch.object(provider.token_loader, 'load_token',
side_effect=AssertionError('unnecessary refresh')), \
- native_only(catalog):
- result = getattr(catalog, operation)('db.t', *args)
- assert result == (['dev'] if operation == 'list_branches' else None)
- assert len(server.requests) == 1
- assert server.requests[0][2]['x-acs-security-token'] == 'test-sts'
-
-
-def test_native_dlf_refreshes_through_owning_provider(transport_server,
tmp_path):
- server = transport_server
- server.require_v4 = True
- token_file = tmp_path / 'credentials.json'
-
- def write_token(sts):
- token_file.write_text(json.dumps({
- 'AccessKeyId': 'test-key', 'AccessKeySecret': 'test-secret',
- 'SecurityToken': sts, 'Expiration': '2099-01-01T00:00:00Z',
- }))
-
- write_token('test-sts')
- catalog = transport_catalog(server, **{
- 'token.provider': 'dlf', 'dlf.region': 'cn-hangzhou',
'dlf.signing-algorithm': 'openapi-v4',
- 'dlf.token-loader': 'local_file', 'dlf.token-path': str(token_file),
- })
- provider = catalog.rest_api.rest_auth_function.auth_provider
- provider.token.expiration_at_millis = 0
- write_token('refreshed-sts')
- server.expected_token = DLFToken('test-key', 'test-secret',
'refreshed-sts', None)
- server.requests.clear()
- with patch.object(provider.signer, 'authorization',
side_effect=AssertionError('Python signing')), \
- patch.object(provider.token_loader, 'load_token',
wraps=provider.token_loader.load_token) as load, \
- native_only(catalog):
- assert catalog.list_branches('db.t') == ['dev']
- token_file.unlink()
- assert catalog.list_branches('db.t') == ['dev']
- assert load.call_count == 1
- assert len(server.requests) == 2
- assert all(request[2]['x-acs-security-token'] == 'refreshed-sts' for
request in server.requests)
-
-
[email protected]('customization', ['hook', 'adapter', 'trust_env',
'timeout', 'ca_bundle'])
-def test_custom_session_keeps_owning_transport(transport_server, monkeypatch,
customization):
- from pypaimon_rust.datafusion import PaimonCatalog
-
- catalog = transport_catalog(transport_server)
- client = catalog.rest_api.client
- called = []
-
- def hook(response, **kwargs):
- called.append('hook')
-
- class Adapter(requests.adapters.HTTPAdapter):
- def send(self, request, **kwargs):
- called.append('adapter')
- return super().send(request, **kwargs)
-
- if customization == 'hook':
- client.session.hooks['response'].append(hook)
- elif customization == 'adapter':
- client.session.mount('http://', Adapter())
- elif customization == 'trust_env':
- client.session.trust_env = False
- elif customization == 'timeout':
- client._timeout = (12, 34)
- else:
- monkeypatch.setenv('REQUESTS_CA_BUNDLE', '/custom/ca.pem')
- with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog)
as construct:
- assert catalog.list_branches('db.t') == ['dev']
- construct.assert_not_called()
- assert called == ([customization] if customization in ('hook', 'adapter')
else [])
-
-
[email protected]('agent_header', ['User-Agent', 'user-agent',
'USER-AGENT'])
-def
test_native_preserves_session_headers_and_one_explicit_user_agent(transport_server,
agent_header):
- catalog = transport_catalog(transport_server, **{'header.' + agent_header:
'explicit-agent'})
- catalog.rest_api.client.session.headers['X-Tenant'] = 'tenant-a'
- catalog.rest_api.client.session.headers['authorization'] = 'Bearer stale'
- catalog.rest_api.client.session.headers['x-priority'] = 'session'
- catalog.rest_api.rest_auth_function.init_header['X-Priority'] = 'request'
- transport_server.requests.clear()
- with native_only(catalog):
- assert catalog.list_branches('db.t') == ['dev']
- assert len(transport_server.requests) == 1
- headers = transport_server.requests[0][2]
- assert headers['X-Tenant'] == 'tenant-a'
- assert headers.get_all('X-Priority') == ['request']
- assert headers.get_all('User-Agent') == ['explicit-agent']
- assert headers.get_all('Authorization') == ['Bearer test-token']
-
-
[email protected]('operation', ['create_branch', 'drop_branch',
'fast_forward'])
-def test_native_mutations_accept_plain_text_success(transport_server,
operation):
- catalog = transport_catalog(transport_server)
- transport_server.requests.clear()
- with native_only(catalog):
- assert getattr(catalog, operation)('db.t', 'dev') is None
- assert len(transport_server.requests) == 1
-
-
[email protected]('status', [429, 503])
-def test_native_branch_listing_retries_temporary_status(transport_server,
status):
- catalog = transport_catalog(transport_server)
- transport_server.requests.clear()
- transport_server.temporary_status = status
- with native_only(catalog):
- assert catalog.list_branches('db.t') == ['dev']
- assert len(transport_server.requests) == 2
- assert all(request[0] == 'GET' for request in transport_server.requests)
-
-
[email protected]
-def catalog(native_rest_catalog):
- native_rest_catalog.context.options.to_map().update({
- 'read.native.enabled': 'true', 'write.native.enabled': 'true',
- })
- return native_rest_catalog
-
-
-@contextmanager
-def native_only(catalog):
- with ExitStack() as guards:
- for name in _OPERATIONS:
- guards.enter_context(patch.object(
- catalog.rest_api, name, side_effect=AssertionError('Python
branch fallback: ' + name)))
- yield
-
-
-def test_native_rest_branch_lifecycle_with_shared_data(tmp_path, catalog):
- catalog, main = _table(tmp_path, catalog, True)
- seed = dict(id=1, value=10, p='a')
- added = dict(id=2, value=20, p='b')
- _write(main, [seed])
- with native_only(catalog):
- empty = _branch(catalog, main, from_tag=False, name='empty')
- _assert_rows(empty, [])
- branch = _branch(catalog, main)
- _write(branch, [added])
- _assert_rows(main, [seed])
- _assert_rows(branch, [seed, added])
- assert catalog.list_branches(main.identifier) == ['dev', 'empty']
- paths = {path for path in tmp_path.rglob('*.parquet')}
- assert paths
- catalog.fast_forward(main.identifier, 'dev')
- catalog.drop_branch(main.identifier, 'dev')
- catalog.drop_branch(main.identifier, 'empty')
- assert catalog.list_branches(main.identifier) == []
- assert all(path.exists() for path in paths)
- catalog.create_branch(main.identifier, 'restored', 'base')
- restored = catalog.get_table(Identifier('default', 't',
branch='restored'))
- _assert_rows(restored, [seed])
-
-
-def test_native_branch_errors_keep_public_python_types(tmp_path, catalog):
- catalog, table = _table(tmp_path, catalog, True)
- with native_only(catalog):
- catalog.create_branch(table.identifier, 'dev')
- cases = [
- (TableNotExistException, lambda:
catalog.create_branch('default.missing', 'dev')),
- (TableNotExistException, lambda:
catalog.list_branches('default.missing')),
- (TagNotExistException, lambda:
catalog.create_branch(table.identifier, 'tagged', 'missing')),
- (BranchAlreadyExistException, lambda:
catalog.create_branch(table.identifier, 'dev')),
- (BranchNotExistException, lambda:
catalog.drop_branch(table.identifier, 'missing')),
- (BranchNotExistException, lambda:
catalog.fast_forward(table.identifier, 'missing')),
- (IllegalArgumentError, lambda:
catalog.create_branch(table.identifier, 'main')),
- ]
- for expected, invoke in cases:
- with pytest.raises(expected):
- invoke()
- manager = table.branch_manager()
- manager.create_branch('dev', ignore_if_exists=True)
- with pytest.raises(ValueError, match='already exists'):
- manager.create_branch('dev')
- with pytest.raises(ValueError, match="doesn't exist"):
- manager.drop_branch('missing')
-
-
[email protected]('operation', _OPERATIONS)
[email protected]('name', ['.', '..', '', ' '])
-def test_invalid_resource_paths_send_no_requests(catalog, operation, name):
- args = () if operation == 'list_branches' else ('dev',)
- with patch('pypaimon.catalog.rest.native_branch._native_catalog',
- side_effect=AssertionError('Invalid paths must be rejected
before preparing a backend')):
- for identifier in (Identifier(name, 't'), Identifier('default', name)):
- with pytest.raises(ValueError):
- getattr(catalog, operation)(identifier, *args)
- if operation in ('drop_branch', 'fast_forward'):
- with pytest.raises(ValueError):
- getattr(catalog, operation)('default.t', name)
-
-
-def test_native_catalog_constructor_uses_resolved_configuration(catalog):
- from pypaimon_rust.datafusion import PaimonCatalog
-
- with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog)
as construct:
- with native_only(catalog):
- # A valid missing table reaches the catalog, rather than failing
construction.
- with pytest.raises(TableNotExistException):
- catalog.list_branches('default.missing')
- options = construct.call_args.args[0]
- assert options['prefix'] == 'native-test'
- assert options['metastore'] == 'rest'
- assert construct.call_args.kwargs == {'config_required': False}
-
-
-def test_preparation_failure_can_use_python_before_mutation(tmp_path, catalog):
- catalog, table = _table(tmp_path, catalog, True)
- with patch('pypaimon_rust.datafusion.PaimonCatalog',
side_effect=ValueError('preparation failed')):
- catalog.create_branch(table.identifier, 'dev')
- with native_only(catalog):
- assert catalog.list_branches(table.identifier) == ['dev']
-
-
-def test_fork_safety_failure_never_uses_python(catalog):
- from pypaimon_rust import ForkSafetyError
-
- error = ForkSafetyError('inherited native state')
- with patch('pypaimon_rust.datafusion.PaimonCatalog', side_effect=error),
native_only(catalog):
- with pytest.raises(ForkSafetyError) as raised:
- catalog.list_branches('default.t')
- assert raised.value is error
-
-
[email protected]('status,body,exception', [
- (403, '{"message":"denied"}', TableNoPermissionException),
- (429, 'too many requests', RESTException),
- (502, 'bad gateway', RESTException),
- (504, 'gateway timeout', RESTException),
-])
-def test_real_http_errors_preserve_python_exceptions(catalog, status, body,
exception):
- # POST is deliberately used so neither HTTP client retries status
responses.
- with patch.object(RESTCatalogServer, '_branches_handle',
return_value=(body, status)) as handle, \
- native_only(catalog):
- with pytest.raises(exception):
- catalog.create_branch('default.t', 'dev')
- assert handle.call_count == 1
-
-
-def test_native_response_parse_failure_is_a_rest_exception(catalog):
- with patch.object(RESTCatalogServer, '_branches_handle',
return_value=('not JSON', 200)) as handle, \
- native_only(catalog):
- with pytest.raises(RESTException):
- catalog.list_branches('default.t')
- assert handle.call_count == 1
-
-
-def test_native_connection_failure_is_a_rest_exception(catalog):
- import socket
-
- with socket.socket() as listener:
- listener.bind(('localhost', 0))
- # A bound, non-listening port rejects connections without racing port
allocation.
- catalog.rest_api.client.uri = 'http://localhost:%d' %
listener.getsockname()[1]
- with native_only(catalog), pytest.raises(RESTException):
- catalog.drop_branch('default.t', 'dev')
-
-
-def test_published_branch_survives_uncertain_response(tmp_path, catalog):
- catalog, table = _table(tmp_path, catalog, True)
- original = RESTCatalogServer._branches_handle
- published = []
-
- def publish_then_fail(server, method, data, identifier):
- response = original(server, method, data, identifier)
- if method == 'POST':
- published.append(identifier)
- return server._mock_response(ErrorResponse(None, None, 'uncertain
result', 500), 500)
- return response
-
- with patch.object(RESTCatalogServer, '_branches_handle',
publish_then_fail), native_only(catalog):
- with pytest.raises(ServiceFailureException):
- catalog.create_branch(table.identifier, 'dev')
- assert catalog.list_branches(table.identifier) == ['dev']
- assert published == [table.identifier]
-
-
-def test_custom_error_handler_uses_python_before_request(catalog):
- from pypaimon.api.client import ErrorHandler
-
- class CustomFailure(RESTException):
- pass
-
- class Handler(ErrorHandler):
- def accept(self, error, request_id):
- raise CustomFailure('custom error handler')
-
- catalog.rest_api.client.set_error_handler(Handler())
- with patch('pypaimon_rust.datafusion.PaimonCatalog',
side_effect=AssertionError('custom handler bypassed')):
- with pytest.raises(CustomFailure, match='custom error handler'):
- catalog.list_branches('default.missing')
-
-
-def test_native_keeps_live_endpoint_and_original_authentication(catalog):
- # Config merging changes options without rebuilding Python's existing
client
- # and provider. Delegation must keep their endpoint, credentials and
headers.
- from pypaimon_rust.datafusion import PaimonCatalog
-
- original_token = catalog.rest_api.rest_auth_function.auth_provider.token
- catalog.rest_api.options.to_map().update({
- 'token.provider': 'dlf', 'token': 'server-token', 'dlf.token-loader':
'ecs',
- 'uri': 'http://invalid.example', 'header.X-Test': 'stale',
- })
- catalog.rest_api.rest_auth_function.init_header['X-Test'] = 'current'
- catalog.rest_api.client.set_user_agent('branch-client')
- with patch('pypaimon_rust.datafusion.PaimonCatalog', wraps=PaimonCatalog)
as construct, native_only(catalog):
- with pytest.raises(TableNotExistException):
- catalog.list_branches('default.missing')
- options = construct.call_args.args[0]
- assert options['token.provider'] == 'bear'
- assert options['token'] == original_token
- assert options['uri'] == catalog.rest_api.client.uri
- assert 'dlf.token-loader' not in options
- assert options['header.X-Test'] == 'current'
- assert options['header.User-Agent'] == 'branch-client'
-
-
[email protected]('code,exception', [
- (403, TableNoPermissionException), (500, ServiceFailureException), (501,
NotImplementedException),
- (404, NoSuchResourceException),
-])
-def test_native_errors_are_translated_without_python_retry(catalog, code,
exception):
- from pypaimon_rust.datafusion import CatalogError
-
- error = CatalogError('native error')
- error.code, error.resource_type, error.resource_name, error.message =
code, 'OTHER', 'unknown', 'native error'
-
- class Native:
- calls = 0
-
- def create_branch(self, *args):
- self.calls += 1
- raise error
-
- native = Native()
- with patch('pypaimon.catalog.rest.native_branch._native_catalog',
return_value=(native, CatalogError)), \
- native_only(catalog):
- with pytest.raises(exception):
- catalog.create_branch('default.t', 'dev')
- assert native.calls == 1
-
-
-def test_native_branch_blob_files_survive_drop(tmp_path, catalog):
- schema = pa.schema([('id', pa.int32()), ('payload', pa.large_binary())])
- catalog.create_table('default.blobs', Schema.from_pyarrow_schema(schema,
options={
- 'data-evolution.enabled': 'true', 'row-tracking.enabled': 'true',
'blob-field': 'payload',
- 'write.native.enabled': 'true', 'commit.native.enabled': 'true',
- }), False)
- main = catalog.get_table('default.blobs')
-
- def write(table, value):
- builder = table.new_batch_write_builder()
- writer, commit = builder.new_write(), builder.new_commit()
- try:
- writer.write_arrow(pa.table({'id': [value], 'payload': [b'payload'
* value]}, schema=schema))
- commit.commit(writer.prepare_commit())
- finally:
- writer.close()
- commit.close()
-
- write(main, 1)
- catalog.create_tag(main.identifier, 'base')
- with native_only(catalog):
- catalog.create_branch(main.identifier, 'dev', 'base')
- branch = catalog.get_table(Identifier('default', 'blobs',
branch='dev'))
- write(branch, 2)
- files = set(tmp_path.rglob('*.blob'))
- assert files
- catalog.drop_branch(main.identifier, 'dev')
- assert all(path.exists() for path in files)
- catalog.create_branch(main.identifier, 'restored', 'base')
- restored = catalog.get_table(Identifier('default', 'blobs',
branch='restored'))
- builder = restored.new_read_builder()
- assert
builder.new_read().to_arrow(builder.new_scan().plan().splits()).to_pydict() == {
- 'id': [1], 'payload': [b'payload'],
- }
diff --git a/paimon-python/pypaimon/tests/rest/rest_branch_test.py
b/paimon-python/pypaimon/tests/rest/rest_branch_test.py
index ea99508568..cebfcd49b5 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 @@ class RESTCatalogBranchCRUDTest(RESTBaseTest):
from pypaimon.api.rest_exception import NoSuchResourceException
error = NoSuchResourceException('OTHER', 'resource', 'unknown resource
missing')
- with patch('pypaimon.catalog.rest.native_branch._native_catalog',
return_value=None), \
- patch.object(self.rest_catalog.rest_api, 'create_branch',
side_effect=error):
+ with 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 0000000000..304979fd53
--- /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
+
+
[email protected]
+def transport_server():
+ class Handler(BaseHTTPRequestHandler):
+ def operation(self):
+ size = int(self.headers.get('Content-Length', 0))
+ data = self.rfile.read(size).decode() if size else None
+ self.server.requests.append((self.command, self.path,
self.headers))
+ 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
+
+
[email protected]('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]
+
+
[email protected]('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
+
+
[email protected]('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)
+
+
[email protected]('operation', ['create_branch', 'drop_branch',
'fast_forward', 'list_branches'])
[email protected]('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 == []
+
+
[email protected]_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])
+
+
[email protected]_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]
+
+
[email protected]_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/rest/rest_version_test.py
b/paimon-python/pypaimon/tests/rest/rest_version_test.py
new file mode 100644
index 0000000000..19bff6433f
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/rest_version_test.py
@@ -0,0 +1,303 @@
+# 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 version management uses the Python client and Java wire semantics."""
+
+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.rest import rest_branch_transport_test as branch_fixtures
+
+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)
+
+
[email protected]
+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)
+
+
[email protected]('operation', _OPERATIONS)
+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 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 ')
+ 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'
+
+
[email protected]('native', [False, True])
[email protected]('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': [' ']}
+
+
[email protected]('native', [False, True])
[email protected]('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()}]
+
+
[email protected]('operation', _OPERATIONS)
[email protected]('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 branch_fixtures.python_rest_only():
+ with pytest.raises(expected):
+ invoke(catalog, operation)
+ assert len(server.requests) == 1
+
+
[email protected]('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)
+
+
[email protected]('operation', ['create_tag', 'delete_tag',
'rollback_to'])
+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 branch_fixtures.python_rest_only():
+ with pytest.raises(ServiceFailureException):
+ invoke(catalog, operation)
+ assert len(server.requests) == 1
+ assert sentinel.read_bytes() == b'existing data'
+
+
[email protected]('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
+
+
[email protected]('native', [False, True])
[email protected]('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
+
+
[email protected]('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 branch_fixtures.python_rest_only():
+ with pytest.raises(NotImplementedException):
+ invoke(catalog, operation)
+ assert len(server.requests) == 1
+
+
[email protected]('native', [False, True])
[email protected]('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
+
+
[email protected]('native', [False, True])
[email protected]('operation', ['create_tag', 'get_tag', 'delete_tag'])
[email protected]('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)
+
+
[email protected]('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 == []
diff --git a/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
b/paimon-python/pypaimon/tests/rest/rollback_protocol_test.py
index 56ac97c37b..2790806373 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()
