Fix projected fields predicate evaluation

pyiceberg/expressions/visitors.py

@@ -860,7 +860,9 @@ class _ColumnNameTranslator(BooleanExpressionVisitor[BooleanExpression]):
 
     Args:
       file_schema (Schema): The schema of the file.
+      projected_schema (Schema): The schema to project onto the data files.
       case_sensitive (bool): Whether to consider case when binding a reference to a field in a schema, defaults to True.
+      projected_missing_fields(dict[str, Any]): Map of fields missing in file_schema, but present as partition values.
 
     Raises:
         TypeError: In the case of an UnboundPredicate.
@@ -870,9 +872,13 @@ class _ColumnNameTranslator(BooleanExpressionVisitor[BooleanExpression]):
     file_schema: Schema
     case_sensitive: bool
 
-    def __init__(self, file_schema: Schema, case_sensitive: bool) -> None:
+    def __init__(
+        self, file_schema: Schema, projected_schema: Schema, case_sensitive: bool, projected_missing_fields: dict[str, Any]
+    ) -> None:
         self.file_schema = file_schema
+        self.projected_schema = projected_schema
         self.case_sensitive = case_sensitive
+        self.projected_missing_fields = projected_missing_fields
 
     def visit_true(self) -> BooleanExpression:
         return AlwaysTrue()
@@ -900,6 +906,24 @@ def visit_bound_predicate(self, predicate: BoundPredicate[L]) -> BooleanExpressi
             # in the file schema when reading older data
             if isinstance(predicate, BoundIsNull):
                 return AlwaysTrue()
+            # Evaluate projected field by value extracted from partition
+            elif (field_name := predicate.term.ref().field.name) in self.projected_missing_fields:
+                unbound_predicate: BooleanExpression
+                if isinstance(predicate, BoundUnaryPredicate):
+                    unbound_predicate = predicate.as_unbound(field_name)
+                elif isinstance(predicate, BoundLiteralPredicate):
+                    unbound_predicate = predicate.as_unbound(field_name, predicate.literal)
+                elif isinstance(predicate, BoundSetPredicate):
+                    unbound_predicate = predicate.as_unbound(field_name, predicate.literals)
+                else:
+                    raise ValueError(f"Unsupported predicate: {predicate}")
+                field = self.projected_schema.find_field(field_name)
+                schema = Schema(field)
+                evaluator = expression_evaluator(schema, unbound_predicate, self.case_sensitive)
+                if evaluator(Record(self.projected_missing_fields[field_name])):
+                    return AlwaysTrue()
+                else:
+                    return AlwaysFalse()
             else:
                 return AlwaysFalse()
 
@@ -913,8 +937,14 @@ def visit_bound_predicate(self, predicate: BoundPredicate[L]) -> BooleanExpressi
             raise ValueError(f"Unsupported predicate: {predicate}")
 
 
-def translate_column_names(expr: BooleanExpression, file_schema: Schema, case_sensitive: bool) -> BooleanExpression:
-    return visit(expr, _ColumnNameTranslator(file_schema, case_sensitive))
+def translate_column_names(
+    expr: BooleanExpression,
+    file_schema: Schema,
+    projected_schema: Schema,
+    case_sensitive: bool,
+    projected_missing_fields: dict[str, Any],
+) -> BooleanExpression:
+    return visit(expr, _ColumnNameTranslator(file_schema, projected_schema, case_sensitive, projected_missing_fields))
 
 
 class _ExpressionFieldIDs(BooleanExpressionVisitor[Set[int]]):

pyiceberg/io/pyarrow.py

@@ -1404,18 +1404,24 @@ def _task_to_record_batches(
         # the table format version.
         file_schema = pyarrow_to_schema(physical_schema, name_mapping, downcast_ns_timestamp_to_us=True)
 
-        pyarrow_filter = None
-        if bound_row_filter is not AlwaysTrue():
-            translated_row_filter = translate_column_names(bound_row_filter, file_schema, case_sensitive=case_sensitive)
-            bound_file_filter = bind(file_schema, translated_row_filter, case_sensitive=case_sensitive)
-            pyarrow_filter = expression_to_pyarrow(bound_file_filter)
-
         # Apply column projection rules
         # https://iceberg.apache.org/spec/#column-projection
         should_project_columns, projected_missing_fields = _get_column_projection_values(
             task.file, projected_schema, partition_spec, file_schema.field_ids
         )
 
+        pyarrow_filter = None
+        if bound_row_filter is not AlwaysTrue():
+            translated_row_filter = translate_column_names(
+                bound_row_filter,
+                file_schema,
+                projected_schema,
+                case_sensitive=case_sensitive,
+                projected_missing_fields=projected_missing_fields,
+            )
+            bound_file_filter = bind(file_schema, translated_row_filter, case_sensitive=case_sensitive)
+            pyarrow_filter = expression_to_pyarrow(bound_file_filter)
+
         file_project_schema = prune_columns(file_schema, projected_field_ids, select_full_types=False)
 
         fragment_scanner = ds.Scanner.from_fragment(

tests/expressions/test_visitors.py

@@ -68,10 +68,12 @@
     BooleanExpressionVisitor,
     BoundBooleanExpressionVisitor,
     _ManifestEvalVisitor,
+    bind,
     expression_evaluator,
     expression_to_plain_format,
     rewrite_not,
     rewrite_to_dnf,
+    translate_column_names,
     visit,
     visit_bound_predicate,
 )
@@ -1623,3 +1625,40 @@ def test_expression_evaluator_null() -> None:
     assert expression_evaluator(schema, LessThan("a", 1), case_sensitive=True)(struct) is False
     assert expression_evaluator(schema, StartsWith("a", 1), case_sensitive=True)(struct) is False
     assert expression_evaluator(schema, NotStartsWith("a", 1), case_sensitive=True)(struct) is True
+
+
+@pytest.mark.parametrize(
+    "before_expression,after_expression",
+    [
+        (In("id", {1, 2, 3}), AlwaysTrue()),
+        (EqualTo("id", 3), AlwaysFalse()),
+        (
+            And(EqualTo("id", 1), EqualTo("all_same_value_or_null", "string")),
+            And(AlwaysTrue(), EqualTo("all_same_value_or_null", "string")),
+        ),
+        (
+            And(EqualTo("all_same_value_or_null", "string"), GreaterThan("id", 2)),
+            And(EqualTo("all_same_value_or_null", "string"), AlwaysFalse()),
+        ),
+        (
+            Or(
+                And(EqualTo("id", 1), EqualTo("all_same_value_or_null", "string")),
+                And(EqualTo("all_same_value_or_null", "string"), GreaterThan("id", 2)),
+            ),
+            Or(
+                And(AlwaysTrue(), EqualTo("all_same_value_or_null", "string")),
+                And(EqualTo("all_same_value_or_null", "string"), AlwaysFalse()),
+            ),
+        ),
+    ],
+)
+def test_translate_column_names_eval_projected_fields(
+    schema: Schema, before_expression: BooleanExpression, after_expression: BooleanExpression
+) -> None:
+    # exclude id from file_schema pretending that it's part of partition values
+    file_schema = Schema(*[field for field in schema.columns if field.name != "id"])
+    projected_missing_fields = {"id": 1}
+    assert (
+        translate_column_names(bind(schema, before_expression, True), file_schema, schema, True, projected_missing_fields)
+        == after_expression
+    )