table

BaseScan

Bases: ABC

A base class for all table scans.

Source code in pyiceberg/table/__init__.py

class BaseScan(ABC):
    """A base class for all table scans."""

    table_metadata: TableMetadata
    io: FileIO
    row_filter: BooleanExpression
    selected_fields: tuple[str, ...]
    case_sensitive: bool
    options: Properties
    limit: int | None

    def __init__(
        self,
        table_metadata: TableMetadata,
        io: FileIO,
        row_filter: str | BooleanExpression = ALWAYS_TRUE,
        selected_fields: tuple[str, ...] = ("*",),
        case_sensitive: bool = True,
        options: Properties = EMPTY_DICT,
        limit: int | None = None,
    ):
        self.table_metadata = table_metadata
        self.io = io
        self.row_filter = _parse_row_filter(row_filter)
        self.selected_fields = selected_fields
        self.case_sensitive = case_sensitive
        self.options = options
        self.limit = limit

    @abstractmethod
    def plan_files(self) -> Iterable[ScanTask]: ...

    @abstractmethod
    def to_arrow(self) -> pa.Table: ...

    def update(self: A, **overrides: Any) -> A:
        """Create a copy of this table scan with updated fields."""
        from inspect import signature

        # Extract those attributes that are constructor parameters. We don't use self.__dict__ as the kwargs to the
        # constructors because it may contain additional attributes that are not part of the constructor signature.
        params = signature(type(self).__init__).parameters.keys() - {"self"}  # Skip "self" parameter
        kwargs = {param: getattr(self, param) for param in params}  # Assume parameters are attributes

        return type(self)(**{**kwargs, **overrides})

    def select(self: A, *field_names: str) -> A:
        if "*" in self.selected_fields:
            return self.update(selected_fields=field_names)
        return self.update(selected_fields=tuple(set(self.selected_fields).intersection(set(field_names))))

    def filter(self: A, expr: str | BooleanExpression) -> A:
        return self.update(row_filter=And(self.row_filter, _parse_row_filter(expr)))

    def with_case_sensitive(self: A, case_sensitive: bool = True) -> A:
        return self.update(case_sensitive=case_sensitive)

    def to_pandas(self, **kwargs: Any) -> pd.DataFrame:
        """Read a Pandas DataFrame eagerly from this Iceberg table.

        Returns:
            pd.DataFrame: Materialized Pandas Dataframe from the Iceberg table
        """
        return self.to_arrow().to_pandas(**kwargs)

    def to_duckdb(self, table_name: str, connection: DuckDBPyConnection | None = None) -> DuckDBPyConnection:
        """Shorthand for loading the Iceberg Table in DuckDB.

        Returns:
            DuckDBPyConnection: In memory DuckDB connection with the Iceberg table.
        """
        import duckdb

        con = connection or duckdb.connect(database=":memory:")
        con.register(table_name, self.to_arrow())

        return con

    def to_ray(self) -> ray.data.dataset.Dataset:
        """Read a Ray Dataset eagerly from this Iceberg table.

        Returns:
            ray.data.dataset.Dataset: Materialized Ray Dataset from the Iceberg table
        """
        import ray

        return ray.data.from_arrow(self.to_arrow())

    def to_polars(self) -> pl.DataFrame:
        """Read a Polars DataFrame from this Iceberg table.

        Returns:
            pl.DataFrame: Materialized Polars Dataframe from the Iceberg table
        """
        import polars as pl

        result = pl.from_arrow(self.to_arrow())
        if isinstance(result, pl.Series):
            result = result.to_frame()

        return result

to_duckdb(table_name, connection=None)

Shorthand for loading the Iceberg Table in DuckDB.

Returns:

Name	Type	Description
DuckDBPyConnection	DuckDBPyConnection	In memory DuckDB connection with the Iceberg table.

Source code in pyiceberg/table/__init__.py

def to_duckdb(self, table_name: str, connection: DuckDBPyConnection | None = None) -> DuckDBPyConnection:
    """Shorthand for loading the Iceberg Table in DuckDB.

    Returns:
        DuckDBPyConnection: In memory DuckDB connection with the Iceberg table.
    """
    import duckdb

    con = connection or duckdb.connect(database=":memory:")
    con.register(table_name, self.to_arrow())

    return con

to_pandas(**kwargs)

Read a Pandas DataFrame eagerly from this Iceberg table.

Returns:

Type	Description
DataFrame	pd.DataFrame: Materialized Pandas Dataframe from the Iceberg table

Source code in pyiceberg/table/__init__.py

def to_pandas(self, **kwargs: Any) -> pd.DataFrame:
    """Read a Pandas DataFrame eagerly from this Iceberg table.

    Returns:
        pd.DataFrame: Materialized Pandas Dataframe from the Iceberg table
    """
    return self.to_arrow().to_pandas(**kwargs)

to_polars()

Read a Polars DataFrame from this Iceberg table.

Returns:

Type	Description
DataFrame	pl.DataFrame: Materialized Polars Dataframe from the Iceberg table

Source code in pyiceberg/table/__init__.py

def to_polars(self) -> pl.DataFrame:
    """Read a Polars DataFrame from this Iceberg table.

    Returns:
        pl.DataFrame: Materialized Polars Dataframe from the Iceberg table
    """
    import polars as pl

    result = pl.from_arrow(self.to_arrow())
    if isinstance(result, pl.Series):
        result = result.to_frame()

    return result

to_ray()

Read a Ray Dataset eagerly from this Iceberg table.

Returns:

Type	Description
Dataset	ray.data.dataset.Dataset: Materialized Ray Dataset from the Iceberg table

Source code in pyiceberg/table/__init__.py

def to_ray(self) -> ray.data.dataset.Dataset:
    """Read a Ray Dataset eagerly from this Iceberg table.

    Returns:
        ray.data.dataset.Dataset: Materialized Ray Dataset from the Iceberg table
    """
    import ray

    return ray.data.from_arrow(self.to_arrow())

update(**overrides)

Create a copy of this table scan with updated fields.

Source code in pyiceberg/table/__init__.py

def update(self: A, **overrides: Any) -> A:
    """Create a copy of this table scan with updated fields."""
    from inspect import signature

    # Extract those attributes that are constructor parameters. We don't use self.__dict__ as the kwargs to the
    # constructors because it may contain additional attributes that are not part of the constructor signature.
    params = signature(type(self).__init__).parameters.keys() - {"self"}  # Skip "self" parameter
    kwargs = {param: getattr(self, param) for param in params}  # Assume parameters are attributes

    return type(self)(**{**kwargs, **overrides})

CommitTableRequest

Bases: IcebergBaseModel

A pydantic BaseModel for a table commit request.

Source code in pyiceberg/table/__init__.py

class CommitTableRequest(IcebergBaseModel):
    """A pydantic BaseModel for a table commit request."""

    identifier: TableIdentifier = Field()
    requirements: tuple[TableRequirement, ...] = Field(default_factory=tuple)
    updates: tuple[TableUpdate, ...] = Field(default_factory=tuple)

CommitTableResponse

Bases: IcebergBaseModel

A pydantic BaseModel for a table commit response.

Source code in pyiceberg/table/__init__.py

class CommitTableResponse(IcebergBaseModel):
    """A pydantic BaseModel for a table commit response."""

    metadata: TableMetadata
    metadata_location: str = Field(alias="metadata-location")

CreateTableTransaction

Bases: Transaction

A transaction that involves the creation of a new table.

Source code in pyiceberg/table/__init__.py

class CreateTableTransaction(Transaction):
    """A transaction that involves the creation of a new table."""

    def _initial_changes(self, table_metadata: TableMetadata) -> None:
        """Set the initial changes that can reconstruct the initial table metadata when creating the CreateTableTransaction."""
        self._updates += (
            AssignUUIDUpdate(uuid=table_metadata.table_uuid),
            UpgradeFormatVersionUpdate(format_version=table_metadata.format_version),
        )

        schema: Schema = table_metadata.schema()
        self._updates += (
            AddSchemaUpdate(schema_=schema),
            SetCurrentSchemaUpdate(schema_id=-1),
        )

        spec: PartitionSpec = table_metadata.spec()
        if spec.is_unpartitioned():
            self._updates += (AddPartitionSpecUpdate(spec=UNPARTITIONED_PARTITION_SPEC),)
        else:
            self._updates += (AddPartitionSpecUpdate(spec=spec),)
        self._updates += (SetDefaultSpecUpdate(spec_id=-1),)

        sort_order: SortOrder | None = table_metadata.sort_order_by_id(table_metadata.default_sort_order_id)
        if sort_order is None or sort_order.is_unsorted:
            self._updates += (AddSortOrderUpdate(sort_order=UNSORTED_SORT_ORDER),)
        else:
            self._updates += (AddSortOrderUpdate(sort_order=sort_order),)
        self._updates += (SetDefaultSortOrderUpdate(sort_order_id=-1),)

        self._updates += (
            SetLocationUpdate(location=table_metadata.location),
            SetPropertiesUpdate(updates=table_metadata.properties),
        )

    def __init__(self, table: StagedTable):
        super().__init__(table, autocommit=False)
        self._initial_changes(table.metadata)

    def commit_transaction(self) -> Table:
        """Commit the changes to the catalog.

        In the case of a CreateTableTransaction, the only requirement is AssertCreate.
        Returns:
            The table with the updates applied.
        """
        if len(self._updates) > 0:
            self._table._do_commit(  # pylint: disable=W0212
                updates=self._updates,
                requirements=(AssertCreate(),),
            )

        self._updates = ()
        self._requirements = ()

        return self._table

commit_transaction()

Commit the changes to the catalog.

In the case of a CreateTableTransaction, the only requirement is AssertCreate. Returns: The table with the updates applied.

Source code in pyiceberg/table/__init__.py

def commit_transaction(self) -> Table:
    """Commit the changes to the catalog.

    In the case of a CreateTableTransaction, the only requirement is AssertCreate.
    Returns:
        The table with the updates applied.
    """
    if len(self._updates) > 0:
        self._table._do_commit(  # pylint: disable=W0212
            updates=self._updates,
            requirements=(AssertCreate(),),
        )

    self._updates = ()
    self._requirements = ()

    return self._table

DataScan

Bases: TableScan

Source code in pyiceberg/table/__init__.py

class DataScan(TableScan):
    @cached_property
    def _manifest_planner(self) -> ManifestGroupPlanner:
        return ManifestGroupPlanner(
            table_metadata=self.table_metadata,
            io=self.io,
            row_filter=self.row_filter,
            case_sensitive=self.case_sensitive,
            options=self.options,
        )

    @cached_property
    def partition_filters(self) -> KeyDefaultDict[int, BooleanExpression]:
        return self._manifest_planner.partition_filters

    def _plan_manifest_entries(self) -> Iterator[list[ManifestEntry]]:
        """Filter and return manifest entries based on partition and metrics evaluators.

        Returns:
            Iterator of ManifestEntry objects that match the scan's partition filter.
        """
        snapshot = self.snapshot()
        if not snapshot:
            return iter([])

        return self._manifest_planner.plan_manifest_entries(snapshot.manifests(self.io))

    def _should_use_server_side_planning(self) -> bool:
        """Check if server-side scan planning should be used for this scan."""
        if not self.catalog:
            return False
        return self.catalog.supports_server_side_planning(self.table_config)

    def _plan_files_server_side(self) -> Iterable[FileScanTask]:
        """Plan files using REST server-side scan planning."""
        from pyiceberg.catalog.rest import RestCatalog
        from pyiceberg.catalog.rest.scan_planning import PlanTableScanRequest

        if not isinstance(self.catalog, RestCatalog):
            raise TypeError("REST scan planning requires a RestCatalog")
        if self.table_identifier is None:
            raise ValueError("REST scan planning requires a table identifier")

        request = PlanTableScanRequest(
            snapshot_id=self.snapshot_id,
            select=list(self.selected_fields) if self.selected_fields != ("*",) else None,
            filter=self.row_filter if self.row_filter != ALWAYS_TRUE else None,
            case_sensitive=self.case_sensitive,
        )

        result = self.catalog._plan_scan_result(self.table_identifier, request)
        location = result.tasks[0].file.file_path if result.tasks else None
        plan_io = self.catalog._file_io_from_plan(self.io.properties, result.storage_credentials, location)
        if plan_io is not None:
            self.io = plan_io
        return result.tasks

    def _plan_files_local(self) -> Iterable[FileScanTask]:
        """Plan files locally by reading manifests."""
        snapshot = self.snapshot()
        if not snapshot:
            return []
        return self._manifest_planner.plan_files(snapshot.manifests(self.io))

    def plan_files(self) -> Iterable[FileScanTask]:
        """Plans the relevant files by filtering on the PartitionSpecs.

        If the table comes from a REST catalog with scan planning enabled,
        this will use server-side scan planning. Otherwise, it falls back
        to local planning.

        Returns:
            List of FileScanTasks that contain both data and delete files.
        """
        if self._should_use_server_side_planning():
            return self._plan_files_server_side()
        return self._plan_files_local()

    def to_arrow(self, dictionary_columns: tuple[str, ...] = ()) -> pa.Table:
        """Read an Arrow table eagerly from this DataScan.

        All rows will be loaded into memory at once.

        Args:
            dictionary_columns:
                A tuple of column names that PyArrow should read as
                dictionary-encoded (``pa.DictionaryArray``).  Dictionary
                encoding can substantially reduce memory usage for columns
                with low-cardinality repeated string values.
                Only applies to Parquet files; silently ignored for ORC.

        Returns:
            pa.Table: Materialized Arrow Table from the Iceberg table's DataScan
        """
        return _to_arrow_via_file_scan_tasks(self, self.projection(), self.plan_files(), dictionary_columns=dictionary_columns)

    def to_arrow_batch_reader(self, dictionary_columns: tuple[str, ...] = ()) -> pa.RecordBatchReader:
        """Return an Arrow RecordBatchReader from this DataScan.

        For large results, using a RecordBatchReader requires less memory than
        loading an Arrow Table for the same DataScan, because a RecordBatch
        is read one at a time.

        Args:
            dictionary_columns:
                A tuple of column names that PyArrow should read as
                dictionary-encoded (``pa.DictionaryArray``).  Dictionary
                encoding can substantially reduce memory usage for columns
                with low-cardinality repeated string values.
                Only applies to Parquet files; silently ignored for ORC.

        Returns:
            pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's DataScan
                which can be used to read a stream of record batches one by one.
        """
        return _to_arrow_batch_reader_via_file_scan_tasks(
            self, self.projection(), self.plan_files(), dictionary_columns=dictionary_columns
        )

    def count(self) -> int:
        from pyiceberg.io.pyarrow import ArrowScan

        # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
        res = 0
        # every task is a FileScanTask
        tasks = self.plan_files()

        for task in tasks:
            # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
            # partition value and task.delete_files represents that positional delete haven't been merged yet
            # hence those files have to read as a pyarrow table applying the filter and deletes
            if task.residual == AlwaysTrue() and len(task.delete_files) == 0:
                # Every File has a metadata stat that stores the file record count
                res += task.file.record_count
            else:
                arrow_scan = ArrowScan(
                    table_metadata=self.table_metadata,
                    io=self.io,
                    projected_schema=self.projection(),
                    row_filter=self.row_filter,
                    case_sensitive=self.case_sensitive,
                )
                tbl = arrow_scan.to_table([task])
                res += len(tbl)
        return res

plan_files()

Plans the relevant files by filtering on the PartitionSpecs.

If the table comes from a REST catalog with scan planning enabled, this will use server-side scan planning. Otherwise, it falls back to local planning.

Returns:

Type	Description
Iterable[FileScanTask]	List of FileScanTasks that contain both data and delete files.

Source code in pyiceberg/table/__init__.py

def plan_files(self) -> Iterable[FileScanTask]:
    """Plans the relevant files by filtering on the PartitionSpecs.

    If the table comes from a REST catalog with scan planning enabled,
    this will use server-side scan planning. Otherwise, it falls back
    to local planning.

    Returns:
        List of FileScanTasks that contain both data and delete files.
    """
    if self._should_use_server_side_planning():
        return self._plan_files_server_side()
    return self._plan_files_local()

to_arrow(dictionary_columns=())

Read an Arrow table eagerly from this DataScan.

All rows will be loaded into memory at once.

Parameters:

Name	Type	Description	Default
dictionary_columns	tuple[str, ...]	A tuple of column names that PyArrow should read as dictionary-encoded (pa.DictionaryArray). Dictionary encoding can substantially reduce memory usage for columns with low-cardinality repeated string values. Only applies to Parquet files; silently ignored for ORC.	()

Returns:

Type	Description
Table	pa.Table: Materialized Arrow Table from the Iceberg table's DataScan

Source code in pyiceberg/table/__init__.py

def to_arrow(self, dictionary_columns: tuple[str, ...] = ()) -> pa.Table:
    """Read an Arrow table eagerly from this DataScan.

    All rows will be loaded into memory at once.

    Args:
        dictionary_columns:
            A tuple of column names that PyArrow should read as
            dictionary-encoded (``pa.DictionaryArray``).  Dictionary
            encoding can substantially reduce memory usage for columns
            with low-cardinality repeated string values.
            Only applies to Parquet files; silently ignored for ORC.

    Returns:
        pa.Table: Materialized Arrow Table from the Iceberg table's DataScan
    """
    return _to_arrow_via_file_scan_tasks(self, self.projection(), self.plan_files(), dictionary_columns=dictionary_columns)

to_arrow_batch_reader(dictionary_columns=())

Return an Arrow RecordBatchReader from this DataScan.

For large results, using a RecordBatchReader requires less memory than loading an Arrow Table for the same DataScan, because a RecordBatch is read one at a time.

Parameters:

Name	Type	Description	Default
dictionary_columns	tuple[str, ...]	A tuple of column names that PyArrow should read as dictionary-encoded (pa.DictionaryArray). Dictionary encoding can substantially reduce memory usage for columns with low-cardinality repeated string values. Only applies to Parquet files; silently ignored for ORC.	()

Returns:

Type	Description
RecordBatchReader	pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's DataScan which can be used to read a stream of record batches one by one.

Source code in pyiceberg/table/__init__.py

def to_arrow_batch_reader(self, dictionary_columns: tuple[str, ...] = ()) -> pa.RecordBatchReader:
    """Return an Arrow RecordBatchReader from this DataScan.

    For large results, using a RecordBatchReader requires less memory than
    loading an Arrow Table for the same DataScan, because a RecordBatch
    is read one at a time.

    Args:
        dictionary_columns:
            A tuple of column names that PyArrow should read as
            dictionary-encoded (``pa.DictionaryArray``).  Dictionary
            encoding can substantially reduce memory usage for columns
            with low-cardinality repeated string values.
            Only applies to Parquet files; silently ignored for ORC.

    Returns:
        pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's DataScan
            which can be used to read a stream of record batches one by one.
    """
    return _to_arrow_batch_reader_via_file_scan_tasks(
        self, self.projection(), self.plan_files(), dictionary_columns=dictionary_columns
    )

FileScanTask dataclass

Bases: ScanTask

Task representing a data file and its corresponding delete files.

Source code in pyiceberg/table/__init__.py

@dataclass(init=False)
class FileScanTask(ScanTask):
    """Task representing a data file and its corresponding delete files."""

    file: DataFile
    delete_files: set[DataFile]
    residual: BooleanExpression

    def __init__(
        self,
        data_file: DataFile,
        delete_files: set[DataFile] | None = None,
        residual: BooleanExpression = ALWAYS_TRUE,
    ) -> None:
        self.file = data_file
        self.delete_files = delete_files or set()
        self.residual = residual

    @staticmethod
    def from_rest_response(
        rest_task: RESTFileScanTask,
        delete_files: list[RESTDeleteFile],
    ) -> FileScanTask:
        """Convert a RESTFileScanTask to a FileScanTask.

        Args:
            rest_task: The REST file scan task.
            delete_files: The list of delete files from the ScanTasks response.

        Returns:
            A FileScanTask with the converted data and delete files.

        Raises:
            NotImplementedError: If equality delete files are encountered.
        """
        from pyiceberg.catalog.rest.scan_planning import RESTEqualityDeleteFile

        data_file = _rest_file_to_data_file(rest_task.data_file)

        resolved_deletes: set[DataFile] = set()
        if rest_task.delete_file_references:
            for idx in rest_task.delete_file_references:
                delete_file = delete_files[idx]
                if isinstance(delete_file, RESTEqualityDeleteFile):
                    raise NotImplementedError(f"PyIceberg does not yet support equality deletes: {delete_file.file_path}")
                resolved_deletes.add(_rest_file_to_data_file(delete_file))

        return FileScanTask(
            data_file=data_file,
            delete_files=resolved_deletes,
            residual=rest_task.residual_filter if rest_task.residual_filter else ALWAYS_TRUE,
        )

from_rest_response(rest_task, delete_files) staticmethod

Convert a RESTFileScanTask to a FileScanTask.

Parameters:

Name	Type	Description	Default
rest_task	RESTFileScanTask	The REST file scan task.	required
delete_files	list[RESTDeleteFile]	The list of delete files from the ScanTasks response.	required

Returns:

Type	Description
FileScanTask	A FileScanTask with the converted data and delete files.

Raises:

Type	Description
NotImplementedError	If equality delete files are encountered.

Source code in pyiceberg/table/__init__.py

@staticmethod
def from_rest_response(
    rest_task: RESTFileScanTask,
    delete_files: list[RESTDeleteFile],
) -> FileScanTask:
    """Convert a RESTFileScanTask to a FileScanTask.

    Args:
        rest_task: The REST file scan task.
        delete_files: The list of delete files from the ScanTasks response.

    Returns:
        A FileScanTask with the converted data and delete files.

    Raises:
        NotImplementedError: If equality delete files are encountered.
    """
    from pyiceberg.catalog.rest.scan_planning import RESTEqualityDeleteFile

    data_file = _rest_file_to_data_file(rest_task.data_file)

    resolved_deletes: set[DataFile] = set()
    if rest_task.delete_file_references:
        for idx in rest_task.delete_file_references:
            delete_file = delete_files[idx]
            if isinstance(delete_file, RESTEqualityDeleteFile):
                raise NotImplementedError(f"PyIceberg does not yet support equality deletes: {delete_file.file_path}")
            resolved_deletes.add(_rest_file_to_data_file(delete_file))

    return FileScanTask(
        data_file=data_file,
        delete_files=resolved_deletes,
        residual=rest_task.residual_filter if rest_task.residual_filter else ALWAYS_TRUE,
    )

IncrementalAppendScan

Bases: BaseScan

An incremental scan of a table's data that accumulates appended data between two snapshots.

Parameters:

Name	Type	Description	Default
from_snapshot_id	int | None	ID of the snapshot to start the incremental scan from. If None, the scan starts from the oldest ancestor of the "to" snapshot (inclusive).	None
from_snapshot_inclusive	bool	Whether from_snapshot_id is included in the scan. If False, the start snapshot is exclusive.	False
to_snapshot_id	int | None	Optional ID of the snapshot to end the incremental scan at, inclusively. Omitting it will default to the table's current snapshot.	None
row_filter	str | BooleanExpression	A string or BooleanExpression that describes the desired rows	ALWAYS_TRUE
selected_fields	tuple[str, ...]	A tuple of strings representing the column names to return in the output dataframe.	('*',)
case_sensitive	bool	If True column matching is case sensitive	True
options	Properties	Additional Table properties as a dictionary of string key value pairs to use for this scan.	EMPTY_DICT
limit	int | None	An integer representing the number of rows to return in the scan result. If None, fetches all matching rows.	None

Source code in pyiceberg/table/__init__.py

class IncrementalAppendScan(BaseScan):
    """An incremental scan of a table's data that accumulates appended data between two snapshots.

    Args:
        from_snapshot_id:
            ID of the snapshot to start the incremental scan from. If None, the scan starts from
            the oldest ancestor of the "to" snapshot (inclusive).
        from_snapshot_inclusive:
            Whether from_snapshot_id is included in the scan. If False, the start snapshot is
            exclusive.
        to_snapshot_id:
            Optional ID of the snapshot to end the incremental scan at, inclusively.
            Omitting it will default to the table's current snapshot.
        row_filter:
            A string or BooleanExpression that describes the
            desired rows
        selected_fields:
            A tuple of strings representing the column names
            to return in the output dataframe.
        case_sensitive:
            If True column matching is case sensitive
        options:
            Additional Table properties as a dictionary of
            string key value pairs to use for this scan.
        limit:
            An integer representing the number of rows to
            return in the scan result. If None, fetches all
            matching rows.
    """

    from_snapshot_id: int | None
    from_snapshot_inclusive: bool
    to_snapshot_id: int | None

    def __init__(
        self,
        table_metadata: TableMetadata,
        io: FileIO,
        row_filter: str | BooleanExpression = ALWAYS_TRUE,
        selected_fields: tuple[str, ...] = ("*",),
        case_sensitive: bool = True,
        from_snapshot_id: int | None = None,
        from_snapshot_inclusive: bool = False,
        to_snapshot_id: int | None = None,
        options: Properties = EMPTY_DICT,
        limit: int | None = None,
    ):
        super().__init__(
            table_metadata=table_metadata,
            io=io,
            row_filter=row_filter,
            selected_fields=selected_fields,
            case_sensitive=case_sensitive,
            options=options,
            limit=limit,
        )
        self.from_snapshot_id = from_snapshot_id
        self.from_snapshot_inclusive = from_snapshot_inclusive
        self.to_snapshot_id = to_snapshot_id

    def from_snapshot_id_exclusive(self: IAS, from_snapshot_id: int) -> IAS:
        """Return a copy of this scan that starts (exclusively) from the given snapshot ID."""
        return self.update(from_snapshot_id=from_snapshot_id, from_snapshot_inclusive=False)

    def from_snapshot_id_inclusive(self: IAS, from_snapshot_id: int) -> IAS:
        """Return a copy of this scan that starts (inclusively) from the given snapshot ID."""
        return self.update(from_snapshot_id=from_snapshot_id, from_snapshot_inclusive=True)

    def to_snapshot_id_inclusive(self: IAS, to_snapshot_id: int) -> IAS:
        """Return a copy of this scan that ends (inclusively) at the given snapshot ID."""
        return self.update(to_snapshot_id=to_snapshot_id)

    def projection(self) -> Schema:
        current_schema = self.table_metadata.schema()
        if "*" in self.selected_fields:
            return current_schema
        return current_schema.select(*self.selected_fields, case_sensitive=self.case_sensitive)

    def plan_files(self) -> Iterable[FileScanTask]:
        """Plans the relevant files added between the specified snapshots."""
        # With neither bound set, an empty table (no current snapshot) has nothing to scan.
        if self.from_snapshot_id is None and self.to_snapshot_id is None and self.table_metadata.current_snapshot() is None:
            return []

        from_snapshot_id_exclusive, to_snapshot_id = self._validate_and_resolve_snapshots()

        append_snapshots = [
            snapshot
            for snapshot in ancestors_between_ids(
                from_snapshot_id_exclusive=from_snapshot_id_exclusive,
                to_snapshot_id_inclusive=to_snapshot_id,
                table_metadata=self.table_metadata,
            )
            if snapshot.summary is not None and snapshot.summary.operation == Operation.APPEND
        ]
        if len(append_snapshots) == 0:
            return []

        append_snapshot_ids = {snapshot.snapshot_id for snapshot in append_snapshots}

        manifests = list(
            {
                manifest_file
                for snapshot in append_snapshots
                for manifest_file in snapshot.manifests(self.io)
                if manifest_file.content == ManifestContent.DATA and manifest_file.added_snapshot_id in append_snapshot_ids
            }
        )

        return ManifestGroupPlanner(
            table_metadata=self.table_metadata,
            io=self.io,
            row_filter=self.row_filter,
            case_sensitive=self.case_sensitive,
            options=self.options,
        ).plan_files(
            manifests=manifests,
            manifest_entry_filter=lambda manifest_entry: manifest_entry.snapshot_id in append_snapshot_ids
            and manifest_entry.status == ManifestEntryStatus.ADDED,
        )

    def to_arrow(self) -> pa.Table:
        """Read an Arrow table eagerly from this IncrementalAppendScan.

        All rows will be loaded into memory at once.

        Returns:
            pa.Table: Materialized Arrow Table from the Iceberg table's IncrementalAppendScan
        """
        return _to_arrow_via_file_scan_tasks(self, self.projection(), self.plan_files())

    def to_arrow_batch_reader(self) -> pa.RecordBatchReader:
        """Return an Arrow RecordBatchReader from this IncrementalAppendScan.

        For large results, using a RecordBatchReader requires less memory than
        loading an Arrow Table for the same IncrementalAppendScan, because a
        RecordBatch is read one at a time.

        Returns:
            pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's IncrementalAppendScan
                which can be used to read a stream of record batches one by one.
        """
        return _to_arrow_batch_reader_via_file_scan_tasks(self, self.projection(), self.plan_files())

    def _validate_and_resolve_snapshots(self) -> tuple[int | None, int]:
        """Resolve the configured range to ``(from_snapshot_id_exclusive, to_snapshot_id_inclusive)``.

        A ``None`` "from" means the scan starts from the oldest ancestor of the end snapshot.
        """
        # Resolve the inclusive end snapshot, defaulting to the table's current snapshot.
        if self.to_snapshot_id is not None:
            if self.table_metadata.snapshot_by_id(self.to_snapshot_id) is None:
                raise ValueError(f"End snapshot not found in table metadata: {self.to_snapshot_id}")
            to_snapshot_id = self.to_snapshot_id
        else:
            current_snapshot = self.table_metadata.current_snapshot()
            if current_snapshot is None:
                raise ValueError("End snapshot is not set and table has no current snapshot")
            to_snapshot_id = current_snapshot.snapshot_id

        # An unset start scans the whole lineage of the end snapshot (from its oldest ancestor).
        if self.from_snapshot_id is None:
            return None, to_snapshot_id

        if self.from_snapshot_inclusive:
            # An inclusive start must be present (its parent becomes the exclusive boundary, and may
            # be None when the start is the root) and an ancestor of the end snapshot.
            from_snapshot = self.table_metadata.snapshot_by_id(self.from_snapshot_id)
            if from_snapshot is None:
                raise ValueError(f"Start snapshot (inclusive) not found in table metadata: {self.from_snapshot_id}")
            if not is_ancestor_of(to_snapshot_id, self.from_snapshot_id, self.table_metadata):
                raise ValueError(
                    f"Starting snapshot (inclusive) {self.from_snapshot_id} is not an ancestor of end snapshot {to_snapshot_id}"
                )
            return from_snapshot.parent_snapshot_id, to_snapshot_id

        # An exclusive start does not need to be present in the table metadata (it may have been
        # expired). It is valid as long as it is the parent of some ancestor of the end snapshot.
        if not is_parent_ancestor_of(to_snapshot_id, self.from_snapshot_id, self.table_metadata):
            raise ValueError(
                f"Starting snapshot (exclusive) {self.from_snapshot_id} is not a parent ancestor of end snapshot {to_snapshot_id}"
            )
        return self.from_snapshot_id, to_snapshot_id

from_snapshot_id_exclusive(from_snapshot_id)

Return a copy of this scan that starts (exclusively) from the given snapshot ID.

Source code in pyiceberg/table/__init__.py

def from_snapshot_id_exclusive(self: IAS, from_snapshot_id: int) -> IAS:
    """Return a copy of this scan that starts (exclusively) from the given snapshot ID."""
    return self.update(from_snapshot_id=from_snapshot_id, from_snapshot_inclusive=False)

from_snapshot_id_inclusive(from_snapshot_id)

Return a copy of this scan that starts (inclusively) from the given snapshot ID.

Source code in pyiceberg/table/__init__.py

def from_snapshot_id_inclusive(self: IAS, from_snapshot_id: int) -> IAS:
    """Return a copy of this scan that starts (inclusively) from the given snapshot ID."""
    return self.update(from_snapshot_id=from_snapshot_id, from_snapshot_inclusive=True)

plan_files()

Plans the relevant files added between the specified snapshots.

Source code in pyiceberg/table/__init__.py

def plan_files(self) -> Iterable[FileScanTask]:
    """Plans the relevant files added between the specified snapshots."""
    # With neither bound set, an empty table (no current snapshot) has nothing to scan.
    if self.from_snapshot_id is None and self.to_snapshot_id is None and self.table_metadata.current_snapshot() is None:
        return []

    from_snapshot_id_exclusive, to_snapshot_id = self._validate_and_resolve_snapshots()

    append_snapshots = [
        snapshot
        for snapshot in ancestors_between_ids(
            from_snapshot_id_exclusive=from_snapshot_id_exclusive,
            to_snapshot_id_inclusive=to_snapshot_id,
            table_metadata=self.table_metadata,
        )
        if snapshot.summary is not None and snapshot.summary.operation == Operation.APPEND
    ]
    if len(append_snapshots) == 0:
        return []

    append_snapshot_ids = {snapshot.snapshot_id for snapshot in append_snapshots}

    manifests = list(
        {
            manifest_file
            for snapshot in append_snapshots
            for manifest_file in snapshot.manifests(self.io)
            if manifest_file.content == ManifestContent.DATA and manifest_file.added_snapshot_id in append_snapshot_ids
        }
    )

    return ManifestGroupPlanner(
        table_metadata=self.table_metadata,
        io=self.io,
        row_filter=self.row_filter,
        case_sensitive=self.case_sensitive,
        options=self.options,
    ).plan_files(
        manifests=manifests,
        manifest_entry_filter=lambda manifest_entry: manifest_entry.snapshot_id in append_snapshot_ids
        and manifest_entry.status == ManifestEntryStatus.ADDED,
    )

to_arrow()

Read an Arrow table eagerly from this IncrementalAppendScan.

All rows will be loaded into memory at once.

Returns:

Type	Description
Table	pa.Table: Materialized Arrow Table from the Iceberg table's IncrementalAppendScan

Source code in pyiceberg/table/__init__.py

def to_arrow(self) -> pa.Table:
    """Read an Arrow table eagerly from this IncrementalAppendScan.

    All rows will be loaded into memory at once.

    Returns:
        pa.Table: Materialized Arrow Table from the Iceberg table's IncrementalAppendScan
    """
    return _to_arrow_via_file_scan_tasks(self, self.projection(), self.plan_files())

to_arrow_batch_reader()

Return an Arrow RecordBatchReader from this IncrementalAppendScan.

For large results, using a RecordBatchReader requires less memory than loading an Arrow Table for the same IncrementalAppendScan, because a RecordBatch is read one at a time.

Returns:

Type	Description
RecordBatchReader	pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's IncrementalAppendScan which can be used to read a stream of record batches one by one.

Source code in pyiceberg/table/__init__.py

def to_arrow_batch_reader(self) -> pa.RecordBatchReader:
    """Return an Arrow RecordBatchReader from this IncrementalAppendScan.

    For large results, using a RecordBatchReader requires less memory than
    loading an Arrow Table for the same IncrementalAppendScan, because a
    RecordBatch is read one at a time.

    Returns:
        pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's IncrementalAppendScan
            which can be used to read a stream of record batches one by one.
    """
    return _to_arrow_batch_reader_via_file_scan_tasks(self, self.projection(), self.plan_files())

to_snapshot_id_inclusive(to_snapshot_id)

Return a copy of this scan that ends (inclusively) at the given snapshot ID.

Source code in pyiceberg/table/__init__.py

def to_snapshot_id_inclusive(self: IAS, to_snapshot_id: int) -> IAS:
    """Return a copy of this scan that ends (inclusively) at the given snapshot ID."""
    return self.update(to_snapshot_id=to_snapshot_id)

ManifestGroupPlanner

Plans the scan tasks for a group of manifests.

Source code in pyiceberg/table/__init__.py

class ManifestGroupPlanner:
    """Plans the scan tasks for a group of manifests."""

    table_metadata: TableMetadata
    io: FileIO
    row_filter: BooleanExpression
    case_sensitive: bool
    options: Properties

    def __init__(
        self,
        table_metadata: TableMetadata,
        io: FileIO,
        row_filter: str | BooleanExpression = ALWAYS_TRUE,
        case_sensitive: bool = True,
        options: Properties = EMPTY_DICT,
    ):
        self.table_metadata = table_metadata
        self.io = io
        self.row_filter = _parse_row_filter(row_filter)
        self.case_sensitive = case_sensitive
        self.options = options

    @cached_property
    def partition_filters(self) -> KeyDefaultDict[int, BooleanExpression]:
        return KeyDefaultDict(self._build_partition_projection)

    def plan_manifest_entries(self, manifests: Iterable[ManifestFile]) -> Iterator[list[ManifestEntry]]:
        """Filter the given manifests using partition summaries and read the matching manifest entries.

        For each manifest that passes the partition-summary filter, returns a list of its
        manifest entries that match the partition and metrics evaluators. The returned iterator
        yields one list per manifest (in parallel).
        """
        # step 1: filter manifests using partition summaries
        # the filter depends on the partition spec used to write the manifest file, so create a cache of filters for each spec id

        manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)

        manifests = [
            manifest_file for manifest_file in manifests if manifest_evaluators[manifest_file.partition_spec_id](manifest_file)
        ]

        # step 2: filter the data files in each manifest
        # this filter depends on the partition spec used to write the manifest file

        partition_evaluators: dict[int, Callable[[DataFile], bool]] = KeyDefaultDict(self._build_partition_evaluator)
        metrics_evaluator = self._build_metrics_evaluator()

        min_sequence_number = _min_sequence_number(manifests)

        executor = ExecutorFactory.get_or_create()
        return executor.map(
            lambda args: _open_manifest(*args),
            [
                (
                    self.io,
                    manifest,
                    partition_evaluators[manifest.partition_spec_id],
                    metrics_evaluator,
                )
                for manifest in manifests
                if self._check_sequence_number(min_sequence_number, manifest)
            ],
        )

    def plan_files(
        self,
        manifests: Iterable[ManifestFile],
        manifest_entry_filter: Callable[[ManifestEntry], bool] = lambda _: True,
    ) -> Iterable[FileScanTask]:
        """Plan the file scan tasks for the given manifests.

        ``manifest_entry_filter`` is an additional predicate applied after the partition
        evaluator; entries for which it returns False are excluded from the result.

        Returns:
            List of FileScanTasks that contain both data and delete files.
        """
        data_entries: list[ManifestEntry] = []
        delete_index = DeleteFileIndex()

        residual_evaluators: dict[int, Callable[[DataFile], ResidualEvaluator]] = KeyDefaultDict(self._build_residual_evaluator)

        for manifest_entry in chain.from_iterable(self.plan_manifest_entries(manifests)):
            if not manifest_entry_filter(manifest_entry):
                continue

            data_file = manifest_entry.data_file
            if data_file.content == DataFileContent.DATA:
                data_entries.append(manifest_entry)
            elif data_file.content == DataFileContent.POSITION_DELETES:
                delete_index.add_delete_file(manifest_entry, partition_key=data_file.partition)
            elif data_file.content == DataFileContent.EQUALITY_DELETES:
                raise ValueError("PyIceberg does not yet support equality deletes: https://github.com/apache/iceberg/issues/6568")
            else:
                raise ValueError(f"Unknown DataFileContent ({data_file.content}): {manifest_entry}")

        return [
            FileScanTask(
                data_entry.data_file,
                delete_files=delete_index.for_data_file(
                    data_entry.sequence_number or INITIAL_SEQUENCE_NUMBER,
                    data_entry.data_file,
                    partition_key=data_entry.data_file.partition,
                ),
                residual=residual_evaluators[data_entry.data_file.spec_id](data_entry.data_file).residual_for(
                    data_entry.data_file.partition
                ),
            )
            for data_entry in data_entries
        ]

    def _build_partition_projection(self, spec_id: int) -> BooleanExpression:
        project = inclusive_projection(self.table_metadata.schema(), self.table_metadata.specs()[spec_id], self.case_sensitive)
        return project(self.row_filter)

    def _build_manifest_evaluator(self, spec_id: int) -> Callable[[ManifestFile], bool]:
        spec = self.table_metadata.specs()[spec_id]
        return manifest_evaluator(spec, self.table_metadata.schema(), self.partition_filters[spec_id], self.case_sensitive)

    def _build_partition_evaluator(self, spec_id: int) -> Callable[[DataFile], bool]:
        spec = self.table_metadata.specs()[spec_id]
        partition_type = spec.partition_type(self.table_metadata.schema())
        partition_schema = Schema(*partition_type.fields)
        partition_expr = self.partition_filters[spec_id]
        evaluator = expression_evaluator(partition_schema, partition_expr, self.case_sensitive)

        # Expression evaluators keep input-specific state local to each call, so the
        # prepared evaluator can be shared by every manifest using this spec.
        return lambda data_file: evaluator(data_file.partition)

    def _build_metrics_evaluator(self) -> Callable[[DataFile], bool]:
        schema = self.table_metadata.schema()
        include_empty_files = strtobool(self.options.get("include_empty_files", "false"))

        # Metrics evaluators keep file-specific state local to each call, so one
        # prepared evaluator can be shared across all manifest tasks in this plan.
        return _InclusiveMetricsEvaluator(
            schema,
            self.row_filter,
            self.case_sensitive,
            include_empty_files,
        ).eval

    def _build_residual_evaluator(self, spec_id: int) -> Callable[[DataFile], ResidualEvaluator]:
        spec = self.table_metadata.specs()[spec_id]

        from pyiceberg.expressions.visitors import residual_evaluator_of

        # The lambda created here is run in multiple threads.
        # So we avoid creating _EvaluatorExpression methods bound to a single
        # shared instance across multiple threads.
        return lambda datafile: residual_evaluator_of(
            spec=spec,
            expr=self.row_filter,
            case_sensitive=self.case_sensitive,
            schema=self.table_metadata.schema(),
        )

    @staticmethod
    def _check_sequence_number(min_sequence_number: int, manifest: ManifestFile) -> bool:
        """Ensure that no manifests are loaded that contain deletes that are older than the data.

        Args:
            min_sequence_number (int): The minimal sequence number.
            manifest (ManifestFile): A ManifestFile that can be either data or deletes.

        Returns:
            Boolean indicating if it is either a data file, or a relevant delete file.
        """
        return manifest.content == ManifestContent.DATA or (
            # Not interested in deletes that are older than the data
            manifest.content == ManifestContent.DELETES
            and (manifest.sequence_number or INITIAL_SEQUENCE_NUMBER) >= min_sequence_number
        )

plan_files(manifests, manifest_entry_filter=lambda _: True)

Plan the file scan tasks for the given manifests.

manifest_entry_filter is an additional predicate applied after the partition evaluator; entries for which it returns False are excluded from the result.

Returns:

Type	Description
Iterable[FileScanTask]	List of FileScanTasks that contain both data and delete files.

Source code in pyiceberg/table/__init__.py

def plan_files(
    self,
    manifests: Iterable[ManifestFile],
    manifest_entry_filter: Callable[[ManifestEntry], bool] = lambda _: True,
) -> Iterable[FileScanTask]:
    """Plan the file scan tasks for the given manifests.

    ``manifest_entry_filter`` is an additional predicate applied after the partition
    evaluator; entries for which it returns False are excluded from the result.

    Returns:
        List of FileScanTasks that contain both data and delete files.
    """
    data_entries: list[ManifestEntry] = []
    delete_index = DeleteFileIndex()

    residual_evaluators: dict[int, Callable[[DataFile], ResidualEvaluator]] = KeyDefaultDict(self._build_residual_evaluator)

    for manifest_entry in chain.from_iterable(self.plan_manifest_entries(manifests)):
        if not manifest_entry_filter(manifest_entry):
            continue

        data_file = manifest_entry.data_file
        if data_file.content == DataFileContent.DATA:
            data_entries.append(manifest_entry)
        elif data_file.content == DataFileContent.POSITION_DELETES:
            delete_index.add_delete_file(manifest_entry, partition_key=data_file.partition)
        elif data_file.content == DataFileContent.EQUALITY_DELETES:
            raise ValueError("PyIceberg does not yet support equality deletes: https://github.com/apache/iceberg/issues/6568")
        else:
            raise ValueError(f"Unknown DataFileContent ({data_file.content}): {manifest_entry}")

    return [
        FileScanTask(
            data_entry.data_file,
            delete_files=delete_index.for_data_file(
                data_entry.sequence_number or INITIAL_SEQUENCE_NUMBER,
                data_entry.data_file,
                partition_key=data_entry.data_file.partition,
            ),
            residual=residual_evaluators[data_entry.data_file.spec_id](data_entry.data_file).residual_for(
                data_entry.data_file.partition
            ),
        )
        for data_entry in data_entries
    ]

plan_manifest_entries(manifests)

Filter the given manifests using partition summaries and read the matching manifest entries.

For each manifest that passes the partition-summary filter, returns a list of its manifest entries that match the partition and metrics evaluators. The returned iterator yields one list per manifest (in parallel).

Source code in pyiceberg/table/__init__.py

def plan_manifest_entries(self, manifests: Iterable[ManifestFile]) -> Iterator[list[ManifestEntry]]:
    """Filter the given manifests using partition summaries and read the matching manifest entries.

    For each manifest that passes the partition-summary filter, returns a list of its
    manifest entries that match the partition and metrics evaluators. The returned iterator
    yields one list per manifest (in parallel).
    """
    # step 1: filter manifests using partition summaries
    # the filter depends on the partition spec used to write the manifest file, so create a cache of filters for each spec id

    manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)

    manifests = [
        manifest_file for manifest_file in manifests if manifest_evaluators[manifest_file.partition_spec_id](manifest_file)
    ]

    # step 2: filter the data files in each manifest
    # this filter depends on the partition spec used to write the manifest file

    partition_evaluators: dict[int, Callable[[DataFile], bool]] = KeyDefaultDict(self._build_partition_evaluator)
    metrics_evaluator = self._build_metrics_evaluator()

    min_sequence_number = _min_sequence_number(manifests)

    executor = ExecutorFactory.get_or_create()
    return executor.map(
        lambda args: _open_manifest(*args),
        [
            (
                self.io,
                manifest,
                partition_evaluators[manifest.partition_spec_id],
                metrics_evaluator,
            )
            for manifest in manifests
            if self._check_sequence_number(min_sequence_number, manifest)
        ],
    )

Namespace

Bases: IcebergRootModel[list[str]]

Reference to one or more levels of a namespace.

Source code in pyiceberg/table/__init__.py

class Namespace(IcebergRootModel[list[str]]):
    """Reference to one or more levels of a namespace."""

    root: list[str] = Field(
        ...,
        description="Reference to one or more levels of a namespace",
    )

StaticTable

Bases: Table

Load a table directly from a metadata file (i.e., without using a catalog).

Source code in pyiceberg/table/__init__.py

class StaticTable(Table):
    """Load a table directly from a metadata file (i.e., without using a catalog)."""

    def refresh(self) -> Table:
        """Refresh the current table metadata."""
        raise NotImplementedError("To be implemented")

    @classmethod
    def _metadata_location_from_version_hint(cls, metadata_location: str, properties: Properties = EMPTY_DICT) -> str:
        version_hint_location = os.path.join(metadata_location, "metadata", "version-hint.text")
        io = load_file_io(properties=properties, location=version_hint_location)
        file = io.new_input(version_hint_location)

        with file.open() as stream:
            content = stream.read().decode("utf-8")

        if content.endswith(".metadata.json"):
            return os.path.join(metadata_location, "metadata", content)
        elif content.isnumeric():
            return os.path.join(metadata_location, "metadata", f"v{content}.metadata.json")
        else:
            return os.path.join(metadata_location, "metadata", f"{content}.metadata.json")

    @classmethod
    def from_metadata(cls, metadata_location: str, properties: Properties = EMPTY_DICT) -> StaticTable:
        if not metadata_location.endswith(".metadata.json"):
            metadata_location = StaticTable._metadata_location_from_version_hint(metadata_location, properties)

        io = load_file_io(properties=properties, location=metadata_location)
        file = io.new_input(metadata_location)

        from pyiceberg.serializers import FromInputFile

        metadata = FromInputFile.table_metadata(file)

        from pyiceberg.catalog.noop import NoopCatalog

        return cls(
            identifier=("static-table", metadata_location),
            metadata_location=metadata_location,
            metadata=metadata,
            io=load_file_io({**properties, **metadata.properties}, location=metadata_location),
            catalog=NoopCatalog("static-table"),
        )

refresh()

Refresh the current table metadata.

Source code in pyiceberg/table/__init__.py

def refresh(self) -> Table:
    """Refresh the current table metadata."""
    raise NotImplementedError("To be implemented")

Table

An Iceberg table.

Source code in pyiceberg/table/__init__.py

class Table:
    """An Iceberg table."""

    _identifier: Identifier = Field()
    metadata: TableMetadata
    metadata_location: str = Field()
    io: FileIO
    catalog: Catalog
    config: dict[str, str]

    def __init__(
        self,
        identifier: Identifier,
        metadata: TableMetadata,
        metadata_location: str,
        io: FileIO,
        catalog: Catalog,
        config: dict[str, str] = EMPTY_DICT,
    ) -> None:
        self._identifier = identifier
        self.metadata = metadata
        self.metadata_location = metadata_location
        self.io = io
        self.catalog = catalog
        self.config = config

    def transaction(self) -> Transaction:
        """Create a new transaction object to first stage the changes, and then commit them to the catalog.

        Returns:
            The transaction object
        """
        return Transaction(self)

    @property
    def inspect(self) -> InspectTable:
        """Return the InspectTable object to browse the table metadata.

        Returns:
            InspectTable object based on this Table.
        """
        return InspectTable(self)

    @property
    def maintenance(self) -> MaintenanceTable:
        """Return the MaintenanceTable object for maintenance.

        Returns:
            MaintenanceTable object based on this Table.
        """
        return MaintenanceTable(self)

    def refresh(self) -> Table:
        """Refresh the current table metadata.

        Returns:
            An updated instance of the same Iceberg table
        """
        fresh = self.catalog.load_table(self._identifier)
        self._check_uuid(self.metadata, fresh.metadata)
        self.metadata = fresh.metadata
        self.io = fresh.io
        self.metadata_location = fresh.metadata_location
        self.config = fresh.config
        return self

    def name(self) -> Identifier:
        """Return the identifier of this table.

        Returns:
            An Identifier tuple of the table name
        """
        return self._identifier

    def scan(
        self,
        row_filter: str | BooleanExpression = ALWAYS_TRUE,
        selected_fields: tuple[str, ...] = ("*",),
        case_sensitive: bool = True,
        snapshot_id: int | None = None,
        options: Properties = EMPTY_DICT,
        limit: int | None = None,
    ) -> DataScan:
        """Fetch a DataScan based on the table's current metadata.

            The data scan can be used to project the table's data
            that matches the provided row_filter onto the table's
            current schema.

        Args:
            row_filter:
                A string or BooleanExpression that describes the
                desired rows
            selected_fields:
                A tuple of strings representing the column names
                to return in the output dataframe.
            case_sensitive:
                If True column matching is case sensitive
            snapshot_id:
                Optional Snapshot ID to time travel to. If None,
                scans the table as of the current snapshot ID.
            options:
                Additional Table properties as a dictionary of
                string key value pairs to use for this scan.
            limit:
                An integer representing the number of rows to
                return in the scan result. If None, fetches all
                matching rows.

        Returns:
            A DataScan based on the table's current metadata.
        """
        return DataScan(
            table_metadata=self.metadata,
            io=self.io,
            row_filter=row_filter,
            selected_fields=selected_fields,
            case_sensitive=case_sensitive,
            snapshot_id=snapshot_id,
            options=options,
            limit=limit,
            catalog=self.catalog,
            table_identifier=self._identifier,
            table_config=self.config,
        )

    def incremental_append_scan(
        self,
        *,
        from_snapshot_id_exclusive: int | None = None,
        to_snapshot_id_inclusive: int | None = None,
        row_filter: str | BooleanExpression = ALWAYS_TRUE,
        selected_fields: tuple[str, ...] = ("*",),
        case_sensitive: bool = True,
        options: Properties = EMPTY_DICT,
        limit: int | None = None,
    ) -> IncrementalAppendScan:
        """Fetch an IncrementalAppendScan based on the table's current metadata.

        The incremental append scan returns the rows added by append snapshots in a snapshot
        range that match the provided row_filter, projected onto the table's current schema.

        Args:
            from_snapshot_id_exclusive:
                Optional ID of the snapshot to start the incremental scan from, exclusively. If not set, the scan
                starts from the oldest ancestor of the end snapshot (inclusive).
            to_snapshot_id_inclusive:
                Optional ID of the snapshot to end the incremental scan at, inclusively. If not set, it defaults to
                the table's current snapshot.
            row_filter:
                A string or BooleanExpression that describes the
                desired rows.
            selected_fields:
                A tuple of strings representing the column names
                to return in the output dataframe.
            case_sensitive:
                If True column matching is case sensitive.
            options:
                Additional Table properties as a dictionary of
                string key value pairs to use for this scan.
            limit:
                An integer representing the number of rows to
                return in the scan result. If None, fetches all
                matching rows.

        Returns:
            An IncrementalAppendScan based on the table's current metadata and provided parameters.
        """
        return IncrementalAppendScan(
            table_metadata=self.metadata,
            io=self.io,
            row_filter=row_filter,
            selected_fields=selected_fields,
            case_sensitive=case_sensitive,
            from_snapshot_id=from_snapshot_id_exclusive,
            from_snapshot_inclusive=False,
            to_snapshot_id=to_snapshot_id_inclusive,
            options=options,
            limit=limit,
        )

    @property
    def format_version(self) -> TableVersion:
        return self.metadata.format_version

    def schema(self) -> Schema:
        """Return the schema for this table."""
        return next(schema for schema in self.metadata.schemas if schema.schema_id == self.metadata.current_schema_id)

    def schemas(self) -> dict[int, Schema]:
        """Return a dict of the schema of this table."""
        return {schema.schema_id: schema for schema in self.metadata.schemas}

    def spec(self) -> PartitionSpec:
        """Return the partition spec of this table."""
        return next(spec for spec in self.metadata.partition_specs if spec.spec_id == self.metadata.default_spec_id)

    def specs(self) -> dict[int, PartitionSpec]:
        """Return a dict the partition specs this table."""
        return {spec.spec_id: spec for spec in self.metadata.partition_specs}

    def sort_order(self) -> SortOrder:
        """Return the sort order of this table."""
        return next(
            sort_order for sort_order in self.metadata.sort_orders if sort_order.order_id == self.metadata.default_sort_order_id
        )

    def sort_orders(self) -> dict[int, SortOrder]:
        """Return a dict of the sort orders of this table."""
        return {sort_order.order_id: sort_order for sort_order in self.metadata.sort_orders}

    def last_partition_id(self) -> int:
        """Return the highest assigned partition field ID across all specs or 999 if only the unpartitioned spec exists."""
        if self.metadata.last_partition_id:
            return self.metadata.last_partition_id
        return PARTITION_FIELD_ID_START - 1

    @property
    def properties(self) -> dict[str, str]:
        """Properties of the table."""
        return self.metadata.properties

    def location(self) -> str:
        """Return the table's base location."""
        return self.metadata.location

    def location_provider(self) -> LocationProvider:
        """Return the table's location provider."""
        return load_location_provider(table_location=self.metadata.location, table_properties=self.metadata.properties)

    @property
    def last_sequence_number(self) -> int:
        return self.metadata.last_sequence_number

    def current_snapshot(self) -> Snapshot | None:
        """Get the current snapshot for this table, or None if there is no current snapshot."""
        if self.metadata.current_snapshot_id is not None:
            return self.snapshot_by_id(self.metadata.current_snapshot_id)
        return None

    def snapshots(self) -> list[Snapshot]:
        return self.metadata.snapshots

    def snapshot_by_id(self, snapshot_id: int) -> Snapshot | None:
        """Get the snapshot of this table with the given id, or None if there is no matching snapshot."""
        return self.metadata.snapshot_by_id(snapshot_id)

    def snapshot_by_name(self, name: str) -> Snapshot | None:
        """Return the snapshot referenced by the given name or null if no such reference exists."""
        if ref := self.metadata.refs.get(name):
            return self.snapshot_by_id(ref.snapshot_id)
        return None

    def snapshot_as_of_timestamp(self, timestamp_ms: int, inclusive: bool = True) -> Snapshot | None:
        """Get the snapshot that was current as of or right before the given timestamp, or None if there is no matching snapshot.

        Args:
            timestamp_ms: Find snapshot that was current at/before this timestamp
            inclusive: Includes timestamp_ms in search when True. Excludes timestamp_ms when False
        """
        for log_entry in reversed(self.history()):
            if (inclusive and log_entry.timestamp_ms <= timestamp_ms) or log_entry.timestamp_ms < timestamp_ms:
                return self.snapshot_by_id(log_entry.snapshot_id)
        return None

    def history(self) -> list[SnapshotLogEntry]:
        """Get the snapshot history of this table."""
        return self.metadata.snapshot_log

    def manage_snapshots(self) -> ManageSnapshots:
        """
        Shorthand to run snapshot management operations like create branch, create tag, etc.

        Use table.manage_snapshots().<operation>().commit() to run a specific operation.
        Use table.manage_snapshots().<operation-one>().<operation-two>().commit() to run multiple operations.
        Pending changes are applied on commit.

        We can also use context managers to make more changes. For example,

        with table.manage_snapshots() as ms:
           ms.create_tag(snapshot_id1, "Tag_A").create_tag(snapshot_id2, "Tag_B")
        """
        return ManageSnapshots(transaction=Transaction(self, autocommit=True))

    def update_statistics(self) -> UpdateStatistics:
        """
        Shorthand to run statistics management operations like add statistics and remove statistics.

        Use table.update_statistics().<operation>().commit() to run a specific operation.
        Use table.update_statistics().<operation-one>().<operation-two>().commit() to run multiple operations.

        Pending changes are applied on commit.

        We can also use context managers to make more changes. For example:

        with table.update_statistics() as update:
            update.set_statistics(statistics_file=statistics_file)
            update.remove_statistics(snapshot_id=2)
        """
        return UpdateStatistics(transaction=Transaction(self, autocommit=True))

    def update_schema(self, allow_incompatible_changes: bool = False, case_sensitive: bool = True) -> UpdateSchema:
        """Create a new UpdateSchema to alter the columns of this table.

        Args:
            allow_incompatible_changes: If changes are allowed that might break downstream consumers.
            case_sensitive: If field names are case-sensitive.

        Returns:
            A new UpdateSchema.
        """
        return UpdateSchema(
            transaction=Transaction(self, autocommit=True),
            allow_incompatible_changes=allow_incompatible_changes,
            case_sensitive=case_sensitive,
            name_mapping=self.name_mapping(),
        )

    def update_sort_order(self, case_sensitive: bool = True) -> UpdateSortOrder:
        """Create a new UpdateSortOrder to update the sort order of this table.

        Returns:
            A new UpdateSortOrder.
        """
        return UpdateSortOrder(transaction=Transaction(self, autocommit=True), case_sensitive=case_sensitive)

    def name_mapping(self) -> NameMapping | None:
        """Return the table's field-id NameMapping."""
        return self.metadata.name_mapping()

    def upsert(
        self,
        df: pa.Table,
        join_cols: list[str] | None = None,
        when_matched_update_all: bool = True,
        when_not_matched_insert_all: bool = True,
        case_sensitive: bool = True,
        branch: str | None = MAIN_BRANCH,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
    ) -> UpsertResult:
        """Shorthand API for performing an upsert to an iceberg table.

        Args:

            df: The input dataframe to upsert with the table's data.
            join_cols: Columns to join on, if not provided, it will use the identifier-field-ids.
            when_matched_update_all: Bool indicating to update rows that are matched but require an update
                due to a value in a non-key column changing
            when_not_matched_insert_all: Bool indicating new rows to be inserted that do not match any
                existing rows in the table
            case_sensitive: Bool indicating if the match should be case-sensitive
            branch: Branch Reference to run the upsert operation
            snapshot_properties: Custom properties to be added to the snapshot summary

            To learn more about the identifier-field-ids: https://iceberg.apache.org/spec/#identifier-field-ids

                Example Use Cases:
                    Case 1: Both Parameters = True (Full Upsert)
                    Existing row found → Update it
                    New row found → Insert it

                    Case 2: when_matched_update_all = False, when_not_matched_insert_all = True
                    Existing row found → Do nothing (no updates)
                    New row found → Insert it

                    Case 3: when_matched_update_all = True, when_not_matched_insert_all = False
                    Existing row found → Update it
                    New row found → Do nothing (no inserts)

                    Case 4: Both Parameters = False (No Merge Effect)
                    Existing row found → Do nothing
                    New row found → Do nothing
                    (Function effectively does nothing)


        Returns:
            An UpsertResult class (contains details of rows updated and inserted)
        """
        with self.transaction() as tx:
            return tx.upsert(
                df=df,
                join_cols=join_cols,
                when_matched_update_all=when_matched_update_all,
                when_not_matched_insert_all=when_not_matched_insert_all,
                case_sensitive=case_sensitive,
                branch=branch,
                snapshot_properties=snapshot_properties,
            )

    def append(
        self,
        df: pa.Table | pa.RecordBatchReader,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        branch: str | None = MAIN_BRANCH,
    ) -> None:
        """
        Shorthand API for appending PyArrow data to the table.

        Accepts either a ``pa.Table`` or a streaming ``pa.RecordBatchReader``.
        See :meth:`Transaction.append` for streaming semantics and partition
        limitations.

        Args:
            df: An Arrow Table or a RecordBatchReader of records to append.
            snapshot_properties: Custom properties to be added to the snapshot summary
            branch: Branch Reference to run the append operation
        """
        with self.transaction() as tx:
            tx.append(df=df, snapshot_properties=snapshot_properties, branch=branch)

    def dynamic_partition_overwrite(
        self, df: pa.Table, snapshot_properties: dict[str, str] = EMPTY_DICT, branch: str | None = MAIN_BRANCH
    ) -> None:
        """Shorthand for dynamic overwriting the table with a PyArrow table.

        Old partitions are auto detected and replaced with data files created for input arrow table.
        Args:
            df: The Arrow dataframe that will be used to overwrite the table
            snapshot_properties: Custom properties to be added to the snapshot summary
            branch: Branch Reference to run the dynamic partition overwrite operation
        """
        with self.transaction() as tx:
            tx.dynamic_partition_overwrite(df=df, snapshot_properties=snapshot_properties, branch=branch)

    def overwrite(
        self,
        df: pa.Table | pa.RecordBatchReader,
        overwrite_filter: BooleanExpression | str = ALWAYS_TRUE,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        case_sensitive: bool = True,
        branch: str | None = MAIN_BRANCH,
    ) -> None:
        """
        Shorthand for overwriting the table with a PyArrow Table or RecordBatchReader.

        Accepts either a ``pa.Table`` or a streaming ``pa.RecordBatchReader``.
        See :meth:`Transaction.overwrite` for streaming semantics and partition
        limitations.

        An overwrite may produce zero or more snapshots based on the operation:

            - DELETE: In case existing Parquet files can be dropped completely.
            - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter..
            - APPEND: In case new data is being inserted into the table.

        Args:
            df: An Arrow Table or a RecordBatchReader of records to write.
            overwrite_filter: ALWAYS_TRUE when you overwrite all the data,
                              or a boolean expression in case of a partial overwrite
            snapshot_properties: Custom properties to be added to the snapshot summary
            case_sensitive: A bool determine if the provided `overwrite_filter` is case-sensitive
            branch: Branch Reference to run the overwrite operation
        """
        with self.transaction() as tx:
            tx.overwrite(
                df=df,
                overwrite_filter=overwrite_filter,
                case_sensitive=case_sensitive,
                snapshot_properties=snapshot_properties,
                branch=branch,
            )

    def delete(
        self,
        delete_filter: BooleanExpression | str = ALWAYS_TRUE,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        case_sensitive: bool = True,
        branch: str | None = MAIN_BRANCH,
    ) -> None:
        """
        Shorthand for deleting rows from the table.

        Args:
            delete_filter: The predicate that used to remove rows
            snapshot_properties: Custom properties to be added to the snapshot summary
            case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive
            branch: Branch Reference to run the delete operation
        """
        with self.transaction() as tx:
            tx.delete(
                delete_filter=delete_filter, case_sensitive=case_sensitive, snapshot_properties=snapshot_properties, branch=branch
            )

    def add_files(
        self,
        file_paths: list[str],
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        check_duplicate_files: bool = True,
        branch: str | None = MAIN_BRANCH,
    ) -> None:
        """
        Shorthand API for adding files as data files to the table.

        Args:
            file_paths: The list of full file paths to be added as data files to the table

        Raises:
            FileNotFoundError: If the file does not exist.
        """
        with self.transaction() as tx:
            tx.add_files(
                file_paths=file_paths,
                snapshot_properties=snapshot_properties,
                check_duplicate_files=check_duplicate_files,
                branch=branch,
            )

    def update_spec(self, case_sensitive: bool = True) -> UpdateSpec:
        return UpdateSpec(Transaction(self, autocommit=True), case_sensitive=case_sensitive)

    def refs(self) -> dict[str, SnapshotRef]:
        """Return the snapshot references in the table."""
        return self.metadata.refs

    @staticmethod
    def _check_uuid(current_metadata: TableMetadata, new_metadata: TableMetadata) -> None:
        """Validate that the table UUID matches after refresh."""
        current = current_metadata.table_uuid
        refreshed = new_metadata.table_uuid

        if current != refreshed:
            raise ValueError(f"Table UUID does not match: current={current} != refreshed={refreshed}")

    def _do_commit(self, updates: tuple[TableUpdate, ...], requirements: tuple[TableRequirement, ...]) -> None:
        response = self.catalog.commit_table(self, requirements, updates)

        # Ensure table uuid has not changed
        self._check_uuid(self.metadata, response.metadata)

        # https://github.com/apache/iceberg/blob/f6faa58/core/src/main/java/org/apache/iceberg/CatalogUtil.java#L527
        # delete old metadata if METADATA_DELETE_AFTER_COMMIT_ENABLED is set to true and uses
        # TableProperties.METADATA_PREVIOUS_VERSIONS_MAX to determine how many previous versions to keep -
        # everything else will be removed.
        try:
            self.catalog._delete_old_metadata(self.io, self.metadata, response.metadata)
        except Exception as e:
            warnings.warn(f"Failed to delete old metadata after commit: {e}", stacklevel=2)

        self.metadata = response.metadata
        self.metadata_location = response.metadata_location

    def __eq__(self, other: Any) -> bool:
        """Return the equality of two instances of the Table class."""
        return (
            self.name() == other.name() and self.metadata == other.metadata and self.metadata_location == other.metadata_location
            if isinstance(other, Table)
            else False
        )

    def __repr__(self) -> str:
        """Return the string representation of the Table class."""
        table_name = self.catalog.table_name_from(self._identifier)
        schema_str = ",\n  ".join(str(column) for column in self.schema().columns if self.schema())
        partition_str = f"partition by: [{', '.join(field.name for field in self.spec().fields if self.spec())}]"
        sort_order_str = f"sort order: [{', '.join(str(field) for field in self.sort_order().fields if self.sort_order())}]"
        snapshot_str = f"snapshot: {str(self.current_snapshot()) if self.current_snapshot() else 'null'}"
        result_str = f"{table_name}(\n  {schema_str}\n),\n{partition_str},\n{sort_order_str},\n{snapshot_str}"
        return result_str

    def to_daft(self) -> daft.DataFrame:
        """Read a Daft DataFrame lazily from this Iceberg table.

        Returns:
            daft.DataFrame: Unmaterialized Daft Dataframe created from the Iceberg table
        """
        import daft

        return daft.read_iceberg(self)

    def to_bodo(self) -> bd.DataFrame:
        """Read a bodo DataFrame lazily from this Iceberg table.

        Returns:
            bd.DataFrame: Unmaterialized Bodo Dataframe created from the Iceberg table
        """
        import bodo.pandas as bd

        return bd.read_iceberg_table(self)

    def to_polars(self) -> pl.LazyFrame:
        """Lazily read from this Apache Iceberg table.

        Returns:
            pl.LazyFrame: Unmaterialized Polars LazyFrame created from the Iceberg table
        """
        import polars as pl

        return pl.scan_iceberg(self)

    def __datafusion_table_provider__(self, session: Any | None = None) -> IcebergDataFusionTable:
        """Return the DataFusion table provider PyCapsule interface.

        To support DataFusion features such as push down filtering, this function will return a PyCapsule
        interface that conforms to the FFI Table Provider required by DataFusion. From an end user perspective
        you should not need to call this function directly. Instead you can use ``register_table`` in
        the DataFusion SessionContext.

        Returns:
            A PyCapsule DataFusion TableProvider interface.

        Example:
            ```python
            from datafusion import SessionContext
            from pyiceberg.catalog import load_catalog
            import pyarrow as pa
            catalog = load_catalog("catalog", type="in-memory")
            catalog.create_namespace_if_not_exists("default")
            data = pa.table({"x": [1, 2, 3], "y": [4, 5, 6]})
            iceberg_table = catalog.create_table("default.test", schema=data.schema)
            iceberg_table.append(data)
            ctx = SessionContext()
            ctx.register_table("test", iceberg_table)
            ctx.table("test").show()
            ```
            Results in
            ```
            DataFrame()
            +---+---+
            | x | y |
            +---+---+
            | 1 | 4 |
            | 2 | 5 |
            | 3 | 6 |
            +---+---+
            ```
        """
        from pyiceberg_core.datafusion import IcebergDataFusionTable

        provider = IcebergDataFusionTable(
            identifier=self.name(),
            metadata_location=self.metadata_location,
            file_io_properties=self.io.properties,
        ).__datafusion_table_provider__
        return provider(session)

inspect property

Return the InspectTable object to browse the table metadata.

Returns:

Type	Description
InspectTable	InspectTable object based on this Table.

maintenance property

Return the MaintenanceTable object for maintenance.

Returns:

Type	Description
MaintenanceTable	MaintenanceTable object based on this Table.

properties property

Properties of the table.

__datafusion_table_provider__(session=None)

Return the DataFusion table provider PyCapsule interface.

To support DataFusion features such as push down filtering, this function will return a PyCapsule interface that conforms to the FFI Table Provider required by DataFusion. From an end user perspective you should not need to call this function directly. Instead you can use register_table in the DataFusion SessionContext.

Returns:

Type	Description
IcebergDataFusionTable	A PyCapsule DataFusion TableProvider interface.

Example

from datafusion import SessionContext
from pyiceberg.catalog import load_catalog
import pyarrow as pa
catalog = load_catalog("catalog", type="in-memory")
catalog.create_namespace_if_not_exists("default")
data = pa.table({"x": [1, 2, 3], "y": [4, 5, 6]})
iceberg_table = catalog.create_table("default.test", schema=data.schema)
iceberg_table.append(data)
ctx = SessionContext()
ctx.register_table("test", iceberg_table)
ctx.table("test").show()

Results in

DataFrame()
+---+---+
| x | y |
+---+---+
| 1 | 4 |
| 2 | 5 |
| 3 | 6 |
+---+---+

Source code in pyiceberg/table/__init__.py

def __datafusion_table_provider__(self, session: Any | None = None) -> IcebergDataFusionTable:
    """Return the DataFusion table provider PyCapsule interface.

    To support DataFusion features such as push down filtering, this function will return a PyCapsule
    interface that conforms to the FFI Table Provider required by DataFusion. From an end user perspective
    you should not need to call this function directly. Instead you can use ``register_table`` in
    the DataFusion SessionContext.

    Returns:
        A PyCapsule DataFusion TableProvider interface.

    Example:
        ```python
        from datafusion import SessionContext
        from pyiceberg.catalog import load_catalog
        import pyarrow as pa
        catalog = load_catalog("catalog", type="in-memory")
        catalog.create_namespace_if_not_exists("default")
        data = pa.table({"x": [1, 2, 3], "y": [4, 5, 6]})
        iceberg_table = catalog.create_table("default.test", schema=data.schema)
        iceberg_table.append(data)
        ctx = SessionContext()
        ctx.register_table("test", iceberg_table)
        ctx.table("test").show()
        ```
        Results in
        ```
        DataFrame()
        +---+---+
        | x | y |
        +---+---+
        | 1 | 4 |
        | 2 | 5 |
        | 3 | 6 |
        +---+---+
        ```
    """
    from pyiceberg_core.datafusion import IcebergDataFusionTable

    provider = IcebergDataFusionTable(
        identifier=self.name(),
        metadata_location=self.metadata_location,
        file_io_properties=self.io.properties,
    ).__datafusion_table_provider__
    return provider(session)

__eq__(other)

Return the equality of two instances of the Table class.

Source code in pyiceberg/table/__init__.py

def __eq__(self, other: Any) -> bool:
    """Return the equality of two instances of the Table class."""
    return (
        self.name() == other.name() and self.metadata == other.metadata and self.metadata_location == other.metadata_location
        if isinstance(other, Table)
        else False
    )

__repr__()

Return the string representation of the Table class.

Source code in pyiceberg/table/__init__.py

def __repr__(self) -> str:
    """Return the string representation of the Table class."""
    table_name = self.catalog.table_name_from(self._identifier)
    schema_str = ",\n  ".join(str(column) for column in self.schema().columns if self.schema())
    partition_str = f"partition by: [{', '.join(field.name for field in self.spec().fields if self.spec())}]"
    sort_order_str = f"sort order: [{', '.join(str(field) for field in self.sort_order().fields if self.sort_order())}]"
    snapshot_str = f"snapshot: {str(self.current_snapshot()) if self.current_snapshot() else 'null'}"
    result_str = f"{table_name}(\n  {schema_str}\n),\n{partition_str},\n{sort_order_str},\n{snapshot_str}"
    return result_str

add_files(file_paths, snapshot_properties=EMPTY_DICT, check_duplicate_files=True, branch=MAIN_BRANCH)

Shorthand API for adding files as data files to the table.

Parameters:

Name	Type	Description	Default
file_paths	list[str]	The list of full file paths to be added as data files to the table	required

Raises:

Type	Description
FileNotFoundError	If the file does not exist.

Source code in pyiceberg/table/__init__.py

def add_files(
    self,
    file_paths: list[str],
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    check_duplicate_files: bool = True,
    branch: str | None = MAIN_BRANCH,
) -> None:
    """
    Shorthand API for adding files as data files to the table.

    Args:
        file_paths: The list of full file paths to be added as data files to the table

    Raises:
        FileNotFoundError: If the file does not exist.
    """
    with self.transaction() as tx:
        tx.add_files(
            file_paths=file_paths,
            snapshot_properties=snapshot_properties,
            check_duplicate_files=check_duplicate_files,
            branch=branch,
        )

append(df, snapshot_properties=EMPTY_DICT, branch=MAIN_BRANCH)

Shorthand API for appending PyArrow data to the table.

Accepts either a pa.Table or a streaming pa.RecordBatchReader. See :meth:Transaction.append for streaming semantics and partition limitations.

Parameters:

Name	Type	Description	Default
df	Table | RecordBatchReader	An Arrow Table or a RecordBatchReader of records to append.	required
snapshot_properties	dict[str, str]	Custom properties to be added to the snapshot summary	EMPTY_DICT
branch	str | None	Branch Reference to run the append operation	MAIN_BRANCH

Source code in pyiceberg/table/__init__.py

def append(
    self,
    df: pa.Table | pa.RecordBatchReader,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    branch: str | None = MAIN_BRANCH,
) -> None:
    """
    Shorthand API for appending PyArrow data to the table.

    Accepts either a ``pa.Table`` or a streaming ``pa.RecordBatchReader``.
    See :meth:`Transaction.append` for streaming semantics and partition
    limitations.

    Args:
        df: An Arrow Table or a RecordBatchReader of records to append.
        snapshot_properties: Custom properties to be added to the snapshot summary
        branch: Branch Reference to run the append operation
    """
    with self.transaction() as tx:
        tx.append(df=df, snapshot_properties=snapshot_properties, branch=branch)

current_snapshot()

Get the current snapshot for this table, or None if there is no current snapshot.

Source code in pyiceberg/table/__init__.py

def current_snapshot(self) -> Snapshot | None:
    """Get the current snapshot for this table, or None if there is no current snapshot."""
    if self.metadata.current_snapshot_id is not None:
        return self.snapshot_by_id(self.metadata.current_snapshot_id)
    return None

delete(delete_filter=ALWAYS_TRUE, snapshot_properties=EMPTY_DICT, case_sensitive=True, branch=MAIN_BRANCH)

Shorthand for deleting rows from the table.

Parameters:

Name	Type	Description	Default
delete_filter	BooleanExpression | str	The predicate that used to remove rows	ALWAYS_TRUE
snapshot_properties	dict[str, str]	Custom properties to be added to the snapshot summary	EMPTY_DICT
case_sensitive	bool	A bool determine if the provided delete_filter is case-sensitive	True
branch	str | None	Branch Reference to run the delete operation	MAIN_BRANCH

Source code in pyiceberg/table/__init__.py

def delete(
    self,
    delete_filter: BooleanExpression | str = ALWAYS_TRUE,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    case_sensitive: bool = True,
    branch: str | None = MAIN_BRANCH,
) -> None:
    """
    Shorthand for deleting rows from the table.

    Args:
        delete_filter: The predicate that used to remove rows
        snapshot_properties: Custom properties to be added to the snapshot summary
        case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive
        branch: Branch Reference to run the delete operation
    """
    with self.transaction() as tx:
        tx.delete(
            delete_filter=delete_filter, case_sensitive=case_sensitive, snapshot_properties=snapshot_properties, branch=branch
        )

dynamic_partition_overwrite(df, snapshot_properties=EMPTY_DICT, branch=MAIN_BRANCH)

Shorthand for dynamic overwriting the table with a PyArrow table.

Old partitions are auto detected and replaced with data files created for input arrow table. Args: df: The Arrow dataframe that will be used to overwrite the table snapshot_properties: Custom properties to be added to the snapshot summary branch: Branch Reference to run the dynamic partition overwrite operation

Source code in pyiceberg/table/__init__.py

def dynamic_partition_overwrite(
    self, df: pa.Table, snapshot_properties: dict[str, str] = EMPTY_DICT, branch: str | None = MAIN_BRANCH
) -> None:
    """Shorthand for dynamic overwriting the table with a PyArrow table.

    Old partitions are auto detected and replaced with data files created for input arrow table.
    Args:
        df: The Arrow dataframe that will be used to overwrite the table
        snapshot_properties: Custom properties to be added to the snapshot summary
        branch: Branch Reference to run the dynamic partition overwrite operation
    """
    with self.transaction() as tx:
        tx.dynamic_partition_overwrite(df=df, snapshot_properties=snapshot_properties, branch=branch)

history()

Get the snapshot history of this table.

Source code in pyiceberg/table/__init__.py

def history(self) -> list[SnapshotLogEntry]:
    """Get the snapshot history of this table."""
    return self.metadata.snapshot_log

incremental_append_scan(*, from_snapshot_id_exclusive=None, to_snapshot_id_inclusive=None, row_filter=ALWAYS_TRUE, selected_fields=('*',), case_sensitive=True, options=EMPTY_DICT, limit=None)

Fetch an IncrementalAppendScan based on the table's current metadata.

The incremental append scan returns the rows added by append snapshots in a snapshot range that match the provided row_filter, projected onto the table's current schema.

Parameters:

Name	Type	Description	Default
from_snapshot_id_exclusive	int | None	Optional ID of the snapshot to start the incremental scan from, exclusively. If not set, the scan starts from the oldest ancestor of the end snapshot (inclusive).	None
to_snapshot_id_inclusive	int | None	Optional ID of the snapshot to end the incremental scan at, inclusively. If not set, it defaults to the table's current snapshot.	None
row_filter	str | BooleanExpression	A string or BooleanExpression that describes the desired rows.	ALWAYS_TRUE
selected_fields	tuple[str, ...]	A tuple of strings representing the column names to return in the output dataframe.	('*',)
case_sensitive	bool	If True column matching is case sensitive.	True
options	Properties	Additional Table properties as a dictionary of string key value pairs to use for this scan.	EMPTY_DICT
limit	int | None	An integer representing the number of rows to return in the scan result. If None, fetches all matching rows.	None

Returns:

Type	Description
IncrementalAppendScan	An IncrementalAppendScan based on the table's current metadata and provided parameters.

Source code in pyiceberg/table/__init__.py

def incremental_append_scan(
    self,
    *,
    from_snapshot_id_exclusive: int | None = None,
    to_snapshot_id_inclusive: int | None = None,
    row_filter: str | BooleanExpression = ALWAYS_TRUE,
    selected_fields: tuple[str, ...] = ("*",),
    case_sensitive: bool = True,
    options: Properties = EMPTY_DICT,
    limit: int | None = None,
) -> IncrementalAppendScan:
    """Fetch an IncrementalAppendScan based on the table's current metadata.

    The incremental append scan returns the rows added by append snapshots in a snapshot
    range that match the provided row_filter, projected onto the table's current schema.

    Args:
        from_snapshot_id_exclusive:
            Optional ID of the snapshot to start the incremental scan from, exclusively. If not set, the scan
            starts from the oldest ancestor of the end snapshot (inclusive).
        to_snapshot_id_inclusive:
            Optional ID of the snapshot to end the incremental scan at, inclusively. If not set, it defaults to
            the table's current snapshot.
        row_filter:
            A string or BooleanExpression that describes the
            desired rows.
        selected_fields:
            A tuple of strings representing the column names
            to return in the output dataframe.
        case_sensitive:
            If True column matching is case sensitive.
        options:
            Additional Table properties as a dictionary of
            string key value pairs to use for this scan.
        limit:
            An integer representing the number of rows to
            return in the scan result. If None, fetches all
            matching rows.

    Returns:
        An IncrementalAppendScan based on the table's current metadata and provided parameters.
    """
    return IncrementalAppendScan(
        table_metadata=self.metadata,
        io=self.io,
        row_filter=row_filter,
        selected_fields=selected_fields,
        case_sensitive=case_sensitive,
        from_snapshot_id=from_snapshot_id_exclusive,
        from_snapshot_inclusive=False,
        to_snapshot_id=to_snapshot_id_inclusive,
        options=options,
        limit=limit,
    )

last_partition_id()

Return the highest assigned partition field ID across all specs or 999 if only the unpartitioned spec exists.

Source code in pyiceberg/table/__init__.py

def last_partition_id(self) -> int:
    """Return the highest assigned partition field ID across all specs or 999 if only the unpartitioned spec exists."""
    if self.metadata.last_partition_id:
        return self.metadata.last_partition_id
    return PARTITION_FIELD_ID_START - 1

location()

Return the table's base location.

Source code in pyiceberg/table/__init__.py

def location(self) -> str:
    """Return the table's base location."""
    return self.metadata.location

location_provider()

Return the table's location provider.

Source code in pyiceberg/table/__init__.py

def location_provider(self) -> LocationProvider:
    """Return the table's location provider."""
    return load_location_provider(table_location=self.metadata.location, table_properties=self.metadata.properties)

manage_snapshots()

Shorthand to run snapshot management operations like create branch, create tag, etc.

Use table.manage_snapshots().().commit() to run a specific operation. Use table.manage_snapshots().().().commit() to run multiple operations. Pending changes are applied on commit.

We can also use context managers to make more changes. For example,

with table.manage_snapshots() as ms: ms.create_tag(snapshot_id1, "Tag_A").create_tag(snapshot_id2, "Tag_B")

Source code in pyiceberg/table/__init__.py

def manage_snapshots(self) -> ManageSnapshots:
    """
    Shorthand to run snapshot management operations like create branch, create tag, etc.

    Use table.manage_snapshots().<operation>().commit() to run a specific operation.
    Use table.manage_snapshots().<operation-one>().<operation-two>().commit() to run multiple operations.
    Pending changes are applied on commit.

    We can also use context managers to make more changes. For example,

    with table.manage_snapshots() as ms:
       ms.create_tag(snapshot_id1, "Tag_A").create_tag(snapshot_id2, "Tag_B")
    """
    return ManageSnapshots(transaction=Transaction(self, autocommit=True))

name()

Return the identifier of this table.

Returns:

Type	Description
Identifier	An Identifier tuple of the table name

Source code in pyiceberg/table/__init__.py

def name(self) -> Identifier:
    """Return the identifier of this table.

    Returns:
        An Identifier tuple of the table name
    """
    return self._identifier

name_mapping()

Return the table's field-id NameMapping.

Source code in pyiceberg/table/__init__.py

def name_mapping(self) -> NameMapping | None:
    """Return the table's field-id NameMapping."""
    return self.metadata.name_mapping()

overwrite(df, overwrite_filter=ALWAYS_TRUE, snapshot_properties=EMPTY_DICT, case_sensitive=True, branch=MAIN_BRANCH)

Shorthand for overwriting the table with a PyArrow Table or RecordBatchReader.

Accepts either a pa.Table or a streaming pa.RecordBatchReader. See :meth:Transaction.overwrite for streaming semantics and partition limitations.

An overwrite may produce zero or more snapshots based on the operation:

- DELETE: In case existing Parquet files can be dropped completely.
- OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter..
- APPEND: In case new data is being inserted into the table.

Parameters:

Name	Type	Description	Default
df	Table | RecordBatchReader	An Arrow Table or a RecordBatchReader of records to write.	required
overwrite_filter	BooleanExpression | str	ALWAYS_TRUE when you overwrite all the data, or a boolean expression in case of a partial overwrite	ALWAYS_TRUE
snapshot_properties	dict[str, str]	Custom properties to be added to the snapshot summary	EMPTY_DICT
case_sensitive	bool	A bool determine if the provided overwrite_filter is case-sensitive	True
branch	str | None	Branch Reference to run the overwrite operation	MAIN_BRANCH

Source code in pyiceberg/table/__init__.py

def overwrite(
    self,
    df: pa.Table | pa.RecordBatchReader,
    overwrite_filter: BooleanExpression | str = ALWAYS_TRUE,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    case_sensitive: bool = True,
    branch: str | None = MAIN_BRANCH,
) -> None:
    """
    Shorthand for overwriting the table with a PyArrow Table or RecordBatchReader.

    Accepts either a ``pa.Table`` or a streaming ``pa.RecordBatchReader``.
    See :meth:`Transaction.overwrite` for streaming semantics and partition
    limitations.

    An overwrite may produce zero or more snapshots based on the operation:

        - DELETE: In case existing Parquet files can be dropped completely.
        - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter..
        - APPEND: In case new data is being inserted into the table.

    Args:
        df: An Arrow Table or a RecordBatchReader of records to write.
        overwrite_filter: ALWAYS_TRUE when you overwrite all the data,
                          or a boolean expression in case of a partial overwrite
        snapshot_properties: Custom properties to be added to the snapshot summary
        case_sensitive: A bool determine if the provided `overwrite_filter` is case-sensitive
        branch: Branch Reference to run the overwrite operation
    """
    with self.transaction() as tx:
        tx.overwrite(
            df=df,
            overwrite_filter=overwrite_filter,
            case_sensitive=case_sensitive,
            snapshot_properties=snapshot_properties,
            branch=branch,
        )

refresh()

Refresh the current table metadata.

Returns:

Type	Description
Table	An updated instance of the same Iceberg table

Source code in pyiceberg/table/__init__.py

def refresh(self) -> Table:
    """Refresh the current table metadata.

    Returns:
        An updated instance of the same Iceberg table
    """
    fresh = self.catalog.load_table(self._identifier)
    self._check_uuid(self.metadata, fresh.metadata)
    self.metadata = fresh.metadata
    self.io = fresh.io
    self.metadata_location = fresh.metadata_location
    self.config = fresh.config
    return self

refs()

Return the snapshot references in the table.

Source code in pyiceberg/table/__init__.py

def refs(self) -> dict[str, SnapshotRef]:
    """Return the snapshot references in the table."""
    return self.metadata.refs

scan(row_filter=ALWAYS_TRUE, selected_fields=('*',), case_sensitive=True, snapshot_id=None, options=EMPTY_DICT, limit=None)

Fetch a DataScan based on the table's current metadata.

The data scan can be used to project the table's data
that matches the provided row_filter onto the table's
current schema.

Parameters:

Name	Type	Description	Default
row_filter	str | BooleanExpression	A string or BooleanExpression that describes the desired rows	ALWAYS_TRUE
selected_fields	tuple[str, ...]	A tuple of strings representing the column names to return in the output dataframe.	('*',)
case_sensitive	bool	If True column matching is case sensitive	True
snapshot_id	int | None	Optional Snapshot ID to time travel to. If None, scans the table as of the current snapshot ID.	None
options	Properties	Additional Table properties as a dictionary of string key value pairs to use for this scan.	EMPTY_DICT
limit	int | None	An integer representing the number of rows to return in the scan result. If None, fetches all matching rows.	None

Returns:

Type	Description
DataScan	A DataScan based on the table's current metadata.

Source code in pyiceberg/table/__init__.py

def scan(
    self,
    row_filter: str | BooleanExpression = ALWAYS_TRUE,
    selected_fields: tuple[str, ...] = ("*",),
    case_sensitive: bool = True,
    snapshot_id: int | None = None,
    options: Properties = EMPTY_DICT,
    limit: int | None = None,
) -> DataScan:
    """Fetch a DataScan based on the table's current metadata.

        The data scan can be used to project the table's data
        that matches the provided row_filter onto the table's
        current schema.

    Args:
        row_filter:
            A string or BooleanExpression that describes the
            desired rows
        selected_fields:
            A tuple of strings representing the column names
            to return in the output dataframe.
        case_sensitive:
            If True column matching is case sensitive
        snapshot_id:
            Optional Snapshot ID to time travel to. If None,
            scans the table as of the current snapshot ID.
        options:
            Additional Table properties as a dictionary of
            string key value pairs to use for this scan.
        limit:
            An integer representing the number of rows to
            return in the scan result. If None, fetches all
            matching rows.

    Returns:
        A DataScan based on the table's current metadata.
    """
    return DataScan(
        table_metadata=self.metadata,
        io=self.io,
        row_filter=row_filter,
        selected_fields=selected_fields,
        case_sensitive=case_sensitive,
        snapshot_id=snapshot_id,
        options=options,
        limit=limit,
        catalog=self.catalog,
        table_identifier=self._identifier,
        table_config=self.config,
    )

schema()

Return the schema for this table.

Source code in pyiceberg/table/__init__.py

def schema(self) -> Schema:
    """Return the schema for this table."""
    return next(schema for schema in self.metadata.schemas if schema.schema_id == self.metadata.current_schema_id)

schemas()

Return a dict of the schema of this table.

Source code in pyiceberg/table/__init__.py

def schemas(self) -> dict[int, Schema]:
    """Return a dict of the schema of this table."""
    return {schema.schema_id: schema for schema in self.metadata.schemas}

snapshot_as_of_timestamp(timestamp_ms, inclusive=True)

Get the snapshot that was current as of or right before the given timestamp, or None if there is no matching snapshot.

Parameters:

Name	Type	Description	Default
timestamp_ms	int	Find snapshot that was current at/before this timestamp	required
inclusive	bool	Includes timestamp_ms in search when True. Excludes timestamp_ms when False	True

Source code in pyiceberg/table/__init__.py

def snapshot_as_of_timestamp(self, timestamp_ms: int, inclusive: bool = True) -> Snapshot | None:
    """Get the snapshot that was current as of or right before the given timestamp, or None if there is no matching snapshot.

    Args:
        timestamp_ms: Find snapshot that was current at/before this timestamp
        inclusive: Includes timestamp_ms in search when True. Excludes timestamp_ms when False
    """
    for log_entry in reversed(self.history()):
        if (inclusive and log_entry.timestamp_ms <= timestamp_ms) or log_entry.timestamp_ms < timestamp_ms:
            return self.snapshot_by_id(log_entry.snapshot_id)
    return None

snapshot_by_id(snapshot_id)

Get the snapshot of this table with the given id, or None if there is no matching snapshot.

Source code in pyiceberg/table/__init__.py

def snapshot_by_id(self, snapshot_id: int) -> Snapshot | None:
    """Get the snapshot of this table with the given id, or None if there is no matching snapshot."""
    return self.metadata.snapshot_by_id(snapshot_id)

snapshot_by_name(name)

Return the snapshot referenced by the given name or null if no such reference exists.

Source code in pyiceberg/table/__init__.py

def snapshot_by_name(self, name: str) -> Snapshot | None:
    """Return the snapshot referenced by the given name or null if no such reference exists."""
    if ref := self.metadata.refs.get(name):
        return self.snapshot_by_id(ref.snapshot_id)
    return None

sort_order()

Return the sort order of this table.

Source code in pyiceberg/table/__init__.py

def sort_order(self) -> SortOrder:
    """Return the sort order of this table."""
    return next(
        sort_order for sort_order in self.metadata.sort_orders if sort_order.order_id == self.metadata.default_sort_order_id
    )

sort_orders()

Return a dict of the sort orders of this table.

Source code in pyiceberg/table/__init__.py

def sort_orders(self) -> dict[int, SortOrder]:
    """Return a dict of the sort orders of this table."""
    return {sort_order.order_id: sort_order for sort_order in self.metadata.sort_orders}

spec()

Return the partition spec of this table.

Source code in pyiceberg/table/__init__.py

def spec(self) -> PartitionSpec:
    """Return the partition spec of this table."""
    return next(spec for spec in self.metadata.partition_specs if spec.spec_id == self.metadata.default_spec_id)

specs()

Return a dict the partition specs this table.

Source code in pyiceberg/table/__init__.py

def specs(self) -> dict[int, PartitionSpec]:
    """Return a dict the partition specs this table."""
    return {spec.spec_id: spec for spec in self.metadata.partition_specs}

to_bodo()

Read a bodo DataFrame lazily from this Iceberg table.

Returns:

Type	Description
DataFrame	bd.DataFrame: Unmaterialized Bodo Dataframe created from the Iceberg table

Source code in pyiceberg/table/__init__.py

def to_bodo(self) -> bd.DataFrame:
    """Read a bodo DataFrame lazily from this Iceberg table.

    Returns:
        bd.DataFrame: Unmaterialized Bodo Dataframe created from the Iceberg table
    """
    import bodo.pandas as bd

    return bd.read_iceberg_table(self)

to_daft()

Read a Daft DataFrame lazily from this Iceberg table.

Returns:

Type	Description
DataFrame	daft.DataFrame: Unmaterialized Daft Dataframe created from the Iceberg table

Source code in pyiceberg/table/__init__.py

def to_daft(self) -> daft.DataFrame:
    """Read a Daft DataFrame lazily from this Iceberg table.

    Returns:
        daft.DataFrame: Unmaterialized Daft Dataframe created from the Iceberg table
    """
    import daft

    return daft.read_iceberg(self)

to_polars()

Lazily read from this Apache Iceberg table.

Returns:

Type	Description
LazyFrame	pl.LazyFrame: Unmaterialized Polars LazyFrame created from the Iceberg table

Source code in pyiceberg/table/__init__.py

def to_polars(self) -> pl.LazyFrame:
    """Lazily read from this Apache Iceberg table.

    Returns:
        pl.LazyFrame: Unmaterialized Polars LazyFrame created from the Iceberg table
    """
    import polars as pl

    return pl.scan_iceberg(self)

transaction()

Create a new transaction object to first stage the changes, and then commit them to the catalog.

Returns:

Type	Description
Transaction	The transaction object

Source code in pyiceberg/table/__init__.py

def transaction(self) -> Transaction:
    """Create a new transaction object to first stage the changes, and then commit them to the catalog.

    Returns:
        The transaction object
    """
    return Transaction(self)

update_schema(allow_incompatible_changes=False, case_sensitive=True)

Create a new UpdateSchema to alter the columns of this table.

Parameters:

Name	Type	Description	Default
allow_incompatible_changes	bool	If changes are allowed that might break downstream consumers.	False
case_sensitive	bool	If field names are case-sensitive.	True

Returns:

Type	Description
UpdateSchema	A new UpdateSchema.

Source code in pyiceberg/table/__init__.py

def update_schema(self, allow_incompatible_changes: bool = False, case_sensitive: bool = True) -> UpdateSchema:
    """Create a new UpdateSchema to alter the columns of this table.

    Args:
        allow_incompatible_changes: If changes are allowed that might break downstream consumers.
        case_sensitive: If field names are case-sensitive.

    Returns:
        A new UpdateSchema.
    """
    return UpdateSchema(
        transaction=Transaction(self, autocommit=True),
        allow_incompatible_changes=allow_incompatible_changes,
        case_sensitive=case_sensitive,
        name_mapping=self.name_mapping(),
    )

update_sort_order(case_sensitive=True)

Create a new UpdateSortOrder to update the sort order of this table.

Returns:

Type	Description
UpdateSortOrder	A new UpdateSortOrder.

Source code in pyiceberg/table/__init__.py

def update_sort_order(self, case_sensitive: bool = True) -> UpdateSortOrder:
    """Create a new UpdateSortOrder to update the sort order of this table.

    Returns:
        A new UpdateSortOrder.
    """
    return UpdateSortOrder(transaction=Transaction(self, autocommit=True), case_sensitive=case_sensitive)

update_statistics()

Shorthand to run statistics management operations like add statistics and remove statistics.

Use table.update_statistics().().commit() to run a specific operation. Use table.update_statistics().().().commit() to run multiple operations.

Pending changes are applied on commit.

We can also use context managers to make more changes. For example:

with table.update_statistics() as update: update.set_statistics(statistics_file=statistics_file) update.remove_statistics(snapshot_id=2)

Source code in pyiceberg/table/__init__.py

def update_statistics(self) -> UpdateStatistics:
    """
    Shorthand to run statistics management operations like add statistics and remove statistics.

    Use table.update_statistics().<operation>().commit() to run a specific operation.
    Use table.update_statistics().<operation-one>().<operation-two>().commit() to run multiple operations.

    Pending changes are applied on commit.

    We can also use context managers to make more changes. For example:

    with table.update_statistics() as update:
        update.set_statistics(statistics_file=statistics_file)
        update.remove_statistics(snapshot_id=2)
    """
    return UpdateStatistics(transaction=Transaction(self, autocommit=True))

upsert(df, join_cols=None, when_matched_update_all=True, when_not_matched_insert_all=True, case_sensitive=True, branch=MAIN_BRANCH, snapshot_properties=EMPTY_DICT)

Shorthand API for performing an upsert to an iceberg table.

Args:

df: The input dataframe to upsert with the table's data.
join_cols: Columns to join on, if not provided, it will use the identifier-field-ids.
when_matched_update_all: Bool indicating to update rows that are matched but require an update
    due to a value in a non-key column changing
when_not_matched_insert_all: Bool indicating new rows to be inserted that do not match any
    existing rows in the table
case_sensitive: Bool indicating if the match should be case-sensitive
branch: Branch Reference to run the upsert operation
snapshot_properties: Custom properties to be added to the snapshot summary

To learn more about the identifier-field-ids: https://iceberg.apache.org/spec/#identifier-field-ids

    Example Use Cases:
        Case 1: Both Parameters = True (Full Upsert)
        Existing row found → Update it
        New row found → Insert it

        Case 2: when_matched_update_all = False, when_not_matched_insert_all = True
        Existing row found → Do nothing (no updates)
        New row found → Insert it

        Case 3: when_matched_update_all = True, when_not_matched_insert_all = False
        Existing row found → Update it
        New row found → Do nothing (no inserts)

        Case 4: Both Parameters = False (No Merge Effect)
        Existing row found → Do nothing
        New row found → Do nothing
        (Function effectively does nothing)

Returns:

Type	Description
UpsertResult	An UpsertResult class (contains details of rows updated and inserted)

Source code in pyiceberg/table/__init__.py

def upsert(
    self,
    df: pa.Table,
    join_cols: list[str] | None = None,
    when_matched_update_all: bool = True,
    when_not_matched_insert_all: bool = True,
    case_sensitive: bool = True,
    branch: str | None = MAIN_BRANCH,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
) -> UpsertResult:
    """Shorthand API for performing an upsert to an iceberg table.

    Args:

        df: The input dataframe to upsert with the table's data.
        join_cols: Columns to join on, if not provided, it will use the identifier-field-ids.
        when_matched_update_all: Bool indicating to update rows that are matched but require an update
            due to a value in a non-key column changing
        when_not_matched_insert_all: Bool indicating new rows to be inserted that do not match any
            existing rows in the table
        case_sensitive: Bool indicating if the match should be case-sensitive
        branch: Branch Reference to run the upsert operation
        snapshot_properties: Custom properties to be added to the snapshot summary

        To learn more about the identifier-field-ids: https://iceberg.apache.org/spec/#identifier-field-ids

            Example Use Cases:
                Case 1: Both Parameters = True (Full Upsert)
                Existing row found → Update it
                New row found → Insert it

                Case 2: when_matched_update_all = False, when_not_matched_insert_all = True
                Existing row found → Do nothing (no updates)
                New row found → Insert it

                Case 3: when_matched_update_all = True, when_not_matched_insert_all = False
                Existing row found → Update it
                New row found → Do nothing (no inserts)

                Case 4: Both Parameters = False (No Merge Effect)
                Existing row found → Do nothing
                New row found → Do nothing
                (Function effectively does nothing)


    Returns:
        An UpsertResult class (contains details of rows updated and inserted)
    """
    with self.transaction() as tx:
        return tx.upsert(
            df=df,
            join_cols=join_cols,
            when_matched_update_all=when_matched_update_all,
            when_not_matched_insert_all=when_not_matched_insert_all,
            case_sensitive=case_sensitive,
            branch=branch,
            snapshot_properties=snapshot_properties,
        )

TableIdentifier

Bases: IcebergBaseModel

Fully Qualified identifier to a table.

Source code in pyiceberg/table/__init__.py

class TableIdentifier(IcebergBaseModel):
    """Fully Qualified identifier to a table."""

    namespace: Namespace
    name: str

TableScan

Bases: BaseScan

A base class for table scans targeting a single snapshot.

Source code in pyiceberg/table/__init__.py

class TableScan(BaseScan):
    """A base class for table scans targeting a single snapshot."""

    snapshot_id: int | None
    catalog: Catalog | None
    table_identifier: Identifier | None
    table_config: Properties

    def __init__(
        self,
        table_metadata: TableMetadata,
        io: FileIO,
        row_filter: str | BooleanExpression = ALWAYS_TRUE,
        selected_fields: tuple[str, ...] = ("*",),
        case_sensitive: bool = True,
        snapshot_id: int | None = None,
        options: Properties = EMPTY_DICT,
        limit: int | None = None,
        catalog: Catalog | None = None,
        table_identifier: Identifier | None = None,
        table_config: Properties = EMPTY_DICT,
    ):
        super().__init__(
            table_metadata=table_metadata,
            io=io,
            row_filter=row_filter,
            selected_fields=selected_fields,
            case_sensitive=case_sensitive,
            options=options,
            limit=limit,
        )
        self.snapshot_id = snapshot_id
        self.catalog = catalog
        self.table_identifier = table_identifier
        self.table_config = table_config

    def snapshot(self) -> Snapshot | None:
        if self.snapshot_id:
            return self.table_metadata.snapshot_by_id(self.snapshot_id)
        return self.table_metadata.current_snapshot()

    def projection(self) -> Schema:
        current_schema = self.table_metadata.schema()
        if self.snapshot_id is not None:
            snapshot = self.table_metadata.snapshot_by_id(self.snapshot_id)
            if snapshot is not None:
                if snapshot.schema_id is not None:
                    try:
                        current_schema = next(
                            schema for schema in self.table_metadata.schemas if schema.schema_id == snapshot.schema_id
                        )
                    except StopIteration:
                        warnings.warn(f"Metadata does not contain schema with id: {snapshot.schema_id}", stacklevel=2)
            else:
                raise ValueError(f"Snapshot not found: {self.snapshot_id}")

        if "*" in self.selected_fields:
            return current_schema

        return current_schema.select(*self.selected_fields, case_sensitive=self.case_sensitive)

    def use_ref(self: S, name: str) -> S:
        if self.snapshot_id:
            raise ValueError(f"Cannot override ref, already set snapshot id={self.snapshot_id}")
        if snapshot := self.table_metadata.snapshot_by_name(name):
            return self.update(snapshot_id=snapshot.snapshot_id)

        raise ValueError(f"Cannot scan unknown ref={name}")

    @abstractmethod
    def count(self) -> int: ...

Transaction

Source code in pyiceberg/table/__init__.py

class Transaction:
    _table: Table
    _autocommit: bool
    _updates: tuple[TableUpdate, ...]
    _requirements: tuple[TableRequirement, ...]

    def __init__(self, table: Table, autocommit: bool = False):
        """Open a transaction to stage and commit changes to a table.

        Args:
            table: The table that will be altered.
            autocommit: Option to automatically commit the changes when they are staged.
        """
        self._table = table
        self._autocommit = autocommit
        self._updates = ()
        self._requirements = ()
        self._snapshot_producers: list[_SnapshotProducer[Any]] = []
        self._failed = False

    @property
    def table_metadata(self) -> TableMetadata:
        return update_table_metadata(self._table.metadata, self._updates)

    def __enter__(self) -> Transaction:
        """Start a transaction to update the table."""
        return self

    def __exit__(self, exctype: type[BaseException] | None, excinst: BaseException | None, exctb: TracebackType | None) -> None:
        """Close and commit the transaction if no exceptions have been raised."""
        if exctype is None and excinst is None and exctb is None:
            self.commit_transaction()

    def _stage(
        self,
        updates: tuple[TableUpdate, ...],
        requirements: tuple[TableRequirement, ...] = (),
    ) -> Transaction:
        """Stage updates to the transaction state without committing to the catalog.

        Args:
            updates: The updates to stage.
            requirements: The requirements that must be met.

        Returns:
            This transaction for method chaining.
        """
        for requirement in requirements:
            requirement.validate(self.table_metadata)

        self._updates += updates

        # For the requirements, it does not make sense to add a requirement more than once
        # For example, you cannot assert that the current schema has two different IDs
        existing_requirements = {type(requirement) for requirement in self._requirements}
        for new_requirement in requirements:
            if type(new_requirement) not in existing_requirements:
                self._requirements = self._requirements + (new_requirement,)

        return self

    def _register_snapshot_producer(self, producer: _SnapshotProducer[Any]) -> None:
        """Register a snapshot producer for retry support."""
        self._snapshot_producers.append(producer)

    def _apply(
        self,
        updates: tuple[TableUpdate, ...],
        requirements: tuple[TableRequirement, ...] = (),
    ) -> Transaction:
        """Check if the requirements are met, and applies the updates to the metadata."""
        self._stage(updates, requirements)

        if self._autocommit:
            self.commit_transaction()

        return self

    def _scan(self, row_filter: str | BooleanExpression = ALWAYS_TRUE, case_sensitive: bool = True) -> DataScan:
        """Minimal data scan of the table with the current state of the transaction."""
        return DataScan(
            table_metadata=self.table_metadata, io=self._table.io, row_filter=row_filter, case_sensitive=case_sensitive
        )

    def upgrade_table_version(self, format_version: TableVersion) -> Transaction:
        """Set the table to a certain version.

        Args:
            format_version: The newly set version.

        Returns:
            The alter table builder.
        """
        if format_version not in {1, 2}:
            raise ValueError(f"Unsupported table format version: {format_version}")

        if format_version < self.table_metadata.format_version:
            raise ValueError(f"Cannot downgrade v{self.table_metadata.format_version} table to v{format_version}")

        if format_version > self.table_metadata.format_version:
            return self._apply((UpgradeFormatVersionUpdate(format_version=format_version),))

        return self

    def set_properties(self, properties: Properties = EMPTY_DICT, **kwargs: Any) -> Transaction:
        """Set properties.

        When a property is already set, it will be overwritten.

        Args:
            properties: The properties set on the table.
            kwargs: properties can also be pass as kwargs.

        Returns:
            The alter table builder.
        """
        if properties and kwargs:
            raise ValueError("Cannot pass both properties and kwargs")
        updates = properties or kwargs
        return self._apply((SetPropertiesUpdate(updates=updates),))

    def _set_ref_snapshot(
        self,
        snapshot_id: int,
        ref_name: str,
        type: str,
        max_ref_age_ms: int | None = None,
        max_snapshot_age_ms: int | None = None,
        min_snapshots_to_keep: int | None = None,
    ) -> UpdatesAndRequirements:
        """Update a ref to a snapshot.

        Returns:
            The updates and requirements for the set-snapshot-ref staged
        """
        updates = (
            SetSnapshotRefUpdate(
                snapshot_id=snapshot_id,
                ref_name=ref_name,
                type=type,
                max_ref_age_ms=max_ref_age_ms,
                max_snapshot_age_ms=max_snapshot_age_ms,
                min_snapshots_to_keep=min_snapshots_to_keep,
            ),
        )
        requirements = (
            AssertRefSnapshotId(
                snapshot_id=self.table_metadata.refs[ref_name].snapshot_id if ref_name in self.table_metadata.refs else None,
                ref=ref_name,
            ),
        )

        return updates, requirements

    def _build_partition_predicate(self, partition_records: set[Record], partition_fields: list[str]) -> BooleanExpression:
        """Build a filter predicate matching any of the input partition records.

        Args:
            partition_records: A set of partition records to match
            partition_fields: The field names to reference for each position in a partition record

        Returns:
            A predicate matching any of the input partition records.
        """
        if not partition_records or not partition_fields:
            return AlwaysFalse()

        per_record_exprs: list[BooleanExpression] = []
        for partition_record in partition_records:
            predicates: list[BooleanExpression] = [
                EqualTo(Reference(partition_field), partition_record[pos])
                if partition_record[pos] is not None
                else IsNull(Reference(partition_field))
                for pos, partition_field in enumerate(partition_fields)
            ]
            per_record_exprs.append(And(*predicates) if len(predicates) > 1 else predicates[0])

        return Or(*per_record_exprs) if len(per_record_exprs) > 1 else per_record_exprs[0]

    def _append_snapshot_producer(
        self, snapshot_properties: dict[str, str], branch: str | None = MAIN_BRANCH
    ) -> _FastAppendFiles:
        """Determine the append type based on table properties.

        Args:
            snapshot_properties: Custom properties to be added to the snapshot summary
        Returns:
            Either a fast-append or a merge-append snapshot producer.
        """
        manifest_merge_enabled = property_as_bool(
            self.table_metadata.properties,
            TableProperties.MANIFEST_MERGE_ENABLED,
            TableProperties.MANIFEST_MERGE_ENABLED_DEFAULT,
        )
        update_snapshot = self.update_snapshot(snapshot_properties=snapshot_properties, branch=branch)
        return update_snapshot.merge_append() if manifest_merge_enabled else update_snapshot.fast_append()

    def update_schema(self, allow_incompatible_changes: bool = False, case_sensitive: bool = True) -> UpdateSchema:
        """Create a new UpdateSchema to alter the columns of this table.

        Args:
            allow_incompatible_changes: If changes are allowed that might break downstream consumers.
            case_sensitive: If field names are case-sensitive.

        Returns:
            A new UpdateSchema.
        """
        return UpdateSchema(
            self,
            allow_incompatible_changes=allow_incompatible_changes,
            case_sensitive=case_sensitive,
            name_mapping=self.table_metadata.name_mapping(),
        )

    def update_sort_order(self, case_sensitive: bool = True) -> UpdateSortOrder:
        """Create a new UpdateSortOrder to update the sort order of this table.

        Args:
            case_sensitive: If field names are case-sensitive.

        Returns:
            A new UpdateSortOrder.
        """
        return UpdateSortOrder(
            self,
            case_sensitive=case_sensitive,
        )

    def update_snapshot(
        self, snapshot_properties: dict[str, str] = EMPTY_DICT, branch: str | None = MAIN_BRANCH
    ) -> UpdateSnapshot:
        """Create a new UpdateSnapshot to produce a new snapshot for the table.

        Returns:
            A new UpdateSnapshot
        """
        return UpdateSnapshot(self, io=self._table.io, branch=branch, snapshot_properties=snapshot_properties)

    def update_statistics(self) -> UpdateStatistics:
        """
        Create a new UpdateStatistics to update the statistics of the table.

        Returns:
            A new UpdateStatistics
        """
        return UpdateStatistics(transaction=self)

    def append(
        self,
        df: pa.Table | pa.RecordBatchReader,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        branch: str | None = MAIN_BRANCH,
    ) -> None:
        """
        Shorthand API for appending PyArrow data to a table transaction.

        Accepts either a fully materialised ``pa.Table`` or a streaming
        ``pa.RecordBatchReader``. Streaming is microbatched by
        ``write.target-file-size-bytes`` so memory stays bounded; the reader is
        consumed once and cannot be reused.

        Streaming writes are currently only supported on unpartitioned tables;
        passing a ``pa.RecordBatchReader`` for a partitioned table raises
        ``NotImplementedError``. See
        https://github.com/apache/iceberg-python/issues/2152.

        Note:
            When ``df`` is a ``pa.RecordBatchReader`` the reader is consumed
            once and cannot be replayed. If the catalog commit fails (e.g.
            ``CommitFailedException`` from a concurrent writer) the reader is
            already drained and a naive retry will append zero rows. Callers
            that need at-least-once semantics should either:

            - reconstruct the reader on each attempt via a factory callable,
              or
            - use a two-stage pattern — write Parquet files explicitly and
              then call :meth:`add_files` (whose input is a replayable list of
              paths) within a retry loop.

            Failures during the write stage (mid-stream reader exception, S3
            errors) do not commit a snapshot, but may leave orphan data files
            in storage that are not referenced by any snapshot. Clean these
            up with expire/orphan-file maintenance jobs.

            ``write.target-file-size-bytes`` is currently interpreted as
            uncompressed in-memory Arrow bytes (the bin-packing weight) rather
            than compressed on-disk Parquet bytes. The resulting files are
            typically 3-10× smaller than the property suggests after
            compression. This matches the existing ``pa.Table`` write path and
            will be tightened once the writer is switched to a
            rolling-``ParquetWriter`` with ``OutputStream.tell()`` (#2998).

        Args:
            df: An Arrow Table or a RecordBatchReader of records to append.
            snapshot_properties: Custom properties to be added to the snapshot summary
            branch: Branch Reference to run the append operation
        """
        try:
            import pyarrow as pa
        except ModuleNotFoundError as e:
            raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

        from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, _dataframe_to_data_files

        if not isinstance(df, (pa.Table, pa.RecordBatchReader)):
            raise ValueError(f"Expected pa.Table or pa.RecordBatchReader, got: {df}")

        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
        _check_pyarrow_schema_compatible(
            self.table_metadata.schema(),
            provided_schema=df.schema,
            downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
            format_version=self.table_metadata.format_version,
        )

        with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
            # For pa.Table we can short-circuit empty inputs cheaply. For a
            # RecordBatchReader the stream is consumed lazily by
            # _dataframe_to_data_files and an empty reader simply yields zero
            # data files (the snapshot is still committed for symmetry with the
            # pa.Table case where empty inputs also produce a snapshot).
            if isinstance(df, pa.RecordBatchReader) or df.shape[0] > 0:
                data_files = _dataframe_to_data_files(
                    table_metadata=self.table_metadata, write_uuid=append_files.commit_uuid, df=df, io=self._table.io
                )
                for data_file in data_files:
                    append_files.append_data_file(data_file)

    def dynamic_partition_overwrite(
        self, df: pa.Table, snapshot_properties: dict[str, str] = EMPTY_DICT, branch: str | None = MAIN_BRANCH
    ) -> None:
        """
        Shorthand for overwriting existing partitions with a PyArrow table.

        The function detects partition values in the provided arrow table using the current
        partition spec, and deletes existing partitions matching these values. Finally, the
        data in the table is appended to the table.

        Args:
            df: The Arrow dataframe that will be used to overwrite the table
            snapshot_properties: Custom properties to be added to the snapshot summary
            branch: Branch Reference to run the dynamic partition overwrite operation
        """
        try:
            import pyarrow as pa
        except ModuleNotFoundError as e:
            raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

        from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, _dataframe_to_data_files

        if not isinstance(df, pa.Table):
            raise ValueError(f"Expected PyArrow table, got: {df}")

        if self.table_metadata.spec().is_unpartitioned():
            raise ValueError("Cannot apply dynamic overwrite on an unpartitioned table.")

        for field in self.table_metadata.spec().fields:
            if not isinstance(field.transform, IdentityTransform):
                raise ValueError(
                    f"For now dynamic overwrite does not support a table with non-identity-transform field "
                    f"in the latest partition spec: {field}"
                )

        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
        _check_pyarrow_schema_compatible(
            self.table_metadata.schema(),
            provided_schema=df.schema,
            downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
            format_version=self.table_metadata.format_version,
        )

        # If dataframe does not have data, there is no need to overwrite
        if df.shape[0] == 0:
            return

        append_snapshot_commit_uuid = uuid.uuid4()
        data_files: list[DataFile] = list(
            _dataframe_to_data_files(
                table_metadata=self._table.metadata, write_uuid=append_snapshot_commit_uuid, df=df, io=self._table.io
            )
        )

        partitions_to_overwrite = {data_file.partition for data_file in data_files}
        partitions_fields = [
            self.table_metadata.schema().find_field(field.source_id).name for field in self.table_metadata.spec().fields
        ]
        delete_filter = self._build_partition_predicate(
            partition_records=partitions_to_overwrite, partition_fields=partitions_fields
        )
        self.delete(
            delete_filter=delete_filter,
            snapshot_properties=snapshot_properties,
            branch=branch,
            _isolation_operation=Operation.OVERWRITE,
        )

        with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
            append_files.commit_uuid = append_snapshot_commit_uuid
            for data_file in data_files:
                append_files.append_data_file(data_file)

    def overwrite(
        self,
        df: pa.Table | pa.RecordBatchReader,
        overwrite_filter: BooleanExpression | str = ALWAYS_TRUE,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        case_sensitive: bool = True,
        branch: str | None = MAIN_BRANCH,
    ) -> None:
        """
        Shorthand for adding a table overwrite with a PyArrow table or RecordBatchReader to the transaction.

        Accepts either a fully materialised ``pa.Table`` or a streaming
        ``pa.RecordBatchReader``. Streaming is microbatched by
        ``write.target-file-size-bytes`` so memory stays bounded; the reader is
        consumed once and cannot be reused.

        Streaming writes are currently only supported on unpartitioned tables;
        passing a ``pa.RecordBatchReader`` for a partitioned table raises
        ``NotImplementedError``. See
        https://github.com/apache/iceberg-python/issues/2152.

        Note:
            When ``df`` is a ``pa.RecordBatchReader`` the reader is consumed
            once and cannot be replayed. If the catalog commit fails (e.g.
            ``CommitFailedException`` from a concurrent writer) the reader is
            already drained and a naive retry will write zero rows. Callers
            that need at-least-once semantics should either:

            - reconstruct the reader on each attempt via a factory callable,
              or
            - use a two-stage pattern — write Parquet files explicitly and
              then call :meth:`add_files` (whose input is a replayable list
              of paths) within a retry loop.

            Failures during the write stage (mid-stream reader exception, S3
            errors) do not commit a snapshot, but may leave orphan data files
            in storage that are not referenced by any snapshot. Clean these
            up with expire/orphan-file maintenance jobs.

            ``write.target-file-size-bytes`` is currently interpreted as
            uncompressed in-memory Arrow bytes (the bin-packing weight) rather
            than compressed on-disk Parquet bytes. The resulting files are
            typically 3-10× smaller than the property suggests after
            compression. This matches the existing ``pa.Table`` write path and
            will be tightened once the writer is switched to a
            rolling-``ParquetWriter`` with ``OutputStream.tell()`` (#2998).

        An overwrite may produce zero or more snapshots based on the operation:

            - DELETE: In case existing Parquet files can be dropped completely.
            - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter.
            - APPEND: In case new data is being inserted into the table.

        Args:
            df: An Arrow Table or a RecordBatchReader of records to write.
            overwrite_filter: ALWAYS_TRUE when you overwrite all the data,
                              or a boolean expression in case of a partial overwrite
            snapshot_properties: Custom properties to be added to the snapshot summary
            case_sensitive: A bool determine if the provided `overwrite_filter` is case-sensitive
            branch: Branch Reference to run the overwrite operation
        """
        try:
            import pyarrow as pa
        except ModuleNotFoundError as e:
            raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

        from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, _dataframe_to_data_files

        if not isinstance(df, (pa.Table, pa.RecordBatchReader)):
            raise ValueError(f"Expected pa.Table or pa.RecordBatchReader, got: {df}")

        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
        _check_pyarrow_schema_compatible(
            self.table_metadata.schema(),
            provided_schema=df.schema,
            downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
            format_version=self.table_metadata.format_version,
        )

        if overwrite_filter != AlwaysFalse():
            # Only delete when the filter is != AlwaysFalse
            self.delete(
                delete_filter=overwrite_filter,
                case_sensitive=case_sensitive,
                snapshot_properties=snapshot_properties,
                branch=branch,
                _isolation_operation=Operation.OVERWRITE,
            )

        with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
            # See append() for the empty-input handling rationale.
            if isinstance(df, pa.RecordBatchReader) or df.shape[0] > 0:
                data_files = _dataframe_to_data_files(
                    table_metadata=self.table_metadata, write_uuid=append_files.commit_uuid, df=df, io=self._table.io
                )
                for data_file in data_files:
                    append_files.append_data_file(data_file)

    def delete(
        self,
        delete_filter: str | BooleanExpression,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        case_sensitive: bool = True,
        branch: str | None = MAIN_BRANCH,
        _isolation_operation: Operation | None = None,
    ) -> None:
        """
        Shorthand for deleting record from a table.

        A delete may produce zero or more snapshots based on the operation:

            - DELETE: In case existing Parquet files can be dropped completely.
            - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the delete filter.

        Args:
            delete_filter: A boolean expression to delete rows from a table
            snapshot_properties: Custom properties to be added to the snapshot summary
            case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive
            branch: Branch Reference to run the delete operation
        """
        from pyiceberg.io.pyarrow import ArrowScan, _dataframe_to_data_files, _expression_to_complementary_pyarrow

        if (
            self.table_metadata.properties.get(TableProperties.DELETE_MODE, TableProperties.DELETE_MODE_DEFAULT)
            == TableProperties.DELETE_MODE_MERGE_ON_READ
        ):
            warnings.warn("Merge on read is not yet supported, falling back to copy-on-write", stacklevel=2)

        if isinstance(delete_filter, str):
            delete_filter = _parse_row_filter(delete_filter)

        with self.update_snapshot(snapshot_properties=snapshot_properties, branch=branch).delete() as delete_snapshot:
            if _isolation_operation is not None:
                delete_snapshot._isolation_operation = _isolation_operation
            delete_snapshot.delete_by_predicate(delete_filter, case_sensitive)

        # Check if there are any files that require an actual rewrite of a data file
        if delete_snapshot.rewrites_needed is True:
            bound_delete_filter = bind(self.table_metadata.schema(), delete_filter, case_sensitive)
            preserve_row_filter = _expression_to_complementary_pyarrow(bound_delete_filter, self.table_metadata.schema())

            file_scan = self._scan(row_filter=delete_filter, case_sensitive=case_sensitive)
            if branch is not None:
                file_scan = file_scan.use_ref(branch)
            files = file_scan.plan_files()

            commit_uuid = uuid.uuid4()
            counter = itertools.count(0)

            replaced_files: list[tuple[DataFile, list[DataFile]]] = []
            # This will load the Parquet file into memory, including:
            #   - Filter out the rows based on the delete filter
            #   - Projecting it to the current schema
            #   - Applying the positional deletes if they are there
            # When writing
            #   - Apply the latest partition-spec
            #   - And sort order when added
            for original_file in files:
                df = ArrowScan(
                    table_metadata=self.table_metadata,
                    io=self._table.io,
                    projected_schema=self.table_metadata.schema(),
                    row_filter=AlwaysTrue(),
                ).to_table(tasks=[original_file])
                filtered_df = df.filter(preserve_row_filter)

                # Only rewrite if there are records being deleted
                if len(filtered_df) == 0:
                    replaced_files.append((original_file.file, []))
                elif len(df) != len(filtered_df):
                    replaced_files.append(
                        (
                            original_file.file,
                            list(
                                _dataframe_to_data_files(
                                    io=self._table.io,
                                    df=filtered_df,
                                    table_metadata=self.table_metadata,
                                    write_uuid=commit_uuid,
                                    counter=counter,
                                )
                            ),
                        )
                    )

            if len(replaced_files) > 0:
                with self.update_snapshot(
                    snapshot_properties=snapshot_properties, branch=branch
                ).overwrite() as overwrite_snapshot:
                    if _isolation_operation is not None:
                        overwrite_snapshot._isolation_operation = _isolation_operation
                    overwrite_snapshot._starting_snapshot_id = delete_snapshot._starting_snapshot_id
                    overwrite_snapshot.commit_uuid = commit_uuid
                    overwrite_snapshot.delete_by_predicate(delete_filter, case_sensitive)
                    for original_data_file, replaced_data_files in replaced_files:
                        overwrite_snapshot.delete_data_file(original_data_file)
                        for replaced_data_file in replaced_data_files:
                            overwrite_snapshot.append_data_file(replaced_data_file)

        if not delete_snapshot.files_affected and not delete_snapshot.rewrites_needed:
            warnings.warn("Delete operation did not match any records", stacklevel=2)

    def upsert(
        self,
        df: pa.Table,
        join_cols: list[str] | None = None,
        when_matched_update_all: bool = True,
        when_not_matched_insert_all: bool = True,
        case_sensitive: bool = True,
        branch: str | None = MAIN_BRANCH,
        snapshot_properties: dict[str, str] = EMPTY_DICT,
    ) -> UpsertResult:
        """Shorthand API for performing an upsert to an iceberg table.

        Args:

            df: The input dataframe to upsert with the table's data.
            join_cols: Columns to join on, if not provided, it will use the identifier-field-ids.
            when_matched_update_all: Bool indicating to update rows that are matched but require an update
                due to a value in a non-key column changing
            when_not_matched_insert_all: Bool indicating new rows to be inserted that do not match any
                existing rows in the table
            case_sensitive: Bool indicating if the match should be case-sensitive
            branch: Branch Reference to run the upsert operation
            snapshot_properties: Custom properties to be added to the snapshot summary

            To learn more about the identifier-field-ids: https://iceberg.apache.org/spec/#identifier-field-ids

                Example Use Cases:
                    Case 1: Both Parameters = True (Full Upsert)
                    Existing row found → Update it
                    New row found → Insert it

                    Case 2: when_matched_update_all = False, when_not_matched_insert_all = True
                    Existing row found → Do nothing (no updates)
                    New row found → Insert it

                    Case 3: when_matched_update_all = True, when_not_matched_insert_all = False
                    Existing row found → Update it
                    New row found → Do nothing (no inserts)

                    Case 4: Both Parameters = False (No Merge Effect)
                    Existing row found → Do nothing
                    New row found → Do nothing
                    (Function effectively does nothing)


        Returns:
            An UpsertResult class (contains details of rows updated and inserted)
        """
        try:
            import pyarrow as pa  # noqa: F401
        except ModuleNotFoundError as e:
            raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

        from pyiceberg.io.pyarrow import expression_to_pyarrow
        from pyiceberg.table import upsert_util

        if join_cols is None:
            join_cols = []
            for field_id in self.table_metadata.schema().identifier_field_ids:
                col = self.table_metadata.schema().find_column_name(field_id)
                if col is not None:
                    join_cols.append(col)
                else:
                    raise ValueError(f"Field-ID could not be found: {join_cols}")

        if len(join_cols) == 0:
            raise ValueError("Join columns could not be found, please set identifier-field-ids or pass in explicitly.")

        if not when_matched_update_all and not when_not_matched_insert_all:
            raise ValueError("no upsert options selected...exiting")

        if upsert_util.has_duplicate_rows(df, join_cols):
            raise ValueError("Duplicate rows found in source dataset based on the key columns. No upsert executed")

        from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible

        downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
        _check_pyarrow_schema_compatible(
            self.table_metadata.schema(),
            provided_schema=df.schema,
            downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
            format_version=self.table_metadata.format_version,
        )

        # get list of rows that exist so we don't have to load the entire target table
        matched_predicate = upsert_util.create_match_filter(df, join_cols)

        # We must use Transaction.table_metadata for the scan. This includes all uncommitted - but relevant - changes.

        matched_iceberg_record_batches_scan = DataScan(
            table_metadata=self.table_metadata,
            io=self._table.io,
            row_filter=matched_predicate,
            case_sensitive=case_sensitive,
        )

        if branch in self.table_metadata.refs:
            matched_iceberg_record_batches_scan = matched_iceberg_record_batches_scan.use_ref(branch)

        matched_iceberg_record_batches = matched_iceberg_record_batches_scan.to_arrow_batch_reader()

        batches_to_overwrite = []
        overwrite_predicates = []
        rows_to_insert = df

        for batch in matched_iceberg_record_batches:
            rows = pa.Table.from_batches([batch])

            if when_matched_update_all:
                # function get_rows_to_update is doing a check on non-key columns to see if any of the
                # values have actually changed. We don't want to do just a blanket overwrite for matched
                # rows if the actual non-key column data hasn't changed.
                # this extra step avoids unnecessary IO and writes
                rows_to_update = upsert_util.get_rows_to_update(df, rows, join_cols)

                if len(rows_to_update) > 0:
                    # build the match predicate filter
                    overwrite_mask_predicate = upsert_util.create_match_filter(rows_to_update, join_cols)

                    batches_to_overwrite.append(rows_to_update)
                    overwrite_predicates.append(overwrite_mask_predicate)

            if when_not_matched_insert_all:
                expr_match = upsert_util.create_match_filter(rows, join_cols)
                expr_match_bound = bind(self.table_metadata.schema(), expr_match, case_sensitive=case_sensitive)
                expr_match_arrow = expression_to_pyarrow(expr_match_bound)

                # Filter rows per batch.
                rows_to_insert = rows_to_insert.filter(~expr_match_arrow)

        update_row_cnt = 0
        insert_row_cnt = 0

        if batches_to_overwrite:
            rows_to_update = pa.concat_tables(batches_to_overwrite)
            update_row_cnt = len(rows_to_update)
            self.overwrite(
                rows_to_update,
                overwrite_filter=Or(*overwrite_predicates) if len(overwrite_predicates) > 1 else overwrite_predicates[0],
                branch=branch,
                snapshot_properties=snapshot_properties,
            )

        if when_not_matched_insert_all:
            insert_row_cnt = len(rows_to_insert)
            if rows_to_insert:
                self.append(rows_to_insert, branch=branch, snapshot_properties=snapshot_properties)

        return UpsertResult(rows_updated=update_row_cnt, rows_inserted=insert_row_cnt)

    def _find_referenced_data_files(self, file_paths: list[str]) -> list[str]:
        """Return file_paths already referenced by data files in the current snapshot."""
        snapshot = self.table_metadata.current_snapshot()
        if snapshot is None:
            return []

        candidates = set(file_paths)
        io = self._table.io
        data_manifests = [m for m in snapshot.manifests(io) if m.content == ManifestContent.DATA]

        def path_filter(data_file: DataFile) -> bool:
            return data_file.file_path in candidates

        executor = ExecutorFactory.get_or_create()
        entries = chain.from_iterable(
            executor.map(
                lambda args: _open_manifest(*args),
                [(io, manifest, path_filter, lambda _: True) for manifest in data_manifests],
            )
        )
        return [entry.data_file.file_path for entry in entries]

    def add_files(
        self,
        file_paths: list[str],
        snapshot_properties: dict[str, str] = EMPTY_DICT,
        check_duplicate_files: bool = True,
        branch: str | None = MAIN_BRANCH,
    ) -> None:
        """
        Shorthand API for adding files as data files to the table transaction.

        Args:
            file_paths: The list of full file paths to be added as data files to the table

        Raises:
            FileNotFoundError: If the file does not exist.
            ValueError: Raises a ValueError given file_paths contains duplicate files
            ValueError: Raises a ValueError given file_paths already referenced by table
        """
        if len(file_paths) != len(set(file_paths)):
            raise ValueError("File paths must be unique")

        if check_duplicate_files:
            referenced_files = self._find_referenced_data_files(file_paths)
            if referenced_files:
                raise ValueError(f"Cannot add files that are already referenced by table, files: {', '.join(referenced_files)}")

        if self.table_metadata.name_mapping() is None:
            self.set_properties(
                **{TableProperties.DEFAULT_NAME_MAPPING: self.table_metadata.schema().name_mapping.model_dump_json()}
            )
        with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
            data_files = _parquet_files_to_data_files(
                table_metadata=self.table_metadata, file_paths=file_paths, io=self._table.io
            )
            for data_file in data_files:
                append_files.append_data_file(data_file)

    def update_spec(self) -> UpdateSpec:
        """Create a new UpdateSpec to update the partitioning of the table.

        Returns:
            A new UpdateSpec.
        """
        return UpdateSpec(self)

    def remove_properties(self, *removals: str) -> Transaction:
        """Remove properties.

        Args:
            removals: Properties to be removed.

        Returns:
            The alter table builder.
        """
        return self._apply((RemovePropertiesUpdate(removals=removals),))

    def update_location(self, location: str) -> Transaction:
        """Set the new table location.

        Args:
            location: The new location of the table.

        Returns:
            The alter table builder.
        """
        raise NotImplementedError("Not yet implemented")

    def commit_transaction(self) -> Table:
        """Commit the changes to the catalog.

        Returns:
            The table with the updates applied.
        """
        if self._failed:
            raise RuntimeError("This transaction failed to commit and cannot be reused; create a new transaction.")
        if len(self._updates) > 0:
            properties = self._table.metadata.properties
            num_retries: int = max(
                0,
                property_as_int(  # type: ignore  # The default is set with non-None value.
                    properties, TableProperties.COMMIT_NUM_RETRIES, TableProperties.COMMIT_NUM_RETRIES_DEFAULT
                ),
            )
            # All retry properties are clamped to non-negative values: a negative wait
            # would raise ValueError from time.sleep mid-retry and mask the original
            # CommitFailedException.
            min_wait_ms: int = max(
                0,
                property_as_int(  # type: ignore  # The default is set with non-None value.
                    properties, TableProperties.COMMIT_MIN_RETRY_WAIT_MS, TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
                ),
            )
            max_wait_ms: int = max(
                0,
                property_as_int(  # type: ignore  # The default is set with non-None value.
                    properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS, TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
                ),
            )
            total_timeout_ms: int = max(
                0,
                property_as_int(  # type: ignore  # The default is set with non-None value.
                    properties, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
                ),
            )
            start_time = time.monotonic()
            self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),)

            try:
                try:
                    for attempt in range(num_retries + 1):
                        try:
                            self._table._do_commit(  # pylint: disable=W0212
                                updates=self._updates,
                                requirements=self._requirements,
                            )
                            self._cleanup_uncommitted_manifests()
                            break
                        except CommitFailedException:
                            elapsed_ms = (time.monotonic() - start_time) * 1000
                            if attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms:
                                raise

                            wait = min(min_wait_ms * (2**attempt), max_wait_ms)
                            jitter = random.uniform(0, 0.1 * wait)
                            logger.warning(
                                "Commit failed due to a concurrent update, retrying (%s/%s) in %s ms",
                                attempt + 1,
                                num_retries,
                                round(wait + jitter),
                            )
                            time.sleep((wait + jitter) / 1000.0)

                            self._table.refresh()
                            if all(
                                self._table.metadata.snapshot_by_id(producer._snapshot_id) is not None
                                for producer in self._snapshot_producers
                            ):
                                # A previous attempt actually landed even though it was reported as
                                # failed (for example a lost response that the transport layer retried).
                                # The snapshot id is stable across attempts, so finding it in the
                                # refreshed metadata means the commit is already applied. Stop here
                                # instead of committing the same data again.
                                self._cleanup_uncommitted_manifests()
                                break
                            self._rebuild_snapshot_updates()
                except (CommitFailedException, ValidationException):
                    # These exceptions guarantee the commit did not land, so it is safe to delete the
                    # files written for it. Any other exception (unknown outcome, or a commit that already
                    # succeeded) is re-raised without deleting, since those files may be referenced by the
                    # catalog's current snapshot and deleting them would corrupt the table for all readers.
                    for producer in self._snapshot_producers:
                        producer._clean_all_uncommitted()
                    raise
            except Exception:
                # Any failure leaves the transaction in an indeterminate state (files deleted, or the
                # commit outcome unknown), so mark it as failed to refuse reuse.
                self._failed = True
                raise

            self._snapshot_producers = []

        elif self._snapshot_producers:
            # An empty staged output (e.g. a delete whose plan matched nothing) skips the
            # commit loop above, so run concurrency validation explicitly before reporting success.
            from pyiceberg.table.update.snapshot import CommitWindow

            try:
                self._table.refresh()
                commit_window = CommitWindow.resolve(
                    self._table.metadata,
                    self._snapshot_producers[0]._starting_snapshot_id,
                    self._snapshot_producers[0]._target_branch,
                )
                for producer in self._snapshot_producers:
                    producer._commit_window = commit_window
                    producer._validate_concurrency()
            except Exception:
                for producer in self._snapshot_producers:
                    producer._clean_all_uncommitted()
                self._failed = True
                raise

            self._snapshot_producers = []

        self._updates = ()
        self._requirements = ()

        return self._table

    def _cleanup_uncommitted_manifests(self) -> None:
        """Clean up manifests from failed retry attempts after a successful commit."""
        for producer in self._snapshot_producers:
            producer._cleanup_uncommitted()

    def _rebuild_snapshot_updates(self) -> None:
        """Rebuild snapshot updates for retry by re-executing registered producers."""
        from pyiceberg.table.update import AddSnapshotUpdate, AssertRefSnapshotId, SetSnapshotRefUpdate
        from pyiceberg.table.update.snapshot import CommitWindow

        self._updates = tuple(u for u in self._updates if not isinstance(u, (AddSnapshotUpdate, SetSnapshotRefUpdate)))
        self._requirements = tuple(r for r in self._requirements if not isinstance(r, AssertRefSnapshotId))

        starting_id = self._snapshot_producers[0]._starting_snapshot_id if self._snapshot_producers else None
        target_branch = self._snapshot_producers[0]._target_branch if self._snapshot_producers else None
        commit_window = CommitWindow.resolve(self._table.metadata, starting_id, target_branch)

        for producer in self._snapshot_producers:
            producer._commit_window = commit_window
            producer._refresh_for_retry()
            producer._validate_concurrency()
            updates, requirements = producer._commit()
            self._stage(updates, requirements)

__enter__()

Start a transaction to update the table.

Source code in pyiceberg/table/__init__.py

def __enter__(self) -> Transaction:
    """Start a transaction to update the table."""
    return self

__exit__(exctype, excinst, exctb)

Close and commit the transaction if no exceptions have been raised.

Source code in pyiceberg/table/__init__.py

def __exit__(self, exctype: type[BaseException] | None, excinst: BaseException | None, exctb: TracebackType | None) -> None:
    """Close and commit the transaction if no exceptions have been raised."""
    if exctype is None and excinst is None and exctb is None:
        self.commit_transaction()

__init__(table, autocommit=False)

Open a transaction to stage and commit changes to a table.

Parameters:

Name	Type	Description	Default
table	Table	The table that will be altered.	required
autocommit	bool	Option to automatically commit the changes when they are staged.	False

Source code in pyiceberg/table/__init__.py

def __init__(self, table: Table, autocommit: bool = False):
    """Open a transaction to stage and commit changes to a table.

    Args:
        table: The table that will be altered.
        autocommit: Option to automatically commit the changes when they are staged.
    """
    self._table = table
    self._autocommit = autocommit
    self._updates = ()
    self._requirements = ()
    self._snapshot_producers: list[_SnapshotProducer[Any]] = []
    self._failed = False

add_files(file_paths, snapshot_properties=EMPTY_DICT, check_duplicate_files=True, branch=MAIN_BRANCH)

Shorthand API for adding files as data files to the table transaction.

Parameters:

Name	Type	Description	Default
file_paths	list[str]	The list of full file paths to be added as data files to the table	required

Raises:

Type	Description
FileNotFoundError	If the file does not exist.
ValueError	Raises a ValueError given file_paths contains duplicate files
ValueError	Raises a ValueError given file_paths already referenced by table

Source code in pyiceberg/table/__init__.py

def add_files(
    self,
    file_paths: list[str],
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    check_duplicate_files: bool = True,
    branch: str | None = MAIN_BRANCH,
) -> None:
    """
    Shorthand API for adding files as data files to the table transaction.

    Args:
        file_paths: The list of full file paths to be added as data files to the table

    Raises:
        FileNotFoundError: If the file does not exist.
        ValueError: Raises a ValueError given file_paths contains duplicate files
        ValueError: Raises a ValueError given file_paths already referenced by table
    """
    if len(file_paths) != len(set(file_paths)):
        raise ValueError("File paths must be unique")

    if check_duplicate_files:
        referenced_files = self._find_referenced_data_files(file_paths)
        if referenced_files:
            raise ValueError(f"Cannot add files that are already referenced by table, files: {', '.join(referenced_files)}")

    if self.table_metadata.name_mapping() is None:
        self.set_properties(
            **{TableProperties.DEFAULT_NAME_MAPPING: self.table_metadata.schema().name_mapping.model_dump_json()}
        )
    with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
        data_files = _parquet_files_to_data_files(
            table_metadata=self.table_metadata, file_paths=file_paths, io=self._table.io
        )
        for data_file in data_files:
            append_files.append_data_file(data_file)

append(df, snapshot_properties=EMPTY_DICT, branch=MAIN_BRANCH)

Shorthand API for appending PyArrow data to a table transaction.

Accepts either a fully materialised pa.Table or a streaming pa.RecordBatchReader. Streaming is microbatched by write.target-file-size-bytes so memory stays bounded; the reader is consumed once and cannot be reused.

Streaming writes are currently only supported on unpartitioned tables; passing a pa.RecordBatchReader for a partitioned table raises NotImplementedError. See https://github.com/apache/iceberg-python/issues/2152.

Note

When df is a pa.RecordBatchReader the reader is consumed once and cannot be replayed. If the catalog commit fails (e.g. CommitFailedException from a concurrent writer) the reader is already drained and a naive retry will append zero rows. Callers that need at-least-once semantics should either:

• reconstruct the reader on each attempt via a factory callable, or
• use a two-stage pattern — write Parquet files explicitly and then call :meth:add_files (whose input is a replayable list of paths) within a retry loop.

Failures during the write stage (mid-stream reader exception, S3 errors) do not commit a snapshot, but may leave orphan data files in storage that are not referenced by any snapshot. Clean these up with expire/orphan-file maintenance jobs.

write.target-file-size-bytes is currently interpreted as uncompressed in-memory Arrow bytes (the bin-packing weight) rather than compressed on-disk Parquet bytes. The resulting files are typically 3-10× smaller than the property suggests after compression. This matches the existing pa.Table write path and will be tightened once the writer is switched to a rolling-ParquetWriter with OutputStream.tell() (#2998).

Parameters:

Name	Type	Description	Default
df	Table | RecordBatchReader	An Arrow Table or a RecordBatchReader of records to append.	required
snapshot_properties	dict[str, str]	Custom properties to be added to the snapshot summary	EMPTY_DICT
branch	str | None	Branch Reference to run the append operation	MAIN_BRANCH

Source code in pyiceberg/table/__init__.py

def append(
    self,
    df: pa.Table | pa.RecordBatchReader,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    branch: str | None = MAIN_BRANCH,
) -> None:
    """
    Shorthand API for appending PyArrow data to a table transaction.

    Accepts either a fully materialised ``pa.Table`` or a streaming
    ``pa.RecordBatchReader``. Streaming is microbatched by
    ``write.target-file-size-bytes`` so memory stays bounded; the reader is
    consumed once and cannot be reused.

    Streaming writes are currently only supported on unpartitioned tables;
    passing a ``pa.RecordBatchReader`` for a partitioned table raises
    ``NotImplementedError``. See
    https://github.com/apache/iceberg-python/issues/2152.

    Note:
        When ``df`` is a ``pa.RecordBatchReader`` the reader is consumed
        once and cannot be replayed. If the catalog commit fails (e.g.
        ``CommitFailedException`` from a concurrent writer) the reader is
        already drained and a naive retry will append zero rows. Callers
        that need at-least-once semantics should either:

        - reconstruct the reader on each attempt via a factory callable,
          or
        - use a two-stage pattern — write Parquet files explicitly and
          then call :meth:`add_files` (whose input is a replayable list of
          paths) within a retry loop.

        Failures during the write stage (mid-stream reader exception, S3
        errors) do not commit a snapshot, but may leave orphan data files
        in storage that are not referenced by any snapshot. Clean these
        up with expire/orphan-file maintenance jobs.

        ``write.target-file-size-bytes`` is currently interpreted as
        uncompressed in-memory Arrow bytes (the bin-packing weight) rather
        than compressed on-disk Parquet bytes. The resulting files are
        typically 3-10× smaller than the property suggests after
        compression. This matches the existing ``pa.Table`` write path and
        will be tightened once the writer is switched to a
        rolling-``ParquetWriter`` with ``OutputStream.tell()`` (#2998).

    Args:
        df: An Arrow Table or a RecordBatchReader of records to append.
        snapshot_properties: Custom properties to be added to the snapshot summary
        branch: Branch Reference to run the append operation
    """
    try:
        import pyarrow as pa
    except ModuleNotFoundError as e:
        raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

    from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, _dataframe_to_data_files

    if not isinstance(df, (pa.Table, pa.RecordBatchReader)):
        raise ValueError(f"Expected pa.Table or pa.RecordBatchReader, got: {df}")

    downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
    _check_pyarrow_schema_compatible(
        self.table_metadata.schema(),
        provided_schema=df.schema,
        downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
        format_version=self.table_metadata.format_version,
    )

    with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
        # For pa.Table we can short-circuit empty inputs cheaply. For a
        # RecordBatchReader the stream is consumed lazily by
        # _dataframe_to_data_files and an empty reader simply yields zero
        # data files (the snapshot is still committed for symmetry with the
        # pa.Table case where empty inputs also produce a snapshot).
        if isinstance(df, pa.RecordBatchReader) or df.shape[0] > 0:
            data_files = _dataframe_to_data_files(
                table_metadata=self.table_metadata, write_uuid=append_files.commit_uuid, df=df, io=self._table.io
            )
            for data_file in data_files:
                append_files.append_data_file(data_file)

commit_transaction()

Commit the changes to the catalog.

Returns:

Type	Description
Table	The table with the updates applied.

Source code in pyiceberg/table/__init__.py

def commit_transaction(self) -> Table:
    """Commit the changes to the catalog.

    Returns:
        The table with the updates applied.
    """
    if self._failed:
        raise RuntimeError("This transaction failed to commit and cannot be reused; create a new transaction.")
    if len(self._updates) > 0:
        properties = self._table.metadata.properties
        num_retries: int = max(
            0,
            property_as_int(  # type: ignore  # The default is set with non-None value.
                properties, TableProperties.COMMIT_NUM_RETRIES, TableProperties.COMMIT_NUM_RETRIES_DEFAULT
            ),
        )
        # All retry properties are clamped to non-negative values: a negative wait
        # would raise ValueError from time.sleep mid-retry and mask the original
        # CommitFailedException.
        min_wait_ms: int = max(
            0,
            property_as_int(  # type: ignore  # The default is set with non-None value.
                properties, TableProperties.COMMIT_MIN_RETRY_WAIT_MS, TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT
            ),
        )
        max_wait_ms: int = max(
            0,
            property_as_int(  # type: ignore  # The default is set with non-None value.
                properties, TableProperties.COMMIT_MAX_RETRY_WAIT_MS, TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT
            ),
        )
        total_timeout_ms: int = max(
            0,
            property_as_int(  # type: ignore  # The default is set with non-None value.
                properties, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT
            ),
        )
        start_time = time.monotonic()
        self._requirements += (AssertTableUUID(uuid=self.table_metadata.table_uuid),)

        try:
            try:
                for attempt in range(num_retries + 1):
                    try:
                        self._table._do_commit(  # pylint: disable=W0212
                            updates=self._updates,
                            requirements=self._requirements,
                        )
                        self._cleanup_uncommitted_manifests()
                        break
                    except CommitFailedException:
                        elapsed_ms = (time.monotonic() - start_time) * 1000
                        if attempt == num_retries or not self._snapshot_producers or elapsed_ms >= total_timeout_ms:
                            raise

                        wait = min(min_wait_ms * (2**attempt), max_wait_ms)
                        jitter = random.uniform(0, 0.1 * wait)
                        logger.warning(
                            "Commit failed due to a concurrent update, retrying (%s/%s) in %s ms",
                            attempt + 1,
                            num_retries,
                            round(wait + jitter),
                        )
                        time.sleep((wait + jitter) / 1000.0)

                        self._table.refresh()
                        if all(
                            self._table.metadata.snapshot_by_id(producer._snapshot_id) is not None
                            for producer in self._snapshot_producers
                        ):
                            # A previous attempt actually landed even though it was reported as
                            # failed (for example a lost response that the transport layer retried).
                            # The snapshot id is stable across attempts, so finding it in the
                            # refreshed metadata means the commit is already applied. Stop here
                            # instead of committing the same data again.
                            self._cleanup_uncommitted_manifests()
                            break
                        self._rebuild_snapshot_updates()
            except (CommitFailedException, ValidationException):
                # These exceptions guarantee the commit did not land, so it is safe to delete the
                # files written for it. Any other exception (unknown outcome, or a commit that already
                # succeeded) is re-raised without deleting, since those files may be referenced by the
                # catalog's current snapshot and deleting them would corrupt the table for all readers.
                for producer in self._snapshot_producers:
                    producer._clean_all_uncommitted()
                raise
        except Exception:
            # Any failure leaves the transaction in an indeterminate state (files deleted, or the
            # commit outcome unknown), so mark it as failed to refuse reuse.
            self._failed = True
            raise

        self._snapshot_producers = []

    elif self._snapshot_producers:
        # An empty staged output (e.g. a delete whose plan matched nothing) skips the
        # commit loop above, so run concurrency validation explicitly before reporting success.
        from pyiceberg.table.update.snapshot import CommitWindow

        try:
            self._table.refresh()
            commit_window = CommitWindow.resolve(
                self._table.metadata,
                self._snapshot_producers[0]._starting_snapshot_id,
                self._snapshot_producers[0]._target_branch,
            )
            for producer in self._snapshot_producers:
                producer._commit_window = commit_window
                producer._validate_concurrency()
        except Exception:
            for producer in self._snapshot_producers:
                producer._clean_all_uncommitted()
            self._failed = True
            raise

        self._snapshot_producers = []

    self._updates = ()
    self._requirements = ()

    return self._table

delete(delete_filter, snapshot_properties=EMPTY_DICT, case_sensitive=True, branch=MAIN_BRANCH, _isolation_operation=None)

Shorthand for deleting record from a table.

A delete may produce zero or more snapshots based on the operation:

- DELETE: In case existing Parquet files can be dropped completely.
- OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the delete filter.

Parameters:

Name	Type	Description	Default
delete_filter	str | BooleanExpression	A boolean expression to delete rows from a table	required
snapshot_properties	dict[str, str]	Custom properties to be added to the snapshot summary	EMPTY_DICT
case_sensitive	bool	A bool determine if the provided delete_filter is case-sensitive	True
branch	str | None	Branch Reference to run the delete operation	MAIN_BRANCH

Source code in pyiceberg/table/__init__.py

def delete(
    self,
    delete_filter: str | BooleanExpression,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    case_sensitive: bool = True,
    branch: str | None = MAIN_BRANCH,
    _isolation_operation: Operation | None = None,
) -> None:
    """
    Shorthand for deleting record from a table.

    A delete may produce zero or more snapshots based on the operation:

        - DELETE: In case existing Parquet files can be dropped completely.
        - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the delete filter.

    Args:
        delete_filter: A boolean expression to delete rows from a table
        snapshot_properties: Custom properties to be added to the snapshot summary
        case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive
        branch: Branch Reference to run the delete operation
    """
    from pyiceberg.io.pyarrow import ArrowScan, _dataframe_to_data_files, _expression_to_complementary_pyarrow

    if (
        self.table_metadata.properties.get(TableProperties.DELETE_MODE, TableProperties.DELETE_MODE_DEFAULT)
        == TableProperties.DELETE_MODE_MERGE_ON_READ
    ):
        warnings.warn("Merge on read is not yet supported, falling back to copy-on-write", stacklevel=2)

    if isinstance(delete_filter, str):
        delete_filter = _parse_row_filter(delete_filter)

    with self.update_snapshot(snapshot_properties=snapshot_properties, branch=branch).delete() as delete_snapshot:
        if _isolation_operation is not None:
            delete_snapshot._isolation_operation = _isolation_operation
        delete_snapshot.delete_by_predicate(delete_filter, case_sensitive)

    # Check if there are any files that require an actual rewrite of a data file
    if delete_snapshot.rewrites_needed is True:
        bound_delete_filter = bind(self.table_metadata.schema(), delete_filter, case_sensitive)
        preserve_row_filter = _expression_to_complementary_pyarrow(bound_delete_filter, self.table_metadata.schema())

        file_scan = self._scan(row_filter=delete_filter, case_sensitive=case_sensitive)
        if branch is not None:
            file_scan = file_scan.use_ref(branch)
        files = file_scan.plan_files()

        commit_uuid = uuid.uuid4()
        counter = itertools.count(0)

        replaced_files: list[tuple[DataFile, list[DataFile]]] = []
        # This will load the Parquet file into memory, including:
        #   - Filter out the rows based on the delete filter
        #   - Projecting it to the current schema
        #   - Applying the positional deletes if they are there
        # When writing
        #   - Apply the latest partition-spec
        #   - And sort order when added
        for original_file in files:
            df = ArrowScan(
                table_metadata=self.table_metadata,
                io=self._table.io,
                projected_schema=self.table_metadata.schema(),
                row_filter=AlwaysTrue(),
            ).to_table(tasks=[original_file])
            filtered_df = df.filter(preserve_row_filter)

            # Only rewrite if there are records being deleted
            if len(filtered_df) == 0:
                replaced_files.append((original_file.file, []))
            elif len(df) != len(filtered_df):
                replaced_files.append(
                    (
                        original_file.file,
                        list(
                            _dataframe_to_data_files(
                                io=self._table.io,
                                df=filtered_df,
                                table_metadata=self.table_metadata,
                                write_uuid=commit_uuid,
                                counter=counter,
                            )
                        ),
                    )
                )

        if len(replaced_files) > 0:
            with self.update_snapshot(
                snapshot_properties=snapshot_properties, branch=branch
            ).overwrite() as overwrite_snapshot:
                if _isolation_operation is not None:
                    overwrite_snapshot._isolation_operation = _isolation_operation
                overwrite_snapshot._starting_snapshot_id = delete_snapshot._starting_snapshot_id
                overwrite_snapshot.commit_uuid = commit_uuid
                overwrite_snapshot.delete_by_predicate(delete_filter, case_sensitive)
                for original_data_file, replaced_data_files in replaced_files:
                    overwrite_snapshot.delete_data_file(original_data_file)
                    for replaced_data_file in replaced_data_files:
                        overwrite_snapshot.append_data_file(replaced_data_file)

    if not delete_snapshot.files_affected and not delete_snapshot.rewrites_needed:
        warnings.warn("Delete operation did not match any records", stacklevel=2)

dynamic_partition_overwrite(df, snapshot_properties=EMPTY_DICT, branch=MAIN_BRANCH)

Shorthand for overwriting existing partitions with a PyArrow table.

The function detects partition values in the provided arrow table using the current partition spec, and deletes existing partitions matching these values. Finally, the data in the table is appended to the table.

Parameters:

Name	Type	Description	Default
df	Table	The Arrow dataframe that will be used to overwrite the table	required
snapshot_properties	dict[str, str]	Custom properties to be added to the snapshot summary	EMPTY_DICT
branch	str | None	Branch Reference to run the dynamic partition overwrite operation	MAIN_BRANCH

Source code in pyiceberg/table/__init__.py

def dynamic_partition_overwrite(
    self, df: pa.Table, snapshot_properties: dict[str, str] = EMPTY_DICT, branch: str | None = MAIN_BRANCH
) -> None:
    """
    Shorthand for overwriting existing partitions with a PyArrow table.

    The function detects partition values in the provided arrow table using the current
    partition spec, and deletes existing partitions matching these values. Finally, the
    data in the table is appended to the table.

    Args:
        df: The Arrow dataframe that will be used to overwrite the table
        snapshot_properties: Custom properties to be added to the snapshot summary
        branch: Branch Reference to run the dynamic partition overwrite operation
    """
    try:
        import pyarrow as pa
    except ModuleNotFoundError as e:
        raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

    from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, _dataframe_to_data_files

    if not isinstance(df, pa.Table):
        raise ValueError(f"Expected PyArrow table, got: {df}")

    if self.table_metadata.spec().is_unpartitioned():
        raise ValueError("Cannot apply dynamic overwrite on an unpartitioned table.")

    for field in self.table_metadata.spec().fields:
        if not isinstance(field.transform, IdentityTransform):
            raise ValueError(
                f"For now dynamic overwrite does not support a table with non-identity-transform field "
                f"in the latest partition spec: {field}"
            )

    downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
    _check_pyarrow_schema_compatible(
        self.table_metadata.schema(),
        provided_schema=df.schema,
        downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
        format_version=self.table_metadata.format_version,
    )

    # If dataframe does not have data, there is no need to overwrite
    if df.shape[0] == 0:
        return

    append_snapshot_commit_uuid = uuid.uuid4()
    data_files: list[DataFile] = list(
        _dataframe_to_data_files(
            table_metadata=self._table.metadata, write_uuid=append_snapshot_commit_uuid, df=df, io=self._table.io
        )
    )

    partitions_to_overwrite = {data_file.partition for data_file in data_files}
    partitions_fields = [
        self.table_metadata.schema().find_field(field.source_id).name for field in self.table_metadata.spec().fields
    ]
    delete_filter = self._build_partition_predicate(
        partition_records=partitions_to_overwrite, partition_fields=partitions_fields
    )
    self.delete(
        delete_filter=delete_filter,
        snapshot_properties=snapshot_properties,
        branch=branch,
        _isolation_operation=Operation.OVERWRITE,
    )

    with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
        append_files.commit_uuid = append_snapshot_commit_uuid
        for data_file in data_files:
            append_files.append_data_file(data_file)

overwrite(df, overwrite_filter=ALWAYS_TRUE, snapshot_properties=EMPTY_DICT, case_sensitive=True, branch=MAIN_BRANCH)

Shorthand for adding a table overwrite with a PyArrow table or RecordBatchReader to the transaction.

Accepts either a fully materialised pa.Table or a streaming pa.RecordBatchReader. Streaming is microbatched by write.target-file-size-bytes so memory stays bounded; the reader is consumed once and cannot be reused.

Streaming writes are currently only supported on unpartitioned tables; passing a pa.RecordBatchReader for a partitioned table raises NotImplementedError. See https://github.com/apache/iceberg-python/issues/2152.

Note

When df is a pa.RecordBatchReader the reader is consumed once and cannot be replayed. If the catalog commit fails (e.g. CommitFailedException from a concurrent writer) the reader is already drained and a naive retry will write zero rows. Callers that need at-least-once semantics should either:

• reconstruct the reader on each attempt via a factory callable, or
• use a two-stage pattern — write Parquet files explicitly and then call :meth:add_files (whose input is a replayable list of paths) within a retry loop.

Failures during the write stage (mid-stream reader exception, S3 errors) do not commit a snapshot, but may leave orphan data files in storage that are not referenced by any snapshot. Clean these up with expire/orphan-file maintenance jobs.

write.target-file-size-bytes is currently interpreted as uncompressed in-memory Arrow bytes (the bin-packing weight) rather than compressed on-disk Parquet bytes. The resulting files are typically 3-10× smaller than the property suggests after compression. This matches the existing pa.Table write path and will be tightened once the writer is switched to a rolling-ParquetWriter with OutputStream.tell() (#2998).

An overwrite may produce zero or more snapshots based on the operation:

- DELETE: In case existing Parquet files can be dropped completely.
- OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter.
- APPEND: In case new data is being inserted into the table.

Parameters:

Name	Type	Description	Default
df	Table | RecordBatchReader	An Arrow Table or a RecordBatchReader of records to write.	required
overwrite_filter	BooleanExpression | str	ALWAYS_TRUE when you overwrite all the data, or a boolean expression in case of a partial overwrite	ALWAYS_TRUE
snapshot_properties	dict[str, str]	Custom properties to be added to the snapshot summary	EMPTY_DICT
case_sensitive	bool	A bool determine if the provided overwrite_filter is case-sensitive	True
branch	str | None	Branch Reference to run the overwrite operation	MAIN_BRANCH

Source code in pyiceberg/table/__init__.py

def overwrite(
    self,
    df: pa.Table | pa.RecordBatchReader,
    overwrite_filter: BooleanExpression | str = ALWAYS_TRUE,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
    case_sensitive: bool = True,
    branch: str | None = MAIN_BRANCH,
) -> None:
    """
    Shorthand for adding a table overwrite with a PyArrow table or RecordBatchReader to the transaction.

    Accepts either a fully materialised ``pa.Table`` or a streaming
    ``pa.RecordBatchReader``. Streaming is microbatched by
    ``write.target-file-size-bytes`` so memory stays bounded; the reader is
    consumed once and cannot be reused.

    Streaming writes are currently only supported on unpartitioned tables;
    passing a ``pa.RecordBatchReader`` for a partitioned table raises
    ``NotImplementedError``. See
    https://github.com/apache/iceberg-python/issues/2152.

    Note:
        When ``df`` is a ``pa.RecordBatchReader`` the reader is consumed
        once and cannot be replayed. If the catalog commit fails (e.g.
        ``CommitFailedException`` from a concurrent writer) the reader is
        already drained and a naive retry will write zero rows. Callers
        that need at-least-once semantics should either:

        - reconstruct the reader on each attempt via a factory callable,
          or
        - use a two-stage pattern — write Parquet files explicitly and
          then call :meth:`add_files` (whose input is a replayable list
          of paths) within a retry loop.

        Failures during the write stage (mid-stream reader exception, S3
        errors) do not commit a snapshot, but may leave orphan data files
        in storage that are not referenced by any snapshot. Clean these
        up with expire/orphan-file maintenance jobs.

        ``write.target-file-size-bytes`` is currently interpreted as
        uncompressed in-memory Arrow bytes (the bin-packing weight) rather
        than compressed on-disk Parquet bytes. The resulting files are
        typically 3-10× smaller than the property suggests after
        compression. This matches the existing ``pa.Table`` write path and
        will be tightened once the writer is switched to a
        rolling-``ParquetWriter`` with ``OutputStream.tell()`` (#2998).

    An overwrite may produce zero or more snapshots based on the operation:

        - DELETE: In case existing Parquet files can be dropped completely.
        - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter.
        - APPEND: In case new data is being inserted into the table.

    Args:
        df: An Arrow Table or a RecordBatchReader of records to write.
        overwrite_filter: ALWAYS_TRUE when you overwrite all the data,
                          or a boolean expression in case of a partial overwrite
        snapshot_properties: Custom properties to be added to the snapshot summary
        case_sensitive: A bool determine if the provided `overwrite_filter` is case-sensitive
        branch: Branch Reference to run the overwrite operation
    """
    try:
        import pyarrow as pa
    except ModuleNotFoundError as e:
        raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

    from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible, _dataframe_to_data_files

    if not isinstance(df, (pa.Table, pa.RecordBatchReader)):
        raise ValueError(f"Expected pa.Table or pa.RecordBatchReader, got: {df}")

    downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
    _check_pyarrow_schema_compatible(
        self.table_metadata.schema(),
        provided_schema=df.schema,
        downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
        format_version=self.table_metadata.format_version,
    )

    if overwrite_filter != AlwaysFalse():
        # Only delete when the filter is != AlwaysFalse
        self.delete(
            delete_filter=overwrite_filter,
            case_sensitive=case_sensitive,
            snapshot_properties=snapshot_properties,
            branch=branch,
            _isolation_operation=Operation.OVERWRITE,
        )

    with self._append_snapshot_producer(snapshot_properties, branch=branch) as append_files:
        # See append() for the empty-input handling rationale.
        if isinstance(df, pa.RecordBatchReader) or df.shape[0] > 0:
            data_files = _dataframe_to_data_files(
                table_metadata=self.table_metadata, write_uuid=append_files.commit_uuid, df=df, io=self._table.io
            )
            for data_file in data_files:
                append_files.append_data_file(data_file)

remove_properties(*removals)

Remove properties.

Parameters:

Name	Type	Description	Default
removals	str	Properties to be removed.	()

Returns:

Type	Description
Transaction	The alter table builder.

Source code in pyiceberg/table/__init__.py

def remove_properties(self, *removals: str) -> Transaction:
    """Remove properties.

    Args:
        removals: Properties to be removed.

    Returns:
        The alter table builder.
    """
    return self._apply((RemovePropertiesUpdate(removals=removals),))

set_properties(properties=EMPTY_DICT, **kwargs)

Set properties.

When a property is already set, it will be overwritten.

Parameters:

Name	Type	Description	Default
properties	Properties	The properties set on the table.	EMPTY_DICT
kwargs	Any	properties can also be pass as kwargs.	{}

Returns:

Type	Description
Transaction	The alter table builder.

Source code in pyiceberg/table/__init__.py

def set_properties(self, properties: Properties = EMPTY_DICT, **kwargs: Any) -> Transaction:
    """Set properties.

    When a property is already set, it will be overwritten.

    Args:
        properties: The properties set on the table.
        kwargs: properties can also be pass as kwargs.

    Returns:
        The alter table builder.
    """
    if properties and kwargs:
        raise ValueError("Cannot pass both properties and kwargs")
    updates = properties or kwargs
    return self._apply((SetPropertiesUpdate(updates=updates),))

update_location(location)

Set the new table location.

Parameters:

Name	Type	Description	Default
location	str	The new location of the table.	required

Returns:

Type	Description
Transaction	The alter table builder.

Source code in pyiceberg/table/__init__.py

def update_location(self, location: str) -> Transaction:
    """Set the new table location.

    Args:
        location: The new location of the table.

    Returns:
        The alter table builder.
    """
    raise NotImplementedError("Not yet implemented")

update_schema(allow_incompatible_changes=False, case_sensitive=True)

Create a new UpdateSchema to alter the columns of this table.

Parameters:

Name	Type	Description	Default
allow_incompatible_changes	bool	If changes are allowed that might break downstream consumers.	False
case_sensitive	bool	If field names are case-sensitive.	True

Returns:

Type	Description
UpdateSchema	A new UpdateSchema.

Source code in pyiceberg/table/__init__.py

def update_schema(self, allow_incompatible_changes: bool = False, case_sensitive: bool = True) -> UpdateSchema:
    """Create a new UpdateSchema to alter the columns of this table.

    Args:
        allow_incompatible_changes: If changes are allowed that might break downstream consumers.
        case_sensitive: If field names are case-sensitive.

    Returns:
        A new UpdateSchema.
    """
    return UpdateSchema(
        self,
        allow_incompatible_changes=allow_incompatible_changes,
        case_sensitive=case_sensitive,
        name_mapping=self.table_metadata.name_mapping(),
    )

update_snapshot(snapshot_properties=EMPTY_DICT, branch=MAIN_BRANCH)

Create a new UpdateSnapshot to produce a new snapshot for the table.

Returns:

Type	Description
UpdateSnapshot	A new UpdateSnapshot

Source code in pyiceberg/table/__init__.py

def update_snapshot(
    self, snapshot_properties: dict[str, str] = EMPTY_DICT, branch: str | None = MAIN_BRANCH
) -> UpdateSnapshot:
    """Create a new UpdateSnapshot to produce a new snapshot for the table.

    Returns:
        A new UpdateSnapshot
    """
    return UpdateSnapshot(self, io=self._table.io, branch=branch, snapshot_properties=snapshot_properties)

update_sort_order(case_sensitive=True)

Create a new UpdateSortOrder to update the sort order of this table.

Parameters:

Name	Type	Description	Default
case_sensitive	bool	If field names are case-sensitive.	True

Returns:

Type	Description
UpdateSortOrder	A new UpdateSortOrder.

Source code in pyiceberg/table/__init__.py

def update_sort_order(self, case_sensitive: bool = True) -> UpdateSortOrder:
    """Create a new UpdateSortOrder to update the sort order of this table.

    Args:
        case_sensitive: If field names are case-sensitive.

    Returns:
        A new UpdateSortOrder.
    """
    return UpdateSortOrder(
        self,
        case_sensitive=case_sensitive,
    )

update_spec()

Create a new UpdateSpec to update the partitioning of the table.

Returns:

Type	Description
UpdateSpec	A new UpdateSpec.

Source code in pyiceberg/table/__init__.py

def update_spec(self) -> UpdateSpec:
    """Create a new UpdateSpec to update the partitioning of the table.

    Returns:
        A new UpdateSpec.
    """
    return UpdateSpec(self)

update_statistics()

Create a new UpdateStatistics to update the statistics of the table.

Returns:

Type	Description
UpdateStatistics	A new UpdateStatistics

Source code in pyiceberg/table/__init__.py

def update_statistics(self) -> UpdateStatistics:
    """
    Create a new UpdateStatistics to update the statistics of the table.

    Returns:
        A new UpdateStatistics
    """
    return UpdateStatistics(transaction=self)

upgrade_table_version(format_version)

Set the table to a certain version.

Parameters:

Name	Type	Description	Default
format_version	TableVersion	The newly set version.	required

Returns:

Type	Description
Transaction	The alter table builder.

Source code in pyiceberg/table/__init__.py

def upgrade_table_version(self, format_version: TableVersion) -> Transaction:
    """Set the table to a certain version.

    Args:
        format_version: The newly set version.

    Returns:
        The alter table builder.
    """
    if format_version not in {1, 2}:
        raise ValueError(f"Unsupported table format version: {format_version}")

    if format_version < self.table_metadata.format_version:
        raise ValueError(f"Cannot downgrade v{self.table_metadata.format_version} table to v{format_version}")

    if format_version > self.table_metadata.format_version:
        return self._apply((UpgradeFormatVersionUpdate(format_version=format_version),))

    return self

upsert(df, join_cols=None, when_matched_update_all=True, when_not_matched_insert_all=True, case_sensitive=True, branch=MAIN_BRANCH, snapshot_properties=EMPTY_DICT)

Shorthand API for performing an upsert to an iceberg table.

Args:

df: The input dataframe to upsert with the table's data.
join_cols: Columns to join on, if not provided, it will use the identifier-field-ids.
when_matched_update_all: Bool indicating to update rows that are matched but require an update
    due to a value in a non-key column changing
when_not_matched_insert_all: Bool indicating new rows to be inserted that do not match any
    existing rows in the table
case_sensitive: Bool indicating if the match should be case-sensitive
branch: Branch Reference to run the upsert operation
snapshot_properties: Custom properties to be added to the snapshot summary

To learn more about the identifier-field-ids: https://iceberg.apache.org/spec/#identifier-field-ids

    Example Use Cases:
        Case 1: Both Parameters = True (Full Upsert)
        Existing row found → Update it
        New row found → Insert it

        Case 2: when_matched_update_all = False, when_not_matched_insert_all = True
        Existing row found → Do nothing (no updates)
        New row found → Insert it

        Case 3: when_matched_update_all = True, when_not_matched_insert_all = False
        Existing row found → Update it
        New row found → Do nothing (no inserts)

        Case 4: Both Parameters = False (No Merge Effect)
        Existing row found → Do nothing
        New row found → Do nothing
        (Function effectively does nothing)

Returns:

Type	Description
UpsertResult	An UpsertResult class (contains details of rows updated and inserted)

Source code in pyiceberg/table/__init__.py

def upsert(
    self,
    df: pa.Table,
    join_cols: list[str] | None = None,
    when_matched_update_all: bool = True,
    when_not_matched_insert_all: bool = True,
    case_sensitive: bool = True,
    branch: str | None = MAIN_BRANCH,
    snapshot_properties: dict[str, str] = EMPTY_DICT,
) -> UpsertResult:
    """Shorthand API for performing an upsert to an iceberg table.

    Args:

        df: The input dataframe to upsert with the table's data.
        join_cols: Columns to join on, if not provided, it will use the identifier-field-ids.
        when_matched_update_all: Bool indicating to update rows that are matched but require an update
            due to a value in a non-key column changing
        when_not_matched_insert_all: Bool indicating new rows to be inserted that do not match any
            existing rows in the table
        case_sensitive: Bool indicating if the match should be case-sensitive
        branch: Branch Reference to run the upsert operation
        snapshot_properties: Custom properties to be added to the snapshot summary

        To learn more about the identifier-field-ids: https://iceberg.apache.org/spec/#identifier-field-ids

            Example Use Cases:
                Case 1: Both Parameters = True (Full Upsert)
                Existing row found → Update it
                New row found → Insert it

                Case 2: when_matched_update_all = False, when_not_matched_insert_all = True
                Existing row found → Do nothing (no updates)
                New row found → Insert it

                Case 3: when_matched_update_all = True, when_not_matched_insert_all = False
                Existing row found → Update it
                New row found → Do nothing (no inserts)

                Case 4: Both Parameters = False (No Merge Effect)
                Existing row found → Do nothing
                New row found → Do nothing
                (Function effectively does nothing)


    Returns:
        An UpsertResult class (contains details of rows updated and inserted)
    """
    try:
        import pyarrow as pa  # noqa: F401
    except ModuleNotFoundError as e:
        raise ModuleNotFoundError("For writes PyArrow needs to be installed") from e

    from pyiceberg.io.pyarrow import expression_to_pyarrow
    from pyiceberg.table import upsert_util

    if join_cols is None:
        join_cols = []
        for field_id in self.table_metadata.schema().identifier_field_ids:
            col = self.table_metadata.schema().find_column_name(field_id)
            if col is not None:
                join_cols.append(col)
            else:
                raise ValueError(f"Field-ID could not be found: {join_cols}")

    if len(join_cols) == 0:
        raise ValueError("Join columns could not be found, please set identifier-field-ids or pass in explicitly.")

    if not when_matched_update_all and not when_not_matched_insert_all:
        raise ValueError("no upsert options selected...exiting")

    if upsert_util.has_duplicate_rows(df, join_cols):
        raise ValueError("Duplicate rows found in source dataset based on the key columns. No upsert executed")

    from pyiceberg.io.pyarrow import _check_pyarrow_schema_compatible

    downcast_ns_timestamp_to_us = Config().get_bool(DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE) or False
    _check_pyarrow_schema_compatible(
        self.table_metadata.schema(),
        provided_schema=df.schema,
        downcast_ns_timestamp_to_us=downcast_ns_timestamp_to_us,
        format_version=self.table_metadata.format_version,
    )

    # get list of rows that exist so we don't have to load the entire target table
    matched_predicate = upsert_util.create_match_filter(df, join_cols)

    # We must use Transaction.table_metadata for the scan. This includes all uncommitted - but relevant - changes.

    matched_iceberg_record_batches_scan = DataScan(
        table_metadata=self.table_metadata,
        io=self._table.io,
        row_filter=matched_predicate,
        case_sensitive=case_sensitive,
    )

    if branch in self.table_metadata.refs:
        matched_iceberg_record_batches_scan = matched_iceberg_record_batches_scan.use_ref(branch)

    matched_iceberg_record_batches = matched_iceberg_record_batches_scan.to_arrow_batch_reader()

    batches_to_overwrite = []
    overwrite_predicates = []
    rows_to_insert = df

    for batch in matched_iceberg_record_batches:
        rows = pa.Table.from_batches([batch])

        if when_matched_update_all:
            # function get_rows_to_update is doing a check on non-key columns to see if any of the
            # values have actually changed. We don't want to do just a blanket overwrite for matched
            # rows if the actual non-key column data hasn't changed.
            # this extra step avoids unnecessary IO and writes
            rows_to_update = upsert_util.get_rows_to_update(df, rows, join_cols)

            if len(rows_to_update) > 0:
                # build the match predicate filter
                overwrite_mask_predicate = upsert_util.create_match_filter(rows_to_update, join_cols)

                batches_to_overwrite.append(rows_to_update)
                overwrite_predicates.append(overwrite_mask_predicate)

        if when_not_matched_insert_all:
            expr_match = upsert_util.create_match_filter(rows, join_cols)
            expr_match_bound = bind(self.table_metadata.schema(), expr_match, case_sensitive=case_sensitive)
            expr_match_arrow = expression_to_pyarrow(expr_match_bound)

            # Filter rows per batch.
            rows_to_insert = rows_to_insert.filter(~expr_match_arrow)

    update_row_cnt = 0
    insert_row_cnt = 0

    if batches_to_overwrite:
        rows_to_update = pa.concat_tables(batches_to_overwrite)
        update_row_cnt = len(rows_to_update)
        self.overwrite(
            rows_to_update,
            overwrite_filter=Or(*overwrite_predicates) if len(overwrite_predicates) > 1 else overwrite_predicates[0],
            branch=branch,
            snapshot_properties=snapshot_properties,
        )

    if when_not_matched_insert_all:
        insert_row_cnt = len(rows_to_insert)
        if rows_to_insert:
            self.append(rows_to_insert, branch=branch, snapshot_properties=snapshot_properties)

    return UpsertResult(rows_updated=update_row_cnt, rows_inserted=insert_row_cnt)

UpsertResult dataclass

Summary the upsert operation.

Source code in pyiceberg/table/__init__.py

@dataclass()
class UpsertResult:
    """Summary the upsert operation."""

    rows_updated: int = 0
    rows_inserted: int = 0

WriteTask dataclass

Task with the parameters for writing a DataFile.

Source code in pyiceberg/table/__init__.py

@dataclass(frozen=True)
class WriteTask:
    """Task with the parameters for writing a DataFile."""

    write_uuid: uuid.UUID
    task_id: int
    schema: Schema
    record_batches: list[pa.RecordBatch]
    sort_order_id: int | None = None
    partition_key: PartitionKey | None = None

    def generate_data_file_filename(self, extension: str) -> str:
        # Mimics the behavior in the Java API:
        # https://github.com/apache/iceberg/blob/a582968975dd30ff4917fbbe999f1be903efac02/core/src/main/java/org/apache/iceberg/io/OutputFileFactory.java#L92-L101
        return f"00000-{self.task_id}-{self.write_uuid}.{extension}"