inspect.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

import itertools
from collections.abc import Iterator
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any

from pyiceberg.conversions import from_bytes
from pyiceberg.expressions import AlwaysTrue, BooleanExpression
from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestFile, PartitionFieldSummary
from pyiceberg.partitioning import PartitionSpec
from pyiceberg.table.snapshots import Snapshot, ancestors_of
from pyiceberg.types import PrimitiveType
from pyiceberg.utils.concurrent import ExecutorFactory
from pyiceberg.utils.singleton import _convert_to_hashable_type

if TYPE_CHECKING:
    import pyarrow as pa

    from pyiceberg.table import Table

ALWAYS_TRUE = AlwaysTrue()


def _readable_bound(field_type: PrimitiveType, bound: bytes | None) -> Any | None:
    return from_bytes(field_type, bound) if bound is not None else None


class InspectTable:
    tbl: Table

    def __init__(self, tbl: Table) -> None:
        """Initialize the InspectTable helper.

        Args:
            tbl: The Iceberg :class:`~pyiceberg.table.Table` whose metadata
                will be inspected.

        Raises:
            ModuleNotFoundError: If PyArrow is not installed.
        """
        self.tbl = tbl

        try:
            import pyarrow as pa  # noqa
        except ModuleNotFoundError as e:
            raise ModuleNotFoundError("For metadata operations PyArrow needs to be installed") from e

    def _get_snapshot(self, snapshot_id: int | None = None) -> Snapshot:
        """Return the snapshot identified by *snapshot_id*, or the current snapshot.

        Args:
            snapshot_id: The snapshot ID to look up.  When ``None`` the current
                table snapshot is returned.

        Returns:
            The requested :class:`~pyiceberg.table.snapshots.Snapshot`.

        Raises:
            ValueError: If *snapshot_id* does not exist, or if the table has no
                snapshots at all.
        """
        if snapshot_id is not None:
            if snapshot := self.tbl.metadata.snapshot_by_id(snapshot_id):
                return snapshot
            else:
                raise ValueError(f"Cannot find snapshot with ID {snapshot_id}")

        if snapshot := self.tbl.metadata.current_snapshot():
            return snapshot
        else:
            raise ValueError("Cannot get a snapshot as the table does not have any.")

    def snapshots(self) -> pa.Table:
        """Return all snapshots of the table as a PyArrow table.

        Each row represents one snapshot and includes the commit timestamp,
        snapshot ID, parent snapshot ID, the write operation that produced the
        snapshot (e.g. ``append``, ``overwrite``, ``delete``), the path to the
        manifest list file, and a map of additional summary properties.

        Returns:
            A :class:`pyarrow.Table` with schema::

                committed_at    timestamp[ms, tz=UTC]  not null
                snapshot_id     int64                  not null
                parent_id       int64                  nullable
                operation       string                 nullable
                manifest_list   string                 not null
                summary         map<string, string>    nullable

        Example:
            >>> tbl.inspect.snapshots().to_pandas()
        """
        import pyarrow as pa

        snapshots_schema = pa.schema(
            [
                pa.field("committed_at", pa.timestamp(unit="ms"), nullable=False),
                pa.field("snapshot_id", pa.int64(), nullable=False),
                pa.field("parent_id", pa.int64(), nullable=True),
                pa.field("operation", pa.string(), nullable=True),
                pa.field("manifest_list", pa.string(), nullable=False),
                pa.field("summary", pa.map_(pa.string(), pa.string()), nullable=True),
            ]
        )
        snapshots = []
        for snapshot in self.tbl.metadata.snapshots:
            if summary := snapshot.summary:
                operation = summary.operation.value
                additional_properties = snapshot.summary.additional_properties
            else:
                operation = None
                additional_properties = None

            snapshots.append(
                {
                    "committed_at": datetime.fromtimestamp(snapshot.timestamp_ms / 1000.0, tz=timezone.utc),
                    "snapshot_id": snapshot.snapshot_id,
                    "parent_id": snapshot.parent_snapshot_id,
                    "operation": str(operation),
                    "manifest_list": snapshot.manifest_list,
                    "summary": additional_properties,
                }
            )

        return pa.Table.from_pylist(
            snapshots,
            schema=snapshots_schema,
        )

    def entries(self, snapshot_id: int | None = None) -> pa.Table:
        """Return all manifest entries for a snapshot as a PyArrow table.

        Each row represents one manifest entry (a single data or delete file
        tracked within a manifest).  Readable per-column metrics (lower/upper
        bounds, null counts, etc.) are decoded from their binary representation
        into native Python types.

        Args:
            snapshot_id: The snapshot to inspect.  Defaults to the current
                snapshot when ``None``.

        Returns:
            A :class:`pyarrow.Table` with schema::

                status               int8     not null  (0=EXISTING, 1=ADDED, 2=DELETED)
                snapshot_id          int64    not null
                sequence_number      int64    not null
                file_sequence_number int64    not null
                data_file            struct   not null  (content, file_path, format, partition, metrics…)
                readable_metrics     struct   nullable  (one sub-struct per column)

        Raises:
            ValueError: If *snapshot_id* does not exist or the table has no snapshots.

        Example:
            >>> tbl.inspect.entries().to_pandas()
        """

        schema = self.tbl.metadata.schema()

        readable_metrics_struct = []

        def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
            pa_bound_type = schema_to_pyarrow(bound_type)
            return pa.struct(
                [
                    pa.field("column_size", pa.int64(), nullable=True),
                    pa.field("value_count", pa.int64(), nullable=True),
                    pa.field("null_value_count", pa.int64(), nullable=True),
                    pa.field("nan_value_count", pa.int64(), nullable=True),
                    pa.field("lower_bound", pa_bound_type, nullable=True),
                    pa.field("upper_bound", pa_bound_type, nullable=True),
                ]
            )

        for field in self.tbl.metadata.schema().fields:
            readable_metrics_struct.append(
                pa.field(schema.find_column_name(field.field_id), _readable_metrics_struct(field.field_type), nullable=False)
            )

        partition_record = self.tbl.metadata.specs_struct()
        pa_record_struct = schema_to_pyarrow(partition_record)

        entries_schema = pa.schema(
            [
                pa.field("status", pa.int8(), nullable=False),
                pa.field("snapshot_id", pa.int64(), nullable=False),
                pa.field("sequence_number", pa.int64(), nullable=False),
                pa.field("file_sequence_number", pa.int64(), nullable=False),
                pa.field(
                    "data_file",
                    pa.struct(
                        [
                            pa.field("content", pa.int8(), nullable=False),
                            pa.field("file_path", pa.string(), nullable=False),
                            pa.field("file_format", pa.string(), nullable=False),
                            pa.field("partition", pa_record_struct, nullable=False),
                            pa.field("record_count", pa.int64(), nullable=False),
                            pa.field("file_size_in_bytes", pa.int64(), nullable=False),
                            pa.field("column_sizes", pa.map_(pa.int32(), pa.int64()), nullable=True),
                            pa.field("value_counts", pa.map_(pa.int32(), pa.int64()), nullable=True),
                            pa.field("null_value_counts", pa.map_(pa.int32(), pa.int64()), nullable=True),
                            pa.field("nan_value_counts", pa.map_(pa.int32(), pa.int64()), nullable=True),
                            pa.field("lower_bounds", pa.map_(pa.int32(), pa.binary()), nullable=True),
                            pa.field("upper_bounds", pa.map_(pa.int32(), pa.binary()), nullable=True),
                            pa.field("key_metadata", pa.binary(), nullable=True),
                            pa.field("split_offsets", pa.list_(pa.int64()), nullable=True),
                            pa.field("equality_ids", pa.list_(pa.int32()), nullable=True),
                            pa.field("sort_order_id", pa.int32(), nullable=True),
                        ]
                    ),
                    nullable=False,
                ),
                pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
            ]
        )

        entries = []
        snapshot = self._get_snapshot(snapshot_id)
        for manifest in snapshot.manifests(self.tbl.io):
            for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=False):
                column_sizes = entry.data_file.column_sizes or {}
                value_counts = entry.data_file.value_counts or {}
                null_value_counts = entry.data_file.null_value_counts or {}
                nan_value_counts = entry.data_file.nan_value_counts or {}
                lower_bounds = entry.data_file.lower_bounds or {}
                upper_bounds = entry.data_file.upper_bounds or {}
                readable_metrics = {
                    schema.find_column_name(field.field_id): {
                        "column_size": column_sizes.get(field.field_id),
                        "value_count": value_counts.get(field.field_id),
                        "null_value_count": null_value_counts.get(field.field_id),
                        "nan_value_count": nan_value_counts.get(field.field_id),
                        # Makes them readable
                        "lower_bound": _readable_bound(field.field_type, lower_bounds.get(field.field_id)),
                        "upper_bound": _readable_bound(field.field_type, upper_bounds.get(field.field_id)),
                    }
                    for field in self.tbl.metadata.schema().fields
                }

                partition = entry.data_file.partition
                partition_record_dict = {
                    field.name: partition[pos]
                    for pos, field in enumerate(self.tbl.metadata.specs()[manifest.partition_spec_id].fields)
                }

                entries.append(
                    {
                        "status": entry.status.value,
                        "snapshot_id": entry.snapshot_id,
                        "sequence_number": entry.sequence_number,
                        "file_sequence_number": entry.file_sequence_number,
                        "data_file": {
                            "content": entry.data_file.content,
                            "file_path": entry.data_file.file_path,
                            "file_format": entry.data_file.file_format,
                            "partition": partition_record_dict,
                            "record_count": entry.data_file.record_count,
                            "file_size_in_bytes": entry.data_file.file_size_in_bytes,
                            "column_sizes": dict(entry.data_file.column_sizes),
                            "value_counts": dict(entry.data_file.value_counts or {}),
                            "null_value_counts": dict(entry.data_file.null_value_counts or {}),
                            "nan_value_counts": dict(entry.data_file.nan_value_counts or {}),
                            "lower_bounds": entry.data_file.lower_bounds,
                            "upper_bounds": entry.data_file.upper_bounds,
                            "key_metadata": entry.data_file.key_metadata,
                            "split_offsets": entry.data_file.split_offsets,
                            "equality_ids": entry.data_file.equality_ids,
                            "sort_order_id": entry.data_file.sort_order_id,
                            "spec_id": entry.data_file.spec_id,
                        },
                        "readable_metrics": readable_metrics,
                    }
                )

        return pa.Table.from_pylist(
            entries,
            schema=entries_schema,
        )

    def refs(self) -> pa.Table:
        """Return all named references (branches and tags) of the table.

        Returns:
            A :class:`pyarrow.Table` with schema::

                name                    string                  not null
                type                    dictionary<int32,str>   not null  ("branch" or "tag")
                snapshot_id             int64                   not null
                max_reference_age_in_ms int64                   nullable
                min_snapshots_to_keep   int32                   nullable
                max_snapshot_age_in_ms  int64                   nullable

        Example:
            >>> tbl.inspect.refs().to_pandas()
        """
        import pyarrow as pa

        ref_schema = pa.schema(
            [
                pa.field("name", pa.string(), nullable=False),
                pa.field("type", pa.dictionary(pa.int32(), pa.string()), nullable=False),
                pa.field("snapshot_id", pa.int64(), nullable=False),
                pa.field("max_reference_age_in_ms", pa.int64(), nullable=True),
                pa.field("min_snapshots_to_keep", pa.int32(), nullable=True),
                pa.field("max_snapshot_age_in_ms", pa.int64(), nullable=True),
            ]
        )

        ref_results = []
        for ref in self.tbl.metadata.refs:
            if snapshot_ref := self.tbl.metadata.refs.get(ref):
                ref_results.append(
                    {
                        "name": ref,
                        "type": snapshot_ref.snapshot_ref_type.upper(),
                        "snapshot_id": snapshot_ref.snapshot_id,
                        "max_reference_age_in_ms": snapshot_ref.max_ref_age_ms,
                        "min_snapshots_to_keep": snapshot_ref.min_snapshots_to_keep,
                        "max_snapshot_age_in_ms": snapshot_ref.max_snapshot_age_ms,
                    }
                )

        return pa.Table.from_pylist(ref_results, schema=ref_schema)

    def partitions(
        self,
        snapshot_id: int | None = None,
        row_filter: str | BooleanExpression = ALWAYS_TRUE,
        case_sensitive: bool = True,
    ) -> pa.Table:
        """Return partition-level statistics for a snapshot.

        Aggregates record counts, file counts, total data file sizes and delete
        file information per distinct partition value.  When a *row_filter* is
        supplied only the partitions that could contain matching rows are
        included.

        Args:
            snapshot_id: The snapshot to inspect.  Defaults to the current
                snapshot when ``None``.
            row_filter: A predicate (as a string expression or
                :class:`~pyiceberg.expressions.BooleanExpression`) used to
                prune which partitions are returned.  Defaults to
                :data:`ALWAYS_TRUE` (all partitions).
            case_sensitive: Whether column name matching in *row_filter* is
                case-sensitive.  Defaults to ``True``.

        Returns:
            A :class:`pyarrow.Table` with one row per distinct partition.
            The schema includes ``partition``, ``spec_id``, ``record_count``,
            ``file_count``, ``total_data_file_size_in_bytes``,
            ``position_delete_record_count``, ``position_delete_file_count``,
            ``equality_delete_record_count``, ``equality_delete_file_count``,
            ``last_updated_at``, and ``last_updated_snapshot_id``.

        Raises:
            ValueError: If *snapshot_id* does not exist or the table has no
                snapshots.

        Example:
            >>> tbl.inspect.partitions().to_pandas()
        """

        table_schema = pa.schema(
            [
                pa.field("record_count", pa.int64(), nullable=False),
                pa.field("file_count", pa.int32(), nullable=False),
                pa.field("total_data_file_size_in_bytes", pa.int64(), nullable=False),
                pa.field("position_delete_record_count", pa.int64(), nullable=False),
                pa.field("position_delete_file_count", pa.int32(), nullable=False),
                pa.field("equality_delete_record_count", pa.int64(), nullable=False),
                pa.field("equality_delete_file_count", pa.int32(), nullable=False),
                pa.field("last_updated_at", pa.timestamp(unit="ms"), nullable=True),
                pa.field("last_updated_snapshot_id", pa.int64(), nullable=True),
            ]
        )

        snapshot = self._get_snapshot(snapshot_id)
        spec_ids = {manifest.partition_spec_id for manifest in snapshot.manifests(self.tbl.io)}
        partition_record = self.tbl.metadata.specs_struct(spec_ids=spec_ids)
        has_partitions = len(partition_record.fields) > 0

        if has_partitions:
            pa_record_struct = schema_to_pyarrow(partition_record)
            partitions_schema = pa.schema(
                [
                    pa.field("partition", pa_record_struct, nullable=False),
                    pa.field("spec_id", pa.int32(), nullable=False),
                ]
            )

            table_schema = pa.unify_schemas([partitions_schema, table_schema])

        scan = DataScan(
            table_metadata=self.tbl.metadata,
            io=self.tbl.io,
            row_filter=row_filter,
            case_sensitive=case_sensitive,
            snapshot_id=snapshot.snapshot_id,
        )

        partitions_map: dict[tuple[str, Any], Any] = {}

        for entry in itertools.chain.from_iterable(scan.scan_plan_helper()):
            partition = entry.data_file.partition
            partition_record_dict = {
                field.name: partition[pos] for pos, field in enumerate(self.tbl.metadata.specs()[entry.data_file.spec_id].fields)
            }
            entry_snapshot = self.tbl.snapshot_by_id(entry.snapshot_id) if entry.snapshot_id is not None else None
            self._update_partitions_map_from_manifest_entry(
                partitions_map, entry.data_file, partition_record_dict, entry_snapshot
            )

        return pa.Table.from_pylist(
            partitions_map.values(),
            schema=table_schema,
        )

    def _update_partitions_map_from_manifest_entry(
        self,
        partitions_map: dict[tuple[str, Any], Any],
        file: DataFile,
        partition_record_dict: dict[str, Any],
        snapshot: Snapshot | None,
    ) -> None:
        partition_record_key = _convert_to_hashable_type(partition_record_dict)
        if partition_record_key not in partitions_map:
            partitions_map[partition_record_key] = {
                "partition": partition_record_dict,
                "spec_id": file.spec_id,
                "record_count": 0,
                "file_count": 0,
                "total_data_file_size_in_bytes": 0,
                "position_delete_record_count": 0,
                "position_delete_file_count": 0,
                "equality_delete_record_count": 0,
                "equality_delete_file_count": 0,
                "last_updated_at": snapshot.timestamp_ms if snapshot else None,
                "last_updated_snapshot_id": snapshot.snapshot_id if snapshot else None,
            }

        partition_row = partitions_map[partition_record_key]

        if snapshot is not None:
            if partition_row["last_updated_at"] is None or partition_row["last_updated_snapshot_id"] < snapshot.timestamp_ms:
                partition_row["last_updated_at"] = snapshot.timestamp_ms
                partition_row["last_updated_snapshot_id"] = snapshot.snapshot_id

        if file.content == DataFileContent.DATA:
            partition_row["record_count"] += file.record_count
            partition_row["file_count"] += 1
            partition_row["total_data_file_size_in_bytes"] += file.file_size_in_bytes
        elif file.content == DataFileContent.POSITION_DELETES:
            partition_row["position_delete_record_count"] += file.record_count
            partition_row["position_delete_file_count"] += 1
        elif file.content == DataFileContent.EQUALITY_DELETES:
            partition_row["equality_delete_record_count"] += file.record_count
            partition_row["equality_delete_file_count"] += 1
        else:
            raise ValueError(f"Unknown DataFileContent ({file.content})")

    def _get_manifests_schema(self) -> pa.Schema:
        import pyarrow as pa

        partition_summary_schema = pa.struct(
            [
                pa.field("contains_null", pa.bool_(), nullable=False),
                pa.field("contains_nan", pa.bool_(), nullable=True),
                pa.field("lower_bound", pa.string(), nullable=True),
                pa.field("upper_bound", pa.string(), nullable=True),
            ]
        )

        manifest_schema = pa.schema(
            [
                pa.field("content", pa.int8(), nullable=False),
                pa.field("path", pa.string(), nullable=False),
                pa.field("length", pa.int64(), nullable=False),
                pa.field("partition_spec_id", pa.int32(), nullable=False),
                pa.field("added_snapshot_id", pa.int64(), nullable=False),
                pa.field("added_data_files_count", pa.int32(), nullable=False),
                pa.field("existing_data_files_count", pa.int32(), nullable=False),
                pa.field("deleted_data_files_count", pa.int32(), nullable=False),
                pa.field("added_delete_files_count", pa.int32(), nullable=False),
                pa.field("existing_delete_files_count", pa.int32(), nullable=False),
                pa.field("deleted_delete_files_count", pa.int32(), nullable=False),
                pa.field("partition_summaries", pa.list_(partition_summary_schema), nullable=False),
            ]
        )
        return manifest_schema

    def _get_all_manifests_schema(self) -> pa.Schema:
        import pyarrow as pa

        all_manifests_schema = self._get_manifests_schema()
        all_manifests_schema = all_manifests_schema.append(pa.field("reference_snapshot_id", pa.int64(), nullable=False))
        return all_manifests_schema

    def _generate_manifests_table(self, snapshot: Snapshot | None, is_all_manifests_table: bool = False) -> pa.Table:
        import pyarrow as pa

        def _partition_summaries_to_rows(
            spec: PartitionSpec, partition_summaries: list[PartitionFieldSummary]
        ) -> list[dict[str, Any]]:
            rows = []
            for i, field_summary in enumerate(partition_summaries):
                field = spec.fields[i]
                partition_field_type = spec.partition_type(self.tbl.schema()).fields[i].field_type
                lower_bound = (
                    (
                        field.transform.to_human_string(
                            partition_field_type, from_bytes(partition_field_type, field_summary.lower_bound)
                        )
                    )
                    if field_summary.lower_bound
                    else None
                )
                upper_bound = (
                    (
                        field.transform.to_human_string(
                            partition_field_type, from_bytes(partition_field_type, field_summary.upper_bound)
                        )
                    )
                    if field_summary.upper_bound
                    else None
                )
                rows.append(
                    {
                        "contains_null": field_summary.contains_null,
                        "contains_nan": field_summary.contains_nan,
                        "lower_bound": lower_bound,
                        "upper_bound": upper_bound,
                    }
                )
            return rows

        specs = self.tbl.metadata.specs()
        manifests = []
        if snapshot:
            for manifest in snapshot.manifests(self.tbl.io):
                is_data_file = manifest.content == ManifestContent.DATA
                is_delete_file = manifest.content == ManifestContent.DELETES
                manifest_row = {
                    "content": manifest.content,
                    "path": manifest.manifest_path,
                    "length": manifest.manifest_length,
                    "partition_spec_id": manifest.partition_spec_id,
                    "added_snapshot_id": manifest.added_snapshot_id,
                    "added_data_files_count": manifest.added_files_count if is_data_file else 0,
                    "existing_data_files_count": manifest.existing_files_count if is_data_file else 0,
                    "deleted_data_files_count": manifest.deleted_files_count if is_data_file else 0,
                    "added_delete_files_count": manifest.added_files_count if is_delete_file else 0,
                    "existing_delete_files_count": manifest.existing_files_count if is_delete_file else 0,
                    "deleted_delete_files_count": manifest.deleted_files_count if is_delete_file else 0,
                    "partition_summaries": _partition_summaries_to_rows(specs[manifest.partition_spec_id], manifest.partitions)
                    if manifest.partitions
                    else [],
                }
                if is_all_manifests_table:
                    manifest_row["reference_snapshot_id"] = snapshot.snapshot_id
                manifests.append(manifest_row)

        return pa.Table.from_pylist(
            manifests,
            schema=self._get_all_manifests_schema() if is_all_manifests_table else self._get_manifests_schema(),
        )

    def manifests(self) -> pa.Table:
        """Return the manifest files for the current snapshot.

        Each row describes one manifest file referenced by the current snapshot,
        including file counts (added, existing, deleted) for both data files
        and delete files, as well as per-partition bound summaries.

        Returns:
            A :class:`pyarrow.Table` — see :meth:`_get_manifests_schema` for
            the full column list.

        Example:
            >>> tbl.inspect.manifests().to_pandas()
        """
        return self._generate_manifests_table(self.tbl.current_snapshot())

    def metadata_log_entries(self) -> pa.Table:
        """Return the metadata log for the table.

        The metadata log records every metadata file that has been the current
        metadata for the table, along with the snapshot that was current when
        each metadata file was written.  The most recent entry corresponds to
        the current metadata location.

        Returns:
            A :class:`pyarrow.Table` with schema::

                timestamp               timestamp[ms]   not null
                file                    string          not null
                latest_snapshot_id      int64           nullable
                latest_schema_id        int32           nullable
                latest_sequence_number  int64           nullable

        Example:
            >>> tbl.inspect.metadata_log_entries().to_pandas()
        """

        table_schema = pa.schema(
            [
                pa.field("timestamp", pa.timestamp(unit="ms"), nullable=False),
                pa.field("file", pa.string(), nullable=False),
                pa.field("latest_snapshot_id", pa.int64(), nullable=True),
                pa.field("latest_schema_id", pa.int32(), nullable=True),
                pa.field("latest_sequence_number", pa.int64(), nullable=True),
            ]
        )

        def metadata_log_entry_to_row(metadata_entry: MetadataLogEntry) -> dict[str, Any]:
            latest_snapshot = self.tbl.snapshot_as_of_timestamp(metadata_entry.timestamp_ms)
            return {
                "timestamp": metadata_entry.timestamp_ms,
                "file": metadata_entry.metadata_file,
                "latest_snapshot_id": latest_snapshot.snapshot_id if latest_snapshot else None,
                "latest_schema_id": latest_snapshot.schema_id if latest_snapshot else None,
                "latest_sequence_number": latest_snapshot.sequence_number if latest_snapshot else None,
            }

        # similar to MetadataLogEntriesTable in Java
        # https://github.com/apache/iceberg/blob/8a70fe0ff5f241aec8856f8091c77fdce35ad256/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java#L62-L66
        metadata_log_entries = self.tbl.metadata.metadata_log + [
            MetadataLogEntry(metadata_file=self.tbl.metadata_location, timestamp_ms=self.tbl.metadata.last_updated_ms)
        ]

        return pa.Table.from_pylist(
            [metadata_log_entry_to_row(entry) for entry in metadata_log_entries],
            schema=table_schema,
        )

    def history(self) -> pa.Table:
        """Return the snapshot history of the table.

        Each row in the result corresponds to a snapshot-log entry, i.e. a
        point in time when a snapshot became the current snapshot of the table.
        The ``is_current_ancestor`` column is ``True`` when the snapshot is on
        the ancestry chain of the current snapshot (useful for detecting
        expired or replaced snapshots after branch operations).

        Returns:
            A :class:`pyarrow.Table` with schema::

                made_current_at     timestamp[ms, tz=UTC]   not null
                snapshot_id         int64                   not null
                parent_id           int64                   nullable
                is_current_ancestor bool                    not null

        Example:
            >>> tbl.inspect.history().to_pandas()
        """
        import pyarrow as pa

        history_schema = pa.schema(
            [
                pa.field("made_current_at", pa.timestamp(unit="ms"), nullable=False),
                pa.field("snapshot_id", pa.int64(), nullable=False),
                pa.field("parent_id", pa.int64(), nullable=True),
                pa.field("is_current_ancestor", pa.bool_(), nullable=False),
            ]
        )

        ancestors_ids = {snapshot.snapshot_id for snapshot in ancestors_of(self.tbl.current_snapshot(), self.tbl.metadata)}

        history = []
        metadata = self.tbl.metadata

        for snapshot_entry in metadata.snapshot_log:
            snapshot = metadata.snapshot_by_id(snapshot_entry.snapshot_id)

            history.append(
                {
                    "made_current_at": datetime.fromtimestamp(snapshot_entry.timestamp_ms / 1000.0, tz=timezone.utc),
                    "snapshot_id": snapshot_entry.snapshot_id,
                    "parent_id": snapshot.parent_snapshot_id if snapshot else None,
                    "is_current_ancestor": snapshot_entry.snapshot_id in ancestors_ids,
                }
            )

        return pa.Table.from_pylist(history, schema=history_schema)

    def _get_files_from_manifest(
        self, manifest_list: ManifestFile, data_file_filter: set[DataFileContent] | None = None
    ) -> pa.Table:
        import pyarrow as pa

        files: list[dict[str, Any]] = []
        schema = self.tbl.metadata.schema()
        io = self.tbl.io

        for manifest_entry in manifest_list.fetch_manifest_entry(io):
            data_file = manifest_entry.data_file
            if data_file_filter and data_file.content not in data_file_filter:
                continue
            column_sizes = data_file.column_sizes or {}
            value_counts = data_file.value_counts or {}
            null_value_counts = data_file.null_value_counts or {}
            nan_value_counts = data_file.nan_value_counts or {}
            lower_bounds = data_file.lower_bounds or {}
            upper_bounds = data_file.upper_bounds or {}
            readable_metrics = {
                schema.find_column_name(field.field_id): {
                    "column_size": column_sizes.get(field.field_id),
                    "value_count": value_counts.get(field.field_id),
                    "null_value_count": null_value_counts.get(field.field_id),
                    "nan_value_count": nan_value_counts.get(field.field_id),
                    "lower_bound": _readable_bound(field.field_type, lower_bounds.get(field.field_id)),
                    "upper_bound": _readable_bound(field.field_type, upper_bounds.get(field.field_id)),
                }
                for field in self.tbl.metadata.schema().fields
            }
            partition = data_file.partition
            partition_record_dict = {
                field.name: partition[pos]
                for pos, field in enumerate(self.tbl.metadata.specs()[manifest_list.partition_spec_id].fields)
            }
            files.append(
                {
                    "content": data_file.content,
                    "file_path": data_file.file_path,
                    "file_format": data_file.file_format,
                    "spec_id": data_file.spec_id,
                    "partition": partition_record_dict,
                    "record_count": data_file.record_count,
                    "file_size_in_bytes": data_file.file_size_in_bytes,
                    "column_sizes": dict(data_file.column_sizes) if data_file.column_sizes is not None else None,
                    "value_counts": dict(data_file.value_counts) if data_file.value_counts is not None else None,
                    "null_value_counts": dict(data_file.null_value_counts) if data_file.null_value_counts is not None else None,
                    "nan_value_counts": dict(data_file.nan_value_counts) if data_file.nan_value_counts is not None else None,
                    "lower_bounds": dict(data_file.lower_bounds) if data_file.lower_bounds is not None else None,
                    "upper_bounds": dict(data_file.upper_bounds) if data_file.upper_bounds is not None else None,
                    "key_metadata": data_file.key_metadata,
                    "split_offsets": data_file.split_offsets,
                    "equality_ids": data_file.equality_ids,
                    "sort_order_id": data_file.sort_order_id,
                    "readable_metrics": readable_metrics,
                }
            )
        return pa.Table.from_pylist(
            files,
            schema=self._get_files_schema(),
        )

    def _get_files_schema(self) -> pa.Schema:
        import pyarrow as pa

        from pyiceberg.io.pyarrow import schema_to_pyarrow

        schema = self.tbl.metadata.schema()
        readable_metrics_struct = []

        def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
            pa_bound_type = schema_to_pyarrow(bound_type)
            return pa.struct(
                [
                    pa.field("column_size", pa.int64(), nullable=True),
                    pa.field("value_count", pa.int64(), nullable=True),
                    pa.field("null_value_count", pa.int64(), nullable=True),
                    pa.field("nan_value_count", pa.int64(), nullable=True),
                    pa.field("lower_bound", pa_bound_type, nullable=True),
                    pa.field("upper_bound", pa_bound_type, nullable=True),
                ]
            )

        partition_record = self.tbl.metadata.specs_struct()
        pa_record_struct = schema_to_pyarrow(partition_record)

        for field in self.tbl.metadata.schema().fields:
            readable_metrics_struct.append(
                pa.field(schema.find_column_name(field.field_id), _readable_metrics_struct(field.field_type), nullable=False)
            )

        files_schema = pa.schema(
            [
                pa.field("content", pa.int8(), nullable=False),
                pa.field("file_path", pa.string(), nullable=False),
                pa.field("file_format", pa.dictionary(pa.int32(), pa.string()), nullable=False),
                pa.field("spec_id", pa.int32(), nullable=False),
                pa.field("partition", pa_record_struct, nullable=False),
                pa.field("record_count", pa.int64(), nullable=False),
                pa.field("file_size_in_bytes", pa.int64(), nullable=False),
                pa.field("column_sizes", pa.map_(pa.int32(), pa.int64()), nullable=True),
                pa.field("value_counts", pa.map_(pa.int32(), pa.int64()), nullable=True),
                pa.field("null_value_counts", pa.map_(pa.int32(), pa.int64()), nullable=True),
                pa.field("nan_value_counts", pa.map_(pa.int32(), pa.int64()), nullable=True),
                pa.field("lower_bounds", pa.map_(pa.int32(), pa.binary()), nullable=True),
                pa.field("upper_bounds", pa.map_(pa.int32(), pa.binary()), nullable=True),
                pa.field("key_metadata", pa.binary(), nullable=True),
                pa.field("split_offsets", pa.list_(pa.int64()), nullable=True),
                pa.field("equality_ids", pa.list_(pa.int32()), nullable=True),
                pa.field("sort_order_id", pa.int32(), nullable=True),
                pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
            ]
        )
        return files_schema

    def _files(self, snapshot_id: int | None = None, data_file_filter: set[DataFileContent] | None = None) -> pa.Table:
        import pyarrow as pa

        if not snapshot_id and not self.tbl.metadata.current_snapshot():
            return self._get_files_schema().empty_table()

        snapshot = self._get_snapshot(snapshot_id)
        io = self.tbl.io

        executor = ExecutorFactory.get_or_create()
        results = list(
            executor.map(
                lambda manifest_list: self._get_files_from_manifest(manifest_list, data_file_filter), snapshot.manifests(io)
            )
        )
        return pa.concat_tables(results)

    def files(self, snapshot_id: int | None = None) -> pa.Table:
        """Return all data and delete files tracked by a snapshot.

        Args:
            snapshot_id: The snapshot to inspect.  Defaults to the current
                snapshot when ``None``.

        Returns:
            A :class:`pyarrow.Table` with one row per file — see
            :meth:`_get_files_schema` for the full schema.  The ``content``
            column distinguishes data files (``0``) from position-delete files
            (``1``) and equality-delete files (``2``).

        Raises:
            ValueError: If *snapshot_id* does not exist or the table has no
                snapshots.

        Example:
            >>> tbl.inspect.files().to_pandas()
        """
        return self._files(snapshot_id)

    def data_files(self, snapshot_id: int | None = None) -> pa.Table:
        """Return only data files tracked by a snapshot.

        Convenience wrapper around :meth:`files` that filters to rows where
        ``content == 0`` (``DATA``).

        Args:
            snapshot_id: The snapshot to inspect.  Defaults to the current
                snapshot when ``None``.

        Returns:
            A :class:`pyarrow.Table` — same schema as :meth:`files`.

        Raises:
            ValueError: If *snapshot_id* does not exist or the table has no
                snapshots.

        Example:
            >>> tbl.inspect.data_files().to_pandas()
        """
        return self._files(snapshot_id, {DataFileContent.DATA})

    def delete_files(self, snapshot_id: int | None = None) -> pa.Table:
        """Return only delete files (position and equality) tracked by a snapshot.

        Convenience wrapper around :meth:`files` that filters to rows where
        ``content`` is ``1`` (``POSITION_DELETES``) or ``2``
        (``EQUALITY_DELETES``).

        Args:
            snapshot_id: The snapshot to inspect.  Defaults to the current
                snapshot when ``None``.

        Returns:
            A :class:`pyarrow.Table` — same schema as :meth:`files`.

        Raises:
            ValueError: If *snapshot_id* does not exist or the table has no
                snapshots.

        Example:
            >>> tbl.inspect.delete_files().to_pandas()
        """
        return self._files(snapshot_id, {DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES})

    def all_manifests(self) -> pa.Table:
        """Return manifest files across *all* snapshots of the table.

        Unlike :meth:`manifests`, which only covers the current snapshot, this
        method collects manifests from every snapshot and adds a
        ``reference_snapshot_id`` column so that each row can be traced back to
        the snapshot it belongs to.

        Returns:
            A :class:`pyarrow.Table` with the same columns as :meth:`manifests`
            plus ``reference_snapshot_id int64 not null``.  Returns an empty
            table when the table has no snapshots.

        Example:
            >>> tbl.inspect.all_manifests().to_pandas()
        """
        import pyarrow as pa

        snapshots = self.tbl.snapshots()
        if not snapshots:
            return pa.Table.from_pylist([], schema=self._get_all_manifests_schema())

        executor = ExecutorFactory.get_or_create()
        manifests_by_snapshots: Iterator[pa.Table] = executor.map(
            lambda args: self._generate_manifests_table(*args), [(snapshot, True) for snapshot in snapshots]
        )
        return pa.concat_tables(manifests_by_snapshots)

    def _all_files(self, data_file_filter: set[DataFileContent] | None = None) -> pa.Table:
        import pyarrow as pa

        snapshots = self.tbl.snapshots()
        if not snapshots:
            return pa.Table.from_pylist([], schema=self._get_files_schema())

        executor = ExecutorFactory.get_or_create()
        manifest_lists = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io), snapshots)

        unique_manifests = {(manifest.manifest_path, manifest) for manifest_list in manifest_lists for manifest in manifest_list}

        file_lists = executor.map(
            lambda args: self._get_files_from_manifest(*args), [(manifest, data_file_filter) for _, manifest in unique_manifests]
        )

        return pa.concat_tables(file_lists)

    def all_files(self) -> pa.Table:
        """Return all files (data and delete) across *all* snapshots.

        De-duplicates manifests by path so that files shared between snapshots
        are only emitted once.  Use :meth:`files` when you only need the
        current snapshot.

        Returns:
            A :class:`pyarrow.Table` — same schema as :meth:`files`.  Returns
            an empty table when the table has no snapshots.

        Example:
            >>> tbl.inspect.all_files().to_pandas()
        """
        return self._all_files()

    def all_data_files(self) -> pa.Table:
        """Return all data files across *all* snapshots.

        Convenience wrapper around :meth:`all_files` that filters to data files
        only (``content == 0``).

        Returns:
            A :class:`pyarrow.Table` — same schema as :meth:`files`.

        Example:
            >>> tbl.inspect.all_data_files().to_pandas()
        """
        return self._all_files({DataFileContent.DATA})

    def all_delete_files(self) -> pa.Table:
        """Return all delete files (position and equality) across *all* snapshots.

        Convenience wrapper around :meth:`all_files` that filters to delete
        files only (``content`` is ``1`` or ``2``).

        Returns:
            A :class:`pyarrow.Table` — same schema as :meth:`files`.

        Example:
            >>> tbl.inspect.all_delete_files().to_pandas()
        """
        return self._all_files({DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES})