| # 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: | |
| """Provides metadata inspection for Apache Iceberg tables. | |
| Exposes table metadata (snapshots, manifests, partitions, files, | |
| schema history) as PyArrow Tables, mirroring the metadata tables | |
| available in the Java Iceberg implementation. | |
| Args: | |
| tbl: The Iceberg table to inspect. | |
| Raises: | |
| ModuleNotFoundError: If PyArrow is not installed. | |
| """ | |
| tbl: Table | |
| def __init__(self, tbl: Table) -> None: | |
| """Initialize InspectTable with the given Iceberg table. | |
| Args: | |
| tbl: The Iceberg table instance to inspect. | |
| 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: | |
| """Retrieve a snapshot by ID, or return the current snapshot. | |
| Args: | |
| snapshot_id: The snapshot ID to look up. If None, the current | |
| snapshot is returned. | |
| Returns: | |
| The requested Snapshot. | |
| Raises: | |
| ValueError: If the snapshot ID is not found, or if no current | |
| snapshot exists when snapshot_id is None. | |
| """ | |
| 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 _get_snapshots_by_id(self) -> dict[int, Snapshot]: | |
| """Index the snapshots by ID.""" | |
| return {snapshot.snapshot_id: snapshot for snapshot in self.tbl.metadata.snapshots} | |
| def snapshots(self) -> pa.Table: | |
| """Return all snapshots of the table as a PyArrow Table. | |
| Returns: | |
| pa.Table: A table with columns: ``committed_at`` (timestamp[ms]), | |
| ``snapshot_id`` (int64), ``parent_id`` (int64, nullable), | |
| ``operation`` (string, nullable), ``manifest_list`` (string), | |
| and ``summary`` (map<string, string>, nullable). | |
| """ | |
| 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 is one manifest entry (data or delete file), including | |
| raw and human-readable column-level statistics. | |
| Args: | |
| snapshot_id: The snapshot to inspect. Defaults to the current snapshot. | |
| Returns: | |
| pa.Table: A table with columns: ``status``, ``snapshot_id``, | |
| ``sequence_number``, ``file_sequence_number``, ``data_file`` | |
| (struct), and ``readable_metrics`` (struct). | |
| Raises: | |
| ValueError: If the specified snapshot does not exist, or if the | |
| table has no snapshots and snapshot_id is None. | |
| """ | |
| 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), | |
| ] | |
| ) | |
| 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 snapshot references (branches and tags) as a PyArrow Table. | |
| Returns: | |
| pa.Table: A table with columns: ``name`` (string), ``type`` | |
| (dictionary<int32, string>), ``snapshot_id`` (int64), | |
| ``max_reference_age_in_ms`` (int64, nullable), | |
| ``min_snapshots_to_keep`` (int32, nullable), and | |
| ``max_snapshot_age_in_ms`` (int64, nullable). | |
| """ | |
| 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 as a PyArrow Table. | |
| Each row represents a unique partition and aggregates statistics | |
| across all data and delete files in that partition. For unpartitioned | |
| tables, a single row covers the whole table. | |
| Args: | |
| snapshot_id: The snapshot to inspect. Defaults to the current snapshot. | |
| row_filter: A filter to limit which partitions are included. | |
| Accepts a SQL-style string or a ``BooleanExpression``. | |
| case_sensitive: Whether column name matching is case-sensitive. | |
| Returns: | |
| pa.Table: A table with columns: ``partition`` (struct, for | |
| partitioned tables), ``spec_id`` (int32, for partitioned | |
| tables), ``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 the specified snapshot does not exist, or if the | |
| table has no snapshots and snapshot_id is None. | |
| """ | |
| import pyarrow as pa | |
| from pyiceberg.io.pyarrow import schema_to_pyarrow | |
| from pyiceberg.table import DataScan | |
| 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] = {} | |
| snapshots_by_id = self._get_snapshots_by_id() | |
| for entry in itertools.chain.from_iterable(scan._plan_manifest_entries()): | |
| 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 = snapshots_by_id.get(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: | |
| """Update the partition statistics map with data from a single manifest entry. | |
| Initialises a new partition row if the key has not been seen, then | |
| increments count and size statistics based on the file content type. | |
| Args: | |
| partitions_map: Mutable dict keyed by hashable partition values, | |
| accumulating per-partition statistics. | |
| file: The data file entry from the manifest. | |
| partition_record_dict: Maps partition field names to their values. | |
| snapshot: The owning snapshot, used to track last-updated timestamps. | |
| """ | |
| 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_at"] < 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: | |
| """Return the PyArrow schema for the manifests metadata table. | |
| Returns: | |
| pa.Schema: Schema with fields: ``content``, ``path``, ``length``, | |
| ``partition_spec_id``, ``added_snapshot_id``, | |
| ``added_data_files_count``, ``existing_data_files_count``, | |
| ``deleted_data_files_count``, ``added_delete_files_count``, | |
| ``existing_delete_files_count``, ``deleted_delete_files_count``, | |
| and ``partition_summaries``. | |
| """ | |
| 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: | |
| """Return the PyArrow schema for the all_manifests metadata table. | |
| Extends the manifests schema with a ``reference_snapshot_id`` column | |
| that identifies which snapshot each manifest was retrieved from, and a | |
| ``key_metadata`` column holding the manifest's encryption key metadata. | |
| Returns: | |
| pa.Schema: The manifests schema plus ``reference_snapshot_id`` | |
| (int64) and ``key_metadata`` (binary, nullable) fields. | |
| """ | |
| 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)) | |
| all_manifests_schema = all_manifests_schema.append(pa.field("key_metadata", pa.binary(), nullable=True)) | |
| return all_manifests_schema | |
| def _generate_manifests_table(self, snapshot: Snapshot | None, is_all_manifests_table: bool = False) -> pa.Table: | |
| """Build a manifests PyArrow Table for the given snapshot. | |
| Args: | |
| snapshot: The snapshot whose manifests to include. If None, an | |
| empty table is returned. | |
| is_all_manifests_table: If True, appends ``reference_snapshot_id`` | |
| and ``key_metadata`` columns and uses the all-manifests schema. | |
| Defaults to False. | |
| Returns: | |
| pa.Table: One row per manifest file in the snapshot. | |
| """ | |
| 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 is not None | |
| 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 is not None | |
| 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 | |
| manifest_row["key_metadata"] = manifest.key_metadata | |
| 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 as a PyArrow Table. | |
| Returns: | |
| pa.Table: Manifest metadata for the current snapshot, or an empty | |
| table if the table has no snapshots. | |
| See ``_get_manifests_schema`` for the full column list. | |
| """ | |
| return self._generate_manifests_table(self.tbl.current_snapshot()) | |
| def metadata_log_entries(self) -> pa.Table: | |
| """Return the metadata log of the table as a PyArrow Table. | |
| Each row corresponds to a metadata file that was previously current, | |
| plus the current metadata file appended as the final row. | |
| Returns: | |
| pa.Table: A table with columns: ``timestamp`` (timestamp[ms]), | |
| ``file`` (string), ``latest_snapshot_id`` (int64, nullable), | |
| ``latest_schema_id`` (int32, nullable), and | |
| ``latest_sequence_number`` (int64, nullable). | |
| """ | |
| import pyarrow as pa | |
| from pyiceberg.table.snapshots import MetadataLogEntry | |
| 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 as a PyArrow Table. | |
| Each row is one snapshot-log entry, showing when the snapshot became | |
| current and whether it is an ancestor of the current snapshot. | |
| Returns: | |
| pa.Table: A table with columns: ``made_current_at`` (timestamp[ms]), | |
| ``snapshot_id`` (int64), ``parent_id`` (int64, nullable), and | |
| ``is_current_ancestor`` (bool). | |
| """ | |
| 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 | |
| snapshots_by_id = self._get_snapshots_by_id() | |
| for snapshot_entry in metadata.snapshot_log: | |
| snapshot = snapshots_by_id.get(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: | |
| """Read file-level metadata entries from a single manifest file. | |
| Args: | |
| manifest_list: The manifest file to read entries from. | |
| data_file_filter: If provided, only files whose ``DataFileContent`` | |
| is in this set are included. If None, all file types are returned. | |
| Returns: | |
| pa.Table: File metadata rows conforming to the files schema. | |
| See ``_get_files_schema`` for the full column list. | |
| """ | |
| 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: | |
| """Return the PyArrow schema for file-level metadata tables. | |
| The schema is table-specific because the ``readable_metrics`` struct | |
| is derived from the current table schema's field names and types. | |
| Returns: | |
| pa.Schema: Schema with fields: ``content``, ``file_path``, | |
| ``file_format``, ``spec_id``, ``partition``, ``record_count``, | |
| ``file_size_in_bytes``, ``column_sizes``, ``value_counts``, | |
| ``null_value_counts``, ``nan_value_counts``, ``lower_bounds``, | |
| ``upper_bounds``, ``key_metadata``, ``split_offsets``, | |
| ``equality_ids``, ``sort_order_id``, and ``readable_metrics``. | |
| """ | |
| 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: | |
| """Return file metadata for a snapshot, optionally filtered by content type. | |
| Reads all manifests in the snapshot concurrently using the configured executor. | |
| Args: | |
| snapshot_id: The snapshot to inspect. Defaults to the current snapshot. | |
| Returns an empty table if no current snapshot exists. | |
| data_file_filter: If provided, only files whose ``DataFileContent`` | |
| is in this set are returned. If None, all file types are returned. | |
| Returns: | |
| pa.Table: Concatenated file metadata from all manifests in the snapshot. | |
| """ | |
| 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 data and delete files for a snapshot as a PyArrow Table. | |
| Args: | |
| snapshot_id: The snapshot to inspect. Defaults to the current snapshot. | |
| Returns: | |
| pa.Table: File metadata for all content types. | |
| See ``_get_files_schema`` for the full column list. | |
| """ | |
| return self._files(snapshot_id) | |
| def data_files(self, snapshot_id: int | None = None) -> pa.Table: | |
| """Return only data files for a snapshot as a PyArrow Table. | |
| Args: | |
| snapshot_id: The snapshot to inspect. Defaults to the current snapshot. | |
| Returns: | |
| pa.Table: File metadata for data files only. | |
| See ``_get_files_schema`` for the full column list. | |
| """ | |
| return self._files(snapshot_id, {DataFileContent.DATA}) | |
| def delete_files(self, snapshot_id: int | None = None) -> pa.Table: | |
| """Return only delete files for a snapshot as a PyArrow Table. | |
| Includes both position delete and equality delete files. | |
| Args: | |
| snapshot_id: The snapshot to inspect. Defaults to the current snapshot. | |
| Returns: | |
| pa.Table: File metadata for position and equality delete files. | |
| See ``_get_files_schema`` for the full column list. | |
| """ | |
| return self._files(snapshot_id, {DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES}) | |
| def all_manifests(self) -> pa.Table: | |
| """Return manifests across all snapshots as a PyArrow Table. | |
| Includes a ``reference_snapshot_id`` column to identify which snapshot | |
| each manifest belongs to, and a ``key_metadata`` column. Manifests | |
| shared across snapshots may appear more than once. | |
| Returns: | |
| pa.Table: Manifest metadata across all snapshots, or an empty table | |
| if the table has no snapshots. | |
| See ``_get_all_manifests_schema`` for the full column list. | |
| """ | |
| 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: | |
| """Return file metadata across all snapshots, optionally filtered by content type. | |
| Deduplicates manifests by path so each unique manifest is read only once, | |
| regardless of how many snapshots reference it. | |
| Args: | |
| data_file_filter: If provided, only files whose ``DataFileContent`` | |
| is in this set are returned. If None, all file types are returned. | |
| Returns: | |
| pa.Table: File metadata from all unique manifests across all snapshots, | |
| or an empty table if the table has no snapshots. | |
| """ | |
| 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 data and delete files across all snapshots as a PyArrow Table. | |
| Returns: | |
| pa.Table: File metadata for all content types across all snapshots. | |
| See ``_get_files_schema`` for the full column list. | |
| """ | |
| return self._all_files() | |
| def all_data_files(self) -> pa.Table: | |
| """Return all data files across all snapshots as a PyArrow Table. | |
| Returns: | |
| pa.Table: File metadata for data files only, across all snapshots. | |
| See ``_get_files_schema`` for the full column list. | |
| """ | |
| return self._all_files({DataFileContent.DATA}) | |
| def all_delete_files(self) -> pa.Table: | |
| """Return all delete files across all snapshots as a PyArrow Table. | |
| Includes both position delete and equality delete files. | |
| Returns: | |
| pa.Table: File metadata for position and equality delete files, | |
| across all snapshots. | |
| See ``_get_files_schema`` for the full column list. | |
| """ | |
| return self._all_files({DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES}) |