fix(expr): prune In predicates that straddle metrics bounds

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

@@ -437,28 +437,17 @@ impl BoundPredicateVisitor for InclusiveMetricsEvaluator<'_> {
             return ROWS_MIGHT_MATCH;
         }
 
-        if let Some(lower_bound) = self.lower_bound(field_id) {
-            if lower_bound.is_nan() {
-                // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more.
-                return ROWS_MIGHT_MATCH;
-            }
-
-            if !literals.iter().any(|datum| datum.ge(lower_bound)) {
-                // if all values are less than lower bound, rows cannot match.
-                return ROWS_CANNOT_MATCH;
-            }
-        }
-
-        if let Some(upper_bound) = self.upper_bound(field_id) {
-            if upper_bound.is_nan() {
-                // NaN indicates unreliable bounds. See the InclusiveMetricsEvaluator docs for more.
-                return ROWS_MIGHT_MATCH;
-            }
-
-            if !literals.iter().any(|datum| datum.le(upper_bound)) {
-                // if all values are greater than upper bound, rows cannot match.
-                return ROWS_CANNOT_MATCH;
-            }
+        let lower_bound = self.lower_bound(field_id);
+        let upper_bound = self.upper_bound(field_id);
+
+        // A NaN bound is unreliable on that side only. Drop it to unbounded
+        // so a valid bound can still prune.
+        if !super::any_literal_in_bounds(
+            super::finite_bound(lower_bound),
+            super::finite_bound(upper_bound),
+            literals,
+        ) {
+            return ROWS_CANNOT_MATCH;
         }
 
         ROWS_MIGHT_MATCH
@@ -1551,6 +1540,59 @@ mod test {
         );
     }
 
+    #[test]
+    fn test_integer_in_straddling_bounds() {
+        let result = InclusiveMetricsEvaluator::eval(
+            &r#in_int("id", &[INT_MIN_VALUE - 25, INT_MAX_VALUE + 25]),
+            &get_test_file_1(),
+            true,
+        )
+        .unwrap();
+        assert!(!result, "Should skip: id in (5, 104), bounds are [30, 79]");
+    }
+
+    #[test]
+    fn test_float_in_nan_upper_bound_prunes_below_lower() {
+        let result = InclusiveMetricsEvaluator::eval(
+            &r#in_float("no_nans", &[2.0, 3.0]),
+            &get_test_file_float_nan_upper(),
+            true,
+        )
+        .unwrap();
+        assert!(
+            !result,
+            "Should skip: NaN upper is unbounded, both literals are below lower 4.0"
+        );
+    }
+
+    #[test]
+    fn test_float_in_nan_lower_bound_prunes_above_upper() {
+        let result = InclusiveMetricsEvaluator::eval(
+            &r#in_float("no_nans", &[2.0, 3.0]),
+            &get_test_file_float_nan_lower(),
+            true,
+        )
+        .unwrap();
+        assert!(
+            !result,
+            "Should skip: NaN lower is unbounded, both literals are above upper 1.0"
+        );
+    }
+
+    #[test]
+    fn test_float_in_nan_lower_bound_does_not_prune_when_a_literal_is_inside_upper() {
+        let result = InclusiveMetricsEvaluator::eval(
+            &r#in_float("no_nans", &[2.0, 4.0]),
+            &get_test_file_float_nan_lower_upper_3(),
+            true,
+        )
+        .unwrap();
+        assert!(
+            result,
+            "Should read: NaN lower is unbounded, 2.0 is inside upper 3.0"
+        );
+    }
+
     #[test]
     fn test_integer_not_in() {
         let result = InclusiveMetricsEvaluator::eval(
@@ -1880,6 +1922,16 @@ mod test {
         filter.bind(schema.clone(), true).unwrap()
     }
 
+    fn in_float(reference: &str, float_literals: &[f32]) -> BoundPredicate {
+        let schema = create_test_schema();
+        let filter = Predicate::Set(SetExpression::new(
+            In,
+            Reference::new(reference),
+            FnvHashSet::from_iter(float_literals.iter().copied().map(Datum::float)),
+        ));
+        filter.bind(schema.clone(), true).unwrap()
+    }
+
     fn not_in_int(reference: &str, int_literals: &[i32]) -> BoundPredicate {
         let schema = create_test_schema();
         let filter = Predicate::Set(SetExpression::new(
@@ -2093,6 +2145,85 @@ mod test {
             content_size_in_bytes: None,
         }
     }
+
+    fn get_test_file_float_nan_upper() -> DataFile {
+        DataFile {
+            content: DataContentType::Data,
+            file_path: "/test/path".to_string(),
+            file_format: DataFileFormat::Parquet,
+            partition: Struct::empty(),
+            record_count: 10,
+            file_size_in_bytes: 10,
+            column_sizes: Default::default(),
+            value_counts: HashMap::from([(9, 10)]),
+            null_value_counts: HashMap::from([(9, 0)]),
+            nan_value_counts: HashMap::from([(9, 0)]),
+            lower_bounds: HashMap::from([(9, Datum::float(4.0_f32))]),
+            upper_bounds: HashMap::from([(9, Datum::float(f32::NAN))]),
+            key_metadata: None,
+            split_offsets: None,
+            equality_ids: None,
+            sort_order_id: None,
+            partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
+        }
+    }
+
+    fn get_test_file_float_nan_lower() -> DataFile {
+        DataFile {
+            content: DataContentType::Data,
+            file_path: "/test/path".to_string(),
+            file_format: DataFileFormat::Parquet,
+            partition: Struct::empty(),
+            record_count: 10,
+            file_size_in_bytes: 10,
+            column_sizes: Default::default(),
+            value_counts: HashMap::from([(9, 10)]),
+            null_value_counts: HashMap::from([(9, 0)]),
+            nan_value_counts: HashMap::from([(9, 0)]),
+            lower_bounds: HashMap::from([(9, Datum::float(f32::NAN))]),
+            upper_bounds: HashMap::from([(9, Datum::float(1.0_f32))]),
+            key_metadata: None,
+            split_offsets: None,
+            equality_ids: None,
+            sort_order_id: None,
+            partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
+        }
+    }
+
+    fn get_test_file_float_nan_lower_upper_3() -> DataFile {
+        DataFile {
+            content: DataContentType::Data,
+            file_path: "/test/path".to_string(),
+            file_format: DataFileFormat::Parquet,
+            partition: Struct::empty(),
+            record_count: 10,
+            file_size_in_bytes: 10,
+            column_sizes: Default::default(),
+            value_counts: HashMap::from([(9, 10)]),
+            null_value_counts: HashMap::from([(9, 0)]),
+            nan_value_counts: HashMap::from([(9, 0)]),
+            lower_bounds: HashMap::from([(9, Datum::float(f32::NAN))]),
+            upper_bounds: HashMap::from([(9, Datum::float(3.0_f32))]),
+            key_metadata: None,
+            split_offsets: None,
+            equality_ids: None,
+            sort_order_id: None,
+            partition_spec_id: 0,
+            first_row_id: None,
+            referenced_data_file: None,
+            content_offset: None,
+            content_size_in_bytes: None,
+        }
+    }
+
     fn get_test_file_2() -> DataFile {
         DataFile {
             content: DataContentType::Data,

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

@@ -401,6 +401,9 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
         _predicate: &BoundPredicate,
     ) -> Result<bool> {
         let field = self.field_summary_for_reference(reference);
+        // A missing lower_bound covers all-null, all-NaN, and mixed null+NaN
+        // summaries (Iceberg spec). Those cases prune here and never reach
+        // any_literal_in_bounds.
         if field.lower_bound.is_none() {
             return ROWS_CANNOT_MATCH;
         }
@@ -409,20 +412,15 @@ impl BoundPredicateVisitor for ManifestFilterVisitor<'_> {
             return ROWS_MIGHT_MATCH;
         }
 
-        if let Some(lower_bound) = &field.lower_bound {
-            let lower_bound =
-                ManifestFilterVisitor::bytes_to_datum(lower_bound, &reference.field().field_type);
-            if literals.iter().all(|datum| &lower_bound > datum) {
-                return ROWS_CANNOT_MATCH;
-            }
-        }
+        let lower_bound = field.lower_bound.as_ref().map(|bound| {
+            ManifestFilterVisitor::bytes_to_datum(bound, reference.field().field_type.as_ref())
+        });
+        let upper_bound = field.upper_bound.as_ref().map(|bound| {
+            ManifestFilterVisitor::bytes_to_datum(bound, reference.field().field_type.as_ref())
+        });
 
-        if let Some(upper_bound) = &field.upper_bound {
-            let upper_bound =
-                ManifestFilterVisitor::bytes_to_datum(upper_bound, &reference.field().field_type);
-            if literals.iter().all(|datum| &upper_bound < datum) {
-                return ROWS_CANNOT_MATCH;
-            }
+        if !super::any_literal_in_bounds(lower_bound.as_ref(), upper_bound.as_ref(), literals) {
+            return ROWS_CANNOT_MATCH;
         }
 
         ROWS_MIGHT_MATCH
@@ -1371,6 +1369,30 @@ mod test {
         Ok(())
     }
 
+    #[test]
+    fn test_in_straddling_bounds() -> Result<()> {
+        let case_sensitive = true;
+        let schema = create_schema()?;
+        let manifest_file = create_manifest_file(create_partitions());
+
+        let filter = Predicate::Set(SetExpression::new(
+            PredicateOperator::In,
+            Reference::new("id"),
+            FnvHashSet::from_iter(vec![
+                Datum::int(INT_MIN_VALUE - 25),
+                Datum::int(INT_MAX_VALUE + 25),
+            ]),
+        ))
+        .bind(schema.clone(), case_sensitive)?;
+        assert!(
+            !ManifestEvaluator::builder(filter)
+                .build()
+                .eval(&manifest_file)?,
+            "Should not read: id in (5, 104), summary is [30, 79]"
+        );
+        Ok(())
+    }
+
     #[test]
     fn test_not_in() -> Result<()> {
         let case_sensitive = true;

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

@@ -15,6 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use fnv::FnvHashSet;
+
+use crate::spec::Datum;
+
 pub(crate) mod bloom_filter_evaluator;
 pub(crate) mod bound_predicate_visitor;
 pub(crate) mod expression_evaluator;
@@ -27,3 +31,102 @@ pub(crate) mod rewrite_not;
 pub(crate) mod row_group_metrics_evaluator;
 pub(crate) mod strict_metrics_evaluator;
 pub(crate) mod strict_projection;
+
+/// Returns true if any literal could match the inclusive `[lower, upper]` range.
+/// Missing bounds are treated as unbounded on that side.
+///
+/// `(None, None)` returns true because no bound is available to prune against.
+pub(crate) fn any_literal_in_bounds(
+    lower: Option<&Datum>,
+    upper: Option<&Datum>,
+    literals: &FnvHashSet<Datum>,
+) -> bool {
+    match (lower, upper) {
+        (Some(lower), Some(upper)) => literals
+            .iter()
+            .any(|datum| datum.ge(lower) && datum.le(upper)),
+        (Some(lower), None) => literals.iter().any(|datum| datum.ge(lower)),
+        (None, Some(upper)) => literals.iter().any(|datum| datum.le(upper)),
+        (None, None) => true,
+    }
+}
+
+/// Drops a NaN bound so that side is treated as unbounded.
+///
+/// A NaN min or max is unreliable, but the other bound may still prune.
+/// Inclusive evaluators in Java and PyIceberg bail to might-match when
+/// either bound is NaN. Dropping only the NaN side is a deliberate
+/// divergence: `total_cmp` treats NaN as the maximum, so the remaining
+/// finite bound is still a valid prune.
+pub(crate) fn finite_bound(bound: Option<&Datum>) -> Option<&Datum> {
+    bound.filter(|datum| !datum.is_nan())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn floats(vals: &[f32]) -> FnvHashSet<Datum> {
+        vals.iter().copied().map(Datum::float).collect()
+    }
+
+    #[test]
+    fn both_bounds_require_a_literal_inside_the_range() {
+        let lower = Datum::float(4.0_f32);
+        let upper = Datum::float(6.0_f32);
+        assert!(!any_literal_in_bounds(
+            Some(&lower),
+            Some(&upper),
+            &floats(&[2.0, 8.0])
+        ));
+        assert!(any_literal_in_bounds(
+            Some(&lower),
+            Some(&upper),
+            &floats(&[2.0, 5.0])
+        ));
+    }
+
+    #[test]
+    fn lower_only_prunes_literals_below_the_bound() {
+        let lower = Datum::float(4.0_f32);
+        assert!(!any_literal_in_bounds(
+            Some(&lower),
+            None,
+            &floats(&[2.0, 3.0])
+        ));
+        assert!(any_literal_in_bounds(
+            Some(&lower),
+            None,
+            &floats(&[2.0, 4.0])
+        ));
+    }
+
+    #[test]
+    fn upper_only_prunes_literals_above_the_bound() {
+        let upper = Datum::float(1.0_f32);
+        assert!(!any_literal_in_bounds(
+            None,
+            Some(&upper),
+            &floats(&[2.0, 3.0])
+        ));
+        assert!(any_literal_in_bounds(
+            None,
+            Some(&upper),
+            &floats(&[0.5, 3.0])
+        ));
+    }
+
+    #[test]
+    fn neither_bound_cannot_prune() {
+        assert!(any_literal_in_bounds(None, None, &floats(&[2.0, 3.0])));
+    }
+
+    #[test]
+    fn finite_bound_drops_nan() {
+        let nan = Datum::float(f32::NAN);
+        let finite = Datum::float(4.0_f32);
+        assert!(finite_bound(Some(&nan)).is_none());
+        assert_eq!(finite_bound(Some(&finite)), Some(&finite));
+        assert!(finite_bound(None).is_none());
+    }
+}