small refactor, and comments

Author: gabeiglio

pyiceberg/table/__init__.py

@@ -26,25 +26,12 @@
 from functools import cached_property
 from itertools import chain
 from types import TracebackType
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    TypeVar,
-)
+from typing import TYPE_CHECKING, Any, TypeVar
 
 from pydantic import Field
 
 import pyiceberg.expressions.parser as parser
-from pyiceberg.expressions import (
-    AlwaysFalse,
-    AlwaysTrue,
-    And,
-    BooleanExpression,
-    EqualTo,
-    IsNull,
-    Or,
-    Reference,
-)
+from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference
 from pyiceberg.expressions.visitors import (
     ResidualEvaluator,
     _InclusiveMetricsEvaluator,
@@ -54,36 +41,17 @@
     manifest_evaluator,
 )
 from pyiceberg.io import FileIO, load_file_io
-from pyiceberg.manifest import (
-    DataFile,
-    DataFileContent,
-    ManifestContent,
-    ManifestEntry,
-    ManifestFile,
-)
-from pyiceberg.partitioning import (
-    PARTITION_FIELD_ID_START,
-    UNPARTITIONED_PARTITION_SPEC,
-    PartitionKey,
-    PartitionSpec,
-)
+from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestFile
+from pyiceberg.partitioning import PARTITION_FIELD_ID_START, UNPARTITIONED_PARTITION_SPEC, PartitionKey, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.table.delete_file_index import DeleteFileIndex
 from pyiceberg.table.inspect import InspectTable
 from pyiceberg.table.locations import LocationProvider, load_location_provider
 from pyiceberg.table.maintenance import MaintenanceTable
-from pyiceberg.table.metadata import (
-    INITIAL_SEQUENCE_NUMBER,
-    TableMetadata,
-)
-from pyiceberg.table.name_mapping import (
-    NameMapping,
-)
+from pyiceberg.table.metadata import INITIAL_SEQUENCE_NUMBER, TableMetadata
+from pyiceberg.table.name_mapping import NameMapping
 from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef
-from pyiceberg.table.snapshots import (
-    Snapshot,
-    SnapshotLogEntry,
-)
+from pyiceberg.table.snapshots import Snapshot, SnapshotLogEntry
 from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
 from pyiceberg.table.update import (
     AddPartitionSpecUpdate,
@@ -107,11 +75,7 @@
     update_table_metadata,
 )
 from pyiceberg.table.update.schema import UpdateSchema
-from pyiceberg.table.update.snapshot import (
-    ManageSnapshots,
-    UpdateSnapshot,
-    _FastAppendFiles,
-)
+from pyiceberg.table.update.snapshot import ManageSnapshots, UpdateSnapshot, _FastAppendFiles
 from pyiceberg.table.update.sorting import UpdateSortOrder
 from pyiceberg.table.update.spec import UpdateSpec
 from pyiceberg.table.update.statistics import UpdateStatistics
@@ -126,9 +90,7 @@
     Record,
     TableVersion,
 )
-from pyiceberg.types import (
-    strtobool,
-)
+from pyiceberg.types import strtobool
 from pyiceberg.utils.concurrent import ExecutorFactory
 from pyiceberg.utils.config import Config
 from pyiceberg.utils.properties import property_as_bool
@@ -144,11 +106,7 @@
     from pyiceberg_core.datafusion import IcebergDataFusionTable
 
     from pyiceberg.catalog import Catalog
-    from pyiceberg.catalog.rest.scan_planning import (
-        RESTContentFile,
-        RESTDeleteFile,
-        RESTFileScanTask,
-    )
+    from pyiceberg.catalog.rest.scan_planning import RESTContentFile, RESTDeleteFile, RESTFileScanTask
 
 ALWAYS_TRUE = AlwaysTrue()
 DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE = "downcast-ns-timestamp-to-us-on-write"
@@ -403,6 +361,8 @@ def _build_partition_predicate(
 
         Args:
             partition_records: A set of partition records to match
+            spec: An optional partition spec, if none then defaults to current
+            schema: An optional schema, if none then defaults to current
         Returns:
             A predicate matching any of the input partition records.
         """
@@ -675,11 +635,7 @@ def delete(
             case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive
             branch: Branch Reference to run the delete operation
         """
-        from pyiceberg.io.pyarrow import (
-            ArrowScan,
-            _dataframe_to_data_files,
-            _expression_to_complementary_pyarrow,
-        )
+        from pyiceberg.io.pyarrow import ArrowScan, _dataframe_to_data_files, _expression_to_complementary_pyarrow
 
         if (
             self.table_metadata.properties.get(TableProperties.DELETE_MODE, TableProperties.DELETE_MODE_DEFAULT)

pyiceberg/table/update/snapshot.py

@@ -211,7 +211,7 @@ def _write_delete_manifest() -> list[ManifestFile]:
                 for spec_id, entries in partition_groups.items():
                     with write_manifest(
                         format_version=self._transaction.table_metadata.format_version,
-                        spec=self._transaction.table_metadata.specs()[spec_id],
+                        spec=self.spec(spec_id),
                         schema=self.schema(),
                         output_file=self.new_manifest_output(),
                         snapshot_id=self._snapshot_id,
@@ -369,17 +369,15 @@ def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted: bool = T
         return manifest.fetch_manifest_entry(io=self._io, discard_deleted=discard_deleted)
 
     def _build_partition_projection(self, spec_id: int) -> BooleanExpression:
-        spec = self._transaction.table_metadata.specs()[spec_id]
-        project = inclusive_projection(self.schema(), spec, self._case_sensitive)
+        project = inclusive_projection(self.schema(), self.spec(spec_id), self._case_sensitive)
         return project(self._predicate)
 
     @cached_property
     def partition_filters(self) -> KeyDefaultDict[int, BooleanExpression]:
         return KeyDefaultDict(self._build_partition_projection)
 
     def _build_manifest_evaluator(self, spec_id: int) -> Callable[[ManifestFile], bool]:
-        spec = self._transaction.table_metadata.specs()[spec_id]
-        return manifest_evaluator(spec, self.schema(), self.partition_filters[spec_id], self._case_sensitive)
+        return manifest_evaluator(self.spec(spec_id), self.schema(), self.partition_filters[spec_id], self._case_sensitive)
 
     def delete_by_predicate(self, predicate: BooleanExpression, case_sensitive: bool = True) -> None:
         self._predicate = Or(self._predicate, predicate)