@@ -355,29 +355,46 @@ def _set_ref_snapshot(
         return updates, requirements
 
     def _build_partition_predicate(
-        self, partition_records: set[Record], spec: PartitionSpec, schema: Schema
+        self,
+        partition_records: set[Record],
+        spec: PartitionSpec,
+        schema: Schema,
+        evolved_source_ids: set[int] | None = None,
     ) -> BooleanExpression:
         """Build a filter predicate matching any of the input partition records.
 
         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
+            evolved_source_ids: Source IDs of partition fields that were added via spec
+                evolution and therefore may be absent (NULL) in data written under older
+                specs.  When a field is in this set and the partition value is non-NULL,
+                the predicate also accepts NULL so that pre-evolution files are included.
         Returns:
             A predicate matching any of the input partition records.
         """
-        partition_fields = [schema.find_field(field.source_id).name for field in spec.fields]
-        if not partition_records or not partition_fields:
+        spec_fields = spec.fields
+        partition_field_names = [schema.find_field(field.source_id).name for field in spec_fields]
+        if not partition_records or not partition_field_names:
             return AlwaysFalse()
 
+        nullable_source_ids: set[int] = evolved_source_ids or set()
+
         per_record_exprs: list[BooleanExpression] = []
         for partition_record in partition_records:
-            predicates: list[BooleanExpression] = [
-                EqualTo(Reference(partition_field), partition_record[pos])
-                if partition_record[pos] is not None
-                else IsNull(Reference(partition_field))
-                for pos, partition_field in enumerate(partition_fields)
-            ]
+            predicates: list[BooleanExpression] = []
+            for pos, (field_name, spec_field) in enumerate(zip(partition_field_names, spec_fields, strict=True)):
+                value = partition_record[pos]
+                if value is not None:
+                    field_pred: BooleanExpression = EqualTo(Reference(field_name), value)
+                    if spec_field.source_id in nullable_source_ids:
+                        # Also match NULL: pre-evolution files may have no value for this
+                        # field because it was not in the schema when the data was written.
+                        field_pred = Or(field_pred, IsNull(Reference(field_name)))
+                else:
+                    field_pred = IsNull(Reference(field_name))
+                predicates.append(field_pred)
             per_record_exprs.append(And(*predicates) if len(predicates) > 1 else predicates[0])
 
         return Or(*per_record_exprs) if len(per_record_exprs) > 1 else per_record_exprs[0]
@@ -586,8 +603,24 @@ def dynamic_partition_overwrite(
         )
 
         partitions_to_overwrite = {data_file.partition for data_file in data_files}
+
+        # Determine which partition fields were introduced via spec evolution.  Data
+        # written under older specs may have NULL for those fields even when the new
+        # partition value is non-NULL, so the delete predicate must also match NULL.
+        current_spec = self.table_metadata.spec()
+        all_specs = self.table_metadata.specs()
+        # A field is "evolved" if it is absent from at least one historical spec.
+        evolved_source_ids = {
+            f.source_id
+            for f in current_spec.fields
+            if not all(any(hf.source_id == f.source_id for hf in hs.fields) for hs in all_specs.values())
+        }
+
         delete_filter = self._build_partition_predicate(
-            partition_records=partitions_to_overwrite, spec=self.table_metadata.spec(), schema=self.table_metadata.schema()
+            partition_records=partitions_to_overwrite,
+            spec=current_spec,
+            schema=self.table_metadata.schema(),
+            evolved_source_ids=evolved_source_ids,
         )
         self.delete(delete_filter=delete_filter, snapshot_properties=snapshot_properties, branch=branch)
 
@@ -2072,13 +2105,11 @@ def _build_residual_evaluator(self, spec_id: int) -> Callable[[DataFile], Residu
         # The lambda created here is run in multiple threads.
         # So we avoid creating _EvaluatorExpression methods bound to a single
         # shared instance across multiple threads.
-        return lambda datafile: (
-            residual_evaluator_of(
-                spec=spec,
-                expr=self.row_filter,
-                case_sensitive=self.case_sensitive,
-                schema=self.table_metadata.schema(),
-            )
+        return lambda datafile: residual_evaluator_of(
+            spec=spec,
+            expr=self.row_filter,
+            case_sensitive=self.case_sensitive,
+            schema=self.table_metadata.schema(),
         )
 
     @staticmethod
 
@@ -31,6 +31,8 @@
     And,
     EqualTo,
     In,
+    IsNull,
+    Or,
 )
 from pyiceberg.expressions.visitors import bind
 from pyiceberg.io import PY_IO_IMPL, load_file_io
@@ -1766,3 +1768,63 @@ def test_build_large_partition_predicate(table_v2: Table) -> None:
         )
 
     bind(table_v2.metadata.schema(), expr, case_sensitive=True)
+
+
+def test_build_partition_predicate_with_evolved_source_ids(table_v2: Table) -> None:
+    """Regression test for https://github.com/apache/iceberg-python/issues/3148.
+
+    When a partition field is added via spec evolution, data written under older specs
+    may have NULL for that field.  _build_partition_predicate must include
+    ``OR field IS NULL`` for evolved fields so that pre-evolution files are matched by
+    the delete predicate and are not silently left behind.
+    """
+    from pyiceberg.transforms import IdentityTransform
+
+    schema = Schema(
+        NestedField(1, "category", StringType(), required=False),
+        NestedField(2, "region", StringType(), required=False),
+    )
+    spec = PartitionSpec(
+        PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category"),
+        PartitionField(source_id=2, field_id=1001, transform=IdentityTransform(), name="region"),
+    )
+
+    # region (source_id=2) was added via spec evolution → it is an evolved field
+    with table_v2.transaction() as tx:
+        expr = tx._build_partition_predicate(
+            partition_records={Record("A", "us")},
+            spec=spec,
+            schema=schema,
+            evolved_source_ids={2},
+        )
+
+    # category=A AND (region=us OR region IS NULL)
+    expected = And(
+        EqualTo("category", "A"),
+        Or(EqualTo("region", "us"), IsNull("region")),
+    )
+    assert repr(expr) == repr(expected)
+
+
+def test_build_partition_predicate_without_evolved_source_ids(table_v2: Table) -> None:
+    """Without evolved_source_ids, the predicate matches exact values only."""
+    from pyiceberg.transforms import IdentityTransform
+
+    schema = Schema(
+        NestedField(1, "category", StringType(), required=False),
+        NestedField(2, "region", StringType(), required=False),
+    )
+    spec = PartitionSpec(
+        PartitionField(source_id=1, field_id=1000, transform=IdentityTransform(), name="category"),
+        PartitionField(source_id=2, field_id=1001, transform=IdentityTransform(), name="region"),
+    )
+
+    with table_v2.transaction() as tx:
+        expr = tx._build_partition_predicate(
+            partition_records={Record("A", "us")},
+            spec=spec,
+            schema=schema,
+        )
+
+    expected = And(EqualTo("category", "A"), EqualTo("region", "us"))
+    assert repr(expr) == repr(expected)