project value in _ColumnNameTranslator

Author: kevinjqliu

pyiceberg/expressions/visitors.py

@@ -23,6 +23,7 @@
     Dict,
     Generic,
     List,
+    Optional,
     Set,
     SupportsFloat,
     Tuple,
@@ -861,6 +862,7 @@ class _ColumnNameTranslator(BooleanExpressionVisitor[BooleanExpression]):
     Args:
       file_schema (Schema): The schema of the file.
       case_sensitive (bool): Whether to consider case when binding a reference to a field in a schema, defaults to True.
+      projected_fields (Dict[str, Any]): Partition field values for missing fields from projection.
 
     Raises:
         TypeError: In the case of an UnboundPredicate.
@@ -869,10 +871,12 @@ class _ColumnNameTranslator(BooleanExpressionVisitor[BooleanExpression]):
 
     file_schema: Schema
     case_sensitive: bool
+    projected_fields: Dict[str, Any]
 
-    def __init__(self, file_schema: Schema, case_sensitive: bool) -> None:
+    def __init__(self, file_schema: Schema, case_sensitive: bool, projected_fields: Optional[Dict[str, Any]] = None) -> None:
         self.file_schema = file_schema
         self.case_sensitive = case_sensitive
+        self.projected_fields = projected_fields or {}
 
     def visit_true(self) -> BooleanExpression:
         return AlwaysTrue()
@@ -912,7 +916,9 @@ def visit_bound_predicate(self, predicate: BoundPredicate[L]) -> BooleanExpressi
 
             return (
                 AlwaysTrue()
-                if expression_evaluator(Schema(field), pred, case_sensitive=self.case_sensitive)(Record(field.initial_default))
+                if expression_evaluator(Schema(field), pred, case_sensitive=self.case_sensitive)(
+                    Record(field.initial_default or self.projected_fields.get(field.name, None))
+                )
                 else AlwaysFalse()
             )
 
@@ -926,8 +932,10 @@ 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, case_sensitive: bool, projected_fields: Optional[Dict[str, Any]] = None
+) -> BooleanExpression:
+    return visit(expr, _ColumnNameTranslator(file_schema, case_sensitive, projected_fields))
 
 
 class _ProjectedColumnsEvaluator(BooleanExpressionVisitor[BooleanExpression]):

pyiceberg/io/pyarrow.py

@@ -78,7 +78,6 @@
 from pyiceberg.expressions.visitors import (
     BoundBooleanExpressionVisitor,
     bind,
-    evaluate_projected_columns,
     extract_field_ids,
     translate_column_names,
 )
@@ -1469,17 +1468,8 @@ def _task_to_record_batches(
 
         pyarrow_filter = None
         if bound_row_filter is not AlwaysTrue():
-            evaluated_projected_columns_filter = evaluate_projected_columns(
-                bound_row_filter,
-                file_schema,
-                projected_schema,
-                case_sensitive=case_sensitive,
-                projected_missing_fields=projected_missing_fields,
-            )
             translated_row_filter = translate_column_names(
-                evaluated_projected_columns_filter,
-                file_schema,
-                case_sensitive=case_sensitive,
+                bound_row_filter, file_schema, case_sensitive=case_sensitive, projected_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)