fix(arrow): evaluate filters on columns missing from a data file against their identity partition value or initial-default

crates/iceberg/src/arrow/reader/pipeline.rs

@@ -33,6 +33,7 @@ use parquet::arrow::{
 };
 use parquet::encryption::decrypt::FileDecryptionProperties;
 
+use super::predicate_visitor::residual_for_missing_fields;
 use super::row_lineage::synthesize_row_id_column;
 use super::{
     ArrowFileReader, ArrowReader, ParquetReadOptions, add_fallback_field_ids_to_arrow_schema,
@@ -535,6 +536,16 @@ impl FileScanTaskReader {
                 &predicate,
                 use_position_fallback,
             )?;
+            // Apply the residual before the row filter and every pruning layer below, so they all
+            // see the same predicate.
+            let predicate = residual_for_missing_fields(
+                predicate,
+                &iceberg_field_ids,
+                &field_id_map,
+                task.schema(),
+                task.partition_spec().map(Arc::as_ref),
+                task.partition(),
+            )?;
 
             let row_filter = ArrowReader::get_row_filter(
                 &predicate,