[python] Use Native REST catalog branch operations

.github/workflows/ci-python.yml

@@ -231,8 +231,11 @@ jobs:
               normalize_array_for_write,
               read_arrow_schema,
           )
-          from pypaimon_rust.datafusion import ReadBuilder
+          from pypaimon_rust.datafusion import CatalogError, PaimonCatalog, ReadBuilder
           assert callable(getattr(ReadBuilder, 'with_partition_filter', None))
+          assert issubclass(CatalogError, ValueError)
+          for method in ('create_branch', 'drop_branch', 'fast_forward', 'list_branches'):
+              assert callable(getattr(PaimonCatalog, method, None))
           PY
 
       - name: Run Python tests with native planning, reading, writing, updating and committing

paimon-python/README.md

@@ -139,6 +139,28 @@ table = table.copy({"parquet.filter.columnindex.enabled": "false"})
 Unsupported reads use the normal path. Reading fewer bytes may require more
 object-store requests.
 
+# Native REST branch management
+
+REST catalogs delegate `list_branches` to Rust when the catalog option
+`read.native.enabled=true`, and `create_branch`, `drop_branch` and `fast_forward`
+when `write.native.enabled=true`. Existing `table.branch_manager()` calls use
+the same route. Install the optional `pypaimon-rust` runtime to enable it.
+
+The bridge reuses the catalog's resolved prefix, endpoint, authentication and
+request headers, and preserves the public Python exception types. Custom clients,
+authentication providers or error handlers use Python. Once a Native operation
+starts, failures propagate without replaying the request through Python.
+
+OpenAPI V4 (`dlf.signing-algorithm=openapi-v4`) is signed in Rust core, including
+the action header. Session headers are merged case-insensitively, preserving an
+explicit User-Agent. Sessions with Cookie, custom authentication or TLS settings
+use Python to retain their transport state. Native listing retries HTTP 429/503;
+mutations are sent once and accept successful responses without a JSON body.
+
+As in Java, creation without a tag produces an empty branch; creation from a tag
+uses its snapshot. REST branch renaming is unsupported. Dropping branch metadata
+does not remove data files shared with another branch or tag.
+
 # Native write and commit
 
 PyPaimon can write Arrow batches through the optional `pypaimon-rust` runtime.

paimon-python/pypaimon/api/api_request.py

@@ -212,13 +212,6 @@ class CreateBranchRequest(RESTRequest):
     from_tag: Optional[str] = json_field(FIELD_FROM_TAG, default=None)
 
 
-@dataclass
-class RenameBranchRequest(RESTRequest):
-    FIELD_TO_BRANCH = "toBranch"
-
-    to_branch: str = json_field(FIELD_TO_BRANCH)
-
-
 @dataclass
 class ForwardBranchRequest(RESTRequest):
     """Empty body request; serializes to ``{}`` per Java ForwardBranchRequest."""

paimon-python/pypaimon/api/resource_paths.py

@@ -33,7 +33,6 @@ class ResourcePaths:
     FUNCTION_DETAILS = "function-details"
     TAGS = "tags"
     BRANCHES = "branches"
-    RENAME = "rename"
     FORWARD = "forward"
     PERMISSIONS = "permissions"
     POLICIES = "policies"
@@ -135,10 +134,6 @@ def branch(self, database_name: str, table_name: str, branch_name: str) -> str:
             self.TABLES, RESTUtil.encode_string(table_name), self.BRANCHES,
             RESTUtil.encode_string(branch_name))
 
-    def rename_branch(self, database_name: str, table_name: str, branch_name: str) -> str:
-        return "{}/{}".format(
-            self.branch(database_name, table_name, branch_name), self.RENAME)
-
     def forward_branch(self, database_name: str, table_name: str, branch_name: str) -> str:
         return "{}/{}".format(
             self.branch(database_name, table_name, branch_name), self.FORWARD)

paimon-python/pypaimon/api/rest_api.py

@@ -27,7 +27,7 @@
                                       CreateTableRequest, CreateTagRequest,
                                       DropPolicyRequest, ForwardBranchRequest,
                                       GrantPermissionRequest, PolicyRequest,
-                                      RenameBranchRequest, RenameTableRequest,
+                                      RenameTableRequest,
                                       RevokePermissionRequest,
                                       RollbackTableRequest)
 from pypaimon.api.api_response import (CommitTableResponse, ConfigResponse,
@@ -128,6 +128,10 @@ def __init__(self, options: Union[Options, Dict[str, str]], config_required: boo
         self.options = options
         self.resource_paths = ResourcePaths.for_catalog_properties(options)
 
+    @staticmethod
+    def _is_auth_option(key):
+        return key in ('uri', 'token.provider', 'token') or key.startswith('dlf.')
+
     def __build_paged_query_params(
             self,
             max_results: Optional[int],
@@ -545,13 +549,24 @@ def delete_tag(self, identifier: Identifier, tag_name: str) -> None:
         )
 
     # Branch CRUD wrappers — mirror Java RESTApi branch methods.
+    @staticmethod
+    def _validate_branch_path(identifier, branch_name=None):
+        database_name, table_name = RESTApi.__validate_identifier(identifier)
+        components = [(database_name, 'Database name'), (table_name, 'Table name')]
+        if branch_name is not None:
+            components.append((branch_name, 'Branch name'))
+        for value, label in components:
+            if not value or not value.strip() or value in ('.', '..'):
+                raise ValueError('{} must be a non-empty resource path component'.format(label))
+        return database_name, table_name
+
     def create_branch(
             self,
             identifier: Identifier,
             branch_name: str,
             tag_name: Optional[str] = None,
     ) -> None:
-        database_name, table_name = self.__validate_identifier(identifier)
+        database_name, table_name = self._validate_branch_path(identifier)
         if not branch_name or not branch_name.strip():
             raise ValueError("Branch name cannot be empty")
         request = CreateBranchRequest(branch=branch_name, from_tag=tag_name)
@@ -562,38 +577,22 @@ def create_branch(
         )
 
     def drop_branch(self, identifier: Identifier, branch_name: str) -> None:
-        database_name, table_name = self.__validate_identifier(identifier)
+        database_name, table_name = self._validate_branch_path(identifier, branch_name)
         self.client.delete(
             self.resource_paths.branch(database_name, table_name, branch_name),
             self.rest_auth_function,
         )
 
-    def rename_branch(
-            self,
-            identifier: Identifier,
-            from_branch: str,
-            to_branch: str,
-    ) -> None:
-        database_name, table_name = self.__validate_identifier(identifier)
-        if not to_branch or not to_branch.strip():
-            raise ValueError("Target branch name cannot be empty")
-        request = RenameBranchRequest(to_branch=to_branch)
-        self.client.post(
-            self.resource_paths.rename_branch(database_name, table_name, from_branch),
-            request,
-            self.rest_auth_function,
-        )
-
     def fast_forward(self, identifier: Identifier, branch_name: str) -> None:
-        database_name, table_name = self.__validate_identifier(identifier)
+        database_name, table_name = self._validate_branch_path(identifier, branch_name)
         self.client.post(
             self.resource_paths.forward_branch(database_name, table_name, branch_name),
             ForwardBranchRequest(),
             self.rest_auth_function,
         )
 
     def list_branches(self, identifier: Identifier) -> List[str]:
-        database_name, table_name = self.__validate_identifier(identifier)
+        database_name, table_name = self._validate_branch_path(identifier)
         response = self.client.get(
             self.resource_paths.branches(database_name, table_name),
             ListBranchesResponse,

paimon-python/pypaimon/branch/branch_manager.py

@@ -40,11 +40,11 @@ def create_branch(
         ignore_if_exists: bool = False
     ) -> None:
         """
-        Create a branch from the current state or from a tag.
+        Create an empty branch or initialize it from a tag.
 
         Args:
             branch_name: Name of the branch to create
-            tag_name: Optional tag name to create branch from, None for current state
+            tag_name: Optional tag name to create branch from, None for an empty branch
             ignore_if_exists: If true, do nothing when branch already exists;
                             if false, throw exception
 

paimon-python/pypaimon/branch/catalog_branch_manager.py

@@ -85,22 +85,19 @@ def create_branch(
         ignore_if_exists: bool = False
     ) -> None:
         """
-        Create a branch from the current state or from a tag.
+        Create an empty branch or initialize it from a tag.
 
         Args:
             branch_name: Name of the branch to create
-            tag_name: Optional tag name to create branch from, None for current state
+            tag_name: Optional tag name to create branch from, None for an empty branch
             ignore_if_exists: If true, do nothing when branch already exists
 
         Raises:
             ValueError: If branch name is invalid or branch already exists (and ignore_if_exists=False)
         """
         def _create(catalog: Catalog):
             BranchManager.validate_branch(branch_name)
-            existing_branches = catalog.list_branches(self.identifier)
-            if branch_name in existing_branches:
-                if not ignore_if_exists:
-                    raise ValueError(f"Branch name '{branch_name}' already exists.")
+            if ignore_if_exists and branch_name in catalog.list_branches(self.identifier):
                 return
             catalog.create_branch(self.identifier, branch_name, tag_name)
 

paimon-python/pypaimon/catalog/rest/native_branch.py

@@ -0,0 +1,137 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Transport REST branch operations and structured errors to/from Rust core."""
+
+import logging
+import os
+
+import requests
+from requests.structures import CaseInsensitiveDict
+
+from pypaimon.api.api_response import ErrorResponse
+from pypaimon.api.auth import RESTAuthFunction
+from pypaimon.api.auth.bearer import BearTokenAuthProvider
+from pypaimon.api.auth.dlf_provider import DLFAuthProvider
+from pypaimon.api.client import DefaultErrorHandler, HttpClient
+from pypaimon.api.rest_api import RESTApi
+from pypaimon.catalog.catalog_context import CatalogContext
+from pypaimon.common.options.core_options import CoreOptions
+
+logger = logging.getLogger(__name__)
+
+
+def _native_headers(api):
+    session = api.client.session
+    if (type(session) is not requests.Session or session.auth is not None
+            or session.cert is not None or session.verify is not True or session.proxies or session.params
+            or not session.trust_env or any(session.hooks.values())
+            or set(session.adapters) != {'http://', 'https://'}
+            or any(type(adapter) is not requests.adapters.HTTPAdapter for adapter in session.adapters.values())
+            or api.client._timeout != (180, 180)
+            or os.environ.get('REQUESTS_CA_BUNDLE') or os.environ.get('CURL_CA_BUNDLE')):
+        # Keep custom transport credentials and policies in their owning client.
+        return None
+    headers = CaseInsensitiveDict(session.headers)
+    # Let each HTTP transport advertise only the codecs it can decode and manage
+    # its own connections. An explicitly customized transport header stays Python.
+    for name in ('Accept-Encoding', 'Connection'):
+        if (headers.get(name) != requests.utils.default_headers().get(name)
+                or any(key.lower() == name.lower() for key in api.rest_auth_function.init_header)):
+            return None
+        headers.pop(name, None)
+    headers.update(api.rest_auth_function.init_header)
+    return {key: value for key, value in headers.items() if value is not None}
+
+
+def _native_catalog(catalog, operation):
+    from pypaimon.catalog.rest.rest_catalog import RESTCatalog
+
+    if (type(catalog) is not RESTCatalog or type(catalog.context) is not CatalogContext
+            or type(catalog.rest_api) is not RESTApi
+            or any(getattr(catalog.context, attr, None) is not None for attr in (
+                'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader'))):
+        return None
+    options = CoreOptions(catalog.context.options)
+    enabled = options.native_read_enabled() if operation == 'list_branches' else options.native_write_enabled()
+    if not enabled:
+        return None
+    api = catalog.rest_api
+    if (type(api.client) is not HttpClient
+            or type(api.client.error_handler) is not DefaultErrorHandler
+            or type(api.rest_auth_function) is not RESTAuthFunction
+            or type(api.rest_auth_function.auth_provider) not in (BearTokenAuthProvider, DLFAuthProvider)):
+        return None
+    headers = _native_headers(api)
+    if headers is None:
+        return None
+    try:
+        from pypaimon_rust.datafusion import CatalogError, PaimonCatalog
+    except ImportError:
+        return None
+    from pypaimon.read.native_plan import _option_value_to_string, _raise_if_native_fork_safety_error
+    resolved = {key: value for key, value in api.options.to_map().items()
+                if not RESTApi._is_auth_option(key) and not key.startswith(RESTApi.HEADER_PREFIX)}
+    resolved['uri'] = api.client.uri
+    resolved.update({RESTApi.HEADER_PREFIX + key: value for key, value in headers.items()})
+    resolved = {str(key): _option_value_to_string(value)
+                for key, value in resolved.items() if value is not None}
+    resolved['metastore'] = 'rest'
+    try:
+        provider = api.rest_auth_function.auth_provider
+        if type(provider) is DLFAuthProvider:
+            # The owning provider retains Java's STS cache/refresh policy. Pass
+            # its current credentials for this call; Rust still signs the request.
+            token = provider.get_token()
+            resolved.update({
+                'token.provider': 'dlf', 'dlf.region': provider.region,
+                'dlf.signing-algorithm': provider.signing_algorithm,
+                'dlf.access-key-id': token.access_key_id,
+                'dlf.access-key-secret': token.access_key_secret,
+            })
+            if token.security_token is not None:
+                resolved['dlf.security-token'] = token.security_token
+        else:
+            resolved.update({'token.provider': 'bear', 'token': provider.token})
+        # Reuse the resolved configuration, as RESTCatalogLoader does. Keep no
+        # opaque native state on the Python catalog across pickle or process fork.
+        return PaimonCatalog(resolved, config_required=False), CatalogError
+    except Exception as error:
+        _raise_if_native_fork_safety_error(error)
+        logger.debug('Native branch preparation failed: %s', error)
+        return None
+
+
+def call_branch(catalog, operation, identifier, *args):
+    # Both backends must reject a path that HTTP clients normalize to another
+    # resource. CREATE's branch is in JSON; only DROP/FORWARD put it in the URL.
+    path_branch = args[0] if operation in ('drop_branch', 'fast_forward') else None
+    RESTApi._validate_branch_path(identifier, path_branch)
+    if operation == 'create_branch' and (not args[0] or not args[0].strip()):
+        raise ValueError('Branch name cannot be empty')
+    prepared = _native_catalog(catalog, operation)
+    if prepared is None:
+        return getattr(catalog.rest_api, operation)(identifier, *args)
+    native, error_type = prepared
+    try:
+        # Once an operation starts, never execute it again through Python:
+        # the REST server may have published it before returning an error.
+        return getattr(native, operation)((identifier.get_database_name(), identifier.get_object_name()), *args)
+    except error_type as error:
+        DefaultErrorHandler.get_instance().accept(ErrorResponse(
+            resource_type=error.resource_type, resource_name=error.resource_name,
+            message=error.message, code=error.code), HttpClient.DEFAULT_REQUEST_ID)