You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "amitgilad3 (via GitHub)" <gi...@apache.org> on 2025/02/04 17:10:59 UTC
[PR] support all_entries in pyiceberg [iceberg-python]
amitgilad3 opened a new pull request, #1608:
URL: https://github.com/apache/iceberg-python/pull/1608
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#discussion_r2173826903
##########
pyiceberg/table/inspect.py:
##########
@@ -704,3 +731,30 @@ def all_data_files(self) -> "pa.Table":
def all_delete_files(self) -> "pa.Table":
return self._all_files({DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES})
+
+ def all_entries(self) -> "pa.Table":
+ import pyarrow as pa
+
+ snapshots = self.tbl.snapshots()
+ if not snapshots:
+ return pa.Table.from_pylist([], self._get_entries_schema())
+
+ schemas = self.tbl.schemas()
+ snapshot_schemas: Dict[int, "pa.Schema"] = {}
+ for snapshot in snapshots:
+ if snapshot.schema_id is None:
+ raise ValueError(f"Cannot find schema_id for snapshot: {snapshot.snapshot_id}")
+ else:
+ snapshot_schemas[snapshot.snapshot_id] = schemas.get(snapshot.schema_id)
+
+ executor = ExecutorFactory.get_or_create()
+ all_manifests: Iterator[List[ManifestFile]] = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io), snapshots)
+ unique_flattened_manifests = list(
+ {manifest.manifest_path: manifest for manifest_list in all_manifests for manifest in manifest_list}.values()
+ )
+
Review Comment:
can we use the `all_manifests` method of the `InspectTable` class instead of "re-defining" it to get the `unique_flattened_manifests` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on code in PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#discussion_r1947916711
##########
tests/integration/test_inspect_table.py:
##########
@@ -938,3 +938,111 @@ def test_inspect_all_manifests(spark: SparkSession, session_catalog: Catalog, fo
lhs = spark.table(f"{identifier}.all_manifests").toPandas()
rhs = df.to_pandas()
assert_frame_equal(lhs, rhs, check_dtype=False)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_inspect_all_entries(spark: SparkSession, session_catalog: Catalog, format_version: int) -> None:
+ identifier = "default.table_metadata_all_entries"
+ try:
+ session_catalog.drop_table(identifier=identifier)
+ except NoSuchTableError:
+ pass
+
+ spark.sql(
+ f"""
+ CREATE TABLE {identifier} (
+ id int,
+ data string
+ )
+ PARTITIONED BY (data)
+ TBLPROPERTIES ('write.update.mode'='merge-on-read',
+ 'write.delete.mode'='merge-on-read')
+ """
+ )
+ tbl = session_catalog.load_table(identifier)
+
+ spark.sql(f"INSERT INTO {identifier} VALUES (1, 'a')")
+ spark.sql(f"INSERT INTO {identifier} VALUES (2, 'b')")
+
+ spark.sql(f"UPDATE {identifier} SET data = 'c' WHERE id = 1")
+
+ spark.sql(f"DELETE FROM {identifier} WHERE id = 2")
+
+ spark.sql(f"INSERT OVERWRITE {identifier} VALUES (1, 'a')")
+
+ def check_pyiceberg_df_equals_spark_df(df: pa.Table, spark_df: DataFrame) -> None:
+ assert df.column_names == [
+ "status",
+ "snapshot_id",
+ "sequence_number",
+ "file_sequence_number",
+ "data_file",
+ "readable_metrics",
+ ]
+
+ # Make sure that they are filled properly
Review Comment:
nit: add a comment to signal that this checks the first 4 columns. and the rest of the tests check for the last 2 `data_file` and `readable_metrics`
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,92 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+ def _get_entries(self, manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
+ import pyarrow as pa
+
+ schema = self.tbl.metadata.schema()
Review Comment:
something im concerned about with the `all_*` metadata tables is taking into account the table schema/partition evolution.
since we're looking across all snapshots, it might not be right to use the current snapshot here
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,92 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+ def _get_entries(self, manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
+ import pyarrow as pa
+
+ schema = self.tbl.metadata.schema()
+ entries_schema = self._get_entries_schema()
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):
- 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": from_bytes(field.field_type, lower_bound)
- if (lower_bound := lower_bounds.get(field.field_id))
- else None,
- "upper_bound": from_bytes(field.field_type, upper_bound)
- if (upper_bound := upper_bounds.get(field.field_id))
- else None,
- }
- for field in self.tbl.metadata.schema().fields
+ for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=discard_deleted):
+ 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": from_bytes(field.field_type, lower_bound)
+ if (lower_bound := lower_bounds.get(field.field_id))
+ else None,
+ "upper_bound": from_bytes(field.field_type, upper_bound)
+ if (upper_bound := upper_bounds.get(field.field_id))
+ else None,
}
+ 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)
- }
+ 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),
- "null_value_counts": dict(entry.data_file.null_value_counts),
- "nan_value_counts": dict(entry.data_file.nan_value_counts),
- "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,
- }
- )
+ 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) if entry.data_file.column_sizes is not None else None,
Review Comment:
how about
```
"column_sizes": dict(entry.data_file.column_sizes) or None
```
##########
tests/integration/test_inspect_table.py:
##########
@@ -938,3 +938,111 @@ def test_inspect_all_manifests(spark: SparkSession, session_catalog: Catalog, fo
lhs = spark.table(f"{identifier}.all_manifests").toPandas()
rhs = df.to_pandas()
assert_frame_equal(lhs, rhs, check_dtype=False)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_inspect_all_entries(spark: SparkSession, session_catalog: Catalog, format_version: int) -> None:
+ identifier = "default.table_metadata_all_entries"
+ try:
+ session_catalog.drop_table(identifier=identifier)
+ except NoSuchTableError:
+ pass
+
+ spark.sql(
+ f"""
+ CREATE TABLE {identifier} (
+ id int,
+ data string
+ )
+ PARTITIONED BY (data)
+ TBLPROPERTIES ('write.update.mode'='merge-on-read',
+ 'write.delete.mode'='merge-on-read')
+ """
+ )
+ tbl = session_catalog.load_table(identifier)
+
+ spark.sql(f"INSERT INTO {identifier} VALUES (1, 'a')")
+ spark.sql(f"INSERT INTO {identifier} VALUES (2, 'b')")
+
+ spark.sql(f"UPDATE {identifier} SET data = 'c' WHERE id = 1")
+
+ spark.sql(f"DELETE FROM {identifier} WHERE id = 2")
+
+ spark.sql(f"INSERT OVERWRITE {identifier} VALUES (1, 'a')")
+
+ def check_pyiceberg_df_equals_spark_df(df: pa.Table, spark_df: DataFrame) -> None:
+ assert df.column_names == [
+ "status",
+ "snapshot_id",
+ "sequence_number",
+ "file_sequence_number",
+ "data_file",
+ "readable_metrics",
+ ]
+
+ # Make sure that they are filled properly
+ for int_column in ["status", "snapshot_id", "sequence_number", "file_sequence_number"]:
+ for value in df[int_column]:
+ assert isinstance(value.as_py(), int)
+
+ for snapshot_id in df["snapshot_id"]:
+ assert isinstance(snapshot_id.as_py(), int)
+
+ lhs = df.to_pandas()
+ lhs["content"] = lhs["data_file"].apply(lambda x: x.get("content"))
+ lhs["file_path"] = lhs["data_file"].apply(lambda x: x.get("file_path"))
+ lhs = lhs.sort_values(["status", "snapshot_id", "sequence_number", "content", "file_path"]).drop(
+ columns=["file_path", "content"]
+ )
Review Comment:
nit: maybe inline these operation to show that the same operations are performed on lhs and rhs
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,92 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+ def _get_entries(self, manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
+ import pyarrow as pa
+
+ schema = self.tbl.metadata.schema()
+ entries_schema = self._get_entries_schema()
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):
- 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": from_bytes(field.field_type, lower_bound)
- if (lower_bound := lower_bounds.get(field.field_id))
- else None,
- "upper_bound": from_bytes(field.field_type, upper_bound)
- if (upper_bound := upper_bounds.get(field.field_id))
- else None,
- }
- for field in self.tbl.metadata.schema().fields
+ for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=discard_deleted):
+ 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": from_bytes(field.field_type, lower_bound)
+ if (lower_bound := lower_bounds.get(field.field_id))
+ else None,
+ "upper_bound": from_bytes(field.field_type, upper_bound)
+ if (upper_bound := upper_bounds.get(field.field_id))
+ else None,
}
+ 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)
- }
+ 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),
- "null_value_counts": dict(entry.data_file.null_value_counts),
- "nan_value_counts": dict(entry.data_file.nan_value_counts),
- "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,
- }
- )
+ 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) if entry.data_file.column_sizes is not None else None,
+ "value_counts": dict(entry.data_file.value_counts) if entry.data_file.value_counts is not None else None,
+ "null_value_counts": dict(entry.data_file.null_value_counts)
+ if entry.data_file.null_value_counts is not None
+ else None,
+ "nan_value_counts": dict(entry.data_file.nan_value_counts)
+ if entry.data_file.nan_value_counts is not None
+ else None,
+ "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 entries(self, snapshot_id: Optional[int] = None, discard_deleted: bool = True) -> "pa.Table":
Review Comment:
what is `discard_deleted` used for?
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,92 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+ def _get_entries(self, manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
+ import pyarrow as pa
+
+ schema = self.tbl.metadata.schema()
+ entries_schema = self._get_entries_schema()
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):
- 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": from_bytes(field.field_type, lower_bound)
- if (lower_bound := lower_bounds.get(field.field_id))
- else None,
- "upper_bound": from_bytes(field.field_type, upper_bound)
- if (upper_bound := upper_bounds.get(field.field_id))
- else None,
- }
- for field in self.tbl.metadata.schema().fields
+ for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=discard_deleted):
+ 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": from_bytes(field.field_type, lower_bound)
+ if (lower_bound := lower_bounds.get(field.field_id))
+ else None,
+ "upper_bound": from_bytes(field.field_type, upper_bound)
+ if (upper_bound := upper_bounds.get(field.field_id))
+ else None,
}
+ 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)
- }
+ 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),
- "null_value_counts": dict(entry.data_file.null_value_counts),
- "nan_value_counts": dict(entry.data_file.nan_value_counts),
- "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,
- }
- )
+ 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) if entry.data_file.column_sizes is not None else None,
+ "value_counts": dict(entry.data_file.value_counts) if entry.data_file.value_counts is not None else None,
+ "null_value_counts": dict(entry.data_file.null_value_counts)
+ if entry.data_file.null_value_counts is not None
+ else None,
+ "nan_value_counts": dict(entry.data_file.nan_value_counts)
+ if entry.data_file.nan_value_counts is not None
+ else None,
+ "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 entries(self, snapshot_id: Optional[int] = None, discard_deleted: bool = True) -> "pa.Table":
Review Comment:
`entries` metadata table should show both
https://iceberg.apache.org/docs/nightly/spark-queries/#entries
```
To show all the table's current manifest entries for both data and delete files.
```
also `discard_deleted==True` is essentially getting just the data files, which we can do with the files metadata table
##########
tests/integration/test_inspect_table.py:
##########
@@ -938,3 +938,111 @@ def test_inspect_all_manifests(spark: SparkSession, session_catalog: Catalog, fo
lhs = spark.table(f"{identifier}.all_manifests").toPandas()
rhs = df.to_pandas()
assert_frame_equal(lhs, rhs, check_dtype=False)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_inspect_all_entries(spark: SparkSession, session_catalog: Catalog, format_version: int) -> None:
+ identifier = "default.table_metadata_all_entries"
+ try:
+ session_catalog.drop_table(identifier=identifier)
+ except NoSuchTableError:
+ pass
+
+ spark.sql(
+ f"""
+ CREATE TABLE {identifier} (
+ id int,
+ data string
+ )
+ PARTITIONED BY (data)
+ TBLPROPERTIES ('write.update.mode'='merge-on-read',
+ 'write.delete.mode'='merge-on-read')
+ """
+ )
+ tbl = session_catalog.load_table(identifier)
+
+ spark.sql(f"INSERT INTO {identifier} VALUES (1, 'a')")
+ spark.sql(f"INSERT INTO {identifier} VALUES (2, 'b')")
+
+ spark.sql(f"UPDATE {identifier} SET data = 'c' WHERE id = 1")
+
+ spark.sql(f"DELETE FROM {identifier} WHERE id = 2")
+
+ spark.sql(f"INSERT OVERWRITE {identifier} VALUES (1, 'a')")
+
+ def check_pyiceberg_df_equals_spark_df(df: pa.Table, spark_df: DataFrame) -> None:
+ assert df.column_names == [
+ "status",
+ "snapshot_id",
+ "sequence_number",
+ "file_sequence_number",
+ "data_file",
+ "readable_metrics",
+ ]
+
+ # Make sure that they are filled properly
+ for int_column in ["status", "snapshot_id", "sequence_number", "file_sequence_number"]:
+ for value in df[int_column]:
+ assert isinstance(value.as_py(), int)
+
+ for snapshot_id in df["snapshot_id"]:
+ assert isinstance(snapshot_id.as_py(), int)
Review Comment:
this is redundant, right? already included in the above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "amitgilad3 (via GitHub)" <gi...@apache.org>.
amitgilad3 commented on PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#issuecomment-2634621892
will try
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "amitgilad3 (via GitHub)" <gi...@apache.org>.
amitgilad3 commented on PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#issuecomment-2638045119
Ready for review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#discussion_r2173826547
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +160,98 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+
+ def _get_entries(self, schema: Schema, manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
+ import pyarrow as pa
+ entries_schema = self._get_entries_schema(schema=schema)
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": from_bytes(field.field_type, lower_bound)
- if (lower_bound := lower_bounds.get(field.field_id))
- else None,
- "upper_bound": from_bytes(field.field_type, upper_bound)
- if (upper_bound := upper_bounds.get(field.field_id))
- else None,
- }
- for field in self.tbl.metadata.schema().fields
+ for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=discard_deleted):
+ 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": from_bytes(field.field_type, lower_bound)
+ if (lower_bound := lower_bounds.get(field.field_id))
+ else None,
+ "upper_bound": from_bytes(field.field_type, upper_bound)
+ if (upper_bound := upper_bounds.get(field.field_id))
+ else None,
}
+ for field in 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)
- }
+ 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,
- }
- )
+ 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) if entry.data_file.column_sizes is not None else None,
+ "value_counts": dict(entry.data_file.value_counts) if entry.data_file.value_counts is not None else None,
+ "null_value_counts": dict(entry.data_file.null_value_counts)
+ if entry.data_file.null_value_counts is not None
+ else None,
+ "nan_value_counts": dict(entry.data_file.nan_value_counts)
+ if entry.data_file.nan_value_counts is not None
+ else None,
+ "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 entries(self, snapshot_id: Optional[int] = None) -> "pa.Table":
+ import pyarrow as pa
+
+ entries = []
+ snapshot = self._get_snapshot(snapshot_id)
+
+ if snapshot.schema_id is None:
+ raise ValueError(f"Cannot find schema_id for snapshot {snapshot.snapshot_id}")
+
+ schema = self.tbl.schemas().get(snapshot.schema_id)
+ if not schema:
+ raise ValueError(f"Cannot find schema with ID {snapshot.schema_id}")
+ for manifest in snapshot.manifests(self.tbl.io):
+ manifest_entries = self._get_entries(schema=schema, manifest=manifest, discard_deleted=True)
+ entries.append(manifest_entries)
+ return pa.concat_tables(entries)
Review Comment:
I think its safe to use an executor for parallelism in this for loop right? `entries` isnt used anywhere else
##########
pyiceberg/table/inspect.py:
##########
@@ -704,3 +731,30 @@ def all_data_files(self) -> "pa.Table":
def all_delete_files(self) -> "pa.Table":
return self._all_files({DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES})
+
+ def all_entries(self) -> "pa.Table":
+ import pyarrow as pa
+
+ snapshots = self.tbl.snapshots()
+ if not snapshots:
+ return pa.Table.from_pylist([], self._get_entries_schema())
+
+ schemas = self.tbl.schemas()
+ snapshot_schemas: Dict[int, "pa.Schema"] = {}
+ for snapshot in snapshots:
+ if snapshot.schema_id is None:
+ raise ValueError(f"Cannot find schema_id for snapshot: {snapshot.snapshot_id}")
+ else:
+ snapshot_schemas[snapshot.snapshot_id] = schemas.get(snapshot.schema_id)
+
+ executor = ExecutorFactory.get_or_create()
+ all_manifests: Iterator[List[ManifestFile]] = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io), snapshots)
+ unique_flattened_manifests = list(
+ {manifest.manifest_path: manifest for manifest_list in all_manifests for manifest in manifest_list}.values()
+ )
+
Review Comment:
can we use the `all_manifests` method of the `InspectTable` class instead of doing it again to get the `unique_flattened_manifests` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "amitgilad3 (via GitHub)" <gi...@apache.org>.
amitgilad3 commented on code in PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#discussion_r1958750651
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,96 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+
+ def _get_entries(self, schema: "pa.Schema", manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
+ import pyarrow as pa
+ entries_schema = self._get_entries_schema()
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):
- 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": from_bytes(field.field_type, lower_bound)
- if (lower_bound := lower_bounds.get(field.field_id))
- else None,
- "upper_bound": from_bytes(field.field_type, upper_bound)
- if (upper_bound := upper_bounds.get(field.field_id))
- else None,
- }
- for field in self.tbl.metadata.schema().fields
+ for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=discard_deleted):
+ 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": from_bytes(field.field_type, lower_bound)
+ if (lower_bound := lower_bounds.get(field.field_id))
+ else None,
+ "upper_bound": from_bytes(field.field_type, upper_bound)
+ if (upper_bound := upper_bounds.get(field.field_id))
+ else None,
}
+ 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)
- }
+ 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),
- "null_value_counts": dict(entry.data_file.null_value_counts),
- "nan_value_counts": dict(entry.data_file.nan_value_counts),
- "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,
- }
- )
+ 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) or None,
+ "value_counts": dict(entry.data_file.value_counts) if entry.data_file.value_counts is not None else None,
+ "null_value_counts": dict(entry.data_file.null_value_counts)
+ if entry.data_file.null_value_counts is not None
+ else None,
+ "nan_value_counts": dict(entry.data_file.nan_value_counts)
+ if entry.data_file.nan_value_counts is not None
+ else None,
Review Comment:
i used the same logic you used in lines L645 - L650 , if i use dict() or None then the code breaks when trying todo dict(None)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#issuecomment-4122324342
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#issuecomment-4071509633
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that's incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #1608: support all_entries in pyiceberg
URL: https://github.com/apache/iceberg-python/pull/1608
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "amitgilad3 (via GitHub)" <gi...@apache.org>.
amitgilad3 commented on PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#issuecomment-2648831468
Ready for review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on code in PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#discussion_r1957418226
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,96 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+
+ def _get_entries(self, schema: "pa.Schema", manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
Review Comment:
schema here should be iceberg schema instead of pa.schema
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,96 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+
+ def _get_entries(self, schema: "pa.Schema", manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
Review Comment:
there are still a couple places in this function using `self.tbl.` which is the table's current metadata
L189 & L195.
We should use the schema and partition spec at the time of the snapshot instead
##########
mkdocs/docs/api.md:
##########
@@ -716,6 +716,8 @@ readable_metrics: [
[6.0989]]
```
+To show all the table's current manifest entries for both data and delete files, use `table.inspect.all_entries()`.
Review Comment:
i'd be great to add this to its own section, similar to https://iceberg.apache.org/docs/nightly/spark-queries/#all-metadata-tables
##########
pyiceberg/table/inspect.py:
##########
@@ -157,74 +158,96 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
pa.field("readable_metrics", pa.struct(readable_metrics_struct), nullable=True),
]
)
+ return entries_schema
+
+ def _get_entries(self, schema: "pa.Schema", manifest: ManifestFile, discard_deleted: bool = True) -> "pa.Table":
+ import pyarrow as pa
+ entries_schema = self._get_entries_schema()
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):
- 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": from_bytes(field.field_type, lower_bound)
- if (lower_bound := lower_bounds.get(field.field_id))
- else None,
- "upper_bound": from_bytes(field.field_type, upper_bound)
- if (upper_bound := upper_bounds.get(field.field_id))
- else None,
- }
- for field in self.tbl.metadata.schema().fields
+ for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=discard_deleted):
+ 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": from_bytes(field.field_type, lower_bound)
+ if (lower_bound := lower_bounds.get(field.field_id))
+ else None,
+ "upper_bound": from_bytes(field.field_type, upper_bound)
+ if (upper_bound := upper_bounds.get(field.field_id))
+ else None,
}
+ 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)
- }
+ 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),
- "null_value_counts": dict(entry.data_file.null_value_counts),
- "nan_value_counts": dict(entry.data_file.nan_value_counts),
- "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,
- }
- )
+ 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) or None,
+ "value_counts": dict(entry.data_file.value_counts) if entry.data_file.value_counts is not None else None,
+ "null_value_counts": dict(entry.data_file.null_value_counts)
+ if entry.data_file.null_value_counts is not None
+ else None,
+ "nan_value_counts": dict(entry.data_file.nan_value_counts)
+ if entry.data_file.nan_value_counts is not None
+ else None,
Review Comment:
nit: can we use the same dict() or None pattern here
##########
pyiceberg/table/inspect.py:
##########
@@ -94,7 +94,7 @@ def snapshots(self) -> "pa.Table":
schema=snapshots_schema,
)
- def entries(self, snapshot_id: Optional[int] = None) -> "pa.Table":
+ def _get_entries_schema(self) -> "pa.Schema":
Review Comment:
same problem here using the table's current metadata (`self.tbl.`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#discussion_r2173968530
##########
pyiceberg/table/inspect.py:
##########
@@ -704,3 +731,30 @@ def all_data_files(self) -> "pa.Table":
def all_delete_files(self) -> "pa.Table":
return self._all_files({DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES})
+
+ def all_entries(self) -> "pa.Table":
+ import pyarrow as pa
+
+ snapshots = self.tbl.snapshots()
+ if not snapshots:
+ return pa.Table.from_pylist([], self._get_entries_schema())
+
+ schemas = self.tbl.schemas()
+ snapshot_schemas: Dict[int, "pa.Schema"] = {}
+ for snapshot in snapshots:
+ if snapshot.schema_id is None:
+ raise ValueError(f"Cannot find schema_id for snapshot: {snapshot.snapshot_id}")
+ else:
+ snapshot_schemas[snapshot.snapshot_id] = schemas.get(snapshot.schema_id)
+
+ executor = ExecutorFactory.get_or_create()
+ all_manifests: Iterator[List[ManifestFile]] = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io), snapshots)
+ unique_flattened_manifests = list(
+ {manifest.manifest_path: manifest for manifest_list in all_manifests for manifest in manifest_list}.values()
+ )
+
Review Comment:
I guess yeah but will probably be slower...would need to do `_generate_manifests_table` for each snapshot and then concat all the `path` fields
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on code in PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#discussion_r2173968530
##########
pyiceberg/table/inspect.py:
##########
@@ -704,3 +731,30 @@ def all_data_files(self) -> "pa.Table":
def all_delete_files(self) -> "pa.Table":
return self._all_files({DataFileContent.POSITION_DELETES, DataFileContent.EQUALITY_DELETES})
+
+ def all_entries(self) -> "pa.Table":
+ import pyarrow as pa
+
+ snapshots = self.tbl.snapshots()
+ if not snapshots:
+ return pa.Table.from_pylist([], self._get_entries_schema())
+
+ schemas = self.tbl.schemas()
+ snapshot_schemas: Dict[int, "pa.Schema"] = {}
+ for snapshot in snapshots:
+ if snapshot.schema_id is None:
+ raise ValueError(f"Cannot find schema_id for snapshot: {snapshot.snapshot_id}")
+ else:
+ snapshot_schemas[snapshot.snapshot_id] = schemas.get(snapshot.schema_id)
+
+ executor = ExecutorFactory.get_or_create()
+ all_manifests: Iterator[List[ManifestFile]] = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io), snapshots)
+ unique_flattened_manifests = list(
+ {manifest.manifest_path: manifest for manifest_list in all_manifests for manifest in manifest_list}.values()
+ )
+
Review Comment:
I guess yeah but will probably be slower...would need to do `_generate_manifests_table` for each snapshot and then concat all the `path` fields
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "soumya-ghosh (via GitHub)" <gi...@apache.org>.
soumya-ghosh commented on PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#issuecomment-2867718906
@amitgilad3 @Fokko @kevinjqliu Let's get this PR to completion to complete remaining metadata tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support all_entries in pyiceberg [iceberg-python]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on PR #1608:
URL: https://github.com/apache/iceberg-python/pull/1608#issuecomment-2634618707
> the number of entries returned by spark is less then what pyiceberg returns and i'm not sure why
looks like there are a lot of dups in pyiceberg. if you do set and sort, it should match
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org