[python] Keep REST catalog operations on Python

.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

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
 

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")
 
 

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],
@@ -142,11 +138,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
@@ -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(
@@ -538,23 +534,23 @@ 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)
+        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,
         )
 
     # 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 @@ 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,22 +573,22 @@ 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(),
             self.rest_auth_function,
         )
 
     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,

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

@@ -648,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_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 @@ def create_branch(self, identifier: Union[str, Identifier], branch_name: str,
             raise IllegalArgumentError(str(e)) from e
 
     def drop_branch(self, identifier: Union[str, Identifier], branch_name: str) -> None:
-        from pypaimon.catalog.rest.native_branch import call_branch
         if not isinstance(identifier, Identifier):
             identifier = Identifier.from_string(identifier)
         try:
-            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 @@ 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
         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:

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