dynamodb

DynamoDbCatalog

Bases: MetastoreCatalog

Source code in pyiceberg/catalog/dynamodb.py

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

        session = boto3.Session(
            profile_name=properties.get(DYNAMODB_PROFILE_NAME),
            region_name=get_first_property_value(properties, DYNAMODB_REGION, AWS_REGION),
            botocore_session=properties.get(DEPRECATED_BOTOCORE_SESSION),
            aws_access_key_id=get_first_property_value(properties, DYNAMODB_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
            aws_secret_access_key=get_first_property_value(properties, DYNAMODB_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
            aws_session_token=get_first_property_value(properties, DYNAMODB_SESSION_TOKEN, AWS_SESSION_TOKEN),
        )
        self.dynamodb = session.client(DYNAMODB_CLIENT)
        self.dynamodb_table_name = self.properties.get(DYNAMODB_TABLE_NAME, DYNAMODB_TABLE_NAME_DEFAULT)
        self._ensure_catalog_table_exists_or_create()

    def _ensure_catalog_table_exists_or_create(self) -> None:
        if self._dynamodb_table_exists():
            return None

        try:
            self.dynamodb.create_table(
                TableName=self.dynamodb_table_name,
                AttributeDefinitions=CREATE_CATALOG_ATTRIBUTE_DEFINITIONS,
                KeySchema=CREATE_CATALOG_KEY_SCHEMA,
                GlobalSecondaryIndexes=CREATE_CATALOG_GLOBAL_SECONDARY_INDEXES,
                BillingMode=DYNAMODB_PAY_PER_REQUEST,
            )
        except (
            self.dynamodb.exceptions.ResourceInUseException,
            self.dynamodb.exceptions.LimitExceededException,
            self.dynamodb.exceptions.InternalServerError,
        ) as e:
            raise GenericDynamoDbError(e.message) from e

    def _dynamodb_table_exists(self) -> bool:
        try:
            response = self.dynamodb.describe_table(TableName=self.dynamodb_table_name)
        except self.dynamodb.exceptions.ResourceNotFoundException:
            return False
        except self.dynamodb.exceptions.InternalServerError as e:
            raise GenericDynamoDbError(e.message) from e

        if response["Table"]["TableStatus"] != ACTIVE:
            raise GenericDynamoDbError(f"DynamoDB table for catalog {self.dynamodb_table_name} is not {ACTIVE}")
        else:
            return True

    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.

        """
        schema: Schema = self._convert_schema_if_needed(schema)  # type: ignore

        database_name, table_name = self.identifier_to_database_and_table(identifier)

        location = self._resolve_table_location(location, database_name, table_name)
        metadata_location = self._get_metadata_location(location=location)
        metadata = new_table_metadata(
            location=location, schema=schema, partition_spec=partition_spec, sort_order=sort_order, properties=properties
        )
        io = load_file_io(properties=self.properties, location=metadata_location)
        self._write_metadata(metadata, io, metadata_location)

        self._ensure_namespace_exists(database_name=database_name)

        try:
            self._put_dynamo_item(
                item=_get_create_table_item(
                    database_name=database_name, table_name=table_name, properties=properties, metadata_location=metadata_location
                ),
                condition_expression=f"attribute_not_exists({DYNAMODB_COL_IDENTIFIER})",
            )
        except ConditionalCheckFailedException as e:
            raise TableAlreadyExistsError(f"Table {database_name}.{table_name} already exists") from e

        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
        """
        raise NotImplementedError

    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.
        """
        raise NotImplementedError

    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)
        dynamo_table_item = self._get_iceberg_table_item(database_name=database_name, table_name=table_name)
        return self._convert_dynamo_table_item_to_iceberg_table(dynamo_table_item=dynamo_table_item)

    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._delete_dynamo_item(
                namespace=database_name,
                identifier=f"{database_name}.{table_name}",
                condition_expression=f"attribute_exists({DYNAMODB_COL_IDENTIFIER})",
            )
        except ConditionalCheckFailedException 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)

        from_table_item = self._get_iceberg_table_item(database_name=from_database_name, table_name=from_table_name)

        try:
            # Verify that from_identifier is a valid iceberg table
            self._convert_dynamo_table_item_to_iceberg_table(dynamo_table_item=from_table_item)
        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

        self._ensure_namespace_exists(database_name=from_database_name)
        self._ensure_namespace_exists(database_name=to_database_name)

        try:
            self._put_dynamo_item(
                item=_get_rename_table_item(
                    from_dynamo_table_item=from_table_item, to_database_name=to_database_name, to_table_name=to_table_name
                ),
                condition_expression=f"attribute_not_exists({DYNAMODB_COL_IDENTIFIER})",
            )
        except ConditionalCheckFailedException as e:
            raise TableAlreadyExistsError(f"Table {to_database_name}.{to_table_name} already exists") from e

        try:
            self.drop_table(from_identifier_tuple)
        except (NoSuchTableError, GenericDynamoDbError) 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, GenericDynamoDbError):
                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._put_dynamo_item(
                item=_get_create_database_item(database_name=database_name, properties=properties),
                condition_expression=f"attribute_not_exists({DYNAMODB_COL_NAMESPACE})",
            )
        except ConditionalCheckFailedException 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)
        table_identifiers = self.list_tables(namespace=database_name)

        if len(table_identifiers) > 0:
            raise NamespaceNotEmptyError(f"Database {database_name} is not empty")

        try:
            self._delete_dynamo_item(
                namespace=database_name,
                identifier=DYNAMODB_NAMESPACE,
                condition_expression=f"attribute_exists({DYNAMODB_COL_IDENTIFIER})",
            )
        except ConditionalCheckFailedException as e:
            raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e

    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
        """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.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)

        paginator = self.dynamodb.get_paginator("query")

        try:
            page_iterator = paginator.paginate(
                TableName=self.dynamodb_table_name,
                IndexName=DYNAMODB_NAMESPACE_GSI,
                KeyConditionExpression=f"{DYNAMODB_COL_NAMESPACE} = :namespace ",
                ExpressionAttributeValues={
                    ":namespace": {
                        "S": database_name,
                    }
                },
            )
        except (
            self.dynamodb.exceptions.ProvisionedThroughputExceededException,
            self.dynamodb.exceptions.RequestLimitExceeded,
            self.dynamodb.exceptions.InternalServerError,
            self.dynamodb.exceptions.ResourceNotFoundException,
        ) as e:
            raise GenericDynamoDbError(e.message) from e

        table_identifiers = []
        for page in page_iterator:
            for item in page["Items"]:
                _dict = _convert_dynamo_item_to_regular_dict(item)
                identifier_col = _dict[DYNAMODB_COL_IDENTIFIER]
                if identifier_col == DYNAMODB_NAMESPACE:
                    continue

                table_identifiers.append(self.identifier_to_tuple(identifier_col))

        return table_identifiers

    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
        """List top-level namespaces from the catalog.

        We do not support hierarchical namespace.

        Returns:
            List[Identifier]: a List of namespace identifiers.
        """
        # Hierarchical namespace is not supported. Return an empty list
        if namespace:
            return []

        paginator = self.dynamodb.get_paginator("query")

        try:
            page_iterator = paginator.paginate(
                TableName=self.dynamodb_table_name,
                ConsistentRead=True,
                KeyConditionExpression=f"{DYNAMODB_COL_IDENTIFIER} = :identifier",
                ExpressionAttributeValues={
                    ":identifier": {
                        "S": DYNAMODB_NAMESPACE,
                    }
                },
            )
        except (
            self.dynamodb.exceptions.ProvisionedThroughputExceededException,
            self.dynamodb.exceptions.RequestLimitExceeded,
            self.dynamodb.exceptions.InternalServerError,
            self.dynamodb.exceptions.ResourceNotFoundException,
        ) as e:
            raise GenericDynamoDbError(e.message) from e

        database_identifiers = []
        for page in page_iterator:
            for item in page["Items"]:
                _dict = _convert_dynamo_item_to_regular_dict(item)
                namespace_col = _dict[DYNAMODB_COL_NAMESPACE]
                database_identifiers.append(self.identifier_to_tuple(namespace_col))

        return database_identifiers

    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)
        namespace_item = self._get_iceberg_namespace_item(database_name=database_name)
        namespace_dict = _convert_dynamo_item_to_regular_dict(namespace_item)
        return _get_namespace_properties(namespace_dict=namespace_dict)

    def update_namespace_properties(
        self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = None, updates: Properties = EMPTY_DICT
    ) -> PropertiesUpdateSummary:
        """
        Remove or update provided property keys 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.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        namespace_item = self._get_iceberg_namespace_item(database_name=database_name)
        namespace_dict = _convert_dynamo_item_to_regular_dict(namespace_item)
        current_properties = _get_namespace_properties(namespace_dict=namespace_dict)

        properties_update_summary, updated_properties = self._get_updated_props_and_update_summary(
            current_properties=current_properties, removals=removals, updates=updates
        )

        try:
            self._put_dynamo_item(
                item=_get_update_database_item(
                    namespace_item=namespace_item,
                    updated_properties=updated_properties,
                ),
                condition_expression=f"attribute_exists({DYNAMODB_COL_NAMESPACE})",
            )
        except ConditionalCheckFailedException as e:
            raise NoSuchNamespaceError(f"Database {database_name} does not exist") from e

        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

    def _get_iceberg_table_item(self, database_name: str, table_name: str) -> Dict[str, Any]:
        try:
            return self._get_dynamo_item(identifier=f"{database_name}.{table_name}", namespace=database_name)
        except ValueError as e:
            raise NoSuchTableError(f"Table does not exist: {database_name}.{table_name}") from e

    def _get_iceberg_namespace_item(self, database_name: str) -> Dict[str, Any]:
        try:
            return self._get_dynamo_item(identifier=DYNAMODB_NAMESPACE, namespace=database_name)
        except ValueError as e:
            raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}") from e

    def _ensure_namespace_exists(self, database_name: str) -> Dict[str, Any]:
        return self._get_iceberg_namespace_item(database_name)

    def _get_dynamo_item(self, identifier: str, namespace: str) -> Dict[str, Any]:
        try:
            response = self.dynamodb.get_item(
                TableName=self.dynamodb_table_name,
                ConsistentRead=True,
                Key={
                    DYNAMODB_COL_IDENTIFIER: {
                        "S": identifier,
                    },
                    DYNAMODB_COL_NAMESPACE: {
                        "S": namespace,
                    },
                },
            )
            if ITEM in response:
                return response[ITEM]
            else:
                raise ValueError(f"Item not found. identifier: {identifier} - namespace: {namespace}")
        except self.dynamodb.exceptions.ResourceNotFoundException as e:
            raise ValueError(f"Item not found. identifier: {identifier} - namespace: {namespace}") from e
        except (
            self.dynamodb.exceptions.ProvisionedThroughputExceededException,
            self.dynamodb.exceptions.RequestLimitExceeded,
            self.dynamodb.exceptions.InternalServerError,
        ) as e:
            raise GenericDynamoDbError(e.message) from e

    def _put_dynamo_item(self, item: Dict[str, Any], condition_expression: str) -> None:
        try:
            self.dynamodb.put_item(TableName=self.dynamodb_table_name, Item=item, ConditionExpression=condition_expression)
        except self.dynamodb.exceptions.ConditionalCheckFailedException as e:
            raise ConditionalCheckFailedException(f"Condition expression check failed: {condition_expression} - {item}") from e
        except (
            self.dynamodb.exceptions.ProvisionedThroughputExceededException,
            self.dynamodb.exceptions.RequestLimitExceeded,
            self.dynamodb.exceptions.InternalServerError,
            self.dynamodb.exceptions.ResourceNotFoundException,
            self.dynamodb.exceptions.ItemCollectionSizeLimitExceededException,
            self.dynamodb.exceptions.TransactionConflictException,
        ) as e:
            raise GenericDynamoDbError(e.message) from e

    def _delete_dynamo_item(self, namespace: str, identifier: str, condition_expression: str) -> None:
        try:
            self.dynamodb.delete_item(
                TableName=self.dynamodb_table_name,
                Key={
                    DYNAMODB_COL_IDENTIFIER: {
                        "S": identifier,
                    },
                    DYNAMODB_COL_NAMESPACE: {
                        "S": namespace,
                    },
                },
                ConditionExpression=condition_expression,
            )
        except self.dynamodb.exceptions.ConditionalCheckFailedException as e:
            raise ConditionalCheckFailedException(
                f"Condition expression check failed: {condition_expression} - {identifier}"
            ) from e
        except (
            self.dynamodb.exceptions.ProvisionedThroughputExceededException,
            self.dynamodb.exceptions.RequestLimitExceeded,
            self.dynamodb.exceptions.InternalServerError,
            self.dynamodb.exceptions.ResourceNotFoundException,
            self.dynamodb.exceptions.ItemCollectionSizeLimitExceededException,
            self.dynamodb.exceptions.TransactionConflictException,
        ) as e:
            raise GenericDynamoDbError(e.message) from e

    def _convert_dynamo_table_item_to_iceberg_table(self, dynamo_table_item: Dict[str, Any]) -> Table:
        table_dict = _convert_dynamo_item_to_regular_dict(dynamo_table_item)

        for prop in [_add_property_prefix(prop) for prop in (TABLE_TYPE, METADATA_LOCATION)] + [
            DYNAMODB_COL_IDENTIFIER,
            DYNAMODB_COL_NAMESPACE,
            DYNAMODB_COL_CREATED_AT,
        ]:
            if prop not in table_dict.keys():
                raise NoSuchPropertyException(f"Iceberg required property {prop} is missing: {dynamo_table_item}")

        table_type = table_dict[_add_property_prefix(TABLE_TYPE)]
        identifier = table_dict[DYNAMODB_COL_IDENTIFIER]
        metadata_location = table_dict[_add_property_prefix(METADATA_LOCATION)]
        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)

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

        io = load_file_io(properties=self.properties, 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,
        )

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/dynamodb.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.
    """
    raise NotImplementedError

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/dynamodb.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._put_dynamo_item(
            item=_get_create_database_item(database_name=database_name, properties=properties),
            condition_expression=f"attribute_not_exists({DYNAMODB_COL_NAMESPACE})",
        )
    except ConditionalCheckFailedException 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/dynamodb.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.

    """
    schema: Schema = self._convert_schema_if_needed(schema)  # type: ignore

    database_name, table_name = self.identifier_to_database_and_table(identifier)

    location = self._resolve_table_location(location, database_name, table_name)
    metadata_location = self._get_metadata_location(location=location)
    metadata = new_table_metadata(
        location=location, schema=schema, partition_spec=partition_spec, sort_order=sort_order, properties=properties
    )
    io = load_file_io(properties=self.properties, location=metadata_location)
    self._write_metadata(metadata, io, metadata_location)

    self._ensure_namespace_exists(database_name=database_name)

    try:
        self._put_dynamo_item(
            item=_get_create_table_item(
                database_name=database_name, table_name=table_name, properties=properties, metadata_location=metadata_location
            ),
            condition_expression=f"attribute_not_exists({DYNAMODB_COL_IDENTIFIER})",
        )
    except ConditionalCheckFailedException as e:
        raise TableAlreadyExistsError(f"Table {database_name}.{table_name} already exists") from e

    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/dynamodb.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)
    table_identifiers = self.list_tables(namespace=database_name)

    if len(table_identifiers) > 0:
        raise NamespaceNotEmptyError(f"Database {database_name} is not empty")

    try:
        self._delete_dynamo_item(
            namespace=database_name,
            identifier=DYNAMODB_NAMESPACE,
            condition_expression=f"attribute_exists({DYNAMODB_COL_IDENTIFIER})",
        )
    except ConditionalCheckFailedException as e:
        raise NoSuchNamespaceError(f"Database does not exist: {database_name}") from e

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/dynamodb.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._delete_dynamo_item(
            namespace=database_name,
            identifier=f"{database_name}.{table_name}",
            condition_expression=f"attribute_exists({DYNAMODB_COL_IDENTIFIER})",
        )
    except ConditionalCheckFailedException as e:
        raise NoSuchTableError(f"Table does not exist: {database_name}.{table_name}") from e

list_namespaces(namespace=())

List top-level namespaces from the catalog.

We do not support hierarchical namespace.

Returns:

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

Source code in pyiceberg/catalog/dynamodb.py

def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
    """List top-level namespaces from the catalog.

    We do not support hierarchical namespace.

    Returns:
        List[Identifier]: a List of namespace identifiers.
    """
    # Hierarchical namespace is not supported. Return an empty list
    if namespace:
        return []

    paginator = self.dynamodb.get_paginator("query")

    try:
        page_iterator = paginator.paginate(
            TableName=self.dynamodb_table_name,
            ConsistentRead=True,
            KeyConditionExpression=f"{DYNAMODB_COL_IDENTIFIER} = :identifier",
            ExpressionAttributeValues={
                ":identifier": {
                    "S": DYNAMODB_NAMESPACE,
                }
            },
        )
    except (
        self.dynamodb.exceptions.ProvisionedThroughputExceededException,
        self.dynamodb.exceptions.RequestLimitExceeded,
        self.dynamodb.exceptions.InternalServerError,
        self.dynamodb.exceptions.ResourceNotFoundException,
    ) as e:
        raise GenericDynamoDbError(e.message) from e

    database_identifiers = []
    for page in page_iterator:
        for item in page["Items"]:
            _dict = _convert_dynamo_item_to_regular_dict(item)
            namespace_col = _dict[DYNAMODB_COL_NAMESPACE]
            database_identifiers.append(self.identifier_to_tuple(namespace_col))

    return database_identifiers

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.

Source code in pyiceberg/catalog/dynamodb.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.
    """
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)

    paginator = self.dynamodb.get_paginator("query")

    try:
        page_iterator = paginator.paginate(
            TableName=self.dynamodb_table_name,
            IndexName=DYNAMODB_NAMESPACE_GSI,
            KeyConditionExpression=f"{DYNAMODB_COL_NAMESPACE} = :namespace ",
            ExpressionAttributeValues={
                ":namespace": {
                    "S": database_name,
                }
            },
        )
    except (
        self.dynamodb.exceptions.ProvisionedThroughputExceededException,
        self.dynamodb.exceptions.RequestLimitExceeded,
        self.dynamodb.exceptions.InternalServerError,
        self.dynamodb.exceptions.ResourceNotFoundException,
    ) as e:
        raise GenericDynamoDbError(e.message) from e

    table_identifiers = []
    for page in page_iterator:
        for item in page["Items"]:
            _dict = _convert_dynamo_item_to_regular_dict(item)
            identifier_col = _dict[DYNAMODB_COL_IDENTIFIER]
            if identifier_col == DYNAMODB_NAMESPACE:
                continue

            table_identifiers.append(self.identifier_to_tuple(identifier_col))

    return table_identifiers

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/dynamodb.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)
    namespace_item = self._get_iceberg_namespace_item(database_name=database_name)
    namespace_dict = _convert_dynamo_item_to_regular_dict(namespace_item)
    return _get_namespace_properties(namespace_dict=namespace_dict)

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/dynamodb.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)
    dynamo_table_item = self._get_iceberg_table_item(database_name=database_name, table_name=table_name)
    return self._convert_dynamo_table_item_to_iceberg_table(dynamo_table_item=dynamo_table_item)

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/dynamodb.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
    """
    raise NotImplementedError

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/dynamodb.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)

    from_table_item = self._get_iceberg_table_item(database_name=from_database_name, table_name=from_table_name)

    try:
        # Verify that from_identifier is a valid iceberg table
        self._convert_dynamo_table_item_to_iceberg_table(dynamo_table_item=from_table_item)
    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

    self._ensure_namespace_exists(database_name=from_database_name)
    self._ensure_namespace_exists(database_name=to_database_name)

    try:
        self._put_dynamo_item(
            item=_get_rename_table_item(
                from_dynamo_table_item=from_table_item, to_database_name=to_database_name, to_table_name=to_table_name
            ),
            condition_expression=f"attribute_not_exists({DYNAMODB_COL_IDENTIFIER})",
        )
    except ConditionalCheckFailedException as e:
        raise TableAlreadyExistsError(f"Table {to_database_name}.{to_table_name} already exists") from e

    try:
        self.drop_table(from_identifier_tuple)
    except (NoSuchTableError, GenericDynamoDbError) 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, GenericDynamoDbError):
            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 or update provided property keys 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/dynamodb.py

def update_namespace_properties(
    self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = None, updates: Properties = EMPTY_DICT
) -> PropertiesUpdateSummary:
    """
    Remove or update provided property keys 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.
    """
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    namespace_item = self._get_iceberg_namespace_item(database_name=database_name)
    namespace_dict = _convert_dynamo_item_to_regular_dict(namespace_item)
    current_properties = _get_namespace_properties(namespace_dict=namespace_dict)

    properties_update_summary, updated_properties = self._get_updated_props_and_update_summary(
        current_properties=current_properties, removals=removals, updates=updates
    )

    try:
        self._put_dynamo_item(
            item=_get_update_database_item(
                namespace_item=namespace_item,
                updated_properties=updated_properties,
            ),
            condition_expression=f"attribute_exists({DYNAMODB_COL_NAMESPACE})",
        )
    except ConditionalCheckFailedException as e:
        raise NoSuchNamespaceError(f"Database {database_name} does not exist") from e

    return properties_update_summary