Fokko commented on code in PR #61:
URL: https://github.com/apache/iceberg-python/pull/61#discussion_r1372143737
##########
pyiceberg/table/snapshots.py:
##########
@@ -116,3 +144,199 @@ 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:
+ if data_file.content == DataFileContent.DATA:
+ self.added_files += 1
+ self.added_records += data_file.record_count
+ self.added_size += data_file.file_size_in_bytes
+ 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:
+ 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
+
+
+properties = ['records', 'files-size', 'data-files', 'delete-files',
'position-deletes', 'equality-deletes']
+
+
+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 == Operation.OVERWRITE and previous_summary is not
None:
+ summary = truncate_table_summary(summary, previous_summary)
Review Comment:
@HonahX Yes, I think we want to add this in the future. I kept the current
version as minimal as possible. In #41 we're writing a single file right now.
So that would always replace all the other files when doing an overwrite.
Once we go to writing to partitioned tables, we'll need to implement this
when overwriting a specific partition.
--
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
