feat: support pagination in list_* methods in rest catalog

pyiceberg/catalog/__init__.py

@@ -28,6 +28,7 @@
     TYPE_CHECKING,
     Callable,
     Dict,
+    Iterator,
     List,
     Optional,
     Set,
@@ -575,42 +576,42 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         """
 
     @abstractmethod
-    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_tables(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         """List tables under the given namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier to search.
 
         Returns:
-            List[Identifier]: list of table identifiers.
+            Iterator[Identifier]: iterator of table identifiers.
 
         Raises:
             NoSuchNamespaceError: If a namespace with the given name does not exist.
         """
 
     @abstractmethod
-    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
+    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> Iterator[Identifier]:
         """List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier to search.
 
         Returns:
-            List[Identifier]: a List of namespace identifiers.
+            Iterator[Identifier]: iterator of namespace identifiers.
 
         Raises:
             NoSuchNamespaceError: If a namespace with the given name does not exist.
         """
 
     @abstractmethod
-    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_views(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         """List views under the given namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier to search.
 
         Returns:
-            List[Identifier]: list of table identifiers.
+            Iterator[Identifier]: iterator of table identifiers.
 
         Raises:
             NoSuchNamespaceError: If a namespace with the given name does not exist.

pyiceberg/catalog/dynamodb.py

@@ -20,7 +20,7 @@
     TYPE_CHECKING,
     Any,
     Dict,
-    List,
+    Iterator,
     Optional,
     Set,
     Tuple,
@@ -385,7 +385,7 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
         table_identifiers = self.list_tables(namespace=database_name)
 
-        if len(table_identifiers) > 0:
+        if len(list(table_identifiers)) > 0:
             raise NamespaceNotEmptyError(f"Database {database_name} is not empty")
 
         try:
@@ -397,14 +397,14 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         except ConditionalCheckFailedException as e:
             raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e
 
-    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_tables(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         """List Iceberg tables under the given namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier to search.
 
         Returns:
-            List[Identifier]: list of table identifiers.
+            Iterator[Identifier]: iterator of table identifiers.
         """
         database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
 
@@ -429,29 +429,26 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         ) as e:
             raise GenericDynamoDbError(e.message) from e
 
-        table_identifiers = []
         for page in page_iterator:
             for item in page["Items"]:
                 _dict = _convert_dynamo_item_to_regular_dict(item)
                 identifier_col = _dict[DYNAMODB_COL_IDENTIFIER]
                 if identifier_col == DYNAMODB_NAMESPACE:
                     continue
 
-                table_identifiers.append(self.identifier_to_tuple(identifier_col))
+                yield self.identifier_to_tuple(identifier_col)
 
-        return table_identifiers
-
-    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
+    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> Iterator[Identifier]:
         """List top-level namespaces from the catalog.
 
         We do not support hierarchical namespace.
 
         Returns:
-            List[Identifier]: a List of namespace identifiers.
+            Iterator[Identifier]: iterator of namespace identifiers.
         """
         # Hierarchical namespace is not supported. Return an empty list
         if namespace:
-            return []
+            return
 
         paginator = self.dynamodb.get_paginator("query")
 
@@ -474,14 +471,11 @@ def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identi
         ) as e:
             raise GenericDynamoDbError(e.message) from e
 
-        database_identifiers = []
         for page in page_iterator:
             for item in page["Items"]:
                 _dict = _convert_dynamo_item_to_regular_dict(item)
                 namespace_col = _dict[DYNAMODB_COL_NAMESPACE]
-                database_identifiers.append(self.identifier_to_tuple(namespace_col))
-
-        return database_identifiers
+                yield self.identifier_to_tuple(namespace_col)
 
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
         """
@@ -538,7 +532,7 @@ def update_namespace_properties(
 
         return properties_update_summary
 
-    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_views(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         raise NotImplementedError
 
     def drop_view(self, identifier: Union[str, Identifier]) -> None:

pyiceberg/catalog/glue.py

@@ -20,6 +20,7 @@
     TYPE_CHECKING,
     Any,
     Dict,
+    Iterator,
     List,
     Optional,
     Set,
@@ -96,7 +97,6 @@
     from mypy_boto3_glue.type_defs import (
         ColumnTypeDef,
         DatabaseInputTypeDef,
-        DatabaseTypeDef,
         StorageDescriptorTypeDef,
         TableInputTypeDef,
         TableTypeDef,
@@ -710,20 +710,19 @@ 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]:
+    def list_tables(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         """List Iceberg tables under the given namespace in the catalog.
 
         Args:
             namespace (str | Identifier): Namespace identifier to search.
 
         Returns:
-            List[Identifier]: list of table identifiers.
+            Iterator[Identifier]: iterator of table identifiers.
 
         Raises:
             NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
         """
         database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
-        table_list: List["TableTypeDef"] = []
         next_token: Optional[str] = None
         try:
             while True:
@@ -732,37 +731,36 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
                     if not next_token
                     else self.glue.get_tables(DatabaseName=database_name, NextToken=next_token)
                 )
-                table_list.extend(table_list_response["TableList"])
+                for table in table_list_response["TableList"]:
+                    if self.__is_iceberg_table(table):
+                        yield (database_name, table["Name"])
                 next_token = table_list_response.get("NextToken")
                 if not next_token:
                     break
 
         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 if self.__is_iceberg_table(table)]
 
-    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
+    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> Iterator[Identifier]:
         """List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.
 
         Returns:
-            List[Identifier]: a List of namespace identifiers.
+            Iterator[Identifier]: iterator of namespace identifiers.
         """
         # Hierarchical namespace is not supported. Return an empty list
         if namespace:
-            return []
+            return
 
-        database_list: List["DatabaseTypeDef"] = []
         next_token: Optional[str] = None
 
         while True:
             databases_response = self.glue.get_databases() if not next_token else self.glue.get_databases(NextToken=next_token)
-            database_list.extend(databases_response["DatabaseList"])
+            for database in databases_response["DatabaseList"]:
+                yield self.identifier_to_tuple(database["Name"])
             next_token = databases_response.get("NextToken")
             if not next_token:
                 break
 
-        return [self.identifier_to_tuple(database["Name"]) for database in database_list]
-
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
         """Get properties for a namespace.
 
@@ -817,7 +815,7 @@ def update_namespace_properties(
 
         return properties_update_summary
 
-    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_views(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         raise NotImplementedError
 
     def drop_view(self, identifier: Union[str, Identifier]) -> None:

pyiceberg/catalog/hive.py

@@ -23,6 +23,7 @@
     TYPE_CHECKING,
     Any,
     Dict,
+    Iterator,
     List,
     Optional,
     Set,
@@ -469,7 +470,7 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
 
         return self._convert_hive_into_iceberg(hive_table)
 
-    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_views(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         raise NotImplementedError
 
     def view_exists(self, identifier: Union[str, Identifier]) -> bool:
@@ -710,7 +711,7 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         except MetaException as e:
             raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e
 
-    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_tables(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         """List Iceberg tables under the given namespace in the catalog.
 
         When the database doesn't exist, it will just return an empty list.
@@ -719,33 +720,33 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
             namespace: Database to list.
 
         Returns:
-            List[Identifier]: list of table identifiers.
+            Iterator[Identifier]: iterator of table identifiers.
 
         Raises:
             NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
         """
         database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
         with self._client as open_client:
-            return [
+            yield from [
                 (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.get(TABLE_TYPE, "").lower() == ICEBERG
             ]
 
-    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
+    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> Iterator[Identifier]:
         """List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.
 
         Returns:
-            List[Identifier]: a List of namespace identifiers.
+            Iterator[Identifier]: an iterator of namespace identifiers.
         """
         # Hierarchical namespace is not supported. Return an empty list
         if namespace:
-            return []
+            return iter([])
 
         with self._client as open_client:
-            return list(map(self.identifier_to_tuple, open_client.get_all_databases()))
+            return iter(map(self.identifier_to_tuple, open_client.get_all_databases()))
 
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
         """Get properties for a namespace.

pyiceberg/catalog/noop.py

@@ -16,7 +16,7 @@
 #  under the License.
 from typing import (
     TYPE_CHECKING,
-    List,
+    Iterator,
     Optional,
     Set,
     Tuple,
@@ -106,10 +106,10 @@ def create_namespace(self, namespace: Union[str, Identifier], properties: Proper
     def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         raise NotImplementedError
 
-    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_tables(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         raise NotImplementedError
 
-    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
+    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> Iterator[Identifier]:
         raise NotImplementedError
 
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
@@ -120,7 +120,7 @@ def update_namespace_properties(
     ) -> PropertiesUpdateSummary:
         raise NotImplementedError
 
-    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
+    def list_views(self, namespace: Union[str, Identifier]) -> Iterator[Identifier]:
         raise NotImplementedError
 
     def view_exists(self, identifier: Union[str, Identifier]) -> bool: