You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "jqin61 (via GitHub)" <gi...@apache.org> on 2024/03/12 17:55:29 UTC
[PR] Add partition stats in snapshot summary [iceberg-python]
jqin61 opened a new pull request, #521:
URL: https://github.com/apache/iceberg-python/pull/521
Changes include:
- add write.summary.partition-limit to table properties
- add changed-partition-count to snapshot summary
- add partition stats. these are key value pairs whose key is like partitiion.<partition_key> and whose value is like a nested mini-summary string concatenated by ','.
For example:
```'partitions.int_field=1': 'added-files-size=1234,removed-files-size=1234,added-data-files=1,deleted-data-files=1,added-records=100,deleted-records=100'```
--
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] Add partition stats in snapshot summary [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on PR #521:
URL: https://github.com/apache/iceberg-python/pull/521#issuecomment-2003194763
This is great, thanks for working on this @jqin61
--
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] Add partition stats in snapshot summary [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on PR #521:
URL: https://github.com/apache/iceberg-python/pull/521#issuecomment-1992323876
> @Fokko As we were discussing last time, it looks like the trustPartitionMetrics flag is always set to true in spark, so I ignored the flag. Please let me know the concerns if there is any!
This is correct! It is only affected when we add manifests directly to the manifest list, but that's something we want to avoid (at almost all cost :)
--
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] Add partition stats in snapshot summary [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko merged PR #521:
URL: https://github.com/apache/iceberg-python/pull/521
--
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] Add partition stats in snapshot summary [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #521:
URL: https://github.com/apache/iceberg-python/pull/521#discussion_r1524719927
##########
pyiceberg/table/snapshots.py:
##########
@@ -172,100 +260,57 @@ class SnapshotLogEntry(IcebergBaseModel):
class SnapshotSummaryCollector:
- added_file_size: int
- removed_file_size: int
- added_data_files: int
- removed_data_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
+ metrics: UpdateMetrics
+ partition_metrics: DefaultDict[str, UpdateMetrics]
+ max_changed_partitions_for_summaries: int
def __init__(self) -> None:
- self.added_file_size = 0
- self.removed_file_size = 0
- self.added_data_files = 0
- self.removed_data_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_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.added_data_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 remove_file(self, data_file: DataFile) -> None:
- self.removed_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.removed_data_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
+ self.metrics = UpdateMetrics()
+ self.partition_metrics = defaultdict(UpdateMetrics)
+ self.max_changed_partitions_for_summaries = 0
+
+ def set_partition_summary_limit(self, limit: int) -> None:
+ self.max_changed_partitions_for_summaries = limit
+
+ def add_file(
+ self, data_file: DataFile, partition_spec: Optional[PartitionSpec] = None, schema: Optional[Schema] = None
Review Comment:
Since this is more-or-less internal, I think it makes sense to make the spec and schema required.
##########
pyiceberg/table/__init__.py:
##########
@@ -2569,9 +2572,21 @@ def _write_delete_manifest() -> List[ManifestFile]:
def _summary(self) -> Summary:
ssc = SnapshotSummaryCollector()
+ partition_summary_limit = self._transaction.table_metadata.properties.get(
Review Comment:
I think they are stored as strings, so maybe we should try to parse it, and raise a ValueError otherwise.
##########
tests/integration/test_writes.py:
##########
@@ -525,12 +525,15 @@ def test_summaries_with_only_nulls(
'total-records': '2',
}
- assert summaries[0] == {
- 'total-data-files': '0',
- 'total-delete-files': '0',
+ assert summaries[2] == {
Review Comment:
Nice catch!
##########
pyiceberg/table/snapshots.py:
##########
@@ -172,100 +260,57 @@ class SnapshotLogEntry(IcebergBaseModel):
class SnapshotSummaryCollector:
- added_file_size: int
- removed_file_size: int
- added_data_files: int
- removed_data_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
+ metrics: UpdateMetrics
+ partition_metrics: DefaultDict[str, UpdateMetrics]
+ max_changed_partitions_for_summaries: int
def __init__(self) -> None:
- self.added_file_size = 0
- self.removed_file_size = 0
- self.added_data_files = 0
- self.removed_data_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_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.added_data_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 remove_file(self, data_file: DataFile) -> None:
- self.removed_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.removed_data_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
+ self.metrics = UpdateMetrics()
+ self.partition_metrics = defaultdict(UpdateMetrics)
+ self.max_changed_partitions_for_summaries = 0
+
+ def set_partition_summary_limit(self, limit: int) -> None:
+ self.max_changed_partitions_for_summaries = limit
+
+ def add_file(
+ self, data_file: DataFile, partition_spec: Optional[PartitionSpec] = None, schema: Optional[Schema] = None
+ ) -> None:
+ self.metrics.add_file(data_file)
+ if getattr(data_file, "partition", None) is not None and len(data_file.partition.record_fields()) != 0:
Review Comment:
The partition is required, and should always be there:
##########
pyiceberg/table/snapshots.py:
##########
@@ -172,100 +260,57 @@ class SnapshotLogEntry(IcebergBaseModel):
class SnapshotSummaryCollector:
- added_file_size: int
- removed_file_size: int
- added_data_files: int
- removed_data_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
+ metrics: UpdateMetrics
+ partition_metrics: DefaultDict[str, UpdateMetrics]
+ max_changed_partitions_for_summaries: int
def __init__(self) -> None:
- self.added_file_size = 0
- self.removed_file_size = 0
- self.added_data_files = 0
- self.removed_data_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_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.added_data_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 remove_file(self, data_file: DataFile) -> None:
- self.removed_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.removed_data_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
+ self.metrics = UpdateMetrics()
+ self.partition_metrics = defaultdict(UpdateMetrics)
+ self.max_changed_partitions_for_summaries = 0
+
+ def set_partition_summary_limit(self, limit: int) -> None:
+ self.max_changed_partitions_for_summaries = limit
+
+ def add_file(
+ self, data_file: DataFile, partition_spec: Optional[PartitionSpec] = None, schema: Optional[Schema] = None
Review Comment:
Or something like:
```suggestion
self, data_file: DataFile, schema: Schema, partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC
```
##########
pyiceberg/table/snapshots.py:
##########
@@ -172,100 +260,57 @@ class SnapshotLogEntry(IcebergBaseModel):
class SnapshotSummaryCollector:
- added_file_size: int
- removed_file_size: int
- added_data_files: int
- removed_data_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
+ metrics: UpdateMetrics
+ partition_metrics: DefaultDict[str, UpdateMetrics]
+ max_changed_partitions_for_summaries: int
def __init__(self) -> None:
- self.added_file_size = 0
- self.removed_file_size = 0
- self.added_data_files = 0
- self.removed_data_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_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.added_data_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 remove_file(self, data_file: DataFile) -> None:
- self.removed_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.removed_data_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
+ self.metrics = UpdateMetrics()
+ self.partition_metrics = defaultdict(UpdateMetrics)
+ self.max_changed_partitions_for_summaries = 0
+
+ def set_partition_summary_limit(self, limit: int) -> None:
+ self.max_changed_partitions_for_summaries = limit
+
+ def add_file(
+ self, data_file: DataFile, partition_spec: Optional[PartitionSpec] = None, schema: Optional[Schema] = None
+ ) -> None:
+ self.metrics.add_file(data_file)
+ if getattr(data_file, "partition", None) is not None and len(data_file.partition.record_fields()) != 0:
+ if partition_spec is None or schema is None:
+ raise ValueError("add data file with partition but without specifying the partiton_spec and schema")
+ self.update_partition_metrics(partition_spec=partition_spec, file=data_file, is_add_file=True, schema=schema)
+
+ def remove_file(self, data_file: DataFile, partition_spec: Optional[PartitionSpec], schema: Optional[Schema]) -> None:
+ self.metrics.remove_file(data_file)
+ if getattr(data_file, "partition", None) is not None and len(data_file.partition.record_fields()) != 0:
+ if partition_spec is None or schema is None:
+ raise ValueError("remove data file with partition but without specifying the partiton_spec and schema")
+ self.update_partition_metrics(partition_spec=partition_spec, file=data_file, is_add_file=False, schema=schema)
+
+ def update_partition_metrics(self, partition_spec: PartitionSpec, file: DataFile, is_add_file: bool, schema: Schema) -> None:
+ partition_path = partition_spec.partition_to_path(file.partition, schema)
+ partition_metrics: UpdateMetrics = self.partition_metrics[partition_path]
+
+ if is_add_file:
+ partition_metrics.add_file(file)
else:
- raise ValueError(f"Unknown data file content: {data_file.content}")
+ partition_metrics.remove_file(file)
def build(self) -> Dict[str, str]:
- def set_when_positive(properties: Dict[str, str], num: int, property_name: str) -> None:
- if num > 0:
- properties[property_name] = str(num)
-
- properties: Dict[str, str] = {}
- set_when_positive(properties, self.added_file_size, ADDED_FILE_SIZE)
- set_when_positive(properties, self.removed_file_size, REMOVED_FILE_SIZE)
- set_when_positive(properties, self.added_data_files, ADDED_DATA_FILES)
- set_when_positive(properties, self.removed_data_files, DELETED_DATA_FILES)
- set_when_positive(properties, self.added_eq_delete_files, ADDED_EQUALITY_DELETE_FILES)
- set_when_positive(properties, self.removed_eq_delete_files, REMOVED_EQUALITY_DELETE_FILES)
- set_when_positive(properties, self.added_pos_delete_files, ADDED_POSITION_DELETE_FILES)
- set_when_positive(properties, self.removed_pos_delete_files, REMOVED_POSITION_DELETE_FILES)
- set_when_positive(properties, self.added_delete_files, ADDED_DELETE_FILES)
- set_when_positive(properties, self.removed_delete_files, REMOVED_DELETE_FILES)
- set_when_positive(properties, self.added_records, ADDED_RECORDS)
- set_when_positive(properties, self.deleted_records, DELETED_RECORDS)
- set_when_positive(properties, self.added_pos_deletes, ADDED_POSITION_DELETES)
- set_when_positive(properties, self.removed_pos_deletes, REMOVED_POSITION_DELETES)
- set_when_positive(properties, self.added_eq_deletes, ADDED_EQUALITY_DELETES)
- set_when_positive(properties, self.removed_eq_deletes, REMOVED_EQUALITY_DELETES)
+ properties = self.metrics.to_dict()
+ changed_partitions_size = len(self.partition_metrics)
+ set_when_positive(properties, changed_partitions_size, CHANGED_PARTITION_COUNT_PROP)
+ if changed_partitions_size <= self.max_changed_partitions_for_summaries:
+ for partition_path, update_metrics_partition in self.partition_metrics.items():
+ if (summary := self.partition_summary(update_metrics_partition)) and len(summary) != 0:
+ properties[CHANGED_PARTITION_PREFIX + partition_path] = summary
return properties
+ def partition_summary(self, update_metrics: UpdateMetrics) -> str:
Review Comment:
```suggestion
def _partition_summary(self, update_metrics: UpdateMetrics) -> str:
```
##########
pyiceberg/table/snapshots.py:
##########
@@ -172,100 +260,57 @@ class SnapshotLogEntry(IcebergBaseModel):
class SnapshotSummaryCollector:
- added_file_size: int
- removed_file_size: int
- added_data_files: int
- removed_data_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
+ metrics: UpdateMetrics
+ partition_metrics: DefaultDict[str, UpdateMetrics]
+ max_changed_partitions_for_summaries: int
def __init__(self) -> None:
- self.added_file_size = 0
- self.removed_file_size = 0
- self.added_data_files = 0
- self.removed_data_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_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.added_data_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 remove_file(self, data_file: DataFile) -> None:
- self.removed_file_size += data_file.file_size_in_bytes
-
- if data_file.content == DataFileContent.DATA:
- self.removed_data_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
+ self.metrics = UpdateMetrics()
+ self.partition_metrics = defaultdict(UpdateMetrics)
+ self.max_changed_partitions_for_summaries = 0
+
+ def set_partition_summary_limit(self, limit: int) -> None:
+ self.max_changed_partitions_for_summaries = limit
+
+ def add_file(
+ self, data_file: DataFile, partition_spec: Optional[PartitionSpec] = None, schema: Optional[Schema] = None
Review Comment:
Same goes for the other methods, would be nice to have less noise there of arguments that we want to pass in anyway.
--
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