[python] Support getting a table by id in the REST catalog (#10697)

Author: plusplusjiajia

docs/docs/pypaimon/catalogs.mdx

@@ -264,6 +264,9 @@ catalog.create_table(
 
 # Get Table
 table = catalog.get_table('database_name.table_name')
+
+# Get Table by id (REST catalog only)
+table = catalog.get_table_by_id('table_id')
 ```
 
 ## Alter Table

paimon-python/pypaimon/api/api_response.py

@@ -247,13 +247,16 @@ class GetTableResponse(AuditRESTResponse):
 
     # Field constants for JSON serialization
     FIELD_ID = "id"
+    FIELD_DATABASE = "database"
     FIELD_NAME = "name"
     FIELD_PATH = "path"
     FIELD_IS_EXTERNAL = "isExternal"
     FIELD_SCHEMA_ID = "schemaId"
     FIELD_SCHEMA = "schema"
 
     id: Optional[str] = json_field(FIELD_ID, default=None)
+    # Omitted when None: native readers compare it with the table's database
+    database: Optional[str] = optional_json_field(FIELD_DATABASE, "non_null")
     name: Optional[str] = json_field(FIELD_NAME, default=None)
     path: Optional[str] = json_field(FIELD_PATH, default=None)
     is_external: Optional[bool] = json_field(FIELD_IS_EXTERNAL, default=None)
@@ -273,9 +276,11 @@ def __init__(
             created_by: Optional[str] = None,
             updated_at: Optional[int] = None,
             updated_by: Optional[str] = None,
+            database: Optional[str] = None,
     ):
         super().__init__(owner, created_at, created_by, updated_at, updated_by)
         self.id = id
+        self.database = database
         self.name = name
         self.path = path
         self.is_external = is_external
@@ -285,6 +290,9 @@ def __init__(
     def get_id(self) -> str:
         return self.id
 
+    def get_database(self) -> Optional[str]:
+        return self.database
+
     def get_name(self) -> str:
         return self.name
 

paimon-python/pypaimon/api/resource_paths.py

@@ -27,6 +27,7 @@ class ResourcePaths:
     V1 = "v1"
     DATABASES = "databases"
     TABLES = "tables"
+    ID = "id"
     TABLE_DETAILS = "table-details"
     PARTITIONS = "partitions"
     FUNCTIONS = "functions"
@@ -70,6 +71,11 @@ def table(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)))
 
+    def table_by_id(self, table_id: str) -> str:
+        # Keep '~' raw, as in table_details.
+        table_id = urllib.parse.quote_plus(table_id, safe="*~", errors="replace")
+        return "{}/{}/{}/{}".format(self.base_path, self.TABLES, self.ID, table_id)
+
     def table_details(self, database_name: str) -> str:
         database_name = urllib.parse.quote_plus(database_name, safe="*~", errors="replace")
         return "{}/{}/{}/{}".format(

paimon-python/pypaimon/api/rest_api.py

@@ -313,6 +313,16 @@ def get_table(self, identifier: Identifier) -> GetTableResponse:
             self.rest_auth_function,
         )
 
+    def get_table_by_id(self, table_id: str) -> GetTableResponse:
+        if not table_id or not table_id.strip():
+            raise ValueError("Table id cannot be empty")
+
+        return self.client.get(
+            self.resource_paths.table_by_id(table_id),
+            GetTableResponse,
+            self.rest_auth_function,
+        )
+
     def drop_table(self, identifier: Identifier) -> GetTableResponse:
         database_name, table_name = self.__validate_identifier(identifier)
 

paimon-python/pypaimon/catalog/catalog.py

@@ -97,6 +97,12 @@ def alter_database(self, name: str, changes: List[PropertyChange]) -> None:
     def get_table(self, identifier: Union[str, Identifier]) -> Table:
         """Get paimon table identified by the given Identifier."""
 
+    def get_table_by_id(self, table_id: str) -> Table:
+        """Get paimon table identified by the given table id."""
+        raise NotImplementedError(
+            "get_table_by_id is not supported by this catalog."
+        )
+
     @abstractmethod
     def create_table(self, identifier: Union[str, Identifier], schema: Schema, ignore_if_exists: bool) -> None:
         """Create table with schema."""

paimon-python/pypaimon/catalog/catalog_exception.py

@@ -71,6 +71,22 @@ def __init__(self, identifier: Identifier):
         super().__init__(f"No permission to access table {identifier.get_full_name()}")
 
 
+class TableIdNotExistException(CatalogException):
+    """Table id not exist exception"""
+
+    def __init__(self, table_id: str):
+        self.table_id = table_id
+        super().__init__(f"Table id {table_id} does not exist")
+
+
+class TableIdNoPermissionException(CatalogException):
+    """Table id no permission exception"""
+
+    def __init__(self, table_id: str):
+        self.table_id = table_id
+        super().__init__(f"No permission to access table id {table_id}")
+
+
 class ViewNotExistException(CatalogException):
     """View not exist exception"""
 

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

@@ -32,6 +32,7 @@
     TableNotExistException, DatabaseAlreadyExistException,
     TableAlreadyExistException, DatabaseNotExistException,
     TableNoPermissionException, DatabaseNoPermissionException,
+    TableIdNotExistException, TableIdNoPermissionException,
     FunctionNotExistException, FunctionAlreadyExistException,
     DefinitionAlreadyExistException, DefinitionNotExistException,
     TagNotExistException, TagAlreadyExistException,
@@ -250,6 +251,22 @@ def get_table(self, identifier: Union[str, Identifier]) -> Table:
             return self._load_system_table(identifier)
         return self._load_data_table(identifier)
 
+    def get_table_by_id(self, table_id: str) -> Table:
+        try:
+            response = self.rest_api.get_table_by_id(table_id)
+        except NoSuchResourceException as e:
+            raise TableIdNotExistException(table_id) from e
+        except ForbiddenException as e:
+            raise TableIdNoPermissionException(table_id) from e
+        database = response.get_database()
+        identifier = Identifier.create(database, response.get_name())
+        return self.load_table(
+            identifier,
+            lambda path: self.file_io_for_data(path, identifier),
+            self.file_io_from_options,
+            lambda _: self.to_table_metadata(database, response),
+        )
+
     def _load_data_table(self, identifier: Identifier) -> Union[FormatTable, IcebergTable, ObjectTable, FileStoreTable]:
         return self.load_table(
             identifier,

paimon-python/pypaimon/tests/native_plan_rest_test.py

@@ -208,8 +208,8 @@ def token_file_io(table):
             assert load.call_count == 1
 
 
-@pytest.mark.parametrize('branch', [None, 'dev'])
-def test_rest_dotted_database_and_table_keep_identity(rest_catalog, branch):
+@pytest.mark.parametrize('branch,by_id', [(None, False), ('dev', False), (None, True)])
+def test_rest_dotted_database_and_table_keep_identity(rest_catalog, branch, by_id):
     from pypaimon.common.identifier import Identifier
     from pypaimon.tests.native_plan_resolved_schema_test import _assert_parity, _write
     catalog, server = rest_catalog
@@ -219,6 +219,8 @@ def test_rest_dotted_database_and_table_keep_identity(rest_catalog, branch):
         pa.schema([('id', pa.int64()), ('value', pa.string())])), False)
     table = catalog.get_table(identifier)
     _write(table, [{'id': 1, 'value': 'old'}])
+    if by_id:
+        table = catalog.get_table_by_id(table.catalog_environment.uuid)
     if branch:
         catalog.create_tag(identifier, 'first', 1)
         catalog.create_branch(identifier, branch, tag_name='first')

paimon-python/pypaimon/tests/rest/api_test.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import json
 import logging
 import unittest
 import uuid
@@ -156,6 +157,22 @@ def test_parse_data(self):
         self.assertFalse(vector_type.element.nullable)
         self.assertEqual(vector_type.length, 7)
 
+    def test_get_table_response_database(self):
+        from pypaimon.api.api_response import GetTableResponse
+        from pypaimon.schema.schema import Schema
+
+        # JSON as the Java server sends it
+        parsed = JSON.from_json(
+            '{"id": "id", "database": "db", "name": "t", "path": "path", "isExternal": false, '
+            '"schemaId": 0, "schema": {"fields": [], "partitionKeys": [], "primaryKeys": [], '
+            '"options": {}, "comment": null}}', GetTableResponse)
+        self.assertEqual(parsed.get_database(), 'db')
+        self.assertEqual(parsed.get_name(), 't')
+        self.assertEqual(json.loads(JSON.to_json(parsed))['database'], 'db')
+
+        response = GetTableResponse('id', 't', 'path', False, 0, Schema())
+        self.assertNotIn('database', json.loads(JSON.to_json(response)))
+
     def test_api(self):
         """Example usage of RESTCatalogServer"""
         # Setup logging
@@ -207,6 +224,10 @@ def test_api(self):
             self.assertEqual(rest_api.get_database('default'), test_databases.get('default'))
             table = rest_api.get_table(Identifier.from_string('default.user'))
             self.assertEqual(table.id, str(test_tables['default.user'].uuid))
+            self.assertEqual(table.get_database(), 'default')
+            table_by_id = rest_api.get_table_by_id(table.id)
+            self.assertEqual(table_by_id.get_database(), 'default')
+            self.assertEqual(table_by_id.get_name(), 'user')
 
         finally:
             # Shutdown server
@@ -293,6 +314,12 @@ def test_rest_api_parameter_validation(self):
             rest_api.get_database("   ")
         self.assertIn("Database name cannot be empty", str(context.exception))
 
+        # Test get_table_by_id with empty id
+        for table_id in ("", "   "):
+            with self.assertRaises(ValueError) as context:
+                rest_api.get_table_by_id(table_id)
+            self.assertIn("Table id cannot be empty", str(context.exception))
+
         # Test drop_database with empty name
         with self.assertRaises(ValueError) as context:
             rest_api.drop_database("")

paimon-python/pypaimon/tests/rest/resource_paths_test.py

@@ -16,6 +16,7 @@
 # under the License.
 
 
+import itertools
 import threading
 import unittest
 from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -48,8 +49,9 @@ def test_database_and_table_names_are_url_encoded(self):
             paths.table("sales db", "orders/all"))
         self.assertEqual(
             "/v1/catalog%2Fid/databases/sales+db/table-details", paths.table_details("sales db"))
+        self.assertEqual("/v1/catalog%2Fid/tables/id/a%2Fb+c", paths.table_by_id("a/b c"))
 
-    def test_table_details_signature_matches_the_path_sent(self):
+    def test_signature_matches_the_path_sent(self):
         received = []
 
         class Handler(BaseHTTPRequestHandler):
@@ -74,22 +76,24 @@ def do_GET(self):
         token = DLFToken("test-ak", "test-sk", "test-token", None)
         paths = ResourcePaths("catalog~id")
 
+        builders = ((paths.table_details, "/v1/catalog~id/databases/{}/table-details"),
+                    (paths.table_by_id, "/v1/catalog~id/tables/id/{}"))
         for algorithm in ("default", "openapi-v4"):
             provider = DLFAuthProvider(uri, "cn-hangzhou", algorithm, token=token)
-            for database, encoded in (("sales~db", "sales~db"),
-                                      ("sales~ db/a+b*", "sales~+db%2Fa%2Bb*"),
-                                      ("sales%7Edb", "sales%257Edb")):
-                with self.subTest(algorithm=algorithm, database=database):
+            for (build, template), (name, encoded) in itertools.product(
+                    builders, (("sales~db", "sales~db"),
+                               ("sales~ db/a+b*", "sales~+db%2Fa%2Bb*"),
+                               ("sales%7Edb", "sales%257Edb"))):
+                with self.subTest(algorithm=algorithm, path=template, name=name):
                     signed = []
 
                     def sign(parameter):
                         signed.append(parameter.path)
                         return provider.merge_auth_header({}, parameter)
 
-                    client.get(paths.table_details(database), None, sign)
+                    client.get(build(name), None, sign)
                     path, headers = received[-1]
-                    self.assertEqual(
-                        "/v1/catalog~id/databases/{}/table-details".format(encoded), path)
+                    self.assertEqual(template.format(encoded), path)
                     expected = provider.signer.authorization(
                         RESTAuthParameter("GET", path, "", {}), token,
                         provider.extract_host(uri), headers)