fix(reader): keep matching rows when filtering a promoted column

A file written before an int -> long, float -> double, or decimal
precision promotion keeps its old physical type, while the predicate is
bound to the table type. The row filter cast the literal down to the
file's Arrow type, so an out-of-range literal became null and dropped
every row, and a double literal was rounded to f32.

Cast a numeric column to the table type before comparing instead, as
RecordBatchTransformer does for the output. Other columns keep the
literal cast. Page-index pruning is left to #3507 and #3521.

Part of #3122.

Author: comphead

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

@@ -34,7 +34,7 @@ use arrow_string::like::starts_with;
 use fnv::FnvHashSet;
 use parquet::schema::types::SchemaDescriptor;
 
-use crate::arrow::get_arrow_datum;
+use crate::arrow::{get_arrow_datum, type_to_arrow_type};
 use crate::error::{Result, invalid_data};
 use crate::expr::visitors::bound_predicate_visitor::BoundPredicateVisitor;
 use crate::expr::{BoundPredicate, BoundReference};
@@ -278,6 +278,22 @@ fn project_column(
     }
 }
 
+/// Like `project_column`, but casts a numeric column the file stores in another type to
+/// `target_type`, as `RecordBatchTransformer` does. Narrowing the literal to the file type
+/// instead can turn it into null or round it.
+fn project_promoted_column(
+    batch: &RecordBatch,
+    column_idx: usize,
+    target_type: &DataType,
+) -> std::result::Result<ArrayRef, ArrowError> {
+    let column = project_column(batch, column_idx)?;
+    if column.data_type().is_numeric() && column.data_type() != target_type {
+        cast(&column, target_type)
+    } else {
+        Ok(column)
+    }
+}
+
 fn compute_is_nan(array: &ArrayRef) -> std::result::Result<BooleanArray, ArrowError> {
     // Compute NaN over the contiguous values slice, then fold the null bitmap
     // in with a single bitwise AND so that null slots become false.
@@ -425,9 +441,10 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
     ) -> Result<Box<PredicateResult>> {
         if let Some(idx) = self.bound_reference(reference)? {
             let literal = get_arrow_datum(literal)?;
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
                 let literal = try_cast_literal(&literal, left.data_type())?;
                 lt(&left, literal.as_ref())
             }))
@@ -445,9 +462,10 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
     ) -> Result<Box<PredicateResult>> {
         if let Some(idx) = self.bound_reference(reference)? {
             let literal = get_arrow_datum(literal)?;
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
                 let literal = try_cast_literal(&literal, left.data_type())?;
                 lt_eq(&left, literal.as_ref())
             }))
@@ -465,9 +483,10 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
     ) -> Result<Box<PredicateResult>> {
         if let Some(idx) = self.bound_reference(reference)? {
             let literal = get_arrow_datum(literal)?;
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
                 let literal = try_cast_literal(&literal, left.data_type())?;
                 gt(&left, literal.as_ref())
             }))
@@ -485,9 +504,10 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
     ) -> Result<Box<PredicateResult>> {
         if let Some(idx) = self.bound_reference(reference)? {
             let literal = get_arrow_datum(literal)?;
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
                 let literal = try_cast_literal(&literal, left.data_type())?;
                 gt_eq(&left, literal.as_ref())
             }))
@@ -505,9 +525,10 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
     ) -> Result<Box<PredicateResult>> {
         if let Some(idx) = self.bound_reference(reference)? {
             let literal = get_arrow_datum(literal)?;
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
                 let literal = try_cast_literal(&literal, left.data_type())?;
                 eq(&left, literal.as_ref())
             }))
@@ -525,9 +546,10 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
     ) -> Result<Box<PredicateResult>> {
         if let Some(idx) = self.bound_reference(reference)? {
             let literal = get_arrow_datum(literal)?;
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
                 let literal = try_cast_literal(&literal, left.data_type())?;
                 neq(&left, literal.as_ref())
             }))
@@ -589,10 +611,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
                 .iter()
                 .map(|lit| get_arrow_datum(lit).unwrap())
                 .collect();
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
                 // update this if arrow ever adds a native is_in kernel
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
 
                 let mut acc = constant_bool_array(false, batch.num_rows());
                 for literal in &literals {
@@ -619,10 +642,11 @@ impl BoundPredicateVisitor for PredicateConverter<'_> {
                 .iter()
                 .map(|lit| get_arrow_datum(lit).unwrap())
                 .collect();
+            let target_type = type_to_arrow_type(&reference.field().field_type)?;
 
             Ok(Box::new(move |batch| {
                 // update this if arrow ever adds a native not_in kernel
-                let left = project_column(&batch, idx)?;
+                let left = project_promoted_column(&batch, idx, &target_type)?;
                 let mut acc = constant_bool_array(true, batch.num_rows());
                 for literal in &literals {
                     let literal = try_cast_literal(literal, left.data_type())?;

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

@@ -228,8 +228,8 @@ mod tests {
 
     use arrow_array::cast::AsArray;
     use arrow_array::{
-        ArrayRef, Decimal128Array, Float32Array, Int32Array, Int64Array, LargeStringArray,
-        RecordBatch, StringArray,
+        ArrayRef, Decimal128Array, Float32Array, Float64Array, Int32Array, Int64Array,
+        LargeStringArray, RecordBatch, StringArray,
     };
     use arrow_schema::{DataType, Field, Schema as ArrowSchema};
     use futures::TryStreamExt;
@@ -460,6 +460,89 @@ mod tests {
         }
     }
 
+    /// Writes `column` as field 1, reads it back as `iceberg_type` under each predicate, and
+    /// checks the values that survive (`None` when no row does).
+    async fn assert_promoted_reads(
+        column: ArrayRef,
+        iceberg_type: PrimitiveType,
+        cases: Vec<(Predicate, Option<ArrayRef>)>,
+    ) {
+        let tmp = TempDir::new().unwrap();
+        let path = format!("{}/column.parquet", tmp.path().to_str().unwrap());
+        let arrow_schema = Arc::new(ArrowSchema::new(vec![field_with_id(
+            "v",
+            column.data_type().clone(),
+            1,
+        )]));
+        let batch = RecordBatch::try_new(arrow_schema.clone(), vec![column]).unwrap();
+        write_row_groups(&path, arrow_schema, vec![batch], false);
+
+        let schema = value_schema(Type::Primitive(iceberg_type));
+        let file_size = std::fs::metadata(&path).unwrap().len();
+        let reader = ArrowReaderBuilder::new(FileIO::new_with_fs(), Runtime::current()).build();
+        for (predicate, expected) in cases {
+            let task = FileScanTask::builder()
+                .with_file_size_in_bytes(file_size)
+                .with_start(0)
+                .with_length(0)
+                .with_data_file_path(path.clone())
+                .with_data_file_format(DataFileFormat::Parquet)
+                .with_schema(schema.clone())
+                .with_project_field_ids(vec![1])
+                .with_predicate(Some(predicate.bind(schema.clone(), true).unwrap()))
+                .with_case_sensitive(true)
+                .build()
+                .unwrap();
+            let tasks = Box::pin(futures::stream::iter(vec![Ok(task)])) as FileScanTaskStream;
+            let batches: Vec<RecordBatch> = reader
+                .clone()
+                .read(tasks)
+                .unwrap()
+                .stream()
+                .try_collect()
+                .await
+                .unwrap();
+            let actual = collapse(&batches).map(|batch| Arc::clone(batch.column(0)));
+            assert_eq!(actual, expected, "{predicate}");
+        }
+    }
+
+    /// A file written before a type promotion keeps the old physical type, so the row
+    /// filter must compare its values promoted to the table type.
+    #[tokio::test]
+    async fn test_predicate_on_promoted_column() {
+        let longs = |values: Vec<i64>| Some(Arc::new(Int64Array::from(values)) as ArrayRef);
+        let all = longs(vec![1, 2]);
+        let above = || Datum::long(i64::from(i32::MAX) + 1);
+        let below = || Datum::long(i64::from(i32::MIN) - 1);
+        let v = || Reference::new("v");
+        assert_promoted_reads(
+            Arc::new(Int32Array::from(vec![1, 2])),
+            PrimitiveType::Long,
+            vec![
+                (v().less_than(above()), all.clone()),
+                (v().less_than_or_equal_to(above()), all.clone()),
+                (v().greater_than_or_equal_to(below()), all.clone()),
+                (v().not_equal_to(above()), all),
+                (v().is_in([Datum::long(1), above()]), longs(vec![1])),
+                (v().is_not_in([Datum::long(1), above()]), longs(vec![2])),
+            ],
+        )
+        .await;
+
+        // 0.1f32 promotes to a double that is greater than 0.1, not equal to it.
+        let promoted = Arc::new(Float64Array::from(vec![f64::from(0.1f32)])) as ArrayRef;
+        assert_promoted_reads(
+            Arc::new(Float32Array::from(vec![0.1])),
+            PrimitiveType::Double,
+            vec![
+                (v().greater_than(Datum::double(0.1)), Some(promoted)),
+                (v().equal_to(Datum::double(0.1)), None),
+            ],
+        )
+        .await;
+    }
+
     /// Verifies that file splits respect byte ranges and only read specific row groups.
     #[tokio::test]
     async fn test_file_splits_respect_byte_ranges() {