fix dynamic overwrite null-row regression after spec evolution

Author: Gayathri Srividya Rajavarapu

pyiceberg/table/__init__.py

@@ -622,6 +622,11 @@ def dynamic_partition_overwrite(
             # _compute_deletes uses the right predicate when evaluating each manifest file.
             source_id_to_pos = {field.source_id: pos for pos, field in enumerate(current_spec.fields)}
             source_id_to_col = {field.source_id: schema.find_field(field.source_id).name for field in current_spec.fields}
+            exact_delete_filter = self._build_partition_predicate(
+                partition_records=partitions_to_overwrite,
+                spec=current_spec,
+                schema=schema,
+            )
 
             per_spec_predicates: dict[int, BooleanExpression] = {}
             for spec_id, hist_spec in all_specs.items():
@@ -645,24 +650,21 @@ def dynamic_partition_overwrite(
 
                 per_spec_predicates[spec_id] = Or(*per_record_exprs) if len(per_record_exprs) > 1 else per_record_exprs[0]
 
-            preds = list(per_spec_predicates.values())
-            global_delete_filter = Or(*preds) if len(preds) > 1 else preds[0] if preds else AlwaysFalse()
-
             # Open the delete snapshot and set per-spec predicates before committing.
             # This mirrors Transaction.delete() but injects per_spec_predicates so that
             # _compute_deletes uses the right predicate for each historical spec.
             from pyiceberg.io.pyarrow import ArrowScan, _dataframe_to_data_files, _expression_to_complementary_pyarrow
 
             with self.update_snapshot(snapshot_properties=snapshot_properties, branch=branch).delete() as delete_snapshot:
                 delete_snapshot._per_spec_predicates = per_spec_predicates
-                delete_snapshot.delete_by_predicate(global_delete_filter)
+                delete_snapshot.delete_by_predicate(exact_delete_filter)
 
             # Handle partial-match files that need to be rewritten (copy-on-write).
             if delete_snapshot.rewrites_needed is True:
-                bound_delete_filter = bind(self.table_metadata.schema(), global_delete_filter, case_sensitive=True)
+                bound_delete_filter = bind(self.table_metadata.schema(), exact_delete_filter, case_sensitive=True)
                 preserve_row_filter = _expression_to_complementary_pyarrow(bound_delete_filter, self.table_metadata.schema())
 
-                file_scan = self._scan(row_filter=global_delete_filter)
+                file_scan = self._scan(row_filter=exact_delete_filter)
                 if branch is not None:
                     file_scan = file_scan.use_ref(branch)
 

pyiceberg/table/update/snapshot.py

@@ -362,7 +362,8 @@ def fetch_manifest_entry(self, manifest: ManifestFile, discard_deleted: bool = T
 
     def _build_partition_projection(self, spec_id: int) -> BooleanExpression:
         project = inclusive_projection(self.schema(), self.spec(spec_id), self._case_sensitive)
-        return project(self._predicate)
+        predicate = self._per_spec_predicates.get(spec_id, self._predicate)
+        return project(predicate)
 
     @cached_property
     def partition_filters(self) -> KeyDefaultDict[int, BooleanExpression]:
@@ -434,13 +435,10 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
 
         manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
 
-        def _strict_metrics_for_spec(spec_id: int) -> Callable[[DataFile], bool]:
-            predicate = self._per_spec_predicates.get(spec_id, self._predicate)
-            return _StrictMetricsEvaluator(schema, predicate, case_sensitive=self._case_sensitive).eval
-
-        def _inclusive_metrics_for_spec(spec_id: int) -> Callable[[DataFile], bool]:
-            predicate = self._per_spec_predicates.get(spec_id, self._predicate)
-            return _InclusiveMetricsEvaluator(schema, predicate, case_sensitive=self._case_sensitive).eval
+        strict_metrics_evaluator = _StrictMetricsEvaluator(schema, self._predicate, case_sensitive=self._case_sensitive).eval
+        inclusive_metrics_evaluator = _InclusiveMetricsEvaluator(
+            schema, self._predicate, case_sensitive=self._case_sensitive
+        ).eval
 
         existing_manifests = []
         total_deleted_entries = []
@@ -460,9 +458,6 @@ def _inclusive_metrics_for_spec(spec_id: int) -> Callable[[DataFile], bool]:
                             existing_manifests.append(manifest_file)
                         else:
                             # It is relevant, let's check out the content
-                            spec_id = manifest_file.partition_spec_id
-                            strict_metrics_evaluator = _strict_metrics_for_spec(spec_id)
-                            inclusive_metrics_evaluator = _inclusive_metrics_for_spec(spec_id)
                             deleted_entries = []
                             existing_entries = []
                             for entry in manifest_file.fetch_manifest_entry(io=self._io, discard_deleted=True):