refactor(expr): evaluate inclusive metrics from borrowed statistics

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

@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::collections::HashMap;
+
 use fnv::FnvHashSet;
 
 use crate::expr::visitors::bound_predicate_visitor::{BoundPredicateVisitor, visit};
@@ -26,15 +28,44 @@ const IN_PREDICATE_LIMIT: usize = 200;
 const ROWS_MIGHT_MATCH: crate::Result<bool> = Ok(true);
 const ROWS_CANNOT_MATCH: crate::Result<bool> = Ok(false);
 
-pub(crate) struct InclusiveMetricsEvaluator<'a> {
-    data_file: &'a DataFile,
+/// Borrowed whole-file column statistics, keyed by Iceberg field ID.
+///
+/// The statistics evaluator only reads these maps, so callers that hold them
+/// outside a [`DataFile`] can be evaluated without building one.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct FileMetrics<'a> {
+    /// Number of records in the file, when known.
+    pub(crate) record_count: Option<u64>,
+    /// Number of values, including nulls and NaNs.
+    pub(crate) value_counts: &'a HashMap<i32, u64>,
+    /// Number of null values.
+    pub(crate) null_value_counts: &'a HashMap<i32, u64>,
+    /// Number of NaN values.
+    pub(crate) nan_value_counts: &'a HashMap<i32, u64>,
+    /// Inclusive lower bounds.
+    pub(crate) lower_bounds: &'a HashMap<i32, Datum>,
+    /// Inclusive upper bounds.
+    pub(crate) upper_bounds: &'a HashMap<i32, Datum>,
 }
 
-impl<'a> InclusiveMetricsEvaluator<'a> {
-    fn new(data_file: &'a DataFile) -> Self {
-        InclusiveMetricsEvaluator { data_file }
+impl<'a> From<&'a DataFile> for FileMetrics<'a> {
+    fn from(data_file: &'a DataFile) -> Self {
+        Self {
+            record_count: Some(data_file.record_count),
+            value_counts: &data_file.value_counts,
+            null_value_counts: &data_file.null_value_counts,
+            nan_value_counts: &data_file.nan_value_counts,
+            lower_bounds: &data_file.lower_bounds,
+            upper_bounds: &data_file.upper_bounds,
+        }
     }
+}
+
+pub(crate) struct InclusiveMetricsEvaluator<'a> {
+    metrics: FileMetrics<'a>,
+}
 
+impl<'a> InclusiveMetricsEvaluator<'a> {
     /// Evaluate this `InclusiveMetricsEvaluator`'s filter predicate against the
     /// provided [`DataFile`]'s metrics. Used by [`TableScan`] to
     /// see if this `DataFile` contains data that could match
@@ -44,32 +75,45 @@ impl<'a> InclusiveMetricsEvaluator<'a> {
         data_file: &'a DataFile,
         include_empty_files: bool,
     ) -> crate::Result<bool> {
-        if !include_empty_files && data_file.record_count == 0 {
+        Self::eval_metrics(filter, data_file.into(), include_empty_files)
+    }
+
+    /// Evaluate `filter` against borrowed whole-file statistics. A file whose
+    /// record count is known to be zero cannot match unless
+    /// `include_empty_files` is set. An unknown record count does not by
+    /// itself exclude the file, but the available bounds and counts may still
+    /// prune it.
+    pub(crate) fn eval_metrics(
+        filter: &'a BoundPredicate,
+        metrics: FileMetrics<'a>,
+        include_empty_files: bool,
+    ) -> crate::Result<bool> {
+        if !include_empty_files && metrics.record_count == Some(0) {
             return ROWS_CANNOT_MATCH;
         }
 
-        let mut evaluator = Self::new(data_file);
+        let mut evaluator = Self { metrics };
         visit(&mut evaluator, filter)
     }
 
     fn nan_count(&self, field_id: i32) -> Option<&u64> {
-        self.data_file.nan_value_counts.get(&field_id)
+        self.metrics.nan_value_counts.get(&field_id)
     }
 
     fn null_count(&self, field_id: i32) -> Option<&u64> {
-        self.data_file.null_value_counts.get(&field_id)
+        self.metrics.null_value_counts.get(&field_id)
     }
 
     fn value_count(&self, field_id: i32) -> Option<&u64> {
-        self.data_file.value_counts.get(&field_id)
+        self.metrics.value_counts.get(&field_id)
     }
 
     fn lower_bound(&self, field_id: i32) -> Option<&Datum> {
-        self.data_file.lower_bounds.get(&field_id)
+        self.metrics.lower_bounds.get(&field_id)
     }
 
     fn upper_bound(&self, field_id: i32) -> Option<&Datum> {
-        self.data_file.upper_bounds.get(&field_id)
+        self.metrics.upper_bounds.get(&field_id)
     }
 
     fn contains_nans_only(&self, field_id: i32) -> bool {
@@ -489,7 +533,9 @@ mod test {
         Eq, GreaterThan, GreaterThanOrEq, In, IsNan, IsNull, LessThan, LessThanOrEq, NotEq, NotIn,
         NotNan, NotNull, NotStartsWith, StartsWith,
     };
-    use crate::expr::visitors::inclusive_metrics_evaluator::InclusiveMetricsEvaluator;
+    use crate::expr::visitors::inclusive_metrics_evaluator::{
+        FileMetrics, InclusiveMetricsEvaluator,
+    };
     use crate::expr::{
         BinaryExpression, Bind, BoundPredicate, Predicate, Reference, SetExpression,
         UnaryExpression,
@@ -946,6 +992,48 @@ mod test {
         assert!(!result, "Should skip: or(false, false)");
     }
 
+    #[test]
+    fn test_file_metrics_from_independent_maps() {
+        // Whole-file statistics held outside any DataFile: `id` (field 1)
+        // spans 10..=20 with no nulls; `no_stats` (field 2) has none.
+        let value_counts = HashMap::from([(1, 5)]);
+        let null_value_counts = HashMap::from([(1, 0)]);
+        let nan_value_counts = HashMap::new();
+        let lower_bounds = HashMap::from([(1, Datum::int(10))]);
+        let upper_bounds = HashMap::from([(1, Datum::int(20))]);
+        let metrics = |record_count| FileMetrics {
+            record_count,
+            value_counts: &value_counts,
+            null_value_counts: &null_value_counts,
+            nan_value_counts: &nan_value_counts,
+            lower_bounds: &lower_bounds,
+            upper_bounds: &upper_bounds,
+        };
+        let eval = |predicate: &BoundPredicate, record_count, include_empty_files| {
+            InclusiveMetricsEvaluator::eval_metrics(
+                predicate,
+                metrics(record_count),
+                include_empty_files,
+            )
+            .unwrap()
+        };
+
+        // Bounds prune or keep the file regardless of the record count.
+        for record_count in [Some(5), None] {
+            assert!(!eval(&less_than_int("id", 5), record_count, false));
+            assert!(!eval(&greater_than_int("id", 20), record_count, false));
+            assert!(eval(&less_than_int("id", 15), record_count, false));
+            assert!(!eval(&is_null("id"), record_count, false));
+            assert!(eval(&not_null("id"), record_count, false));
+            // Missing statistics never exclude.
+            assert!(eval(&is_null("no_stats"), record_count, false));
+        }
+
+        // A known empty file is skipped unless empty files are included.
+        assert!(!eval(&less_than_int("id", 15), Some(0), false));
+        assert!(eval(&less_than_int("id", 15), Some(0), true));
+    }
+
     #[test]
     fn test_integer_lt() {
         let result = InclusiveMetricsEvaluator::eval(