clean up

Author: jqin61

Makefile

@@ -40,7 +40,7 @@ test-integration:
 	docker-compose -f dev/docker-compose-integration.yml kill
 	docker-compose -f dev/docker-compose-integration.yml rm -f
 	docker-compose -f dev/docker-compose-integration.yml up -d
-	sleep 5
+	sleep 10
 	docker-compose -f dev/docker-compose-integration.yml exec -T spark-iceberg ipython ./provision.py
 	poetry run pytest tests/ -v -m integration ${PYTEST_ARGS}
 

pyiceberg/avro/file.py

@@ -276,7 +276,6 @@ def _write_header(self) -> None:
     def write_block(self, objects: List[D]) -> None:
         in_memory = io.BytesIO()
         block_content_encoder = BinaryEncoder(output_stream=in_memory)
-
         for obj in objects:
             self.writer.write(block_content_encoder, obj)
         block_content = in_memory.getvalue()

pyiceberg/io/pyarrow.py

@@ -1719,10 +1719,6 @@ def fill_parquet_file_metadata(
     data_file.split_offsets = split_offsets
 
 
-# write_file(table: Table, tasks: Iterator[WriteTask]) -> Iterator[DataFile]:
-# write_file(io = table.io, table_metadata = table.metadata, tasks: Iterator[WriteTask]) -> Iterator[DataFile]:
-
-
 def write_file(io: FileIO, table_metadata: TableMetadata, tasks: Iterator[WriteTask]) -> Iterator[DataFile]:
     for task in tasks:
         parquet_writer_kwargs = _get_parquet_writer_kwargs(table_metadata.properties)

pyiceberg/table/__init__.py

@@ -232,7 +232,6 @@ def property_as_int(properties: Dict[str, str], property_name: str, default: Opt
             return default
 
 
-# to do
 class PartitionProjector:
     def __init__(
         self,
@@ -2528,8 +2527,8 @@ def _write_filtered_manifest(self, file_num: int = 1) -> List[ManifestFile]:
         ) as writer:
             for manifest_entry in filtered_entries:
                 writer.add_entry(manifest_entry)
-        m = writer.to_manifest_file()
-        return [m]
+
+        return [writer.to_manifest_file()]
 
     def _write_added_manifest(self) -> List[ManifestFile]:
         if self._added_data_files:
@@ -2560,11 +2559,11 @@ def _write_added_manifest(self) -> List[ManifestFile]:
     def _manifests(self) -> List[ManifestFile]:
         executor = ExecutorFactory.get_or_create()
 
-        added_manifests = executor.submit(self._write_added_manifest).result()
-        filtered_manifests = executor.submit(self._write_filtered_manifest).result()
-        existing_manifests = executor.submit(self._existing_manifests).result()
+        added_manifests = executor.submit(self._write_added_manifest)
+        filtered_manifests = executor.submit(self._write_filtered_manifest)
+        existing_manifests = executor.submit(self._existing_manifests)
 
-        return added_manifests + filtered_manifests + existing_manifests
+        return added_manifests.result() + filtered_manifests.result() + existing_manifests.result()
 
     def _summary(self) -> Summary:
         ssc = SnapshotSummaryCollector()
@@ -2573,7 +2572,7 @@ def _summary(self) -> Summary:
             ssc.add_file(data_file=data_file)
 
         # Only delete files caused by partial overwrite.
-        # Full table overwrite uses previous snapshot to update in a more efficient way.
+        # Full table overwrite uses previous snapshot summary to update in a more efficient way.
         if self._deleted_data_files is not None and isinstance(self._deleted_data_files, ExplicitlyDeletedDataFiles):
             for data_file in self._deleted_data_files.deleted_files:
                 ssc.remove_file(data_file=data_file)
@@ -2584,10 +2583,9 @@ def _summary(self) -> Summary:
             else None
         )
 
-        summary_collector_to_fill = ssc.build()
-        summary_to_fill = Summary(operation=self._operation, **summary_collector_to_fill)
+        summary_to_update = Summary(operation=self._operation, **ssc.build())
         return update_snapshot_summaries(
-            summary=summary_to_fill,
+            summary=summary_to_update,
             previous_summary=previous_snapshot.summary if previous_snapshot is not None else None,
             truncate_full_table=isinstance(self._deleted_data_files, DeleteAllDataFiles),
         )
@@ -2700,7 +2698,6 @@ def group_by_partition_scheme(
             f"Not all transforms are supported, get: {[transform in supported for transform in iceberg_table_metadata.spec().fields]}."
         )
 
-    # only works for identity
     sort_options = _get_partition_sort_order(partition_columns, reverse=False)
     sorted_arrow_table = arrow_table.sort_by(sorting=sort_options['sort_keys'], null_placement=sort_options['null_placement'])
     return sorted_arrow_table

pyiceberg/table/snapshots.py

@@ -364,4 +364,5 @@ def _update_totals(total_property: str, added_property: str, removed_property: s
         added_property=ADDED_EQUALITY_DELETES,
         removed_property=REMOVED_EQUALITY_DELETES,
     )
+
     return summary

tests/integration/test_partitioned_writes.py

@@ -170,7 +170,7 @@ def table_v1_with_null_partitioned(session_catalog: Catalog, arrow_table_with_nu
         "timestamptz",
         "timestamp",
         "binary",
-    ]  # [, "date", "timestamptz", "timestamp", "binary", "fixed"]
+    ]
     for partition_col in partition_cols:
         identifier = f"default.arrow_table_v1_with_null_partitioned_on_col_{partition_col}"
 
@@ -555,7 +555,7 @@ def test_data_files_with_table_partitioned_with_null(
     # fourth operation of append manifest list abandons M3 since it has no existing or added entries and keeps M4 and added M5 with 3 added files
     # fifth operation of static overwrite's manifest list is linked to one filtered manifest M7 which filters and merges M5 and M6 where each has 1 entrys are deleted (int=1 matching the filter) and 2 entries marked as existed, this operation
     # also links to M6 which adds 3 entries.
-    # so we have flattened list of [[M1], [M1, M2], [M3, M4], [M4, M5], [M6, M7, M8]]
+    # so we have flattened list of [[M1], [M1, M2], [M3, M4], [M4, M5], [M6, M7]]
     # where: add      exist      delete    added_by
     # M1      3         0           0        S1
     # M2      3         0           0        S2