Deprecate Redundant Identifier Support in TableIdentifier, and row_filter (apache#994)

Author: sungwy

pyiceberg/catalog/__init__.py

@@ -67,7 +67,7 @@
     RecursiveDict,
 )
 from pyiceberg.utils.config import Config, merge_config
-from pyiceberg.utils.deprecated import deprecation_message
+from pyiceberg.utils.deprecated import deprecated, deprecation_message
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -613,6 +613,11 @@ def update_namespace_properties(
             ValueError: If removals and updates have overlapping keys.
         """
 
+    @deprecated(
+        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.",
+    )
     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.
 
@@ -627,6 +632,25 @@ 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.
@@ -769,7 +793,7 @@ 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)
+        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         table = self.load_table(identifier_tuple)
         self.drop_table(identifier_tuple)
         io = load_file_io(self.properties, table.metadata_location)
@@ -823,7 +847,7 @@ def _create_staged_table(
         )
         io = self._load_file_io(properties=properties, location=metadata_location)
         return StagedTable(
-            identifier=(self.name, database_name, table_name),
+            identifier=(database_name, table_name),
             metadata=metadata,
             metadata_location=metadata_location,
             io=io,

pyiceberg/catalog/dynamodb.py

@@ -259,7 +259,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)
+        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
 
         try:
@@ -290,7 +290,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_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)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
 

pyiceberg/catalog/glue.py

@@ -462,7 +462,7 @@ def _commit_table(self, table_request: CommitTableRequest) -> CommitTableRespons
             NoSuchTableError: If a table with the given identifier does not exist.
             CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
         """
-        identifier_tuple = self.identifier_to_tuple_without_catalog(
+        identifier_tuple = self._identifier_to_tuple_without_catalog(
             tuple(table_request.identifier.namespace.root + [table_request.identifier.name])
         )
         database_name, table_name = self.identifier_to_database_and_table(identifier_tuple)
@@ -554,7 +554,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)
+        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         try:
             self.glue.delete_table(DatabaseName=database_name, Name=table_name)
@@ -580,7 +580,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_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)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
         try:

pyiceberg/catalog/hive.py

@@ -297,7 +297,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable) -> Table:
         )
 
     def _convert_iceberg_into_hive(self, table: Table) -> HiveTable:
-        identifier_tuple = self.identifier_to_tuple_without_catalog(table.identifier)
+        identifier_tuple = self._identifier_to_tuple_without_catalog(table.identifier)
         database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         current_time_millis = int(time.time() * 1000)
 
@@ -431,7 +431,7 @@ def _commit_table(self, table_request: CommitTableRequest) -> CommitTableRespons
             NoSuchTableError: If a table with the given identifier does not exist.
             CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
         """
-        identifier_tuple = self.identifier_to_tuple_without_catalog(
+        identifier_tuple = self._identifier_to_tuple_without_catalog(
             tuple(table_request.identifier.namespace.root + [table_request.identifier.name])
         )
         database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
@@ -477,7 +477,7 @@ def _commit_table(self, table_request: CommitTableRequest) -> CommitTableRespons
                     # Table does not exist, create it.
                     hive_table = self._convert_iceberg_into_hive(
                         StagedTable(
-                            identifier=(self.name, database_name, table_name),
+                            identifier=(database_name, table_name),
                             metadata=updated_staged_table.metadata,
                             metadata_location=updated_staged_table.metadata_location,
                             io=updated_staged_table.io,
@@ -525,7 +525,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)
+        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         try:
             with self._client as open_client:
@@ -553,7 +553,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_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)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
         try:

pyiceberg/catalog/rest.py

@@ -672,7 +672,7 @@ 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)
+        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)
@@ -689,7 +689,7 @@ 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)
+        from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
         payload = {
             "source": self._split_identifier_for_json(from_identifier_tuple),
             "destination": self._split_identifier_for_json(to_identifier),
@@ -824,7 +824,7 @@ 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)
+        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))
         )

pyiceberg/catalog/sql.py

@@ -192,7 +192,7 @@ def create_table(
         """
         schema: Schema = self._convert_schema_if_needed(schema)  # type: ignore
 
-        identifier_nocatalog = self.identifier_to_tuple_without_catalog(identifier)
+        identifier_nocatalog = self._identifier_to_tuple_without_catalog(identifier)
         namespace_identifier = Catalog.namespace_from(identifier_nocatalog)
         table_name = Catalog.table_name_from(identifier_nocatalog)
         if not self._namespace_exists(namespace_identifier):
@@ -238,7 +238,7 @@ 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)
+        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         namespace_tuple = Catalog.namespace_from(identifier_tuple)
         namespace = Catalog.namespace_to_string(namespace_tuple)
         table_name = Catalog.table_name_from(identifier_tuple)
@@ -300,7 +300,7 @@ 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)
+        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
         namespace_tuple = Catalog.namespace_from(identifier_tuple)
         namespace = Catalog.namespace_to_string(namespace_tuple)
         table_name = Catalog.table_name_from(identifier_tuple)
@@ -347,8 +347,8 @@ 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_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 = Catalog.namespace_to_string(from_namespace_tuple)
         from_table_name = Catalog.table_name_from(from_identifier_tuple)
@@ -406,7 +406,7 @@ def _commit_table(self, table_request: CommitTableRequest) -> CommitTableRespons
             NoSuchTableError: If a table with the given identifier does not exist.
             CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
         """
-        identifier_tuple = self.identifier_to_tuple_without_catalog(
+        identifier_tuple = self._identifier_to_tuple_without_catalog(
             tuple(table_request.identifier.namespace.root + [table_request.identifier.name])
         )
         namespace_tuple = Catalog.namespace_from(identifier_tuple)

pyiceberg/expressions/parser.py

@@ -64,6 +64,7 @@
 )
 from pyiceberg.typedef import L
 from pyiceberg.types import strtobool
+from pyiceberg.utils.deprecated import deprecation_message
 
 ParserElement.enablePackrat()
 
@@ -84,6 +85,14 @@
 
 @column.set_parse_action
 def _(result: ParseResults) -> Reference:
+    if len(result.column) > 1:
+        deprecation_message(
+            deprecated_in="0.8.0",
+            removed_in="0.9.0",
+            help_message="Parsing expressions with table name is deprecated. Only provide field names in the row_filter.",
+        )
+    # TODO: Once this is removed, we will no longer take just the last index of parsed column result
+    # And introduce support for parsing filter expressions with nested fields.
     return Reference(result.column[-1])
 
 

pyiceberg/table/__init__.py

@@ -152,7 +152,7 @@
 from pyiceberg.utils.concurrent import ExecutorFactory
 from pyiceberg.utils.config import Config
 from pyiceberg.utils.datetime import datetime_to_millis
-from pyiceberg.utils.deprecated import deprecated
+from pyiceberg.utils.deprecated import deprecated, deprecation_message
 from pyiceberg.utils.properties import property_as_bool, property_as_int
 from pyiceberg.utils.singleton import _convert_to_hashable_type
 
@@ -1392,7 +1392,7 @@ class CommitTableResponse(IcebergBaseModel):
 
 
 class Table:
-    identifier: Identifier = Field()
+    _identifier: Identifier = Field()
     metadata: TableMetadata
     metadata_location: str = Field()
     io: FileIO
@@ -1401,7 +1401,7 @@ class Table:
     def __init__(
         self, identifier: Identifier, metadata: TableMetadata, metadata_location: str, io: FileIO, catalog: Catalog
     ) -> None:
-        self.identifier = identifier
+        self._identifier = identifier
         self.metadata = metadata
         self.metadata_location = metadata_location
         self.io = io
@@ -1428,6 +1428,16 @@ def refresh(self) -> Table:
         self.metadata_location = fresh.metadata_location
         return self
 
+    @property
+    def identifier(self) -> Identifier:
+        """Return the identifier of this table."""
+        deprecation_message(
+            deprecated_in="0.8.0",
+            removed_in="0.9.0",
+            help_message="Table.identifier property is deprecated. Please use Table.name() function instead.",
+        )
+        return (self.catalog.name,) + self._identifier
+
     def name(self) -> Identifier:
         """Return the identifier of this table."""
         return self.identifier
@@ -1643,7 +1653,7 @@ def refs(self) -> Dict[str, SnapshotRef]:
     def _do_commit(self, updates: Tuple[TableUpdate, ...], requirements: Tuple[TableRequirement, ...]) -> None:
         response = self.catalog._commit_table(  # pylint: disable=W0212
             CommitTableRequest(
-                identifier=TableIdentifier(namespace=self.identifier[:-1], name=self.identifier[-1]),
+                identifier=TableIdentifier(namespace=self._identifier[:-1], name=self._identifier[-1]),
                 updates=updates,
                 requirements=requirements,
             )
@@ -1654,16 +1664,14 @@ def _do_commit(self, updates: Tuple[TableUpdate, ...], requirements: Tuple[Table
     def __eq__(self, other: Any) -> bool:
         """Return the equality of two instances of the Table class."""
         return (
-            self.identifier == other.identifier
-            and self.metadata == other.metadata
-            and self.metadata_location == other.metadata_location
+            self.name() == other.name() and self.metadata == other.metadata and self.metadata_location == other.metadata_location
             if isinstance(other, Table)
             else False
         )
 
     def __repr__(self) -> str:
         """Return the string representation of the Table class."""
-        table_name = self.catalog.table_name_from(self.identifier)
+        table_name = self.catalog.table_name_from(self._identifier)
         schema_str = ",\n  ".join(str(column) for column in self.schema().columns if self.schema())
         partition_str = f"partition by: [{', '.join(field.name for field in self.spec().fields if self.spec())}]"
         sort_order_str = f"sort order: [{', '.join(str(field) for field in self.sort_order().fields if self.sort_order())}]"