snapshots

Operation

Bases: Enum

Describes the operation.

Possible operation values are

• append: Only data files were added and no files were removed.
• replace: Data and delete files were added and removed without changing table data; i.e., compaction, changing the data file format, or relocating data files.
• overwrite: Data and delete files were added and removed in a logical overwrite operation.
• delete: Data files were removed and their contents logically deleted and/or delete files were added to delete rows.

Source code in pyiceberg/table/snapshots.py

class Operation(Enum):
    """Describes the operation.

    Possible operation values are:
        - append: Only data files were added and no files were removed.
        - replace: Data and delete files were added and removed without changing table data;
            i.e., compaction, changing the data file format, or relocating data files.
        - overwrite: Data and delete files were added and removed in a logical overwrite operation.
        - delete: Data files were removed and their contents logically deleted and/or delete files
            were added to delete rows.
    """

    APPEND = "append"
    REPLACE = "replace"
    OVERWRITE = "overwrite"
    DELETE = "delete"

    def __repr__(self) -> str:
        """Return the string representation of the Operation class."""
        return f"Operation.{self.name}"

__repr__()

Return the string representation of the Operation class.

Source code in pyiceberg/table/snapshots.py

def __repr__(self) -> str:
    """Return the string representation of the Operation class."""
    return f"Operation.{self.name}"

Snapshot

Bases: IcebergBaseModel

Source code in pyiceberg/table/snapshots.py

class Snapshot(IcebergBaseModel):
    snapshot_id: int = Field(alias="snapshot-id")
    parent_snapshot_id: int | None = Field(alias="parent-snapshot-id", default=None)
    sequence_number: int | None = Field(alias="sequence-number", default=INITIAL_SEQUENCE_NUMBER)
    timestamp_ms: int = Field(alias="timestamp-ms", default_factory=lambda: int(time.time() * 1000))
    manifest_list: str = Field(alias="manifest-list", description="Location of the snapshot's manifest list file")
    summary: Summary | None = Field(default=None)
    schema_id: int | None = Field(alias="schema-id", default=None)
    first_row_id: int | None = Field(
        alias="first-row-id", default=None, description="assigned to the first row in the first data file in the first manifest"
    )
    added_rows: int | None = Field(
        alias="added-rows", default=None, description="The upper bound of the number of rows with assigned row IDs"
    )

    def __str__(self) -> str:
        """Return the string representation of the Snapshot class."""
        operation = f"{self.summary.operation}: " if self.summary else ""
        parent_id = f", parent_id={self.parent_snapshot_id}" if self.parent_snapshot_id else ""
        schema_id = f", schema_id={self.schema_id}" if self.schema_id is not None else ""
        result_str = f"{operation}id={self.snapshot_id}{parent_id}{schema_id}"
        return result_str

    def __repr__(self) -> str:
        """Return the string representation of the Snapshot class."""
        fields = [
            f"snapshot_id={self.snapshot_id}",
            f"parent_snapshot_id={self.parent_snapshot_id}",
            f"sequence_number={self.sequence_number}",
            f"timestamp_ms={self.timestamp_ms}",
            f"manifest_list='{self.manifest_list}'",
            f"summary={repr(self.summary)}" if self.summary else None,
            f"schema_id={self.schema_id}" if self.schema_id is not None else None,
            f"first_row_id={self.first_row_id}" if self.first_row_id is not None else None,
            f"added_rows={self.added_rows}" if self.added_rows is not None else None,
        ]
        filtered_fields = [field for field in fields if field is not None]
        return f"Snapshot({', '.join(filtered_fields)})"

    def manifests(self, io: FileIO) -> list[ManifestFile]:
        """Return the manifests for the given snapshot."""
        return list(_manifests(io, self.manifest_list))

__repr__()

Return the string representation of the Snapshot class.

Source code in pyiceberg/table/snapshots.py

def __repr__(self) -> str:
    """Return the string representation of the Snapshot class."""
    fields = [
        f"snapshot_id={self.snapshot_id}",
        f"parent_snapshot_id={self.parent_snapshot_id}",
        f"sequence_number={self.sequence_number}",
        f"timestamp_ms={self.timestamp_ms}",
        f"manifest_list='{self.manifest_list}'",
        f"summary={repr(self.summary)}" if self.summary else None,
        f"schema_id={self.schema_id}" if self.schema_id is not None else None,
        f"first_row_id={self.first_row_id}" if self.first_row_id is not None else None,
        f"added_rows={self.added_rows}" if self.added_rows is not None else None,
    ]
    filtered_fields = [field for field in fields if field is not None]
    return f"Snapshot({', '.join(filtered_fields)})"

__str__()

Return the string representation of the Snapshot class.

Source code in pyiceberg/table/snapshots.py

def __str__(self) -> str:
    """Return the string representation of the Snapshot class."""
    operation = f"{self.summary.operation}: " if self.summary else ""
    parent_id = f", parent_id={self.parent_snapshot_id}" if self.parent_snapshot_id else ""
    schema_id = f", schema_id={self.schema_id}" if self.schema_id is not None else ""
    result_str = f"{operation}id={self.snapshot_id}{parent_id}{schema_id}"
    return result_str

manifests(io)

Return the manifests for the given snapshot.

Source code in pyiceberg/table/snapshots.py

def manifests(self, io: FileIO) -> list[ManifestFile]:
    """Return the manifests for the given snapshot."""
    return list(_manifests(io, self.manifest_list))

Summary

Bases: IcebergBaseModel, Mapping[str, str]

A class that stores the summary information for a Snapshot.

The snapshot summary’s operation field is used by some operations, like snapshot expiration, to skip processing certain snapshots.

Source code in pyiceberg/table/snapshots.py

class Summary(IcebergBaseModel, Mapping[str, str]):
    """A class that stores the summary information for a Snapshot.

    The snapshot summary’s operation field is used by some operations,
    like snapshot expiration, to skip processing certain snapshots.
    """

    operation: Operation = Field()
    _additional_properties: dict[str, str] = PrivateAttr()

    def __init__(self, operation: Operation | None = None, **data: Any) -> None:
        if operation is None:
            warnings.warn("Encountered invalid snapshot summary: operation is missing, defaulting to overwrite", stacklevel=2)
            operation = Operation.OVERWRITE
        super().__init__(operation=operation, **data)
        self._additional_properties = data

    def __getitem__(self, __key: str) -> Any | None:  # type: ignore
        """Return a key as it is a map."""
        if __key.lower() == "operation":
            return self.operation
        else:
            return self._additional_properties.get(__key)

    def __setitem__(self, key: str, value: Any) -> None:
        """Set a key as it is a map."""
        if key.lower() == "operation":
            self.operation = value
        else:
            self._additional_properties[key] = value

    def __len__(self) -> int:
        """Return the number of keys in the summary."""
        # Operation is required
        return 1 + len(self._additional_properties)

    @model_serializer
    def ser_model(self) -> dict[str, str]:
        return {
            "operation": str(self.operation.value),
            **self._additional_properties,
        }

    @property
    def additional_properties(self) -> dict[str, str]:
        return self._additional_properties

    def __repr__(self) -> str:
        """Return the string representation of the Summary class."""
        repr_properties = f", **{repr(self._additional_properties)}" if self._additional_properties else ""
        return f"Summary({repr(self.operation)}{repr_properties})"

    def __eq__(self, other: Any) -> bool:
        """Compare if the summary is equal to another summary."""
        return (
            self.operation == other.operation and self.additional_properties == other.additional_properties
            if isinstance(other, Summary)
            else False
        )

__eq__(other)

Compare if the summary is equal to another summary.

Source code in pyiceberg/table/snapshots.py

def __eq__(self, other: Any) -> bool:
    """Compare if the summary is equal to another summary."""
    return (
        self.operation == other.operation and self.additional_properties == other.additional_properties
        if isinstance(other, Summary)
        else False
    )

__getitem__(__key)

Return a key as it is a map.

Source code in pyiceberg/table/snapshots.py

def __getitem__(self, __key: str) -> Any | None:  # type: ignore
    """Return a key as it is a map."""
    if __key.lower() == "operation":
        return self.operation
    else:
        return self._additional_properties.get(__key)

__len__()

Return the number of keys in the summary.

Source code in pyiceberg/table/snapshots.py

def __len__(self) -> int:
    """Return the number of keys in the summary."""
    # Operation is required
    return 1 + len(self._additional_properties)

__repr__()

Return the string representation of the Summary class.

Source code in pyiceberg/table/snapshots.py

def __repr__(self) -> str:
    """Return the string representation of the Summary class."""
    repr_properties = f", **{repr(self._additional_properties)}" if self._additional_properties else ""
    return f"Summary({repr(self.operation)}{repr_properties})"

__setitem__(key, value)

Set a key as it is a map.

Source code in pyiceberg/table/snapshots.py

def __setitem__(self, key: str, value: Any) -> None:
    """Set a key as it is a map."""
    if key.lower() == "operation":
        self.operation = value
    else:
        self._additional_properties[key] = value

ancestors_between(from_snapshot, to_snapshot, table_metadata)

Get the ancestors of and including the given snapshot between the to and from snapshots.

Source code in pyiceberg/table/snapshots.py

def ancestors_between(from_snapshot: Snapshot | None, to_snapshot: Snapshot, table_metadata: TableMetadata) -> Iterable[Snapshot]:
    """Get the ancestors of and including the given snapshot between the to and from snapshots."""
    if from_snapshot is not None:
        for snapshot in ancestors_of(to_snapshot, table_metadata):
            yield snapshot
            if snapshot == from_snapshot:
                break
    else:
        yield from ancestors_of(to_snapshot, table_metadata)

ancestors_of(current_snapshot, table_metadata)

Get the ancestors of and including the given snapshot.

Source code in pyiceberg/table/snapshots.py

def ancestors_of(current_snapshot: Snapshot | None, table_metadata: TableMetadata) -> Iterable[Snapshot]:
    """Get the ancestors of and including the given snapshot."""
    snapshot = current_snapshot
    while snapshot is not None:
        yield snapshot
        if snapshot.parent_snapshot_id is None:
            break
        snapshot = table_metadata.snapshot_by_id(snapshot.parent_snapshot_id)

latest_ancestor_before_timestamp(table_metadata, timestamp_ms)

Find the latest ancestor snapshot whose timestamp is before the provided timestamp.

Parameters:

Name	Type	Description	Default
table_metadata	TableMetadata	The table metadata for a table	required
timestamp_ms	int	lookup snapshots strictly before this timestamp	required

Returns:

Type	Description
Snapshot | None	The latest ancestor snapshot older than the timestamp, or None if not found.

Source code in pyiceberg/table/snapshots.py

def latest_ancestor_before_timestamp(table_metadata: TableMetadata, timestamp_ms: int) -> Snapshot | None:
    """Find the latest ancestor snapshot whose timestamp is before the provided timestamp.

    Args:
        table_metadata: The table metadata for a table
        timestamp_ms: lookup snapshots strictly before this timestamp

    Returns:
        The latest ancestor snapshot older than the timestamp, or None if not found.
    """
    result: Snapshot | None = None
    result_timestamp: int = 0

    for ancestor in ancestors_of(table_metadata.current_snapshot(), table_metadata):
        if timestamp_ms > ancestor.timestamp_ms > result_timestamp:
            result = ancestor
            result_timestamp = ancestor.timestamp_ms

    return result