fix(scan): prune IN predicates using file stats

crates/paimon/src/predicate_stats.rs

@@ -23,6 +23,9 @@ pub(crate) trait StatsAccessor {
     fn null_count(&self, index: usize) -> Option<i64>;
     fn min_value(&self, index: usize, data_type: &DataType) -> Option<Datum>;
     fn max_value(&self, index: usize, data_type: &DataType) -> Option<Datum>;
+    fn supports_in_min_max_pruning(&self) -> bool {
+        false
+    }
 }
 
 pub(crate) fn predicates_may_match_with_schema<T: StatsAccessor>(
@@ -59,7 +62,11 @@ pub(crate) fn data_leaf_may_match<T: StatsAccessor>(
         PredicateOperator::IsNotNull => {
             return all_null != Some(true);
         }
-        PredicateOperator::In | PredicateOperator::NotIn => {
+        PredicateOperator::In if stats.supports_in_min_max_pruning() => {}
+        PredicateOperator::In => {
+            return true;
+        }
+        PredicateOperator::NotIn => {
             return true;
         }
         PredicateOperator::EndsWith | PredicateOperator::Contains => {
@@ -113,6 +120,18 @@ pub(crate) fn data_leaf_may_match<T: StatsAccessor>(
     };
 
     match op {
+        PredicateOperator::In => {
+            if !matches!(
+                min_value.partial_cmp(&max_value),
+                Some(Ordering::Less | Ordering::Equal)
+            ) {
+                return true;
+            }
+            literals.iter().any(|literal| {
+                !matches!(literal.partial_cmp(&min_value), Some(Ordering::Less))
+                    && !matches!(literal.partial_cmp(&max_value), Some(Ordering::Greater))
+            })
+        }
         PredicateOperator::Eq => {
             !matches!(literal.partial_cmp(&min_value), Some(Ordering::Less))
                 && !matches!(literal.partial_cmp(&max_value), Some(Ordering::Greater))
@@ -180,7 +199,6 @@ pub(crate) fn data_leaf_may_match<T: StatsAccessor>(
         }
         PredicateOperator::IsNull
         | PredicateOperator::IsNotNull
-        | PredicateOperator::In
         | PredicateOperator::NotIn
         | PredicateOperator::EndsWith
         | PredicateOperator::Contains
@@ -526,6 +544,12 @@ mod tests {
         data_leaf_may_match(0, &dt, &dt, op, lits, stats)
     }
 
+    #[test]
+    fn in_falls_open_when_accessor_does_not_opt_in() {
+        let stats = int_stats(10, 20);
+        assert!(run_int(PredicateOperator::In, &[Datum::Int(30)], &stats));
+    }
+
     /// Stage 3 invariant: a `Between` leaf and the equivalent `GtEq+LtEq`
     /// conjunction must produce identical stats-prune verdicts. If they
     /// diverge, the DataFusion translator switch (And-of-comparisons →

crates/paimon/src/table/stats_filter.rs

@@ -32,6 +32,7 @@ pub(super) struct FileStatsRows {
     min_values: Option<BinaryRow>,
     max_values: Option<BinaryRow>,
     null_counts: Vec<Option<i64>>,
+    supports_in_min_max_pruning: bool,
     /// Maps schema field index → stats index. `None` means identity mapping
     /// (stats cover all schema fields in order). `Some` is used when
     /// `value_stats_cols` or `write_cols` is present (dense mode).
@@ -51,6 +52,7 @@ impl FileStatsRows {
             min_values,
             max_values,
             null_counts,
+            supports_in_min_max_pruning: false,
             stats_col_mapping: None,
         }
     }
@@ -92,6 +94,7 @@ impl FileStatsRows {
             min_values: BinaryRow::from_serialized_bytes(file.value_stats.min_values()).ok(),
             max_values: BinaryRow::from_serialized_bytes(file.value_stats.max_values()).ok(),
             null_counts: file.value_stats.null_counts().clone(),
+            supports_in_min_max_pruning: true,
             stats_col_mapping,
         }
     }
@@ -132,6 +135,10 @@ impl StatsAccessor for FileStatsRows {
             .as_ref()
             .and_then(|row| extract_stats_datum(row, stats_index, data_type))
     }
+
+    fn supports_in_min_max_pruning(&self) -> bool {
+        self.supports_in_min_max_pruning
+    }
 }
 
 #[derive(Debug)]

crates/paimon/src/table/table_scan.rs

@@ -2005,6 +2005,122 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_data_file_matches_in_prunes_when_all_literals_out_of_range() {
+        let fields = int_field();
+        let file = test_data_file_meta(
+            int_stats_row(Some(10)),
+            int_stats_row(Some(20)),
+            vec![Some(0)],
+            5,
+        );
+        let predicate = PredicateBuilder::new(&fields)
+            .is_in("id", vec![Datum::Int(1), Datum::Int(30)])
+            .unwrap();
+
+        assert!(!data_file_matches_predicates(
+            &file,
+            &[predicate],
+            TEST_SCHEMA_ID,
+            &test_schema_fields(),
+        ));
+    }
+
+    #[test]
+    fn test_data_file_matches_in_keeps_when_any_literal_in_range() {
+        let fields = int_field();
+        let file = test_data_file_meta(
+            int_stats_row(Some(10)),
+            int_stats_row(Some(20)),
+            vec![Some(0)],
+            5,
+        );
+        let predicate = PredicateBuilder::new(&fields)
+            .is_in("id", vec![Datum::Int(1), Datum::Int(15), Datum::Int(30)])
+            .unwrap();
+
+        assert!(data_file_matches_predicates(
+            &file,
+            &[predicate],
+            TEST_SCHEMA_ID,
+            &test_schema_fields(),
+        ));
+    }
+
+    #[test]
+    fn test_data_file_matches_in_prunes_all_null_file() {
+        let fields = int_field();
+        let file = test_data_file_meta(int_stats_row(None), int_stats_row(None), vec![Some(5)], 5);
+        let predicate = PredicateBuilder::new(&fields)
+            .is_in("id", vec![Datum::Int(10)])
+            .unwrap();
+
+        assert!(!data_file_matches_predicates(
+            &file,
+            &[predicate],
+            TEST_SCHEMA_ID,
+            &test_schema_fields(),
+        ));
+    }
+
+    #[test]
+    fn test_data_file_matches_in_with_corrupt_stats_fails_open() {
+        let fields = int_field();
+        let file = test_data_file_meta(Vec::new(), Vec::new(), vec![Some(0)], 5);
+        let predicate = PredicateBuilder::new(&fields)
+            .is_in("id", vec![Datum::Int(30)])
+            .unwrap();
+
+        assert!(data_file_matches_predicates(
+            &file,
+            &[predicate],
+            TEST_SCHEMA_ID,
+            &test_schema_fields(),
+        ));
+    }
+
+    #[test]
+    fn test_data_file_matches_in_with_inverted_stats_fails_open() {
+        let fields = int_field();
+        let file = test_data_file_meta(
+            int_stats_row(Some(20)),
+            int_stats_row(Some(10)),
+            vec![Some(0)],
+            5,
+        );
+        let predicate = PredicateBuilder::new(&fields)
+            .is_in("id", vec![Datum::Int(15)])
+            .unwrap();
+
+        assert!(data_file_matches_predicates(
+            &file,
+            &[predicate],
+            TEST_SCHEMA_ID,
+            &test_schema_fields(),
+        ));
+    }
+
+    #[test]
+    fn test_data_file_matches_not_in_fails_open() {
+        let fields = int_field();
+        let file = test_data_file_meta(
+            int_stats_row(Some(10)),
+            int_stats_row(Some(20)),
+            vec![Some(0)],
+            5,
+        );
+        let predicate = PredicateBuilder::new(&fields)
+            .is_not_in("id", vec![Datum::Int(10), Datum::Int(20)])
+            .unwrap();
+
+        assert!(data_file_matches_predicates(
+            &file,
+            &[predicate],
+            TEST_SCHEMA_ID,
+            &test_schema_fields(),
+        ));
+    }
+
     #[test]
     fn test_data_file_matches_is_null_prunes_when_null_count_is_zero() {
         let fields = int_field();