fix(expr): widen promoted partition values and bounds to prevent false pruning

crates/iceberg/src/expr/visitors/inclusive_metrics_evaluator.rs

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::borrow::Cow;
+
 use fnv::FnvHashSet;
 
 use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor, visit};
@@ -120,6 +122,7 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
         };
 
         if let Some(bound) = bound {
+            let bound = &*Self::promote_bound(bound, reference);
             if cmp_fn(bound, datum) {
                 return ROWS_MIGHT_MATCH;
             }
@@ -129,6 +132,19 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
 
         ROWS_MIGHT_MATCH
     }
+
+    /// Widens a stored bound to the reference's type so a bound written under a
+    /// narrower type (e.g. an `int` column later promoted to `long`) compares
+    /// against a predicate bound to the wider type. Borrows the bound when nothing
+    /// is promoted, so unpromoted bounds (e.g. strings/binaries) are not copied.
+    fn promote_bound<'b>(bound: &'b Datum, reference: &BoundReference) -> Cow<'b, Datum> {
+        reference
+            .field()
+            .field_type
+            .as_primitive_type()
+            .and_then(|primitive_type| bound.promote_to(primitive_type))
+            .map_or(Cow::Borrowed(bound), Cow::Owned)
+    }
 }
 
 impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
@@ -259,6 +275,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
         }
 
         if let Some(lower_bound) = self.lower_bound(field_id) {
+            let lower_bound = &*Self::promote_bound(lower_bound, reference);
             if lower_bound.is_nan() {
                 // NaN indicates unreliable bounds.
                 // See the InclusiveMetricsEvaluator docs for more.
@@ -269,6 +286,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
         }
 
         if let Some(upper_bound) = self.upper_bound(field_id) {
+            let upper_bound = &*Self::promote_bound(upper_bound, reference);
             if upper_bound.is_nan() {
                 // NaN indicates unreliable bounds.
                 // See the InclusiveMetricsEvaluator docs for more.
@@ -438,6 +456,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
         }
 
         if let Some(lower_bound) = self.lower_bound(field_id) {
+            let lower_bound = &*Self::promote_bound(lower_bound, reference);
             if lower_bound.is_nan() {
                 // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more.
                 return ROWS_MIGHT_MATCH;
@@ -450,6 +469,7 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
         }
 
         if let Some(upper_bound) = self.upper_bound(field_id) {
+            let upper_bound = &*Self::promote_bound(upper_bound, reference);
             if upper_bound.is_nan() {
                 // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more.
                 return ROWS_MIGHT_MATCH;
@@ -521,6 +541,92 @@ mod test {
         assert!(result);
     }
 
+    fn promotion_schema(primitive: PrimitiveType) -> SchemaRef {
+        Arc::new(
+            Schema::builder()
+                .with_fields(vec![Arc::new(NestedField::optional(
+                    1,
+                    "a",
+                    Type::Primitive(primitive),
+                ))])
+                .build()
+                .unwrap(),
+        )
+    }
+
+    fn data_file_with_bounds(lower: Datum, upper: Datum) -> DataFile {
+        DataFile {
+            lower_bounds: HashMap::from([(1, lower)]),
+            upper_bounds: HashMap::from([(1, upper)]),
+            ..create_test_data_file()
+        }
+    }
+
+    #[test]
+    fn eval_promotes_int_bounds_for_inequality() {
+        // A column promoted int->long keeps int-typed column bounds in
+        // pre-promotion manifests; a predicate bound to the promoted long type
+        // must still match a file whose range covers it, not silently prune it.
+        let schema = promotion_schema(PrimitiveType::Long);
+        let data_file = data_file_with_bounds(Datum::int(5), Datum::int(10));
+        let predicate = Predicate::Binary(BinaryExpression::new(
+            GreaterThan,
+            Reference::new("a"),
+            Datum::long(4),
+        ))
+        .bind(schema, true)
+        .unwrap();
+
+        assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+    }
+
+    #[test]
+    fn eval_promotes_int_bounds_for_in() {
+        let schema = promotion_schema(PrimitiveType::Long);
+        let data_file = data_file_with_bounds(Datum::int(5), Datum::int(10));
+        let predicate = Predicate::Set(SetExpression::new(
+            In,
+            Reference::new("a"),
+            FnvHashSet::from_iter([Datum::long(5), Datum::long(6)]),
+        ))
+        .bind(schema, true)
+        .unwrap();
+
+        assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+    }
+
+    #[test]
+    fn eval_promotes_int_bounds_for_eq() {
+        // With the bounds widened, `== 5` correctly prunes a file whose range is
+        // entirely above 5 (int bounds [6, 10]) instead of failing open.
+        let schema = promotion_schema(PrimitiveType::Long);
+        let data_file = data_file_with_bounds(Datum::int(6), Datum::int(10));
+        let predicate = Predicate::Binary(BinaryExpression::new(
+            Eq,
+            Reference::new("a"),
+            Datum::long(5),
+        ))
+        .bind(schema, true)
+        .unwrap();
+
+        assert!(!InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+    }
+
+    #[test]
+    fn eval_promotes_float_bounds_for_inequality() {
+        let schema = promotion_schema(PrimitiveType::Double);
+        let data_file = data_file_with_bounds(Datum::float(1.5_f32), Datum::float(2.0_f32));
+        let predicate = Predicate::Binary(BinaryExpression::new(
+            GreaterThan,
+            Reference::new("a"),
+            Datum::double(1.0),
+        ))
+        .bind(schema, true)
+        .unwrap();
+
+        assert!(InclusiveMetricsEvaluator::eval(&predicate, &data_file, true).unwrap());
+    }
+
     #[test]
     fn test_all_nulls() {
         let result =

crates/iceberg/src/scan/context.rs

@@ -30,7 +30,8 @@ use crate::scan::{
 };
 use crate::spec::{
     ManifestContentType, ManifestEntryRef, ManifestFile, ManifestList, NameMapping,
-    PartitionSpecRef, SchemaRef, SnapshotRef, SortOrderRef, StructType, TableMetadataRef,
+    PartitionSpecRef, PrimitiveType, SchemaRef, SnapshotRef, SortOrderRef, StructType,
+    TableMetadataRef,
 };
 use crate::{Error, ErrorKind, Result};
 
@@ -73,6 +74,28 @@ pub(crate) struct ManifestEntryContext {
     pub sort_order: Option<SortOrderRef>,
 }
 
+/// Returns whether any `int` or `float` field of `written` is widened in `current`, to
+/// `long`/`double` or to `unknown` once its source column is dropped, i.e. partition
+/// values written with `written` need normalizing before they compare against predicates
+/// bound to the current schema or against delete-file partitions.
+fn partition_types_differ(current: &StructType, written: &StructType) -> bool {
+    current.fields().iter().zip(written.fields()).any(|(c, w)| {
+        matches!(
+            (
+                w.field_type.as_primitive_type(),
+                c.field_type.as_primitive_type()
+            ),
+            (
+                Some(PrimitiveType::Int),
+                Some(PrimitiveType::Long | PrimitiveType::Unknown)
+            ) | (
+                Some(PrimitiveType::Float),
+                Some(PrimitiveType::Double | PrimitiveType::Unknown)
+            )
+        )
+    })
+}
+
 impl ManifestFileContext {
     /// Consumes this [`ManifestFileContext`], fetching its Manifest from FileIO and then
     /// streaming its constituent [`ManifestEntries`] to the channel provided in the context
@@ -95,7 +118,33 @@ impl ManifestFileContext {
 
         let manifest = object_cache.get_manifest(&manifest_file).await?;
 
+        // A manifest written under an older schema stores partition values in that
+        // schema's (narrower) types. When the current partition type widens the one the
+        // manifest was written with (e.g. an int source column promoted to long), widen
+        // each entry's partition tuple here so the pruning evaluator and the delete-file
+        // index compare like-typed values. The gate compares the partition types
+        // themselves, not schema ids, since v1 may omit them (defaulting to 0). Column
+        // bounds are widened lazily where the metrics evaluator compares them.
+        let promotion_target: Option<(StructType, StructType)> = match partition_spec.as_ref() {
+            Some(spec) => {
+                let current = spec.partition_type(&snapshot_schema)?;
+                let written = spec.partition_type(manifest.metadata().schema())?;
+                partition_types_differ(&current, &written).then_some((current, written))
+            }
+            None => None,
+        };
+
         for manifest_entry in manifest.entries() {
+            let manifest_entry = match promotion_target.as_ref() {
+                Some((current, written)) => {
+                    let mut entry = manifest_entry.as_ref().clone();
+                    entry.data_file.partition =
+                        entry.data_file.promoted_partition(current, written)?;
+                    Arc::new(entry)
+                }
+                None => manifest_entry.clone(),
+            };
+
             // Carry the raw id through unresolved, then resolve it to an order for the
             // reader. The resolved order is `None` for the unsorted order (empty fields) so
             // that `Some` always means genuinely sorted, matching Java's `isSorted()` gate;
@@ -108,8 +157,7 @@ impl ManifestFileContext {
                 .cloned();
 
             let manifest_entry_context = ManifestEntryContext {
-                // TODO: refactor to avoid the expensive ManifestEntry clone
-                manifest_entry: manifest_entry.clone(),
+                manifest_entry,
                 expression_evaluator_cache: expression_evaluator_cache.clone(),
                 field_ids: field_ids.clone(),
                 partition_spec_id: manifest_file.partition_spec_id,