Evaluate projected field predicate

Author: Roman Shanin

pyiceberg/expressions/visitors.py

@@ -870,8 +870,11 @@ class _ColumnNameTranslator(BooleanExpressionVisitor[BooleanExpression]):
     file_schema: Schema
     case_sensitive: bool
 
-    def __init__(self, file_schema: Schema, case_sensitive: bool, projected_missing_fields: dict[str, Any]) -> 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
 
@@ -895,17 +898,30 @@ def visit_unbound_predicate(self, predicate: UnboundPredicate[L]) -> BooleanExpr
 
     def visit_bound_predicate(self, predicate: BoundPredicate[L]) -> BooleanExpression:
         file_column_name = self.file_schema.find_column_name(predicate.term.ref().field.field_id)
-        field_name = predicate.term.ref().field.name
 
         if file_column_name is None:
             # In the case of schema evolution, the column might not be present
             # in the file schema when reading older data
             if isinstance(predicate, BoundIsNull):
                 return AlwaysTrue()
-            # Projected fields are only available for identity partition fields
-            # Which mean that partition pruning excluded partition field which evaluates to false
-            elif field_name in self.projected_missing_fields:
-                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()
 
@@ -919,8 +935,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, projected_missing_fields: dict[str, Any]) -> BooleanExpression:
-    return visit(expr, _ColumnNameTranslator(file_schema, case_sensitive, projected_missing_fields))
+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

@@ -1412,7 +1412,13 @@ def _task_to_record_batches(
 
         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, projected_missing_fields=projected_missing_fields)
+            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)