glue

GlueCatalog

Bases: MetastoreCatalog

Source code in pyiceberg/catalog/glue.py

class GlueCatalog(MetastoreCatalog):
    def __init__(self, name: str, **properties: Any):
        super().__init__(name, **properties)

        session = boto3.Session(
            profile_name=properties.get(GLUE_PROFILE_NAME),
            region_name=get_first_property_value(properties, GLUE_REGION, AWS_REGION),
            botocore_session=properties.get(DEPRECATED_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))

        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:
        properties: Properties = glue_table["Parameters"]

        assert glue_table["DatabaseName"]
        assert glue_table["Parameters"]
        database_name = glue_table["DatabaseName"]
        table_name = glue_table["Name"]

        if TABLE_TYPE not in properties:
            raise NoSuchPropertyException(
                f"Property {TABLE_TYPE} missing, could not determine type: {database_name}.{table_name}"
            )
        glue_table_type = properties[TABLE_TYPE]

        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 not in properties:
            raise NoSuchPropertyException(
                f"Table property {METADATA_LOCATION} is missing, cannot find metadata for: {database_name}.{table_name}"
            )
        metadata_location = properties[METADATA_LOCATION]

        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 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 create_table(
        self,
        identifier: Union[str, Identifier],
        schema: Union[Schema, "pa.Schema"],
        location: Optional[str] = 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.
            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, or no path is given to store metadata.

        """
        staged_table = self._create_staged_table(
            identifier=identifier,
            schema=schema,
            location=location,
            partition_spec=partition_spec,
            sort_order=sort_order,
            properties=properties,
        )
        database_name, table_name = self.identifier_to_database_and_table(identifier)

        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 self.load_table(identifier=identifier)

    def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> 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

        Returns:
            Table: The newly registered table

        Raises:
            TableAlreadyExistsError: If the table already exists
        """
        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)

    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: Optional[TableTypeDef]
        glue_table_version_id: Optional[str]
        current_table: Optional[Table]
        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)
        self._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
        )

    def load_table(self, identifier: Union[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.
        """
        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)

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

    def drop_table(self, identifier: Union[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.
        """
        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
        try:
            self.glue.delete_table(DatabaseName=database_name, Name=table_name)
        except self.glue.exceptions.EntityNotFoundException as e:
            raise NoSuchTableError(f"Table does not exist: {database_name}.{table_name}") from e

    def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[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_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
        to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
        try:
            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}. " f"Please clean up manually"
                )

            raise ValueError(log_message) from e

        return self.load_table(to_identifier)

    def create_namespace(self, namespace: Union[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

    def drop_namespace(self, namespace: Union[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 = self.list_tables(namespace=database_name)
        except NoSuchNamespaceError 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")

        self.glue.delete_database(Name=database_name)

    def list_tables(self, namespace: Union[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: Optional[str] = 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)]

    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.

        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: 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"])
            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.

        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

    def update_namespace_properties(
        self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = 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

    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
        raise NotImplementedError

    def drop_view(self, identifier: Union[str, Identifier]) -> None:
        raise NotImplementedError

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

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

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: Optional[TableTypeDef]
    glue_table_version_id: Optional[str]
    current_table: Optional[Table]
    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)
    self._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	Union[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

def create_namespace(self, namespace: Union[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	Union[str, Identifier]	Table identifier.	required
schema	Union[Schema, Schema]	Table's schema.	required
location	Optional[str]	Location for the table. Optional Argument.	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, or no path is given to store metadata.

Source code in pyiceberg/catalog/glue.py

def create_table(
    self,
    identifier: Union[str, Identifier],
    schema: Union[Schema, "pa.Schema"],
    location: Optional[str] = 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.
        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, or no path is given to store metadata.

    """
    staged_table = self._create_staged_table(
        identifier=identifier,
        schema=schema,
        location=location,
        partition_spec=partition_spec,
        sort_order=sort_order,
        properties=properties,
    )
    database_name, table_name = self.identifier_to_database_and_table(identifier)

    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 self.load_table(identifier=identifier)

drop_namespace(namespace)

Drop a namespace.

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

Parameters:

Name	Type	Description	Default
namespace	Union[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

def drop_namespace(self, namespace: Union[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 = self.list_tables(namespace=database_name)
    except NoSuchNamespaceError 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")

    self.glue.delete_database(Name=database_name)

drop_table(identifier)

Drop a table.

Parameters:

Name	Type	Description	Default
identifier	Union[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

def drop_table(self, identifier: Union[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.
    """
    identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
    database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
    try:
        self.glue.delete_table(DatabaseName=database_name, Name=table_name)
    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

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.

    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: 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"])
        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

def list_tables(self, namespace: Union[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: Optional[str] = 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	Union[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

def load_namespace_properties(self, namespace: Union[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	Union[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

def load_table(self, identifier: Union[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.
    """
    identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
    database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)

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

register_table(identifier, metadata_location)

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

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

def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> 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

    Returns:
        Table: The newly registered table

    Raises:
        TableAlreadyExistsError: If the table already exists
    """
    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	Union[str, Identifier]	Existing table identifier.	required
to_identifier	Union[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

def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[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_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
    from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
    to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
    try:
        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}. " f"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	Union[str, Identifier]	Namespace identifier.	required
removals	Optional[Set[str]]	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

def update_namespace_properties(
    self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = 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