Merge branch 'apache:main' into fix/hive-client-does-not-update-table-properties

Author: kadai0308

.gitignore

@@ -35,6 +35,7 @@ coverage.xml
 .project
 .settings
 bin/
+.vscode/
 
 # Hive/metastore files
 metastore_db/

dev/docker-compose-integration.yml

@@ -81,7 +81,7 @@ services:
       - AWS_REGION=us-east-1
     entrypoint: >
       /bin/sh -c "
-      until (/usr/bin/mc config host add minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
+      until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
       /usr/bin/mc mb minio/warehouse;
       /usr/bin/mc policy set public minio/warehouse;
       tail -f /dev/null

dev/docker-compose.yml

@@ -38,7 +38,7 @@ services:
       - AWS_REGION=us-east-1
     entrypoint: >
       /bin/sh -c "
-      until (/usr/bin/mc config host add minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
+      until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
       /usr/bin/mc rm -r --force minio/warehouse;
       /usr/bin/mc mb minio/warehouse;
       /usr/bin/mc policy set public minio/warehouse;

mkdocs/docs/configuration.md

@@ -155,6 +155,7 @@ For the FileIO there are several configuration options available:
 | adls.tenant-id         | ad667be4-b811-11ed-afa1-0242ac120002                                                      | The tenant-id                                                                                                                                                                                                                                                                          |
 | adls.client-id         | ad667be4-b811-11ed-afa1-0242ac120002                                                      | The client-id                                                                                                                                                                                                                                                                          |
 | adls.client-secret     | oCA3R6P\*ka#oa1Sms2J74z...                                                                | The client-secret                                                                                                                                                                                                                                                                      |
+| adls.account-host      | accountname1.blob.core.windows.net                                                        | The storage account host. See [AzureBlobFileSystem](https://github.com/fsspec/adlfs/blob/adb9c53b74a0d420625b86dd00fbe615b43201d2/adlfs/spec.py#L125) for reference                                                                                                                   |
 
 <!-- markdown-link-check-enable-->
 

poetry.lock

@@ -66,6 +66,8 @@
 
 if TYPE_CHECKING:
     import pyarrow as pa
+    from mypy_boto3_dynamodb.client import DynamoDBClient
+
 
 DYNAMODB_CLIENT = "dynamodb"
 
@@ -94,18 +96,28 @@
 
 
 class DynamoDbCatalog(MetastoreCatalog):
-    def __init__(self, name: str, **properties: str):
+    def __init__(self, name: str, client: Optional["DynamoDBClient"] = None, **properties: str):
+        """Dynamodb catalog.
+
+        Args:
+            name: Name to identify the catalog.
+            client: An optional boto3 dynamodb client.
+            properties: Properties for dynamodb client construction and configuration.
+        """
         super().__init__(name, **properties)
+        if client is not None:
+            self.dynamodb = client
+        else:
+            session = boto3.Session(
+                profile_name=properties.get(DYNAMODB_PROFILE_NAME),
+                region_name=get_first_property_value(properties, DYNAMODB_REGION, AWS_REGION),
+                botocore_session=properties.get(BOTOCORE_SESSION),
+                aws_access_key_id=get_first_property_value(properties, DYNAMODB_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
+                aws_secret_access_key=get_first_property_value(properties, DYNAMODB_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
+                aws_session_token=get_first_property_value(properties, DYNAMODB_SESSION_TOKEN, AWS_SESSION_TOKEN),
+            )
+            self.dynamodb = session.client(DYNAMODB_CLIENT)
 
-        session = boto3.Session(
-            profile_name=properties.get(DYNAMODB_PROFILE_NAME),
-            region_name=get_first_property_value(properties, DYNAMODB_REGION, AWS_REGION),
-            botocore_session=properties.get(BOTOCORE_SESSION),
-            aws_access_key_id=get_first_property_value(properties, DYNAMODB_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
-            aws_secret_access_key=get_first_property_value(properties, DYNAMODB_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
-            aws_session_token=get_first_property_value(properties, DYNAMODB_SESSION_TOKEN, AWS_SESSION_TOKEN),
-        )
-        self.dynamodb = session.client(DYNAMODB_CLIENT)
         self.dynamodb_table_name = self.properties.get(DYNAMODB_TABLE_NAME, DYNAMODB_TABLE_NAME_DEFAULT)
         self._ensure_catalog_table_exists_or_create()
 
@@ -824,7 +836,9 @@ def _convert_dynamo_item_to_regular_dict(dynamo_json: Dict[str, Any]) -> Dict[st
             raise ValueError("Only S and N data types are supported.")
 
         values = list(val_dict.values())
-        assert len(values) == 1
+        if len(values) != 1:
+            raise ValueError(f"Expecting only 1 value: {values}")
+
         column_value = values[0]
         regular_json[column_name] = column_value
 

pyiceberg/catalog/dynamodb.py

@@ -30,15 +30,6 @@
 
 import boto3
 from botocore.config import Config
-from mypy_boto3_glue.client import GlueClient
-from mypy_boto3_glue.type_defs import (
-    ColumnTypeDef,
-    DatabaseInputTypeDef,
-    DatabaseTypeDef,
-    StorageDescriptorTypeDef,
-    TableInputTypeDef,
-    TableTypeDef,
-)
 
 from pyiceberg.catalog import (
     BOTOCORE_SESSION,
@@ -101,6 +92,15 @@
 
 if TYPE_CHECKING:
     import pyarrow as pa
+    from mypy_boto3_glue.client import GlueClient
+    from mypy_boto3_glue.type_defs import (
+        ColumnTypeDef,
+        DatabaseInputTypeDef,
+        DatabaseTypeDef,
+        StorageDescriptorTypeDef,
+        TableInputTypeDef,
+        TableTypeDef,
+    )
 
 
 # There is a unique Glue metastore in each AWS account and each AWS region. By default, GlueCatalog chooses the Glue
@@ -140,7 +140,7 @@
 
 
 def _construct_parameters(
-    metadata_location: str, glue_table: Optional[TableTypeDef] = None, prev_metadata_location: Optional[str] = None
+    metadata_location: str, glue_table: Optional["TableTypeDef"] = None, prev_metadata_location: Optional[str] = None
 ) -> Properties:
     new_parameters = glue_table.get("Parameters", {}) if glue_table else {}
     new_parameters.update({TABLE_TYPE: ICEBERG.upper(), METADATA_LOCATION: metadata_location})
@@ -190,15 +190,15 @@ def primitive(self, primitive: PrimitiveType) -> str:
         return GLUE_PRIMITIVE_TYPES[primitive_type]
 
 
-def _to_columns(metadata: TableMetadata) -> List[ColumnTypeDef]:
-    results: Dict[str, ColumnTypeDef] = {}
+def _to_columns(metadata: TableMetadata) -> List["ColumnTypeDef"]:
+    results: Dict[str, "ColumnTypeDef"] = {}
 
     def _append_to_results(field: NestedField, is_current: bool) -> None:
         if field.name in results:
             return
 
         results[field.name] = cast(
-            ColumnTypeDef,
+            "ColumnTypeDef",
             {
                 "Name": field.name,
                 "Type": visit(field.field_type, _IcebergSchemaToGlueType()),
@@ -230,10 +230,10 @@ def _construct_table_input(
     metadata_location: str,
     properties: Properties,
     metadata: TableMetadata,
-    glue_table: Optional[TableTypeDef] = None,
+    glue_table: Optional["TableTypeDef"] = None,
     prev_metadata_location: Optional[str] = None,
-) -> TableInputTypeDef:
-    table_input: TableInputTypeDef = {
+) -> "TableInputTypeDef":
+    table_input: "TableInputTypeDef" = {
         "Name": table_name,
         "TableType": EXTERNAL_TABLE,
         "Parameters": _construct_parameters(metadata_location, glue_table, prev_metadata_location),
@@ -249,10 +249,12 @@ def _construct_table_input(
     return table_input
 
 
-def _construct_rename_table_input(to_table_name: str, glue_table: TableTypeDef) -> TableInputTypeDef:
-    rename_table_input: TableInputTypeDef = {"Name": to_table_name}
+def _construct_rename_table_input(to_table_name: str, glue_table: "TableTypeDef") -> "TableInputTypeDef":
+    rename_table_input: "TableInputTypeDef" = {"Name": to_table_name}
     # use the same Glue info to create the new table, pointing to the old metadata
-    assert glue_table["TableType"]
+    if not glue_table["TableType"]:
+        raise ValueError("Glue table type is missing, cannot rename table")
+
     rename_table_input["TableType"] = glue_table["TableType"]
     if "Owner" in glue_table:
         rename_table_input["Owner"] = glue_table["Owner"]
@@ -264,16 +266,16 @@ def _construct_rename_table_input(to_table_name: str, glue_table: TableTypeDef)
         # It turns out the output of StorageDescriptor is not the same as the input type
         # because the Column can have a different type, but for now it seems to work, so
         # silence the type error.
-        rename_table_input["StorageDescriptor"] = cast(StorageDescriptorTypeDef, glue_table["StorageDescriptor"])
+        rename_table_input["StorageDescriptor"] = cast("StorageDescriptorTypeDef", glue_table["StorageDescriptor"])
 
     if "Description" in glue_table:
         rename_table_input["Description"] = glue_table["Description"]
 
     return rename_table_input
 
 
-def _construct_database_input(database_name: str, properties: Properties) -> DatabaseInputTypeDef:
-    database_input: DatabaseInputTypeDef = {"Name": database_name}
+def _construct_database_input(database_name: str, properties: Properties) -> "DatabaseInputTypeDef":
+    database_input: "DatabaseInputTypeDef" = {"Name": database_name}
     parameters = {}
     for k, v in properties.items():
         if k == "Description":
@@ -286,7 +288,7 @@ def _construct_database_input(database_name: str, properties: Properties) -> Dat
     return database_input
 
 
-def _register_glue_catalog_id_with_glue_client(glue: GlueClient, glue_catalog_id: str) -> None:
+def _register_glue_catalog_id_with_glue_client(glue: "GlueClient", glue_catalog_id: str) -> None:
     """
     Register the Glue Catalog ID (AWS Account ID) as a parameter on all Glue client methods.
 
@@ -303,9 +305,9 @@ def add_glue_catalog_id(params: Dict[str, str], **kwargs: Any) -> None:
 
 
 class GlueCatalog(MetastoreCatalog):
-    glue: GlueClient
+    glue: "GlueClient"
 
-    def __init__(self, name: str, client: Optional[GlueClient] = None, **properties: Any):
+    def __init__(self, name: str, client: Optional["GlueClient"] = None, **properties: Any):
         """Glue Catalog.
 
         You either need to provide a boto3 glue client, or one will be constructed from the properties.
@@ -317,7 +319,7 @@ def __init__(self, name: str, client: Optional[GlueClient] = None, **properties:
         """
         super().__init__(name, **properties)
 
-        if client:
+        if client is not None:
             self.glue = client
         else:
             retry_mode_prop_value = get_first_property_value(properties, GLUE_RETRY_MODE)
@@ -344,12 +346,17 @@ def __init__(self, name: str, client: Optional[GlueClient] = None, **properties:
             if glue_catalog_id := properties.get(GLUE_ID):
                 _register_glue_catalog_id_with_glue_client(self.glue, glue_catalog_id)
 
-    def _convert_glue_to_iceberg(self, glue_table: TableTypeDef) -> Table:
+    def _convert_glue_to_iceberg(self, glue_table: "TableTypeDef") -> Table:
         properties: Properties = glue_table["Parameters"]
 
-        assert glue_table["DatabaseName"]
-        assert glue_table["Parameters"]
-        database_name = glue_table["DatabaseName"]
+        database_name = glue_table.get("DatabaseName", None)
+        if database_name is None:
+            raise ValueError("Glue table is missing DatabaseName property")
+
+        parameters = glue_table.get("Parameters", None)
+        if parameters is None:
+            raise ValueError("Glue table is missing Parameters property")
+
         table_name = glue_table["Name"]
 
         if TABLE_TYPE not in properties:
@@ -380,15 +387,15 @@ def _convert_glue_to_iceberg(self, glue_table: TableTypeDef) -> Table:
             catalog=self,
         )
 
-    def _create_glue_table(self, database_name: str, table_name: str, table_input: TableInputTypeDef) -> None:
+    def _create_glue_table(self, database_name: str, table_name: str, table_input: "TableInputTypeDef") -> None:
         try:
             self.glue.create_table(DatabaseName=database_name, TableInput=table_input)
         except self.glue.exceptions.AlreadyExistsException as e:
             raise TableAlreadyExistsError(f"Table {database_name}.{table_name} already exists") from e
         except self.glue.exceptions.EntityNotFoundException as e:
             raise NoSuchNamespaceError(f"Database {database_name} does not exist") from e
 
-    def _update_glue_table(self, database_name: str, table_name: str, table_input: TableInputTypeDef, version_id: str) -> None:
+    def _update_glue_table(self, database_name: str, table_name: str, table_input: "TableInputTypeDef", version_id: str) -> None:
         try:
             self.glue.update_table(
                 DatabaseName=database_name,
@@ -403,7 +410,7 @@ def _update_glue_table(self, database_name: str, table_name: str, table_input: T
                 f"Cannot commit {database_name}.{table_name} because Glue detected concurrent update to table version {version_id}"
             ) from e
 
-    def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
+    def _get_glue_table(self, database_name: str, table_name: str) -> "TableTypeDef":
         try:
             load_table_response = self.glue.get_table(DatabaseName=database_name, Name=table_name)
             return load_table_response["Table"]
@@ -496,7 +503,7 @@ def commit_table(
         table_identifier = table.name()
         database_name, table_name = self.identifier_to_database_and_table(table_identifier, NoSuchTableError)
 
-        current_glue_table: Optional[TableTypeDef]
+        current_glue_table: Optional["TableTypeDef"]
         glue_table_version_id: Optional[str]
         current_table: Optional[Table]
         try:
@@ -680,13 +687,19 @@ def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
         """
         database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
         try:
-            table_list = self.list_tables(namespace=database_name)
-        except NoSuchNamespaceError as e:
+            table_list_response = self.glue.get_tables(DatabaseName=database_name)
+            table_list = table_list_response["TableList"]
+        except self.glue.exceptions.EntityNotFoundException as e:
             raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e
 
         if len(table_list) > 0:
-            raise NamespaceNotEmptyError(f"Database {database_name} is not empty")
-
+            first_table = table_list[0]
+            if self.__is_iceberg_table(first_table):
+                raise NamespaceNotEmptyError(f"Cannot drop namespace {database_name} because it still contains Iceberg tables")
+            else:
+                raise NamespaceNotEmptyError(
+                    f"Cannot drop namespace {database_name} because it still contains non-Iceberg tables"
+                )
         self.glue.delete_database(Name=database_name)
 
     def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
@@ -702,7 +715,7 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
             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] = []
+        table_list: List["TableTypeDef"] = []
         next_token: Optional[str] = None
         try:
             while True:
@@ -730,7 +743,7 @@ def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identi
         if namespace:
             return []
 
-        database_list: List[DatabaseTypeDef] = []
+        database_list: List["DatabaseTypeDef"] = []
         next_token: Optional[str] = None
 
         while True:
@@ -806,5 +819,5 @@ def view_exists(self, identifier: Union[str, Identifier]) -> bool:
         raise NotImplementedError
 
     @staticmethod
-    def __is_iceberg_table(table: TableTypeDef) -> bool:
+    def __is_iceberg_table(table: "TableTypeDef") -> bool:
         return table.get("Parameters", {}).get(TABLE_TYPE, "").lower() == ICEBERG

pyiceberg/catalog/glue.py

@@ -32,6 +32,7 @@
 
 from pyiceberg import __version__
 from pyiceberg.catalog import (
+    BOTOCORE_SESSION,
     TOKEN,
     URI,
     WAREHOUSE_LOCATION,
@@ -53,6 +54,7 @@
     TableAlreadyExistsError,
     UnauthorizedError,
 )
+from pyiceberg.io import AWS_ACCESS_KEY_ID, AWS_REGION, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec, assign_fresh_partition_spec_ids
 from pyiceberg.schema import Schema, assign_fresh_schema_ids
 from pyiceberg.table import (
@@ -72,7 +74,7 @@
 from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel, Identifier, Properties
 from pyiceberg.types import transform_dict_value_to_str
 from pyiceberg.utils.deprecated import deprecation_message
-from pyiceberg.utils.properties import get_header_properties, property_as_bool
+from pyiceberg.utils.properties import get_first_property_value, get_header_properties, property_as_bool
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -390,11 +392,17 @@ class SigV4Adapter(HTTPAdapter):
             def __init__(self, **properties: str):
                 super().__init__()
                 self._properties = properties
+                self._boto_session = boto3.Session(
+                    region_name=get_first_property_value(self._properties, AWS_REGION),
+                    botocore_session=self._properties.get(BOTOCORE_SESSION),
+                    aws_access_key_id=get_first_property_value(self._properties, AWS_ACCESS_KEY_ID),
+                    aws_secret_access_key=get_first_property_value(self._properties, AWS_SECRET_ACCESS_KEY),
+                    aws_session_token=get_first_property_value(self._properties, AWS_SESSION_TOKEN),
+                )
 
             def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None:  # pylint: disable=W0613
-                boto_session = boto3.Session()
-                credentials = boto_session.get_credentials().get_frozen_credentials()
-                region = self._properties.get(SIGV4_REGION, boto_session.region_name)
+                credentials = self._boto_session.get_credentials().get_frozen_credentials()
+                region = self._properties.get(SIGV4_REGION, self._boto_session.region_name)
                 service = self._properties.get(SIGV4_SERVICE, "execute-api")
 
                 url = str(request.url).split("?")[0]