fix spec-evolution overwrite predicate matching

Author: Gayathri Srividya Rajavarapu

pyiceberg/table/__init__.py

@@ -632,6 +632,7 @@ def dynamic_partition_overwrite(
             for spec_id, hist_spec in all_specs.items():
                 hist_source_ids = {field.source_id for field in hist_spec.fields}
                 missing_source_ids = current_source_ids - hist_source_ids
+                has_overlap_with_current = bool(hist_source_ids & current_source_ids)
 
                 per_record_exprs: list[BooleanExpression] = []
                 for partition_record in partitions_to_overwrite:
@@ -640,7 +641,7 @@ def dynamic_partition_overwrite(
                         value = partition_record[source_id_to_pos[source_id]]
                         if value is not None:
                             field_pred: BooleanExpression = EqualTo(Reference(col_name), value)
-                            if source_id in missing_source_ids:
+                            if source_id in missing_source_ids and has_overlap_with_current:
                                 field_pred = Or(field_pred, IsNull(Reference(col_name)))
                         else:
                             field_pred = IsNull(Reference(col_name))

pyiceberg/table/update/snapshot.py

@@ -435,10 +435,13 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
 
         manifest_evaluators: dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
 
-        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
+        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
 
         existing_manifests = []
         total_deleted_entries = []
@@ -458,6 +461,9 @@ def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) ->
                             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):