You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/04/30 09:05:43 UTC
(paimon) branch master updated: [python] Add tag CRUD API to Catalog (REST + abstract) (#7746)
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 c688560bed [python] Add tag CRUD API to Catalog (REST + abstract) (#7746)
c688560bed is described below
commit c688560bed00477535fc8ccbb8980d45e6d5a922
Author: chaoyang <ch...@apache.org>
AuthorDate: Thu Apr 30 17:05:35 2026 +0800
[python] Add tag CRUD API to Catalog (REST + abstract) (#7746)
`pypaimon`'s `Catalog` ABC has stubs for branch CRUD (`create_branch` /
`drop_branch` / `list_branches` / `fast_forward`) but no tag methods at
all — calling `catalog.create_tag(...)` on master raises
`AttributeError`. Java's `Catalog`
(`paimon-core/.../catalog/Catalog.java:914-985`) defines `createTag` /
`getTag` / `listTagsPaged` / `deleteTag`. This PR ports that surface to
Python with a complete `RESTCatalog` implementation.
`FilesystemCatalog` inherits the abstract `NotImplementedError` stubs
(mirrors how it inherits the existing branch stubs). A concrete
filesystem implementation requires a Python-side `TagManager` port and
is tracked as a separate follow-up.
---
paimon-python/pypaimon/api/api_request.py | 16 ++
paimon-python/pypaimon/api/api_response.py | 35 +++++
paimon-python/pypaimon/api/resource_paths.py | 12 ++
paimon-python/pypaimon/api/rest_api.py | 66 +++++++-
paimon-python/pypaimon/catalog/catalog.py | 88 +++++++++++
.../pypaimon/catalog/catalog_exception.py | 8 +
.../pypaimon/catalog/rest/rest_catalog.py | 68 ++++++++
paimon-python/pypaimon/tests/api/__init__.py | 17 ++
.../pypaimon/tests/api/test_tag_dto_serde.py | 143 +++++++++++++++++
paimon-python/pypaimon/tests/rest/rest_server.py | 103 +++++++++++-
paimon-python/pypaimon/tests/rest/rest_tag_test.py | 175 +++++++++++++++++++++
11 files changed, 724 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/api/api_request.py b/paimon-python/pypaimon/api/api_request.py
index b7c7c1966b..9bf542ea27 100644
--- a/paimon-python/pypaimon/api/api_request.py
+++ b/paimon-python/pypaimon/api/api_request.py
@@ -156,3 +156,19 @@ class AlterFunctionRequest(RESTRequest):
return {
self.FIELD_CHANGES: [c.to_dict() for c in self.changes]
}
+
+
+# Wire DTO for ``POST /databases/{db}/tables/{tbl}/tags``. Mirrors Java
+# ``CreateTagRequest`` (paimon-api/.../rest/requests/CreateTagRequest.java) — only
+# three fields are serialized. ``ignoreIfExists`` is intentionally NOT included
+# here; it is a client-side flag handled by ``RESTCatalog.create_tag``, not part
+# of the wire format.
+@dataclass
+class CreateTagRequest(RESTRequest):
+ FIELD_TAG_NAME = "tagName"
+ FIELD_SNAPSHOT_ID = "snapshotId"
+ FIELD_TIME_RETAINED = "timeRetained"
+
+ tag_name: str = json_field(FIELD_TAG_NAME)
+ snapshot_id: Optional[int] = json_field(FIELD_SNAPSHOT_ID, default=None)
+ time_retained: Optional[str] = json_field(FIELD_TIME_RETAINED, default=None)
diff --git a/paimon-python/pypaimon/api/api_response.py b/paimon-python/pypaimon/api/api_response.py
index 819b2c45f3..b8fb8e9a18 100644
--- a/paimon-python/pypaimon/api/api_response.py
+++ b/paimon-python/pypaimon/api/api_response.py
@@ -25,6 +25,7 @@ from pypaimon.common.json_util import T, json_field
from pypaimon.common.options import Options
from pypaimon.schema.data_types import DataField
from pypaimon.schema.schema import Schema
+from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
from pypaimon.snapshot.table_snapshot import TableSnapshot
@@ -181,6 +182,40 @@ class ListTablesResponse(PagedResponse[str]):
return self.next_page_token
+@dataclass
+class GetTagResponse(RESTResponse):
+ """Response for getting a tag.
+
+ Mirrors Java ``GetTagResponse`` (paimon-api/.../rest/responses/GetTagResponse.java).
+ """
+ FIELD_TAG_NAME = "tagName"
+ FIELD_SNAPSHOT = "snapshot"
+ FIELD_TAG_CREATE_TIME = "tagCreateTime"
+ FIELD_TAG_TIME_RETAINED = "tagTimeRetained"
+
+ tag_name: Optional[str] = json_field(FIELD_TAG_NAME, default=None)
+ snapshot: Optional[Snapshot] = json_field(FIELD_SNAPSHOT, default=None)
+ tag_create_time: Optional[int] = json_field(FIELD_TAG_CREATE_TIME, default=None)
+ tag_time_retained: Optional[str] = json_field(FIELD_TAG_TIME_RETAINED, default=None)
+
+
+@dataclass
+class ListTagsResponse(PagedResponse[str]):
+ """Paged response for listing tag names. Mirrors Java
+ ``ListTagsResponse`` (paimon-api/.../rest/responses/ListTagsResponse.java)."""
+ FIELD_TAGS = "tags"
+
+ tags: Optional[List[str]] = json_field(FIELD_TAGS, default=None)
+ next_page_token: Optional[str] = json_field(
+ PagedResponse.FIELD_NEXT_PAGE_TOKEN, default=None)
+
+ def data(self) -> Optional[List[str]]:
+ return self.tags
+
+ def get_next_page_token(self) -> Optional[str]:
+ return self.next_page_token
+
+
@dataclass
class GetTableResponse(AuditRESTResponse):
"""Response for getting table"""
diff --git a/paimon-python/pypaimon/api/resource_paths.py b/paimon-python/pypaimon/api/resource_paths.py
index aba64018e6..5684033626 100644
--- a/paimon-python/pypaimon/api/resource_paths.py
+++ b/paimon-python/pypaimon/api/resource_paths.py
@@ -30,6 +30,7 @@ class ResourcePaths:
PARTITIONS = "partitions"
FUNCTIONS = "functions"
FUNCTION_DETAILS = "function-details"
+ TAGS = "tags"
def __init__(self, prefix: str):
self.base_path = "/{}/{}".format(self.V1, prefix).rstrip("/")
@@ -99,3 +100,14 @@ class ResourcePaths:
def function(self, database_name: str, function_name: str) -> str:
return "{}/{}/{}/{}/{}".format(self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
self.FUNCTIONS, RESTUtil.encode_string(function_name))
+
+ def tags(self, database_name: str, table_name: str) -> str:
+ return "{}/{}/{}/{}/{}/{}".format(
+ self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
+ self.TABLES, RESTUtil.encode_string(table_name), self.TAGS)
+
+ def tag(self, database_name: str, table_name: str, tag_name: str) -> str:
+ return "{}/{}/{}/{}/{}/{}/{}".format(
+ self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
+ self.TABLES, RESTUtil.encode_string(table_name), self.TAGS,
+ RESTUtil.encode_string(tag_name))
diff --git a/paimon-python/pypaimon/api/rest_api.py b/paimon-python/pypaimon/api/rest_api.py
index e224421131..ea0c23bc2a 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -23,18 +23,19 @@ import re
from pypaimon.api.api_request import (AlterDatabaseRequest, AlterFunctionRequest,
AlterTableRequest, CommitTableRequest,
CreateDatabaseRequest, CreateFunctionRequest,
- CreateTableRequest, RenameTableRequest,
- RollbackTableRequest)
+ CreateTableRequest, CreateTagRequest,
+ RenameTableRequest, RollbackTableRequest)
from pypaimon.api.api_response import (CommitTableResponse, ConfigResponse,
GetDatabaseResponse, GetFunctionResponse,
GetTableResponse,
- GetTableTokenResponse,
+ GetTableTokenResponse, GetTagResponse,
ListDatabasesResponse,
ListFunctionDetailsResponse,
ListFunctionsGloballyResponse,
ListFunctionsResponse,
ListPartitionsResponse,
- ListTablesResponse, PagedList,
+ ListTablesResponse, ListTagsResponse,
+ PagedList,
PagedResponse, GetTableSnapshotResponse,
Partition)
from pypaimon.api.auth import AuthProviderFactory, RESTAuthFunction
@@ -59,6 +60,7 @@ class RESTApi:
TABLE_TYPE = "tableType"
FUNCTION_NAME_PATTERN = "functionNamePattern"
PARTITION_NAME_PATTERN = "partitionNamePattern"
+ TAG_NAME_PREFIX = "tagNamePrefix"
TOKEN_EXPIRATION_SAFE_TIME_MILLIS = 3_600_000
# Function name validation pattern
@@ -436,6 +438,62 @@ class RESTApi:
partitions = response.data() or []
return PagedList(partitions, response.get_next_page_token())
+ # Tag CRUD wrappers — mirror Java RESTApi.java:1062-1123.
+ def create_tag(
+ self,
+ identifier: Identifier,
+ tag_name: str,
+ snapshot_id: Optional[int] = None,
+ time_retained: Optional[str] = None,
+ ) -> None:
+ database_name, table_name = self.__validate_identifier(identifier)
+ request = CreateTagRequest(
+ tag_name=tag_name,
+ snapshot_id=snapshot_id,
+ time_retained=time_retained,
+ )
+ self.client.post(
+ self.resource_paths.tags(database_name, table_name),
+ request,
+ self.rest_auth_function,
+ )
+
+ def get_tag(self, identifier: Identifier, tag_name: str) -> GetTagResponse:
+ database_name, table_name = self.__validate_identifier(identifier)
+ return self.client.get(
+ self.resource_paths.tag(database_name, table_name, tag_name),
+ GetTagResponse,
+ self.rest_auth_function,
+ )
+
+ def list_tags_paged(
+ self,
+ identifier: Identifier,
+ max_results: Optional[int] = None,
+ page_token: Optional[str] = None,
+ tag_name_prefix: Optional[str] = None,
+ ) -> PagedList[str]:
+ database_name, table_name = self.__validate_identifier(identifier)
+ response = self.client.get_with_params(
+ self.resource_paths.tags(database_name, table_name),
+ self.__build_paged_query_params(
+ max_results,
+ page_token,
+ {self.TAG_NAME_PREFIX: tag_name_prefix},
+ ),
+ ListTagsResponse,
+ self.rest_auth_function,
+ )
+ tags = response.data() or []
+ return 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)
+ self.client.delete(
+ self.resource_paths.tag(database_name, table_name, tag_name),
+ self.rest_auth_function,
+ )
+
@staticmethod
def is_valid_function_name(name: str) -> bool:
if not name:
diff --git a/paimon-python/pypaimon/catalog/catalog.py b/paimon-python/pypaimon/catalog/catalog.py
index c6fcacd4ef..adabf2ebe2 100644
--- a/paimon-python/pypaimon/catalog/catalog.py
+++ b/paimon-python/pypaimon/catalog/catalog.py
@@ -293,3 +293,91 @@ class Catalog(ABC):
raise NotImplementedError(
"list_branches is not supported by this catalog."
)
+
+ def create_tag(
+ self,
+ identifier: Union[str, Identifier],
+ tag_name: str,
+ snapshot_id: Optional[int] = None,
+ time_retained: Optional[str] = None,
+ ignore_if_exists: bool = False,
+ ) -> None:
+ """Create a tag on a table.
+
+ Args:
+ identifier: Table identifier (Identifier or string).
+ tag_name: Tag name to create.
+ snapshot_id: Optional snapshot id; if not set the latest snapshot is tagged.
+ time_retained: Optional time retained string (e.g., ``"1d"``, ``"12h"``, ``"30m"``).
+ ignore_if_exists: If True, do not raise when the tag already exists.
+
+ Raises:
+ NotImplementedError: If the catalog does not support tag management.
+ """
+ raise NotImplementedError(
+ "create_tag is not supported by this catalog."
+ )
+
+ def delete_tag(
+ self,
+ identifier: Union[str, Identifier],
+ tag_name: str,
+ ) -> None:
+ """Delete a tag from a table.
+
+ Args:
+ identifier: Table identifier (Identifier or string).
+ tag_name: Tag name to delete.
+
+ Raises:
+ NotImplementedError: If the catalog does not support tag management.
+ """
+ raise NotImplementedError(
+ "delete_tag is not supported by this catalog."
+ )
+
+ def get_tag(
+ self,
+ identifier: Union[str, Identifier],
+ tag_name: str,
+ ):
+ """Get a tag of a table.
+
+ Args:
+ identifier: Table identifier (Identifier or string).
+ tag_name: Tag name to look up.
+
+ Returns:
+ GetTagResponse describing the tag.
+
+ Raises:
+ NotImplementedError: If the catalog does not support tag management.
+ """
+ raise NotImplementedError(
+ "get_tag is not supported by this catalog."
+ )
+
+ def list_tags_paged(
+ self,
+ identifier: Union[str, Identifier],
+ max_results: Optional[int] = None,
+ page_token: Optional[str] = None,
+ tag_name_prefix: Optional[str] = None,
+ ):
+ """List tags of a table with pagination.
+
+ Args:
+ identifier: Table identifier (Identifier or string).
+ max_results: Maximum number of results to return per page.
+ page_token: Token for pagination.
+ tag_name_prefix: Optional prefix filter for tag names.
+
+ Returns:
+ PagedList of tag names.
+
+ Raises:
+ NotImplementedError: If the catalog does not support tag management.
+ """
+ raise NotImplementedError(
+ "list_tags_paged is not supported by this catalog."
+ )
diff --git a/paimon-python/pypaimon/catalog/catalog_exception.py b/paimon-python/pypaimon/catalog/catalog_exception.py
index af7636bd64..796d648d03 100644
--- a/paimon-python/pypaimon/catalog/catalog_exception.py
+++ b/paimon-python/pypaimon/catalog/catalog_exception.py
@@ -180,6 +180,14 @@ class TagNotExistException(CatalogException):
super().__init__(f"Tag {tag} does not exist")
+class TagAlreadyExistException(CatalogException):
+ """Tag already exist exception"""
+
+ def __init__(self, tag: str):
+ self.tag = tag
+ super().__init__(f"Tag {tag} already exists")
+
+
class IllegalArgumentError(CatalogException):
"""Illegal argument exception"""
pass
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 810fed963b..4cd0105ad7 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -31,6 +31,7 @@ from pypaimon.catalog.catalog_exception import (
TableNoPermissionException, DatabaseNoPermissionException,
FunctionNotExistException, FunctionAlreadyExistException,
DefinitionAlreadyExistException, DefinitionNotExistException,
+ TagNotExistException, TagAlreadyExistException,
)
from pypaimon.catalog.database import Database
from pypaimon.catalog.rest.property_change import PropertyChange
@@ -468,6 +469,73 @@ class RESTCatalog(Catalog):
except NoSuchResourceException as e:
raise DatabaseNotExistException(database_name) from e
+ # Tag CRUD: each method mirrors the corresponding Java handler in
+ # paimon-core/.../rest/RESTCatalog.java line-for-line. Specifically:
+ # create_tag — RESTCatalog.java:1100-1127
+ # get_tag — RESTCatalog.java:1056-1071 (getTag)
+ # list_tags_paged — RESTCatalog.java:1142-1156
+ # delete_tag — RESTCatalog.java:1167-1182
+
+ def create_tag(self, identifier: Union[str, Identifier], tag_name: str,
+ snapshot_id: Optional[int] = None,
+ time_retained: Optional[str] = None,
+ ignore_if_exists: bool = False) -> None:
+ if not isinstance(identifier, Identifier):
+ identifier = Identifier.from_string(identifier)
+ try:
+ self.rest_api.create_tag(identifier, tag_name, snapshot_id, time_retained)
+ except AlreadyExistsException as e:
+ if not ignore_if_exists:
+ raise TagAlreadyExistException(tag_name) from e
+ except NoSuchResourceException as e:
+ if e.resource_type == ErrorResponse.RESOURCE_TYPE_SNAPSHOT:
+ raise ValueError(
+ "Snapshot {} in table {} doesn't exist.".format(
+ e.resource_name, identifier.get_full_name())) from e
+ raise TableNotExistException(identifier) from e
+ except ForbiddenException as e:
+ raise TableNoPermissionException(identifier) from e
+ except BadRequestException as e:
+ raise IllegalArgumentError(str(e)) from e
+
+ def get_tag(self, identifier: Union[str, Identifier], tag_name: str):
+ if not isinstance(identifier, Identifier):
+ identifier = Identifier.from_string(identifier)
+ try:
+ return self.rest_api.get_tag(identifier, tag_name)
+ except NoSuchResourceException as e:
+ if e.resource_type == ErrorResponse.RESOURCE_TYPE_TAG:
+ raise TagNotExistException(tag_name) from e
+ raise TableNotExistException(identifier) from e
+ except ForbiddenException as e:
+ raise TableNoPermissionException(identifier) from e
+
+ def list_tags_paged(self, identifier: Union[str, Identifier],
+ max_results: Optional[int] = None,
+ page_token: Optional[str] = None,
+ tag_name_prefix: Optional[str] = None) -> PagedList[str]:
+ if not isinstance(identifier, Identifier):
+ identifier = Identifier.from_string(identifier)
+ try:
+ return self.rest_api.list_tags_paged(
+ identifier, max_results, page_token, tag_name_prefix)
+ except NoSuchResourceException as e:
+ raise TableNotExistException(identifier) from e
+ except ForbiddenException as e:
+ raise TableNoPermissionException(identifier) from e
+
+ def delete_tag(self, identifier: Union[str, Identifier], tag_name: str) -> None:
+ if not isinstance(identifier, Identifier):
+ identifier = Identifier.from_string(identifier)
+ try:
+ self.rest_api.delete_tag(identifier, tag_name)
+ except NoSuchResourceException as e:
+ if e.resource_type == ErrorResponse.RESOURCE_TYPE_TAG:
+ raise TagNotExistException(tag_name) from e
+ raise TableNotExistException(identifier) from e
+ except ForbiddenException as e:
+ raise TableNoPermissionException(identifier) from e
+
def load_table_metadata(self, identifier: Identifier) -> TableMetadata:
try:
response = self.rest_api.get_table(identifier)
diff --git a/paimon-python/pypaimon/tests/api/__init__.py b/paimon-python/pypaimon/tests/api/__init__.py
new file mode 100644
index 0000000000..65b48d4d79
--- /dev/null
+++ b/paimon-python/pypaimon/tests/api/__init__.py
@@ -0,0 +1,17 @@
+################################################################################
+# 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.
+################################################################################
diff --git a/paimon-python/pypaimon/tests/api/test_tag_dto_serde.py b/paimon-python/pypaimon/tests/api/test_tag_dto_serde.py
new file mode 100644
index 0000000000..793a6eecf8
--- /dev/null
+++ b/paimon-python/pypaimon/tests/api/test_tag_dto_serde.py
@@ -0,0 +1,143 @@
+################################################################################
+# 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.
+################################################################################
+
+"""DTO and resource-path unit tests for the Catalog tag CRUD wire format.
+
+These tests pin the Python wire format to the Java contract so review
+against ``paimon-api/.../rest/requests/CreateTagRequest.java``,
+``responses/GetTagResponse.java``, ``responses/ListTagsResponse.java``,
+and ``ResourcePaths.java`` does not need to inspect a running mock
+server. They explicitly assert the *exact* JSON field names and URL
+templates Java uses, and that ``ignoreIfExists`` is NOT serialized
+(it is a client-side flag — see ``RESTCatalog.createTag``).
+"""
+
+import json
+import unittest
+
+from pypaimon.api.api_request import CreateTagRequest
+from pypaimon.api.api_response import GetTagResponse, ListTagsResponse
+from pypaimon.api.resource_paths import ResourcePaths
+from pypaimon.common.json_util import JSON
+from pypaimon.snapshot.snapshot import Snapshot
+
+
+class CreateTagRequestSerdeTest(unittest.TestCase):
+
+ def test_to_json_minimal(self):
+ request = CreateTagRequest(tag_name="t1")
+ rendered = JSON.to_json(request)
+ # Field name must be ``tagName`` (Java) — not ``tag_name``.
+ self.assertIn('"tagName": "t1"', rendered)
+ # Strict alignment: ``ignoreIfExists`` MUST NOT appear in the
+ # serialized request. ``ignore_if_exists`` is a client-side flag
+ # handled by RESTCatalog and must never be sent over the wire.
+ self.assertNotIn("ignoreIfExists", rendered)
+ self.assertNotIn("ignore_if_exists", rendered)
+
+ def test_to_json_full_uses_java_field_names(self):
+ request = CreateTagRequest(
+ tag_name="release-1.0",
+ snapshot_id=42,
+ time_retained="1d",
+ )
+ parsed = json.loads(JSON.to_json(request))
+ self.assertEqual(parsed["tagName"], "release-1.0")
+ self.assertEqual(parsed["snapshotId"], 42)
+ self.assertEqual(parsed["timeRetained"], "1d")
+ self.assertNotIn("ignoreIfExists", parsed)
+ self.assertEqual(set(parsed.keys()), {"tagName", "snapshotId", "timeRetained"})
+
+
+class GetTagResponseSerdeTest(unittest.TestCase):
+
+ def _snapshot_payload(self):
+ return {
+ "version": 3,
+ "id": 7,
+ "schemaId": 0,
+ "baseManifestList": "manifest-list-base",
+ "deltaManifestList": "manifest-list-delta",
+ "totalRecordCount": 100,
+ "deltaRecordCount": 100,
+ "commitUser": "rest-server",
+ "commitIdentifier": 1,
+ "commitKind": "APPEND",
+ "timeMillis": 12345,
+ }
+
+ def test_from_json_populates_all_fields(self):
+ payload = json.dumps({
+ "tagName": "release-1.0",
+ "snapshot": self._snapshot_payload(),
+ "tagCreateTime": 99999,
+ "tagTimeRetained": "1d",
+ })
+ response = JSON.from_json(payload, GetTagResponse)
+ self.assertEqual(response.tag_name, "release-1.0")
+ self.assertIsInstance(response.snapshot, Snapshot)
+ self.assertEqual(response.snapshot.id, 7)
+ self.assertEqual(response.tag_create_time, 99999)
+ self.assertEqual(response.tag_time_retained, "1d")
+
+
+class ListTagsResponseSerdeTest(unittest.TestCase):
+
+ def test_from_json_populates_data_and_token(self):
+ response = JSON.from_json(
+ json.dumps({"tags": ["t1", "t2"], "nextPageToken": "tok"}),
+ ListTagsResponse,
+ )
+ self.assertEqual(response.data(), ["t1", "t2"])
+ self.assertEqual(response.get_next_page_token(), "tok")
+
+ def test_from_json_empty_payload(self):
+ response = JSON.from_json(
+ json.dumps({"tags": None, "nextPageToken": None}),
+ ListTagsResponse,
+ )
+ self.assertIn(response.data(), (None, []))
+ self.assertIsNone(response.get_next_page_token())
+
+
+class ResourcePathsTagsTest(unittest.TestCase):
+
+ def test_tags_collection_url(self):
+ paths = ResourcePaths(prefix="mock")
+ self.assertEqual(
+ paths.tags("db", "tbl"),
+ "/v1/mock/databases/db/tables/tbl/tags",
+ )
+
+ def test_single_tag_url(self):
+ paths = ResourcePaths(prefix="mock")
+ self.assertEqual(
+ paths.tag("db", "tbl", "release-1.0"),
+ "/v1/mock/databases/db/tables/tbl/tags/release-1.0",
+ )
+
+ def test_single_tag_url_url_encodes_tag_name(self):
+ # RESTUtil.encode_string escapes characters that are not URL-safe.
+ # A space round-trips to ``%20``; this matches Java's ``RESTUtil``.
+ paths = ResourcePaths(prefix="mock")
+ url = paths.tag("db", "tbl", "release 1.0")
+ self.assertEqual(url, "/v1/mock/databases/db/tables/tbl/tags/release%201.0")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py b/paimon-python/pypaimon/tests/rest/rest_server.py
index c3a4188669..1c8a146bc0 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -31,14 +31,17 @@ if TYPE_CHECKING:
from pypaimon.api.api_request import (AlterDatabaseRequest, AlterTableRequest,
CreateDatabaseRequest,
- CreateTableRequest, RenameTableRequest)
+ CreateTableRequest, CreateTagRequest,
+ RenameTableRequest)
from pypaimon.api.api_response import (ConfigResponse, GetDatabaseResponse,
GetFunctionResponse,
- GetTableResponse, ListDatabasesResponse,
+ GetTableResponse, GetTagResponse,
+ ListDatabasesResponse,
ListFunctionDetailsResponse,
ListFunctionsGloballyResponse,
ListFunctionsResponse,
ListPartitionsResponse, ListTablesResponse,
+ ListTagsResponse,
PagedList, Partition,
RESTResponse, ErrorResponse)
from pypaimon.api.resource_paths import ResourcePaths
@@ -51,7 +54,9 @@ from pypaimon.catalog.catalog_exception import (DatabaseNoPermissionException,
FunctionNotExistException,
FunctionAlreadyExistException,
DefinitionAlreadyExistException,
- DefinitionNotExistException)
+ DefinitionNotExistException,
+ TagNotExistException,
+ TagAlreadyExistException)
from pypaimon.catalog.rest.table_metadata import TableMetadata
from pypaimon.common.identifier import Identifier
from pypaimon.api.typedef import RESTAuthParameter
@@ -72,6 +77,7 @@ TABLE_TYPE = "tableType"
VIEW_NAME_PATTERN = "viewNamePattern"
FUNCTION_NAME_PATTERN = "functionNamePattern"
PARTITION_NAME_PATTERN = "partitionNamePattern"
+TAG_NAME_PREFIX = "tagNamePrefix"
MAX_RESULTS = "maxResults"
PAGE_TOKEN = "pageToken"
@@ -205,6 +211,8 @@ class RESTCatalogServer:
self.table_latest_snapshot_store: Dict[str, str] = {}
self.table_partitions_store: Dict[str, List] = {}
self.function_store: Dict[str, Dict] = {} # key: "db.func_name", value: GetFunctionResponse-like dict
+ # Tag store: key = full table name, value = {tag_name: GetTagResponse}.
+ self.tag_store: Dict[str, Dict[str, GetTagResponse]] = {}
self.no_permission_databases: List[str] = []
self.no_permission_tables: List[str] = []
self.table_token_store: Dict[str, "RESTToken"] = {}
@@ -466,6 +474,16 @@ class RESTCatalogServer:
ErrorResponse.RESOURCE_TYPE_FUNCTION, e.identifier.get_full_name(), str(e), 409
)
return self._mock_response(response, 409)
+ except TagNotExistException as e:
+ response = ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_TAG, e.tag, str(e), 404
+ )
+ return self._mock_response(response, 404)
+ except TagAlreadyExistException as e:
+ response = ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_TAG, e.tag, str(e), 409
+ )
+ return self._mock_response(response, 409)
except DefinitionAlreadyExistException as e:
response = ErrorResponse(
ErrorResponse.RESOURCE_TYPE_DEFINITION, e.name, str(e), 409
@@ -524,8 +542,13 @@ class RESTCatalogServer:
return self._table_snapshot_handle(method, lookup_identifier)
elif operation == ResourcePaths.PARTITIONS:
return self._table_partitions_handle(method, lookup_identifier, parameters)
+ elif operation == ResourcePaths.TAGS:
+ return self._tags_handle(method, data, lookup_identifier, parameters)
else:
return self._mock_response(ErrorResponse(None, None, "Not Found", 404), 404)
+ elif len(path_parts) == 5 and path_parts[3] == ResourcePaths.TAGS:
+ tag_name = RESTUtil.decode_string(path_parts[4])
+ return self._tag_handle(method, lookup_identifier, tag_name)
return self._mock_response(ErrorResponse(None, None, "Not Found", 404), 404)
# ======================= Function Handlers ===============================
@@ -722,6 +745,80 @@ class RESTCatalogServer:
partitions = self._list_partitions(identifier, parameters)
return self._generate_final_list_partitions_response(parameters, partitions)
+ # ======================= Tag Handlers ====================================
+
+ def _tags_handle(self, method: str, data: str, identifier: Identifier,
+ parameters: Dict[str, str]) -> Tuple[str, int]:
+ """Handle the table-scoped tags collection (POST create / GET list-paged)."""
+ if identifier.get_full_name() not in self.table_metadata_store:
+ raise TableNotExistException(identifier)
+
+ if method == "POST":
+ request = JSON.from_json(data, CreateTagRequest)
+ store = self.tag_store.setdefault(identifier.get_full_name(), {})
+ if request.tag_name in store:
+ raise TagAlreadyExistException(request.tag_name)
+ snapshot = self._resolve_tag_snapshot(identifier, request.snapshot_id)
+ store[request.tag_name] = GetTagResponse(
+ tag_name=request.tag_name,
+ snapshot=snapshot,
+ tag_create_time=int(time.time() * 1000),
+ tag_time_retained=request.time_retained,
+ )
+ return self._mock_response("", 200)
+
+ if method == "GET":
+ tags = list(self.tag_store.get(identifier.get_full_name(), {}).keys())
+ tag_name_prefix = parameters.get(TAG_NAME_PREFIX)
+ if tag_name_prefix:
+ tags = [t for t in tags if t.startswith(tag_name_prefix)]
+ if tags:
+ max_results = self._get_max_results(parameters)
+ page_token = parameters.get(PAGE_TOKEN)
+ paged = self._build_paged_entities(tags, max_results, page_token)
+ response = ListTagsResponse(
+ tags=paged.elements, next_page_token=paged.next_page_token)
+ else:
+ response = ListTagsResponse(tags=[], next_page_token=None)
+ return self._mock_response(response, 200)
+
+ return self._mock_response(ErrorResponse(None, None, "Method Not Allowed", 405), 405)
+
+ def _tag_handle(self, method: str, identifier: Identifier,
+ tag_name: str) -> Tuple[str, int]:
+ """Handle a single tag (GET / DELETE)."""
+ if identifier.get_full_name() not in self.table_metadata_store:
+ raise TableNotExistException(identifier)
+ store = self.tag_store.get(identifier.get_full_name(), {})
+
+ if method == "GET":
+ if tag_name not in store:
+ raise TagNotExistException(tag_name)
+ return self._mock_response(store[tag_name], 200)
+ if method == "DELETE":
+ if tag_name not in store:
+ raise TagNotExistException(tag_name)
+ del store[tag_name]
+ return self._mock_response("", 200)
+ return self._mock_response(ErrorResponse(None, None, "Method Not Allowed", 405), 405)
+
+ def _resolve_tag_snapshot(self, identifier: Identifier,
+ snapshot_id: Optional[int]):
+ """Look up the snapshot to embed in GetTagResponse.
+
+ When ``snapshot_id`` is None, fall back to the table's latest snapshot.
+ Mirrors Java behavior of ``TagManager`` resolving the latest snapshot
+ when no explicit id is supplied.
+ """
+ try:
+ table = self._get_file_table(identifier)
+ snapshot_manager = table.snapshot_manager()
+ if snapshot_id is None:
+ return snapshot_manager.get_latest_snapshot()
+ return snapshot_manager.get_snapshot_by_id(snapshot_id)
+ except Exception:
+ return None
+
def _databases_api_handler(self, method: str, data: str,
parameters: Dict[str, str]) -> Tuple[str, int]:
"""Handle databases API requests"""
diff --git a/paimon-python/pypaimon/tests/rest/rest_tag_test.py b/paimon-python/pypaimon/tests/rest/rest_tag_test.py
new file mode 100644
index 0000000000..c403eb0d57
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/rest_tag_test.py
@@ -0,0 +1,175 @@
+"""
+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.
+"""
+
+import os
+import shutil
+import tempfile
+import unittest
+
+from pypaimon.api.api_response import GetTagResponse
+from pypaimon.catalog.catalog_exception import (TableNotExistException,
+ TagAlreadyExistException,
+ TagNotExistException)
+from pypaimon import CatalogFactory
+from pypaimon.common.identifier import Identifier
+from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.tests.rest.rest_base_test import RESTBaseTest
+
+
+class RESTCatalogTagCRUDTest(RESTBaseTest):
+ """End-to-end Tag CRUD against the in-process mock REST server.
+
+ Mirrors the matrix exercised by Java
+ ``RESTCatalogTest::testTags`` (paimon-core/.../rest/RESTCatalogTest.java).
+ """
+
+ def _identifier(self):
+ # The base test creates ``default.test_reader_iterator`` and commits a
+ # snapshot in setUp, so it already has snapshot id 1 to tag.
+ return Identifier.from_string("default.test_reader_iterator")
+
+ def test_create_and_get_tag(self):
+ identifier = self._identifier()
+ self.rest_catalog.create_tag(identifier, "t1")
+ response = self.rest_catalog.get_tag(identifier, "t1")
+ self.assertIsInstance(response, GetTagResponse)
+ self.assertEqual(response.tag_name, "t1")
+ self.assertIsInstance(response.snapshot, Snapshot)
+
+ def test_list_tags_paged_basic(self):
+ identifier = self._identifier()
+ for name in ("t1", "t2", "t3"):
+ self.rest_catalog.create_tag(identifier, name)
+
+ all_tags = self.rest_catalog.list_tags_paged(identifier)
+ self.assertEqual(sorted(all_tags.elements), ["t1", "t2", "t3"])
+
+ first_page = self.rest_catalog.list_tags_paged(identifier, max_results=2)
+ self.assertEqual(len(first_page.elements), 2)
+ self.assertIsNotNone(first_page.next_page_token)
+
+ second_page = self.rest_catalog.list_tags_paged(
+ identifier, max_results=2, page_token=first_page.next_page_token)
+ self.assertEqual(len(second_page.elements), 1)
+ self.assertIsNone(second_page.next_page_token)
+
+ def test_list_tags_paged_with_prefix(self):
+ identifier = self._identifier()
+ for name in ("prod_v1", "prod_v2", "dev_v1"):
+ self.rest_catalog.create_tag(identifier, name)
+
+ result = self.rest_catalog.list_tags_paged(
+ identifier, tag_name_prefix="prod_")
+ self.assertEqual(sorted(result.elements), ["prod_v1", "prod_v2"])
+
+ def test_create_tag_with_snapshot_id(self):
+ identifier = self._identifier()
+ self.rest_catalog.create_tag(identifier, "t1", snapshot_id=1)
+ response = self.rest_catalog.get_tag(identifier, "t1")
+ self.assertIsNotNone(response.snapshot)
+ self.assertEqual(response.snapshot.id, 1)
+
+ def test_create_tag_with_time_retained(self):
+ identifier = self._identifier()
+ self.rest_catalog.create_tag(identifier, "t1", time_retained="1d")
+ response = self.rest_catalog.get_tag(identifier, "t1")
+ self.assertEqual(response.tag_time_retained, "1d")
+
+ def test_create_tag_already_exists_raises(self):
+ identifier = self._identifier()
+ self.rest_catalog.create_tag(identifier, "t1")
+ with self.assertRaises(TagAlreadyExistException):
+ self.rest_catalog.create_tag(identifier, "t1")
+
+ def test_create_tag_already_exists_ignore(self):
+ identifier = self._identifier()
+ self.rest_catalog.create_tag(identifier, "t1")
+ # ignore_if_exists=True swallows the conflict; no exception raised.
+ self.rest_catalog.create_tag(identifier, "t1", ignore_if_exists=True)
+ # Still only one tag.
+ result = self.rest_catalog.list_tags_paged(identifier)
+ self.assertEqual(result.elements, ["t1"])
+
+ def test_create_tag_table_not_exists(self):
+ with self.assertRaises(TableNotExistException):
+ self.rest_catalog.create_tag(
+ Identifier.from_string("default.no_such_table"), "t1")
+
+ def test_get_tag_not_exists(self):
+ identifier = self._identifier()
+ with self.assertRaises(TagNotExistException):
+ self.rest_catalog.get_tag(identifier, "absent")
+
+ def test_delete_tag(self):
+ identifier = self._identifier()
+ self.rest_catalog.create_tag(identifier, "t1")
+ self.rest_catalog.delete_tag(identifier, "t1")
+ result = self.rest_catalog.list_tags_paged(identifier)
+ self.assertNotIn("t1", result.elements)
+
+ def test_delete_tag_not_exists(self):
+ identifier = self._identifier()
+ with self.assertRaises(TagNotExistException):
+ self.rest_catalog.delete_tag(identifier, "absent")
+
+
+class FilesystemCatalogTagInheritsNotImplementedTest(unittest.TestCase):
+ """The filesystem catalog inherits the abstract NotImplementedError stubs.
+
+ A concrete tag implementation requires a Python-side TagManager port and
+ is tracked separately. The point of this test is to confirm the
+ abstract API gap is closed: ``FileSystemCatalog().create_tag(...)``
+ raises ``NotImplementedError`` rather than ``AttributeError``.
+ """
+
+ def setUp(self):
+ self.temp_dir = tempfile.mkdtemp(prefix="unittest_tag_")
+ warehouse = os.path.join(self.temp_dir, "warehouse")
+ os.makedirs(warehouse, exist_ok=True)
+ # Use the default (filesystem) catalog. Without a ``metastore`` option
+ # CatalogFactory returns FileSystemCatalog, which inherits the abstract
+ # ``NotImplementedError`` tag stubs.
+ self.catalog = CatalogFactory.create({"warehouse": warehouse})
+
+ def tearDown(self):
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+ def test_create_tag_raises_not_implemented(self):
+ with self.assertRaises(NotImplementedError) as cm:
+ self.catalog.create_tag(
+ Identifier.from_string("default.tbl"), "t1")
+ self.assertIn("create_tag", str(cm.exception))
+
+ def test_delete_tag_raises_not_implemented(self):
+ with self.assertRaises(NotImplementedError):
+ self.catalog.delete_tag(
+ Identifier.from_string("default.tbl"), "t1")
+
+ def test_get_tag_raises_not_implemented(self):
+ with self.assertRaises(NotImplementedError):
+ self.catalog.get_tag(
+ Identifier.from_string("default.tbl"), "t1")
+
+ def test_list_tags_paged_raises_not_implemented(self):
+ with self.assertRaises(NotImplementedError):
+ self.catalog.list_tags_paged(
+ Identifier.from_string("default.tbl"))
+
+
+if __name__ == "__main__":
+ unittest.main()