Glue and Hive catalog return only Iceberg tables (#1145)

* glue and hive returns iceberg tables

* fix formatting

* Update pyiceberg/catalog/glue.py

Co-authored-by: Sung Yun <[email protected]>

---------

Co-authored-by: Sung Yun <[email protected]>

Author: mark-major

pyiceberg/catalog/dynamodb.py

@@ -397,7 +397,7 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
             raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e
 
     def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
-        """List tables under the given namespace in the catalog (including non-Iceberg tables).
+        """List Iceberg tables under the given namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier to search.

pyiceberg/catalog/glue.py

@@ -670,7 +670,7 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         self.glue.delete_database(Name=database_name)
 
     def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
-        """List tables under the given namespace in the catalog (including non-Iceberg tables).
+        """List Iceberg tables under the given namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier to search.
@@ -698,7 +698,7 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
 
         except self.glue.exceptions.EntityNotFoundException as e:
             raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e
-        return [(database_name, table["Name"]) for table in table_list]
+        return [(database_name, table["Name"]) for table in table_list if self.__is_iceberg_table(table)]
 
     def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
         """List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.
@@ -781,3 +781,7 @@ def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
 
     def drop_view(self, identifier: Union[str, Identifier]) -> None:
         raise NotImplementedError
+
+    @staticmethod
+    def __is_iceberg_table(table: TableTypeDef) -> bool:
+        return table["Parameters"] is not None and table["Parameters"][TABLE_TYPE].lower() == ICEBERG

pyiceberg/catalog/hive.py

@@ -631,7 +631,7 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
             raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e
 
     def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
-        """List tables under the given namespace in the catalog (including non-Iceberg tables).
+        """List Iceberg tables under the given namespace in the catalog.
 
         When the database doesn't exist, it will just return an empty list.
 
@@ -646,7 +646,13 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         """
         database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
         with self._client as open_client:
-            return [(database_name, table_name) for table_name in open_client.get_all_tables(db_name=database_name)]
+            return [
+                (database_name, table.tableName)
+                for table in open_client.get_table_objects_by_name(
+                    dbname=database_name, tbl_names=open_client.get_all_tables(db_name=database_name)
+                )
+                if table.parameters[TABLE_TYPE].lower() == ICEBERG
+            ]
 
     def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
         """List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.

tests/catalog/test_glue.py

@@ -448,9 +448,23 @@ def test_list_tables(
 ) -> None:
     test_catalog = GlueCatalog("glue", **{"s3.endpoint": moto_endpoint_url, "warehouse": f"s3://{BUCKET_NAME}/"})
     test_catalog.create_namespace(namespace=database_name)
+
+    non_iceberg_table_name = "non_iceberg_table"
+    glue_client = boto3.client("glue", endpoint_url=moto_endpoint_url)
+    glue_client.create_table(
+        DatabaseName=database_name,
+        TableInput={
+            "Name": non_iceberg_table_name,
+            "TableType": "EXTERNAL_TABLE",
+            "Parameters": {"table_type": "noniceberg"},
+        },
+    )
+
     for table_name in table_list:
         test_catalog.create_table((database_name, table_name), table_schema_nested)
     loaded_table_list = test_catalog.list_tables(database_name)
+
+    assert (database_name, non_iceberg_table_name) not in loaded_table_list
     for table_name in table_list:
         assert (database_name, table_name) in loaded_table_list
 

tests/catalog/test_hive.py

@@ -17,6 +17,7 @@
 # pylint: disable=protected-access,redefined-outer-name
 import copy
 import uuid
+from copy import deepcopy
 from unittest.mock import MagicMock, call, patch
 
 import pytest
@@ -905,23 +906,30 @@ def test_drop_database_does_not_exists() -> None:
     assert "Database does not exists: does_not_exists" in str(exc_info.value)
 
 
-def test_list_tables() -> None:
+def test_list_tables(hive_table: HiveTable) -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
+    tbl1 = deepcopy(hive_table)
+    tbl1.tableName = "table1"
+    tbl1.dbName = "database"
+    tbl2 = deepcopy(hive_table)
+    tbl2.tableName = "table2"
+    tbl2.dbName = "database"
+    tbl3 = deepcopy(hive_table)
+    tbl3.tableName = "table3"
+    tbl3.dbName = "database"
+    tbl3.parameters["table_type"] = "non_iceberg"
+
     catalog._client = MagicMock()
-    catalog._client.__enter__().get_all_tables.return_value = ["table1", "table2"]
+    catalog._client.__enter__().get_all_tables.return_value = ["table1", "table2", "table3"]
+    catalog._client.__enter__().get_table_objects_by_name.return_value = [tbl1, tbl2, tbl3]
 
-    assert catalog.list_tables("database") == [
-        (
-            "database",
-            "table1",
-        ),
-        (
-            "database",
-            "table2",
-        ),
-    ]
+    got_tables = catalog.list_tables("database")
+    assert got_tables == [("database", "table1"), ("database", "table2")]
     catalog._client.__enter__().get_all_tables.assert_called_with(db_name="database")
+    catalog._client.__enter__().get_table_objects_by_name.assert_called_with(
+        dbname="database", tbl_names=["table1", "table2", "table3"]
+    )
 
 
 def test_list_namespaces() -> None: