update

AssertCreate

Bases: ValidatableTableRequirement

The table must not already exist; used for create transactions.

Source code in pyiceberg/table/update/__init__.py

class AssertCreate(ValidatableTableRequirement):
    """The table must not already exist; used for create transactions."""

    type: Literal["assert-create"] = Field(default="assert-create")

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is not None:
            raise CommitFailedException("Table already exists")

AssertCurrentSchemaId

Bases: ValidatableTableRequirement

The table's current schema id must match the requirement's current-schema-id.

Source code in pyiceberg/table/update/__init__.py

class AssertCurrentSchemaId(ValidatableTableRequirement):
    """The table's current schema id must match the requirement's `current-schema-id`."""

    type: Literal["assert-current-schema-id"] = Field(default="assert-current-schema-id")
    current_schema_id: int = Field(..., alias="current-schema-id")

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is None:
            raise CommitFailedException("Requirement failed: current table metadata is missing")
        elif self.current_schema_id != base_metadata.current_schema_id:
            raise CommitFailedException(
                f"Requirement failed: current schema id has changed: expected {self.current_schema_id}, found {base_metadata.current_schema_id}"
            )

AssertDefaultSortOrderId

Bases: ValidatableTableRequirement

The table's default sort order id must match the requirement's default-sort-order-id.

Source code in pyiceberg/table/update/__init__.py

class AssertDefaultSortOrderId(ValidatableTableRequirement):
    """The table's default sort order id must match the requirement's `default-sort-order-id`."""

    type: Literal["assert-default-sort-order-id"] = Field(default="assert-default-sort-order-id")
    default_sort_order_id: int = Field(..., alias="default-sort-order-id")

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is None:
            raise CommitFailedException("Requirement failed: current table metadata is missing")
        elif self.default_sort_order_id != base_metadata.default_sort_order_id:
            raise CommitFailedException(
                f"Requirement failed: default sort order id has changed: expected {self.default_sort_order_id}, found {base_metadata.default_sort_order_id}"
            )

AssertDefaultSpecId

Bases: ValidatableTableRequirement

The table's default spec id must match the requirement's default-spec-id.

Source code in pyiceberg/table/update/__init__.py

class AssertDefaultSpecId(ValidatableTableRequirement):
    """The table's default spec id must match the requirement's `default-spec-id`."""

    type: Literal["assert-default-spec-id"] = Field(default="assert-default-spec-id")
    default_spec_id: int = Field(..., alias="default-spec-id")

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is None:
            raise CommitFailedException("Requirement failed: current table metadata is missing")
        elif self.default_spec_id != base_metadata.default_spec_id:
            raise CommitFailedException(
                f"Requirement failed: default spec id has changed: expected {self.default_spec_id}, found {base_metadata.default_spec_id}"
            )

AssertLastAssignedFieldId

Bases: ValidatableTableRequirement

The table's last assigned column id must match the requirement's last-assigned-field-id.

Source code in pyiceberg/table/update/__init__.py

class AssertLastAssignedFieldId(ValidatableTableRequirement):
    """The table's last assigned column id must match the requirement's `last-assigned-field-id`."""

    type: Literal["assert-last-assigned-field-id"] = Field(default="assert-last-assigned-field-id")
    last_assigned_field_id: int = Field(..., alias="last-assigned-field-id")

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is None:
            raise CommitFailedException("Requirement failed: current table metadata is missing")
        elif base_metadata.last_column_id != self.last_assigned_field_id:
            raise CommitFailedException(
                f"Requirement failed: last assigned field id has changed: expected {self.last_assigned_field_id}, found {base_metadata.last_column_id}"
            )

AssertLastAssignedPartitionId

Bases: ValidatableTableRequirement

The table's last assigned partition id must match the requirement's last-assigned-partition-id.

Source code in pyiceberg/table/update/__init__.py

class AssertLastAssignedPartitionId(ValidatableTableRequirement):
    """The table's last assigned partition id must match the requirement's `last-assigned-partition-id`."""

    type: Literal["assert-last-assigned-partition-id"] = Field(default="assert-last-assigned-partition-id")
    last_assigned_partition_id: Optional[int] = Field(..., alias="last-assigned-partition-id")

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is None:
            raise CommitFailedException("Requirement failed: current table metadata is missing")
        elif base_metadata.last_partition_id != self.last_assigned_partition_id:
            raise CommitFailedException(
                f"Requirement failed: last assigned partition id has changed: expected {self.last_assigned_partition_id}, found {base_metadata.last_partition_id}"
            )

AssertRefSnapshotId

Bases: ValidatableTableRequirement

The table branch or tag identified by the requirement's ref must reference the requirement's snapshot-id.

if snapshot-id is null or missing, the ref must not already exist.

Source code in pyiceberg/table/update/__init__.py

class AssertRefSnapshotId(ValidatableTableRequirement):
    """The table branch or tag identified by the requirement's `ref` must reference the requirement's `snapshot-id`.

    if `snapshot-id` is `null` or missing, the ref must not already exist.
    """

    type: Literal["assert-ref-snapshot-id"] = Field(default="assert-ref-snapshot-id")
    ref: str = Field(...)
    snapshot_id: Optional[int] = Field(default=None, alias="snapshot-id")

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is None:
            raise CommitFailedException("Requirement failed: current table metadata is missing")
        elif snapshot_ref := base_metadata.refs.get(self.ref):
            ref_type = snapshot_ref.snapshot_ref_type
            if self.snapshot_id is None:
                raise CommitFailedException(f"Requirement failed: {ref_type} {self.ref} was created concurrently")
            elif self.snapshot_id != snapshot_ref.snapshot_id:
                raise CommitFailedException(
                    f"Requirement failed: {ref_type} {self.ref} has changed: expected id {self.snapshot_id}, found {snapshot_ref.snapshot_id}"
                )
        elif self.snapshot_id is not None:
            raise CommitFailedException(f"Requirement failed: branch or tag {self.ref} is missing, expected {self.snapshot_id}")

AssertTableUUID

Bases: ValidatableTableRequirement

The table UUID must match the requirement's uuid.

Source code in pyiceberg/table/update/__init__.py

class AssertTableUUID(ValidatableTableRequirement):
    """The table UUID must match the requirement's `uuid`."""

    type: Literal["assert-table-uuid"] = Field(default="assert-table-uuid")
    uuid: uuid.UUID

    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        if base_metadata is None:
            raise CommitFailedException("Requirement failed: current table metadata is missing")
        elif self.uuid != base_metadata.table_uuid:
            raise CommitFailedException(f"Table UUID does not match: {self.uuid} != {base_metadata.table_uuid}")

UpdateTableMetadata

Bases: ABC, Generic[U]

Source code in pyiceberg/table/update/__init__.py

class UpdateTableMetadata(ABC, Generic[U]):
    _transaction: Transaction

    def __init__(self, transaction: Transaction) -> None:
        self._transaction = transaction

    @abstractmethod
    def _commit(self) -> UpdatesAndRequirements: ...

    def commit(self) -> None:
        self._transaction._apply(*self._commit())

    def __exit__(self, _: Any, value: Any, traceback: Any) -> None:
        """Close and commit the change."""
        self.commit()

    def __enter__(self) -> U:
        """Update the table."""
        return self  # type: ignore

__enter__()

Update the table.

Source code in pyiceberg/table/update/__init__.py

def __enter__(self) -> U:
    """Update the table."""
    return self  # type: ignore

__exit__(_, value, traceback)

Close and commit the change.

Source code in pyiceberg/table/update/__init__.py

def __exit__(self, _: Any, value: Any, traceback: Any) -> None:
    """Close and commit the change."""
    self.commit()

ValidatableTableRequirement

Bases: IcebergBaseModel

Source code in pyiceberg/table/update/__init__.py

class ValidatableTableRequirement(IcebergBaseModel):
    type: str

    @abstractmethod
    def validate(self, base_metadata: Optional[TableMetadata]) -> None:
        """Validate the requirement against the base metadata.

        Args:
            base_metadata: The base metadata to be validated against.

        Raises:
            CommitFailedException: When the requirement is not met.
        """
        ...

validate(base_metadata) abstractmethod

Validate the requirement against the base metadata.

Parameters:

Name	Type	Description	Default
base_metadata	Optional[TableMetadata]	The base metadata to be validated against.	required

Raises:

Type	Description
CommitFailedException	When the requirement is not met.

Source code in pyiceberg/table/update/__init__.py

@abstractmethod
def validate(self, base_metadata: Optional[TableMetadata]) -> None:
    """Validate the requirement against the base metadata.

    Args:
        base_metadata: The base metadata to be validated against.

    Raises:
        CommitFailedException: When the requirement is not met.
    """
    ...

update_table_metadata(base_metadata, updates, enforce_validation=False, metadata_location=None)

Update the table metadata with the given updates in one transaction.

Parameters:

Name	Type	Description	Default
base_metadata	TableMetadata	The base metadata to be updated.	required
updates	Tuple[TableUpdate, ...]	The updates in one transaction.	required
enforce_validation	bool	Whether to trigger validation after applying the updates.	False
metadata_location	Optional[str]	Current metadata location of the table	None

Returns:

Type	Description
TableMetadata	The metadata with the updates applied.

Source code in pyiceberg/table/update/__init__.py

def update_table_metadata(
    base_metadata: TableMetadata,
    updates: Tuple[TableUpdate, ...],
    enforce_validation: bool = False,
    metadata_location: Optional[str] = None,
) -> TableMetadata:
    """Update the table metadata with the given updates in one transaction.

    Args:
        base_metadata: The base metadata to be updated.
        updates: The updates in one transaction.
        enforce_validation: Whether to trigger validation after applying the updates.
        metadata_location: Current metadata location of the table

    Returns:
        The metadata with the updates applied.
    """
    context = _TableMetadataUpdateContext()
    new_metadata = base_metadata

    for update in updates:
        new_metadata = _apply_table_update(update, new_metadata, context)

    # Update last_updated_ms if it was not updated by update operations
    if context.has_changes():
        if metadata_location:
            new_metadata = _update_table_metadata_log(new_metadata, metadata_location, base_metadata.last_updated_ms)
        if base_metadata.last_updated_ms == new_metadata.last_updated_ms:
            new_metadata = new_metadata.model_copy(update={"last_updated_ms": datetime_to_millis(datetime.now().astimezone())})

    if enforce_validation:
        return TableMetadataUtil.parse_obj(new_metadata.model_dump())
    else:
        return new_metadata.model_copy(deep=True)