rdblue commented on code in PR #61:
URL: https://github.com/apache/iceberg-python/pull/61#discussion_r1382619528
##########
pyiceberg/table/snapshots.py:
##########
@@ -116,3 +144,202 @@ class MetadataLogEntry(IcebergBaseModel):
class SnapshotLogEntry(IcebergBaseModel):
snapshot_id: int = Field(alias="snapshot-id")
timestamp_ms: int = Field(alias="timestamp-ms")
+
+
+class SnapshotSummaryCollector:
+ added_size: int
+ removed_size: int
+ added_files: int
+ removed_files: int
+ added_eq_delete_files: int
+ removed_eq_delete_files: int
+ added_pos_delete_files: int
+ removed_pos_delete_files: int
+ added_delete_files: int
+ removed_delete_files: int
+ added_records: int
+ deleted_records: int
+ added_pos_deletes: int
+ removed_pos_deletes: int
+ added_eq_deletes: int
+ removed_eq_deletes: int
+
+ def __init__(self) -> None:
+ self.added_size = 0
+ self.removed_size = 0
+ self.added_files = 0
+ self.removed_files = 0
+ self.added_eq_delete_files = 0
+ self.removed_eq_delete_files = 0
+ self.added_pos_delete_files = 0
+ self.removed_pos_delete_files = 0
+ self.added_delete_files = 0
+ self.removed_delete_files = 0
+ self.added_records = 0
+ self.deleted_records = 0
+ self.added_pos_deletes = 0
+ self.removed_pos_deletes = 0
+ self.added_eq_deletes = 0
+ self.removed_eq_deletes = 0
+
+ def add_file(self, data_file: DataFile) -> None:
+ self.added_size += data_file.file_size_in_bytes
+
+ if data_file.content == DataFileContent.DATA:
+ self.added_files += 1
+ self.added_records += data_file.record_count
+ elif data_file.content == DataFileContent.POSITION_DELETES:
+ self.added_delete_files += 1
+ self.added_pos_delete_files += 1
+ self.added_pos_deletes += data_file.record_count
+ elif data_file.content == DataFileContent.EQUALITY_DELETES:
+ self.added_delete_files += 1
+ self.added_eq_delete_files += 1
+ self.added_eq_deletes += data_file.record_count
+ else:
+ raise ValueError(f"Unknown data file content: {data_file.content}")
+
+ def removed_file(self, data_file: DataFile) -> None:
+ self.removed_size += data_file.file_size_in_bytes
+
+ if data_file.content == DataFileContent.DATA:
+ self.removed_files += 1
+ self.deleted_records += data_file.record_count
+ elif data_file.content == DataFileContent.POSITION_DELETES:
+ self.removed_delete_files += 1
+ self.removed_pos_delete_files += 1
+ self.removed_pos_deletes += data_file.record_count
+ elif data_file.content == DataFileContent.EQUALITY_DELETES:
+ self.removed_delete_files += 1
+ self.removed_eq_delete_files += 1
+ self.removed_eq_deletes += data_file.record_count
+ else:
+ raise ValueError(f"Unknown data file content: {data_file.content}")
+
+ def added_manifest(self, manifest: ManifestFile) -> None:
+ if manifest.content == ManifestContent.DATA:
+ self.added_files += manifest.added_files_count or 0
+ self.added_records += manifest.added_rows_count or 0
+ self.removed_files += manifest.deleted_files_count or 0
+ self.deleted_records += manifest.deleted_rows_count or 0
+ elif manifest.content == ManifestContent.DELETES:
+ self.added_delete_files += manifest.added_files_count or 0
+ self.removed_delete_files += manifest.deleted_files_count or 0
+ else:
+ raise ValueError(f"Unknown manifest file content:
{manifest.content}")
+
+ def build(self) -> Dict[str, str]:
+ def set_non_zero(properties: Dict[str, str], num: int, property_name:
str) -> None:
+ if num > 0:
+ properties[property_name] = str(num)
+
+ properties: Dict[str, str] = {}
+ set_non_zero(properties, self.added_size, 'added-files-size')
+ set_non_zero(properties, self.removed_size, 'removed-files-size')
+ set_non_zero(properties, self.added_files, 'added-data-files')
+ set_non_zero(properties, self.removed_files, 'removed-data-files')
+ set_non_zero(properties, self.added_eq_delete_files,
'added-equality-delete-files')
+ set_non_zero(properties, self.removed_eq_delete_files,
'removed-equality-delete-files')
+ set_non_zero(properties, self.added_pos_delete_files,
'added-position-delete-files')
+ set_non_zero(properties, self.removed_pos_delete_files,
'removed-position-delete-files')
+ set_non_zero(properties, self.added_delete_files, 'added-delete-files')
+ set_non_zero(properties, self.removed_delete_files,
'removed-delete-files')
+ set_non_zero(properties, self.added_records, 'added-records')
+ set_non_zero(properties, self.deleted_records, 'deleted-records')
+ set_non_zero(properties, self.added_pos_deletes,
'added-position-deletes')
+ set_non_zero(properties, self.removed_pos_deletes,
'removed-position-deletes')
+ set_non_zero(properties, self.added_eq_deletes,
'added-equality-deletes')
+ set_non_zero(properties, self.removed_eq_deletes,
'removed-equality-deletes')
+
+ return properties
+
+
+def truncate_table_summary(summary: Summary, previous_summary: Mapping[str,
str]) -> Summary:
+ for prop in {
+ 'total-data-files',
+ 'total-delete-files',
+ 'total-records',
+ 'total-files-size',
+ 'total-position-deletes',
+ 'total-equality-deletes',
+ }:
+ summary[prop] = '0'
+
+ if value := previous_summary.get('total-data-files'):
+ summary['deleted-data-files'] = value
+ if value := previous_summary.get('total-delete-files'):
+ summary['removed-delete-files'] = value
+ if value := previous_summary.get('total-records'):
+ summary['deleted-records'] = value
+ if value := previous_summary.get('total-files-size'):
+ summary['removed-files-size'] = value
+ if value := previous_summary.get('total-position-deletes'):
+ summary['removed-position-deletes'] = value
+ if value := previous_summary.get('total-equality-deletes'):
+ summary['removed-equality-deletes'] = value
+
+ return summary
+
+
+def merge_snapshot_summaries(
+ summary: Summary,
+ previous_summary: Optional[Mapping[str, str]] = None,
+) -> Summary:
+ if summary.operation not in {Operation.APPEND, Operation.OVERWRITE}:
+ raise ValueError(f"Operation not implemented: {summary.operation}")
+
+ if summary.operation == Operation.OVERWRITE and previous_summary is not
None:
+ summary = truncate_table_summary(summary, previous_summary)
+
+ if not previous_summary:
+ previous_summary = {
+ 'total-data-files': '0',
+ 'total-delete-files': '0',
+ 'total-records': '0',
+ 'total-files-size': '0',
+ 'total-position-deletes': '0',
+ 'total-equality-deletes': '0',
+ }
+
+ def _update_totals(total_property: str, added_property: str,
removed_property: str) -> None:
+ if new_total_str := previous_summary.get(total_property):
Review Comment:
Nit: `new_total_str` doesn't make sense because it is getting the property
from the previous summary.
--
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-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
