glue

GlueCatalog

Bases: MetastoreCatalog

Source code in pyiceberg/catalog/glue.py

class GlueCatalog(MetastoreCatalog):
    glue: "GlueClient"

    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.

        Args:
            name: Name to identify the catalog.
            client: An optional boto3 glue client.
            properties: Properties for glue client construction and configuration.
        """
        super().__init__(name, **properties)

        if client is not None:
            self.glue = client
        else:
            retry_mode_prop_value = get_first_property_value(properties, GLUE_RETRY_MODE)

            session = boto3.Session(
                profile_name=get_first_property_value(properties, GLUE_PROFILE_NAME, AWS_PROFILE_NAME),
                region_name=get_first_property_value(properties, GLUE_REGION, AWS_REGION),
                botocore_session=properties.get(BOTOCORE_SESSION),
                aws_access_key_id=get_first_property_value(properties, GLUE_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
                aws_secret_access_key=get_first_property_value(properties, GLUE_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
                aws_session_token=get_first_property_value(properties, GLUE_SESSION_TOKEN, AWS_SESSION_TOKEN),
            )
            self.glue: GlueClient = session.client(
                "glue",
                endpoint_url=properties.get(GLUE_CATALOG_ENDPOINT),
                config=Config(
                    retries={
                        "max_attempts": properties.get(GLUE_MAX_RETRIES, MAX_RETRIES),
                        "mode": retry_mode_prop_value if retry_mode_prop_value in EXISTING_RETRY_MODES else STANDARD_RETRY_MODE,
                    }
                ),
            )

            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:
        if (database_name := glue_table.get("DatabaseName")) is None:
            raise ValueError("Glue table is missing DatabaseName property")

        if (table_name := glue_table.get("Name")) is None:
            raise ValueError("Glue table is missing Name property")

        if (parameters := glue_table.get("Parameters")) is None:
            raise ValueError("Glue table is missing Parameters property")

        if (glue_table_type := parameters.get(TABLE_TYPE)) is None:
            raise NoSuchPropertyException(
                f"Property {TABLE_TYPE} missing, could not determine type: {database_name}.{table_name}"
            )

        if glue_table_type.lower() != ICEBERG:
            raise NoSuchIcebergTableError(
                f"Property table_type is {glue_table_type}, expected {ICEBERG}: {database_name}.{table_name}"
            )

        if (metadata_location := parameters.get(METADATA_LOCATION)) is None:
            raise NoSuchPropertyException(
                f"Table property {METADATA_LOCATION} is missing, cannot find metadata for: {database_name}.{table_name}"
            )

        io = self._load_file_io(location=metadata_location)
        file = io.new_input(metadata_location)
        metadata = FromInputFile.table_metadata(file)
        return Table(
            identifier=(database_name, table_name),
            metadata=metadata,
            metadata_location=metadata_location,
            io=self._load_file_io(metadata.properties, metadata_location),
            catalog=self,
        )

    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:
        try:
            self.glue.update_table(
                DatabaseName=database_name,
                TableInput=table_input,
                SkipArchive=property_as_bool(self.properties, GLUE_SKIP_ARCHIVE, GLUE_SKIP_ARCHIVE_DEFAULT),
                VersionId=version_id,
            )
        except self.glue.exceptions.EntityNotFoundException as e:
            raise NoSuchTableError(f"Table does not exist: {database_name}.{table_name} (Glue table version {version_id})") from e
        except self.glue.exceptions.ConcurrentModificationException as e:
            raise CommitFailedException(
                f"Cannot commit {database_name}.{table_name} because Glue detected concurrent update "
                f"to table version {version_id}"
            ) from e

    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"]
        except self.glue.exceptions.EntityNotFoundException as e:
            raise NoSuchTableError(f"Table does not exist: {database_name}.{table_name}") from e

    def _is_s3tables_database(self, database_name: str) -> bool:
        """Check if a Glue database is federated with S3 Tables.

        S3 Tables databases have a FederatedDatabase property with
        ConnectionType set to aws:s3tables.

        Args:
            database_name: The name of the Glue database.

        Returns:
            True if the database is an S3 Tables federated database.
        """
        try:
            database_response = self.glue.get_database(Name=database_name)
        except self.glue.exceptions.EntityNotFoundException:
            return False
        database = database_response["Database"]
        federated = database.get("FederatedDatabase", {})
        return federated.get("ConnectionType", "") == GLUE_CONNECTION_S3_TABLES

    @staticmethod
    def _write_metadata_no_exist_check(metadata: TableMetadata, io: FileIO, metadata_path: str) -> None:
        ToOutputFile.table_metadata(metadata, io.new_output(metadata_path), overwrite=True)

    def _create_table_s3tables(
        self,
        identifier: str | Identifier,
        schema: Union[Schema, "pa.Schema"],
        location: str | None,
        partition_spec: PartitionSpec,
        sort_order: SortOrder,
        properties: Properties,
    ) -> Table:
        """Create an Iceberg table in an S3 Tables federated database.

        S3 Tables manages storage internally, so the table location is not known until the
        table is created in the service. This method:
          1. Creates a minimal table entry in Glue (format=ICEBERG), which causes S3 Tables
             to allocate storage.
          2. Retrieves the managed storage location via GetTable.
          3. Writes Iceberg metadata to that location.
          4. Updates the Glue table entry with the metadata pointer.

        On failure, the table created in step 1 is deleted.
        """
        database_name, table_name = self.identifier_to_database_and_table(identifier)

        if location is not None:
            raise ValueError(
                f"Cannot specify a location for S3 Tables table {database_name}.{table_name}. "
                "S3 Tables manages the storage location automatically."
            )

        # Create a minimal table in Glue so S3 Tables allocates storage.
        self._create_glue_table(
            database_name=database_name,
            table_name=table_name,
            table_input={
                "Name": table_name,
                "Parameters": {"format": "ICEBERG"},
            },
        )

        try:
            # Retrieve the managed storage location.
            glue_table = self._get_glue_table(database_name=database_name, table_name=table_name)
            storage_descriptor = glue_table.get("StorageDescriptor", {})
            managed_location = storage_descriptor.get("Location")
            if not managed_location:
                raise ValueError(f"S3 Tables did not assign a storage location for {database_name}.{table_name}")

            # Build the Iceberg metadata targeting the managed location.
            staged_table = self._create_staged_table(
                identifier=identifier,
                schema=schema,
                location=managed_location,
                partition_spec=partition_spec,
                sort_order=sort_order,
                properties=properties,
            )

            # Write metadata and update the Glue table with the metadata pointer.
            # Skip the exist check before writing; S3 Tables doesn't support ListObjectsV2.
            self._write_metadata_no_exist_check(staged_table.metadata, staged_table.io, staged_table.metadata_location)
            table_input = _construct_table_input(table_name, staged_table.metadata_location, properties, staged_table.metadata)
            version_id = glue_table.get("VersionId")
            if not version_id:
                raise CommitFailedException(
                    f"Cannot commit {database_name}.{table_name} because Glue table version id is missing"
                )
            self._update_glue_table(
                database_name=database_name,
                table_name=table_name,
                table_input=table_input,
                version_id=version_id,
            )
        except Exception:
            # Clean up the table created in step 1.
            try:
                self.glue.delete_table(DatabaseName=database_name, Name=table_name)
            except Exception:
                logger.warning(
                    f"Failed to clean up S3 Tables table {database_name}.{table_name}",
                    exc_info=logger.isEnabledFor(logging.DEBUG),
                )
            raise

        return Table(
            identifier=self.identifier_to_tuple(identifier),
            metadata=staged_table.metadata,
            metadata_location=staged_table.metadata_location,
            io=self._load_file_io(staged_table.metadata.properties, staged_table.metadata_location),
            catalog=self,
        )

    @override
    def create_table(
        self,
        identifier: str | Identifier,
        schema: Union[Schema, "pa.Schema"],
        location: str | None = None,
        partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
        sort_order: SortOrder = UNSORTED_SORT_ORDER,
        properties: Properties = EMPTY_DICT,
    ) -> Table:
        """
        Create an Iceberg table.

        Args:
            identifier: Table identifier.
            schema: Table's schema.
            location: Location for the table. Optional Argument.
                Must not be set for S3 Tables, which manage their own storage.
            partition_spec: PartitionSpec for the table.
            sort_order: SortOrder for the table.
            properties: Table properties that can be a string based dictionary.

        Returns:
            Table: the created table instance.

        Raises:
            AlreadyExistsError: If a table with the name already exists.
            ValueError: If the identifier is invalid, no path is given to store metadata,
                or a location is specified for an S3 Tables table.

        """
        database_name, table_name = self.identifier_to_database_and_table(identifier)

        if self._is_s3tables_database(database_name):
            return self._create_table_s3tables(
                identifier=identifier,
                schema=schema,
                location=location,
                partition_spec=partition_spec,
                sort_order=sort_order,
                properties=properties,
            )

        staged_table = self._create_staged_table(
            identifier=identifier,
            schema=schema,
            location=location,
            partition_spec=partition_spec,
            sort_order=sort_order,
            properties=properties,
        )

        self._write_metadata(staged_table.metadata, staged_table.io, staged_table.metadata_location)
        table_input = _construct_table_input(table_name, staged_table.metadata_location, properties, staged_table.metadata)
        self._create_glue_table(database_name=database_name, table_name=table_name, table_input=table_input)

        return Table(
            identifier=self.identifier_to_tuple(identifier),
            metadata=staged_table.metadata,
            metadata_location=staged_table.metadata_location,
            io=self._load_file_io(staged_table.metadata.properties, staged_table.metadata_location),
            catalog=self,
        )

    @override
    def register_table(self, identifier: str | Identifier, metadata_location: str, overwrite: bool = False) -> Table:
        """Register a new table using existing metadata.

        Args:
            identifier (Union[str, Identifier]): Table identifier for the table
            metadata_location (str): The location to the metadata
            overwrite (bool): Whether to overwrite the existing table, default False

        Returns:
            Table: The newly registered table

        Raises:
            TableAlreadyExistsError: If the table already exists
        """
        if overwrite:
            raise NotImplementedError("`overwrite` isn't supported")

        database_name, table_name = self.identifier_to_database_and_table(identifier)
        properties = EMPTY_DICT
        io = self._load_file_io(location=metadata_location)
        file = io.new_input(metadata_location)
        metadata = FromInputFile.table_metadata(file)
        table_input = _construct_table_input(table_name, metadata_location, properties, metadata)
        self._create_glue_table(database_name=database_name, table_name=table_name, table_input=table_input)
        return self.load_table(identifier=identifier)

    @override
    def commit_table(
        self, table: Table, requirements: tuple[TableRequirement, ...], updates: tuple[TableUpdate, ...]
    ) -> CommitTableResponse:
        """Commit updates to a table.

        Args:
            table (Table): The table to be updated.
            requirements: (Tuple[TableRequirement, ...]): Table requirements.
            updates: (Tuple[TableUpdate, ...]): Table updates.

        Returns:
            CommitTableResponse: The updated metadata.

        Raises:
            NoSuchTableError: If a table with the given identifier does not exist.
            CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
        """
        table_identifier = table.name()
        database_name, table_name = self.identifier_to_database_and_table(table_identifier, NoSuchTableError)

        current_glue_table: TableTypeDef | None
        glue_table_version_id: str | None
        current_table: Table | None
        try:
            current_glue_table = self._get_glue_table(database_name=database_name, table_name=table_name)
            glue_table_version_id = current_glue_table.get("VersionId")
            current_table = self._convert_glue_to_iceberg(glue_table=current_glue_table)
        except NoSuchTableError:
            current_glue_table = None
            glue_table_version_id = None
            current_table = None

        updated_staged_table = self._update_and_stage_table(current_table, table_identifier, requirements, updates)
        if current_table and updated_staged_table.metadata == current_table.metadata:
            # no changes, do nothing
            return CommitTableResponse(metadata=current_table.metadata, metadata_location=current_table.metadata_location)
        # S3 Tables managed storage doesn't support ListObjectsV2, so skip the exist check.
        write_metadata = (
            self._write_metadata_no_exist_check if self._is_s3tables_database(database_name) else self._write_metadata
        )
        write_metadata(
            metadata=updated_staged_table.metadata,
            io=updated_staged_table.io,
            metadata_path=updated_staged_table.metadata_location,
        )

        if current_table:
            # table exists, update the table
            if not glue_table_version_id:
                raise CommitFailedException(
                    f"Cannot commit {database_name}.{table_name} because Glue table version id is missing"
                )

            # Pass `version_id` to implement optimistic locking: it ensures updates are rejected if concurrent
            # modifications occur. See more details at https://iceberg.apache.org/docs/latest/aws/#optimistic-locking
            update_table_input = _construct_table_input(
                table_name=table_name,
                metadata_location=updated_staged_table.metadata_location,
                properties=updated_staged_table.properties,
                metadata=updated_staged_table.metadata,
                glue_table=current_glue_table,
                prev_metadata_location=current_table.metadata_location,
            )
            self._update_glue_table(
                database_name=database_name,
                table_name=table_name,
                table_input=update_table_input,
                version_id=glue_table_version_id,
            )
        else:
            # table does not exist, create the table
            create_table_input = _construct_table_input(
                table_name=table_name,
                metadata_location=updated_staged_table.metadata_location,
                properties=updated_staged_table.properties,
                metadata=updated_staged_table.metadata,
            )
            self._create_glue_table(database_name=database_name, table_name=table_name, table_input=create_table_input)

        return CommitTableResponse(
            metadata=updated_staged_table.metadata, metadata_location=updated_staged_table.metadata_location
        )

    @override
    def load_table(self, identifier: str | Identifier) -> Table:
        """Load the table's metadata and returns the table instance.

        You can also use this method to check for table existence using 'try catalog.table() except TableNotFoundError'.
        Note: This method doesn't scan data stored in the table.

        Args:
            identifier: Table identifier.

        Returns:
            Table: the table instance with its metadata.

        Raises:
            NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
        """
        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)

        return self._convert_glue_to_iceberg(self._get_glue_table(database_name=database_name, table_name=table_name))

    @override
    def drop_table(self, identifier: str | Identifier) -> None:
        """Drop a table.

        Args:
            identifier: Table identifier.

        Raises:
            NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
        """
        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:
            raise NoSuchTableError(f"Table does not exist: {database_name}.{table_name}") from e

    @override
    def rename_table(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> Table:
        """Rename a fully classified table name.

        This method can only rename Iceberg tables in AWS Glue.

        Args:
            from_identifier: Existing table identifier.
            to_identifier: New table identifier.

        Returns:
            Table: the updated table instance with its metadata.

        Raises:
            ValueError: When from table identifier is invalid.
            NoSuchTableError: When a table with the name does not exist.
            NoSuchIcebergTableError: When from table is not a valid iceberg table.
            NoSuchPropertyException: When from table miss some required properties.
            NoSuchNamespaceError: When the destination namespace doesn't exist.
        """
        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)
        except self.glue.exceptions.EntityNotFoundException as e:
            raise NoSuchTableError(f"Table does not exist: {from_database_name}.{from_table_name}") from e

        glue_table = get_table_response["Table"]

        try:
            # verify that from_identifier is a valid iceberg table
            self._convert_glue_to_iceberg(glue_table=glue_table)
        except NoSuchPropertyException as e:
            raise NoSuchPropertyException(
                f"Failed to rename table {from_database_name}.{from_table_name} since it is missing required properties"
            ) from e
        except NoSuchIcebergTableError as e:
            raise NoSuchIcebergTableError(
                f"Failed to rename table {from_database_name}.{from_table_name} since it is not a valid iceberg table"
            ) from e

        rename_table_input = _construct_rename_table_input(to_table_name=to_table_name, glue_table=glue_table)
        self._create_glue_table(database_name=to_database_name, table_name=to_table_name, table_input=rename_table_input)

        try:
            self.drop_table(from_identifier)
        except Exception as e:
            log_message = f"Failed to drop old table {from_database_name}.{from_table_name}. "

            try:
                self.drop_table(to_identifier)
                log_message += f"Rolled back table creation for {to_database_name}.{to_table_name}."
            except NoSuchTableError:
                log_message += (
                    f"Failed to roll back table creation for {to_database_name}.{to_table_name}. Please clean up manually"
                )

            raise ValueError(log_message) from e

        return self.load_table(to_identifier)

    @override
    def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
        """Create a namespace in the catalog.

        Args:
            namespace: Namespace identifier.
            properties: A string dictionary of properties for the given namespace.

        Raises:
            ValueError: If the identifier is invalid.
            AlreadyExistsError: If a namespace with the given name already exists.
        """
        database_name = self.identifier_to_database(namespace)
        try:
            self.glue.create_database(DatabaseInput=_construct_database_input(database_name, properties))
        except self.glue.exceptions.AlreadyExistsException as e:
            raise NamespaceAlreadyExistsError(f"Database {database_name} already exists") from e

    @override
    def drop_namespace(self, namespace: str | Identifier) -> None:
        """Drop a namespace.

        A Glue namespace can only be dropped if it is empty.

        Args:
            namespace: Namespace identifier.

        Raises:
            NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
            NamespaceNotEmptyError: If the namespace is not empty.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        try:
            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:
            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)

    @override
    def list_tables(self, namespace: str | Identifier) -> list[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.

        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: str | None = None
        try:
            while True:
                table_list_response = (
                    self.glue.get_tables(DatabaseName=database_name)
                    if not next_token
                    else self.glue.get_tables(DatabaseName=database_name, NextToken=next_token)
                )
                table_list.extend(table_list_response["TableList"])
                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)]

    @override
    def list_namespaces(self, namespace: str | Identifier = ()) -> list[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.
        """
        # Hierarchical namespace is not supported. Return an empty list
        if namespace:
            return []

        database_list: list[DatabaseTypeDef] = []
        next_token: str | None = 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"])
            next_token = databases_response.get("NextToken")
            if not next_token:
                break

        return [self.identifier_to_tuple(database["Name"]) for database in database_list]

    @override
    def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
        """Get properties for a namespace.

        Args:
            namespace: Namespace identifier.

        Returns:
            Properties: Properties for the given namespace.

        Raises:
            NoSuchNamespaceError: If a namespace with the given name does not exist, or identifier is invalid.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        try:
            database_response = self.glue.get_database(Name=database_name)
        except self.glue.exceptions.EntityNotFoundException as e:
            raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e
        except self.glue.exceptions.InvalidInputException as e:
            raise NoSuchNamespaceError(f"Invalid input for namespace {database_name}") from e

        database = database_response["Database"]

        properties = dict(database.get("Parameters", {}))
        if "LocationUri" in database:
            properties["location"] = database["LocationUri"]
        if "Description" in database:
            properties["Description"] = database["Description"]

        return properties

    @override
    def update_namespace_properties(
        self, namespace: str | Identifier, removals: set[str] | None = None, updates: Properties = EMPTY_DICT
    ) -> PropertiesUpdateSummary:
        """Remove provided property keys and updates properties for a namespace.

        Args:
            namespace: Namespace identifier.
            removals: Set of property keys that need to be removed. Optional Argument.
            updates: Properties to be updated for the given namespace.

        Raises:
            NoSuchNamespaceError: If a namespace with the given name does not exist， or identifier is invalid.
            ValueError: If removals and updates have overlapping keys.
        """
        current_properties = self.load_namespace_properties(namespace=namespace)
        properties_update_summary, updated_properties = self._get_updated_props_and_update_summary(
            current_properties=current_properties, removals=removals, updates=updates
        )

        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        self.glue.update_database(Name=database_name, DatabaseInput=_construct_database_input(database_name, updated_properties))

        return properties_update_summary

    @override
    def create_view(
        self,
        identifier: str | Identifier,
        schema: Union[Schema, "pa.Schema"],
        view_version: ViewVersion,
        location: str | None = None,
        properties: Properties = EMPTY_DICT,
    ) -> View:
        raise NotImplementedError

    @override
    def list_views(self, namespace: str | Identifier) -> list[Identifier]:
        raise NotImplementedError

    @override
    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
        raise NotImplementedError

    @override
    def drop_view(self, identifier: str | Identifier) -> None:
        raise NotImplementedError

    @override
    def view_exists(self, identifier: str | Identifier) -> bool:
        raise NotImplementedError

    @override
    def load_view(self, identifier: str | Identifier) -> View:
        raise NotImplementedError

    @staticmethod
    def __is_iceberg_table(table: "TableTypeDef") -> bool:
        return table.get("Parameters", {}).get(TABLE_TYPE, "").lower() == ICEBERG

    def _get_default_warehouse_location(self, database_name: str, table_name: str) -> str:
        """Override the default warehouse location to follow Hive-style conventions."""
        return self._get_hive_style_warehouse_location(database_name, table_name)

__init__(name, client=None, **properties)

Glue Catalog.

You either need to provide a boto3 glue client, or one will be constructed from the properties.

Parameters:

Name	Type	Description	Default
name	str	Name to identify the catalog.	required
client	Optional[GlueClient]	An optional boto3 glue client.	None
properties	Any	Properties for glue client construction and configuration.	{}

Source code in pyiceberg/catalog/glue.py

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.

    Args:
        name: Name to identify the catalog.
        client: An optional boto3 glue client.
        properties: Properties for glue client construction and configuration.
    """
    super().__init__(name, **properties)

    if client is not None:
        self.glue = client
    else:
        retry_mode_prop_value = get_first_property_value(properties, GLUE_RETRY_MODE)

        session = boto3.Session(
            profile_name=get_first_property_value(properties, GLUE_PROFILE_NAME, AWS_PROFILE_NAME),
            region_name=get_first_property_value(properties, GLUE_REGION, AWS_REGION),
            botocore_session=properties.get(BOTOCORE_SESSION),
            aws_access_key_id=get_first_property_value(properties, GLUE_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
            aws_secret_access_key=get_first_property_value(properties, GLUE_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
            aws_session_token=get_first_property_value(properties, GLUE_SESSION_TOKEN, AWS_SESSION_TOKEN),
        )
        self.glue: GlueClient = session.client(
            "glue",
            endpoint_url=properties.get(GLUE_CATALOG_ENDPOINT),
            config=Config(
                retries={
                    "max_attempts": properties.get(GLUE_MAX_RETRIES, MAX_RETRIES),
                    "mode": retry_mode_prop_value if retry_mode_prop_value in EXISTING_RETRY_MODES else STANDARD_RETRY_MODE,
                }
            ),
        )

        if glue_catalog_id := properties.get(GLUE_ID):
            _register_glue_catalog_id_with_glue_client(self.glue, glue_catalog_id)

commit_table(table, requirements, updates)

Commit updates to a table.

Parameters:

Name	Type	Description	Default
table	Table	The table to be updated.	required
requirements	tuple[TableRequirement, ...]	(Tuple[TableRequirement, ...]): Table requirements.	required
updates	tuple[TableUpdate, ...]	(Tuple[TableUpdate, ...]): Table updates.	required

Returns:

Name	Type	Description
CommitTableResponse	CommitTableResponse	The updated metadata.

Raises:

Type	Description
NoSuchTableError	If a table with the given identifier does not exist.
CommitFailedException	Requirement not met, or a conflict with a concurrent commit.

Source code in pyiceberg/catalog/glue.py

@override
def commit_table(
    self, table: Table, requirements: tuple[TableRequirement, ...], updates: tuple[TableUpdate, ...]
) -> CommitTableResponse:
    """Commit updates to a table.

    Args:
        table (Table): The table to be updated.
        requirements: (Tuple[TableRequirement, ...]): Table requirements.
        updates: (Tuple[TableUpdate, ...]): Table updates.

    Returns:
        CommitTableResponse: The updated metadata.

    Raises:
        NoSuchTableError: If a table with the given identifier does not exist.
        CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
    """
    table_identifier = table.name()
    database_name, table_name = self.identifier_to_database_and_table(table_identifier, NoSuchTableError)

    current_glue_table: TableTypeDef | None
    glue_table_version_id: str | None
    current_table: Table | None
    try:
        current_glue_table = self._get_glue_table(database_name=database_name, table_name=table_name)
        glue_table_version_id = current_glue_table.get("VersionId")
        current_table = self._convert_glue_to_iceberg(glue_table=current_glue_table)
    except NoSuchTableError:
        current_glue_table = None
        glue_table_version_id = None
        current_table = None

    updated_staged_table = self._update_and_stage_table(current_table, table_identifier, requirements, updates)
    if current_table and updated_staged_table.metadata == current_table.metadata:
        # no changes, do nothing
        return CommitTableResponse(metadata=current_table.metadata, metadata_location=current_table.metadata_location)
    # S3 Tables managed storage doesn't support ListObjectsV2, so skip the exist check.
    write_metadata = (
        self._write_metadata_no_exist_check if self._is_s3tables_database(database_name) else self._write_metadata
    )
    write_metadata(
        metadata=updated_staged_table.metadata,
        io=updated_staged_table.io,
        metadata_path=updated_staged_table.metadata_location,
    )

    if current_table:
        # table exists, update the table
        if not glue_table_version_id:
            raise CommitFailedException(
                f"Cannot commit {database_name}.{table_name} because Glue table version id is missing"
            )

        # Pass `version_id` to implement optimistic locking: it ensures updates are rejected if concurrent
        # modifications occur. See more details at https://iceberg.apache.org/docs/latest/aws/#optimistic-locking
        update_table_input = _construct_table_input(
            table_name=table_name,
            metadata_location=updated_staged_table.metadata_location,
            properties=updated_staged_table.properties,
            metadata=updated_staged_table.metadata,
            glue_table=current_glue_table,
            prev_metadata_location=current_table.metadata_location,
        )
        self._update_glue_table(
            database_name=database_name,
            table_name=table_name,
            table_input=update_table_input,
            version_id=glue_table_version_id,
        )
    else:
        # table does not exist, create the table
        create_table_input = _construct_table_input(
            table_name=table_name,
            metadata_location=updated_staged_table.metadata_location,
            properties=updated_staged_table.properties,
            metadata=updated_staged_table.metadata,
        )
        self._create_glue_table(database_name=database_name, table_name=table_name, table_input=create_table_input)

    return CommitTableResponse(
        metadata=updated_staged_table.metadata, metadata_location=updated_staged_table.metadata_location
    )

create_namespace(namespace, properties=EMPTY_DICT)

Create a namespace in the catalog.

Parameters:

Name	Type	Description	Default
namespace	str | Identifier	Namespace identifier.	required
properties	Properties	A string dictionary of properties for the given namespace.	EMPTY_DICT

Raises:

Type	Description
ValueError	If the identifier is invalid.
AlreadyExistsError	If a namespace with the given name already exists.

Source code in pyiceberg/catalog/glue.py

@override
def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
    """Create a namespace in the catalog.

    Args:
        namespace: Namespace identifier.
        properties: A string dictionary of properties for the given namespace.

    Raises:
        ValueError: If the identifier is invalid.
        AlreadyExistsError: If a namespace with the given name already exists.
    """
    database_name = self.identifier_to_database(namespace)
    try:
        self.glue.create_database(DatabaseInput=_construct_database_input(database_name, properties))
    except self.glue.exceptions.AlreadyExistsException as e:
        raise NamespaceAlreadyExistsError(f"Database {database_name} already exists") from e

create_table(identifier, schema, location=None, partition_spec=UNPARTITIONED_PARTITION_SPEC, sort_order=UNSORTED_SORT_ORDER, properties=EMPTY_DICT)

Create an Iceberg table.

Parameters:

Name	Type	Description	Default
identifier	str | Identifier	Table identifier.	required
schema	Union[Schema, Schema]	Table's schema.	required
location	str | None	Location for the table. Optional Argument. Must not be set for S3 Tables, which manage their own storage.	None
partition_spec	PartitionSpec	PartitionSpec for the table.	UNPARTITIONED_PARTITION_SPEC
sort_order	SortOrder	SortOrder for the table.	UNSORTED_SORT_ORDER
properties	Properties	Table properties that can be a string based dictionary.	EMPTY_DICT

Returns:

Name	Type	Description
Table	Table	the created table instance.

Raises:

Type	Description
AlreadyExistsError	If a table with the name already exists.
ValueError	If the identifier is invalid, no path is given to store metadata, or a location is specified for an S3 Tables table.

Source code in pyiceberg/catalog/glue.py

@override
def create_table(
    self,
    identifier: str | Identifier,
    schema: Union[Schema, "pa.Schema"],
    location: str | None = None,
    partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
    sort_order: SortOrder = UNSORTED_SORT_ORDER,
    properties: Properties = EMPTY_DICT,
) -> Table:
    """
    Create an Iceberg table.

    Args:
        identifier: Table identifier.
        schema: Table's schema.
        location: Location for the table. Optional Argument.
            Must not be set for S3 Tables, which manage their own storage.
        partition_spec: PartitionSpec for the table.
        sort_order: SortOrder for the table.
        properties: Table properties that can be a string based dictionary.

    Returns:
        Table: the created table instance.

    Raises:
        AlreadyExistsError: If a table with the name already exists.
        ValueError: If the identifier is invalid, no path is given to store metadata,
            or a location is specified for an S3 Tables table.

    """
    database_name, table_name = self.identifier_to_database_and_table(identifier)

    if self._is_s3tables_database(database_name):
        return self._create_table_s3tables(
            identifier=identifier,
            schema=schema,
            location=location,
            partition_spec=partition_spec,
            sort_order=sort_order,
            properties=properties,
        )

    staged_table = self._create_staged_table(
        identifier=identifier,
        schema=schema,
        location=location,
        partition_spec=partition_spec,
        sort_order=sort_order,
        properties=properties,
    )

    self._write_metadata(staged_table.metadata, staged_table.io, staged_table.metadata_location)
    table_input = _construct_table_input(table_name, staged_table.metadata_location, properties, staged_table.metadata)
    self._create_glue_table(database_name=database_name, table_name=table_name, table_input=table_input)

    return Table(
        identifier=self.identifier_to_tuple(identifier),
        metadata=staged_table.metadata,
        metadata_location=staged_table.metadata_location,
        io=self._load_file_io(staged_table.metadata.properties, staged_table.metadata_location),
        catalog=self,
    )

drop_namespace(namespace)

Drop a namespace.

A Glue namespace can only be dropped if it is empty.

Parameters:

Name	Type	Description	Default
namespace	str | Identifier	Namespace identifier.	required

Raises:

Type	Description
NoSuchNamespaceError	If a namespace with the given name does not exist, or the identifier is invalid.
NamespaceNotEmptyError	If the namespace is not empty.

Source code in pyiceberg/catalog/glue.py

@override
def drop_namespace(self, namespace: str | Identifier) -> None:
    """Drop a namespace.

    A Glue namespace can only be dropped if it is empty.

    Args:
        namespace: Namespace identifier.

    Raises:
        NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
        NamespaceNotEmptyError: If the namespace is not empty.
    """
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    try:
        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:
        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)

drop_table(identifier)

Drop a table.

Parameters:

Name	Type	Description	Default
identifier	str | Identifier	Table identifier.	required

Raises:

Type	Description
NoSuchTableError	If a table with the name does not exist, or the identifier is invalid.

Source code in pyiceberg/catalog/glue.py

@override
def drop_table(self, identifier: str | Identifier) -> None:
    """Drop a table.

    Args:
        identifier: Table identifier.

    Raises:
        NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
    """
    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:
        raise NoSuchTableError(f"Table does not exist: {database_name}.{table_name}") from e

list_namespaces(namespace=())

List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.

Returns:

Type	Description
list[Identifier]	List[Identifier]: a List of namespace identifiers.

Source code in pyiceberg/catalog/glue.py

@override
def list_namespaces(self, namespace: str | Identifier = ()) -> list[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.
    """
    # Hierarchical namespace is not supported. Return an empty list
    if namespace:
        return []

    database_list: list[DatabaseTypeDef] = []
    next_token: str | None = 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"])
        next_token = databases_response.get("NextToken")
        if not next_token:
            break

    return [self.identifier_to_tuple(database["Name"]) for database in database_list]

list_tables(namespace)

List Iceberg tables under the given namespace in the catalog.

Parameters:

Name	Type	Description	Default
namespace	str | Identifier	Namespace identifier to search.	required

Returns:

Type	Description
list[Identifier]	List[Identifier]: list of table identifiers.

Raises:

Type	Description
NoSuchNamespaceError	If a namespace with the given name does not exist, or the identifier is invalid.

Source code in pyiceberg/catalog/glue.py

@override
def list_tables(self, namespace: str | Identifier) -> list[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.

    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: str | None = None
    try:
        while True:
            table_list_response = (
                self.glue.get_tables(DatabaseName=database_name)
                if not next_token
                else self.glue.get_tables(DatabaseName=database_name, NextToken=next_token)
            )
            table_list.extend(table_list_response["TableList"])
            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)]

load_namespace_properties(namespace)

Get properties for a namespace.

Parameters:

Name	Type	Description	Default
namespace	str | Identifier	Namespace identifier.	required

Returns:

Name	Type	Description
Properties	Properties	Properties for the given namespace.

Raises:

Type	Description
NoSuchNamespaceError	If a namespace with the given name does not exist, or identifier is invalid.

Source code in pyiceberg/catalog/glue.py

@override
def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
    """Get properties for a namespace.

    Args:
        namespace: Namespace identifier.

    Returns:
        Properties: Properties for the given namespace.

    Raises:
        NoSuchNamespaceError: If a namespace with the given name does not exist, or identifier is invalid.
    """
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    try:
        database_response = self.glue.get_database(Name=database_name)
    except self.glue.exceptions.EntityNotFoundException as e:
        raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e
    except self.glue.exceptions.InvalidInputException as e:
        raise NoSuchNamespaceError(f"Invalid input for namespace {database_name}") from e

    database = database_response["Database"]

    properties = dict(database.get("Parameters", {}))
    if "LocationUri" in database:
        properties["location"] = database["LocationUri"]
    if "Description" in database:
        properties["Description"] = database["Description"]

    return properties

load_table(identifier)

Load the table's metadata and returns the table instance.

You can also use this method to check for table existence using 'try catalog.table() except TableNotFoundError'. Note: This method doesn't scan data stored in the table.

Parameters:

Name	Type	Description	Default
identifier	str | Identifier	Table identifier.	required

Returns:

Name	Type	Description
Table	Table	the table instance with its metadata.

Raises:

Type	Description
NoSuchTableError	If a table with the name does not exist, or the identifier is invalid.

Source code in pyiceberg/catalog/glue.py

@override
def load_table(self, identifier: str | Identifier) -> Table:
    """Load the table's metadata and returns the table instance.

    You can also use this method to check for table existence using 'try catalog.table() except TableNotFoundError'.
    Note: This method doesn't scan data stored in the table.

    Args:
        identifier: Table identifier.

    Returns:
        Table: the table instance with its metadata.

    Raises:
        NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
    """
    database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)

    return self._convert_glue_to_iceberg(self._get_glue_table(database_name=database_name, table_name=table_name))

register_table(identifier, metadata_location, overwrite=False)

Register a new table using existing metadata.

Parameters:

Name	Type	Description	Default
identifier	Union[str, Identifier]	Table identifier for the table	required
metadata_location	str	The location to the metadata	required
overwrite	bool	Whether to overwrite the existing table, default False	False

Returns:

Name	Type	Description
Table	Table	The newly registered table

Raises:

Type	Description
TableAlreadyExistsError	If the table already exists

Source code in pyiceberg/catalog/glue.py

@override
def register_table(self, identifier: str | Identifier, metadata_location: str, overwrite: bool = False) -> Table:
    """Register a new table using existing metadata.

    Args:
        identifier (Union[str, Identifier]): Table identifier for the table
        metadata_location (str): The location to the metadata
        overwrite (bool): Whether to overwrite the existing table, default False

    Returns:
        Table: The newly registered table

    Raises:
        TableAlreadyExistsError: If the table already exists
    """
    if overwrite:
        raise NotImplementedError("`overwrite` isn't supported")

    database_name, table_name = self.identifier_to_database_and_table(identifier)
    properties = EMPTY_DICT
    io = self._load_file_io(location=metadata_location)
    file = io.new_input(metadata_location)
    metadata = FromInputFile.table_metadata(file)
    table_input = _construct_table_input(table_name, metadata_location, properties, metadata)
    self._create_glue_table(database_name=database_name, table_name=table_name, table_input=table_input)
    return self.load_table(identifier=identifier)

rename_table(from_identifier, to_identifier)

Rename a fully classified table name.

This method can only rename Iceberg tables in AWS Glue.

Parameters:

Name	Type	Description	Default
from_identifier	str | Identifier	Existing table identifier.	required
to_identifier	str | Identifier	New table identifier.	required

Returns:

Name	Type	Description
Table	Table	the updated table instance with its metadata.

Raises:

Type	Description
ValueError	When from table identifier is invalid.
NoSuchTableError	When a table with the name does not exist.
NoSuchIcebergTableError	When from table is not a valid iceberg table.
NoSuchPropertyException	When from table miss some required properties.
NoSuchNamespaceError	When the destination namespace doesn't exist.

Source code in pyiceberg/catalog/glue.py

@override
def rename_table(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> Table:
    """Rename a fully classified table name.

    This method can only rename Iceberg tables in AWS Glue.

    Args:
        from_identifier: Existing table identifier.
        to_identifier: New table identifier.

    Returns:
        Table: the updated table instance with its metadata.

    Raises:
        ValueError: When from table identifier is invalid.
        NoSuchTableError: When a table with the name does not exist.
        NoSuchIcebergTableError: When from table is not a valid iceberg table.
        NoSuchPropertyException: When from table miss some required properties.
        NoSuchNamespaceError: When the destination namespace doesn't exist.
    """
    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)
    except self.glue.exceptions.EntityNotFoundException as e:
        raise NoSuchTableError(f"Table does not exist: {from_database_name}.{from_table_name}") from e

    glue_table = get_table_response["Table"]

    try:
        # verify that from_identifier is a valid iceberg table
        self._convert_glue_to_iceberg(glue_table=glue_table)
    except NoSuchPropertyException as e:
        raise NoSuchPropertyException(
            f"Failed to rename table {from_database_name}.{from_table_name} since it is missing required properties"
        ) from e
    except NoSuchIcebergTableError as e:
        raise NoSuchIcebergTableError(
            f"Failed to rename table {from_database_name}.{from_table_name} since it is not a valid iceberg table"
        ) from e

    rename_table_input = _construct_rename_table_input(to_table_name=to_table_name, glue_table=glue_table)
    self._create_glue_table(database_name=to_database_name, table_name=to_table_name, table_input=rename_table_input)

    try:
        self.drop_table(from_identifier)
    except Exception as e:
        log_message = f"Failed to drop old table {from_database_name}.{from_table_name}. "

        try:
            self.drop_table(to_identifier)
            log_message += f"Rolled back table creation for {to_database_name}.{to_table_name}."
        except NoSuchTableError:
            log_message += (
                f"Failed to roll back table creation for {to_database_name}.{to_table_name}. Please clean up manually"
            )

        raise ValueError(log_message) from e

    return self.load_table(to_identifier)

update_namespace_properties(namespace, removals=None, updates=EMPTY_DICT)

Remove provided property keys and updates properties for a namespace.

Parameters:

Name	Type	Description	Default
namespace	str | Identifier	Namespace identifier.	required
removals	set[str] | None	Set of property keys that need to be removed. Optional Argument.	None
updates	Properties	Properties to be updated for the given namespace.	EMPTY_DICT

Raises:

Type	Description
NoSuchNamespaceError	If a namespace with the given name does not exist， or identifier is invalid.
ValueError	If removals and updates have overlapping keys.

Source code in pyiceberg/catalog/glue.py

@override
def update_namespace_properties(
    self, namespace: str | Identifier, removals: set[str] | None = None, updates: Properties = EMPTY_DICT
) -> PropertiesUpdateSummary:
    """Remove provided property keys and updates properties for a namespace.

    Args:
        namespace: Namespace identifier.
        removals: Set of property keys that need to be removed. Optional Argument.
        updates: Properties to be updated for the given namespace.

    Raises:
        NoSuchNamespaceError: If a namespace with the given name does not exist， or identifier is invalid.
        ValueError: If removals and updates have overlapping keys.
    """
    current_properties = self.load_namespace_properties(namespace=namespace)
    properties_update_summary, updated_properties = self._get_updated_props_and_update_summary(
        current_properties=current_properties, removals=removals, updates=updates
    )

    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    self.glue.update_database(Name=database_name, DatabaseInput=_construct_database_input(database_name, updated_properties))

    return properties_update_summary