remove use of catalog identifiers

Author: sungwy

pyiceberg/catalog/__init__.py

@@ -649,25 +649,6 @@ def identifier_to_tuple_without_catalog(self, identifier: Union[str, Identifier]
             identifier_tuple = identifier_tuple[1:]
         return identifier_tuple
 
-    def _identifier_to_tuple_without_catalog(self, identifier: Union[str, Identifier]) -> Identifier:
-        """Convert an identifier to a tuple and drop this catalog's name from the first element.
-
-        Args:
-            identifier (str | Identifier): Table identifier.
-
-        Returns:
-            Identifier: a tuple of strings with this catalog's name removed
-        """
-        identifier_tuple = Catalog.identifier_to_tuple(identifier)
-        if len(identifier_tuple) >= 3 and identifier_tuple[0] == self.name:
-            deprecation_message(
-                deprecated_in="0.8.0",
-                removed_in="0.9.0",
-                help_message="Support for parsing catalog level identifier in Catalog identifiers is deprecated. Please refer to the table using only its namespace and its table name.",
-            )
-            identifier_tuple = identifier_tuple[1:]
-        return identifier_tuple
-
     @staticmethod
     def identifier_to_tuple(identifier: Union[str, Identifier]) -> Identifier:
         """Parse an identifier to a tuple.
@@ -809,9 +790,8 @@ def table_exists(self, identifier: Union[str, Identifier]) -> bool:
             return False
 
     def purge_table(self, identifier: Union[str, Identifier]) -> None:
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
-        table = self.load_table(identifier_tuple)
-        self.drop_table(identifier_tuple)
+        table = self.load_table(identifier)
+        self.drop_table(identifier)
         io = load_file_io(self.properties, table.metadata_location)
         metadata = table.metadata
         manifest_lists_to_delete = set()

pyiceberg/catalog/dynamodb.py

@@ -257,8 +257,7 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
-        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
+        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
 
         try:
             self._delete_dynamo_item(
@@ -288,8 +287,7 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             NoSuchPropertyException: When from table miss some required properties.
             NoSuchNamespaceError: When the destination namespace doesn't exist.
         """
-        from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
-        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
+        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
 
         from_table_item = self._get_iceberg_table_item(database_name=from_database_name, table_name=from_table_name)
@@ -320,7 +318,7 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             raise TableAlreadyExistsError(f"Table {to_database_name}.{to_table_name} already exists") from e
 
         try:
-            self.drop_table(from_identifier_tuple)
+            self.drop_table(from_identifier)
         except (NoSuchTableError, GenericDynamoDbError) as e:
             log_message = f"Failed to drop old table {from_database_name}.{from_table_name}. "
 

pyiceberg/catalog/glue.py

@@ -569,8 +569,7 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
-        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
+        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
         try:
             self.glue.delete_table(DatabaseName=database_name, Name=table_name)
         except self.glue.exceptions.EntityNotFoundException as e:
@@ -595,8 +594,7 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             NoSuchPropertyException: When from table miss some required properties.
             NoSuchNamespaceError: When the destination namespace doesn't exist.
         """
-        from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
-        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
+        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
         try:
             get_table_response = self.glue.get_table(DatabaseName=from_database_name, Name=from_table_name)

pyiceberg/catalog/hive.py

@@ -547,8 +547,7 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
-        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
+        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
         try:
             with self._client as open_client:
                 open_client.drop_table(dbname=database_name, name=table_name, deleteData=False)
@@ -575,8 +574,7 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             NoSuchTableError: When a table with the name does not exist.
             NoSuchNamespaceError: When the destination namespace doesn't exist.
         """
-        from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
-        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
+        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
         try:
             with self._client as open_client:

pyiceberg/catalog/rest.py

@@ -577,7 +577,6 @@ def _create_table(
         fresh_partition_spec = assign_fresh_partition_spec_ids(partition_spec, iceberg_schema, fresh_schema)
         fresh_sort_order = assign_fresh_sort_order_ids(sort_order, iceberg_schema, fresh_schema)
 
-        identifier = self._identifier_to_tuple_without_catalog(identifier)
         namespace_and_table = self._split_identifier_for_path(identifier)
         if location:
             location = location.rstrip("/")
@@ -658,7 +657,6 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
         Raises:
             TableAlreadyExistsError: If the table already exists
         """
-        identifier = self._identifier_to_tuple_without_catalog(identifier)
         namespace_and_table = self._split_identifier_for_path(identifier)
         request = RegisterTableRequest(
             name=namespace_and_table["table"],
@@ -701,11 +699,8 @@ def load_table(self, identifier: Union[str, Identifier]) -> Table:
 
     @retry(**_RETRY_ARGS)
     def drop_table(self, identifier: Union[str, Identifier], purge_requested: bool = False) -> None:
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         response = self._session.delete(
-            self.url(
-                Endpoints.drop_table, prefixed=True, purge=purge_requested, **self._split_identifier_for_path(identifier_tuple)
-            ),
+            self.url(Endpoints.drop_table, prefixed=True, purge=purge_requested, **self._split_identifier_for_path(identifier)),
         )
         try:
             response.raise_for_status()
@@ -718,9 +713,8 @@ def purge_table(self, identifier: Union[str, Identifier]) -> None:
 
     @retry(**_RETRY_ARGS)
     def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[str, Identifier]) -> Table:
-        from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
         payload = {
-            "source": self._split_identifier_for_json(from_identifier_tuple),
+            "source": self._split_identifier_for_json(from_identifier),
             "destination": self._split_identifier_for_json(to_identifier),
         }
         response = self._session.post(self.url(Endpoints.rename_table), json=payload)
@@ -877,9 +871,8 @@ def table_exists(self, identifier: Union[str, Identifier]) -> bool:
         Returns:
             bool: True if the table exists, False otherwise.
         """
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         response = self._session.head(
-            self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier_tuple))
+            self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier))
         )
 
         if response.status_code == 404:
@@ -896,11 +889,8 @@ def table_exists(self, identifier: Union[str, Identifier]) -> bool:
 
     @retry(**_RETRY_ARGS)
     def drop_view(self, identifier: Union[str]) -> None:
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         response = self._session.delete(
-            self.url(
-                Endpoints.drop_view, prefixed=True, **self._split_identifier_for_path(identifier_tuple, IdentifierKind.VIEW)
-            ),
+            self.url(Endpoints.drop_view, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW)),
         )
         try:
             response.raise_for_status()

pyiceberg/catalog/sql.py

@@ -200,9 +200,8 @@ def create_table(
         """
         schema: Schema = self._convert_schema_if_needed(schema)  # type: ignore
 
-        identifier_nocatalog = self._identifier_to_tuple_without_catalog(identifier)
-        namespace_identifier = Catalog.namespace_from(identifier_nocatalog)
-        table_name = Catalog.table_name_from(identifier_nocatalog)
+        namespace_identifier = Catalog.namespace_from(identifier)
+        table_name = Catalog.table_name_from(identifier)
         if not self._namespace_exists(namespace_identifier):
             raise NoSuchNamespaceError(f"Namespace does not exist: {namespace_identifier}")
 
@@ -246,10 +245,9 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
             TableAlreadyExistsError: If the table already exists
             NoSuchNamespaceError: If namespace does not exist
         """
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
-        namespace_tuple = Catalog.namespace_from(identifier_tuple)
+        namespace_tuple = Catalog.namespace_from(identifier)
         namespace = Catalog.namespace_to_string(namespace_tuple)
-        table_name = Catalog.table_name_from(identifier_tuple)
+        table_name = Catalog.table_name_from(identifier)
         if not self._namespace_exists(namespace):
             raise NoSuchNamespaceError(f"Namespace does not exist: {namespace}")
 
@@ -308,10 +306,9 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist.
         """
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
-        namespace_tuple = Catalog.namespace_from(identifier_tuple)
+        namespace_tuple = Catalog.namespace_from(identifier)
         namespace = Catalog.namespace_to_string(namespace_tuple)
-        table_name = Catalog.table_name_from(identifier_tuple)
+        table_name = Catalog.table_name_from(identifier)
         with Session(self.engine) as session:
             if self.engine.dialect.supports_sane_rowcount:
                 res = session.execute(
@@ -355,14 +352,12 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             TableAlreadyExistsError: If a table with the new name already exist.
             NoSuchNamespaceError: If the target namespace does not exist.
         """
-        from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
-        to_identifier_tuple = self._identifier_to_tuple_without_catalog(to_identifier)
-        from_namespace_tuple = Catalog.namespace_from(from_identifier_tuple)
+        from_namespace_tuple = Catalog.namespace_from(from_identifier)
         from_namespace = Catalog.namespace_to_string(from_namespace_tuple)
-        from_table_name = Catalog.table_name_from(from_identifier_tuple)
-        to_namespace_tuple = Catalog.namespace_from(to_identifier_tuple)
+        from_table_name = Catalog.table_name_from(from_identifier)
+        to_namespace_tuple = Catalog.namespace_from(to_identifier)
         to_namespace = Catalog.namespace_to_string(to_namespace_tuple)
-        to_table_name = Catalog.table_name_from(to_identifier_tuple)
+        to_table_name = Catalog.table_name_from(to_identifier)
         if not self._namespace_exists(to_namespace):
             raise NoSuchNamespaceError(f"Namespace does not exist: {to_namespace}")
         with Session(self.engine) as session:

tests/catalog/test_base.py

@@ -156,28 +156,28 @@ def commit_table(
         return CommitTableResponse(metadata=updated_metadata, metadata_location=new_metadata_location)
 
     def load_table(self, identifier: Union[str, Identifier]) -> Table:
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         try:
-            return self.__tables[identifier_tuple]
+            identifier = Catalog.identifier_to_tuple(identifier)
+            return self.__tables[identifier]
         except KeyError as error:
-            raise NoSuchTableError(f"Table does not exist: {identifier_tuple}") from error
+            raise NoSuchTableError(f"Table does not exist: {identifier}") from error
 
     def drop_table(self, identifier: Union[str, Identifier]) -> None:
-        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         try:
-            self.__tables.pop(identifier_tuple)
+            identifier = Catalog.identifier_to_tuple(identifier)
+            self.__tables.pop(identifier)
         except KeyError as error:
-            raise NoSuchTableError(f"Table does not exist: {identifier_tuple}") from error
+            raise NoSuchTableError(f"Table does not exist: {identifier}") from error
 
     def purge_table(self, identifier: Union[str, Identifier]) -> None:
         self.drop_table(identifier)
 
     def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[str, Identifier]) -> Table:
-        identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
         try:
-            table = self.__tables.pop(identifier_tuple)
+            from_identifier = Catalog.identifier_to_tuple(from_identifier)
+            table = self.__tables.pop(from_identifier)
         except KeyError as error:
-            raise NoSuchTableError(f"Table does not exist: {identifier_tuple}") from error
+            raise NoSuchTableError(f"Table does not exist: {from_identifier}") from error
 
         to_identifier = Catalog.identifier_to_tuple(to_identifier)
         to_namespace = Catalog.namespace_from(to_identifier)
@@ -514,7 +514,7 @@ def test_rename_table(catalog: InMemoryCatalog) -> None:
 
     # Then
     assert table._identifier == Catalog.identifier_to_tuple(new_table)
-
+    print(f"DEBUG: {table._identifier}")
     # And
     table = catalog.load_table(new_table)
     assert table._identifier == Catalog.identifier_to_tuple(new_table)