Fix fully matched row groups with null counts (#21907)

## Which issue does this PR close?

- Related to #21637

## Rationale for this change

This is split out from review feedback on #21637. Row groups can only be
marked fully matched when all rows are guaranteed to pass the filter.
For nullable predicate columns, proving `NOT(predicate)` is not enough
because rows where the predicate evaluates to NULL do not pass the
filter.

## What changes are included in this PR?

This PR makes the fully matched row-group proof conservative for nulls
by adding `IS NULL` checks for nullable columns referenced by the
predicate before evaluating the inverted pruning predicate.

It also threads `with_missing_null_counts_as_zero` through
`RowGroupPruningStatistics` so normal row-group pruning keeps the
existing default behavior, while fully matched proofs treat missing null
counts as unknown. This reuses the existing statistics conversion path
instead of adding a separate null-count conversion pass.

## Are these changes tested?

Added a regression test covering row groups with known nulls, known zero
nulls, and missing null counts.


## Are there any user-facing changes?

No API changes. This only prevents false positives in the row-group
fully matched optimization.

---------

Co-authored-by: Dmitrii Blaginin <dmitrii@blaginin.me>

Author: xudong963

datafusion/datasource-parquet/src/row_group_filter.rs

@@ -24,8 +24,10 @@ use arrow::datatypes::Schema;
 use datafusion_common::pruning::PruningStatistics;
 use datafusion_common::{Column, Result, ScalarValue};
 use datafusion_datasource::FileRange;
-use datafusion_physical_expr::PhysicalExprSimplifier;
-use datafusion_physical_expr::expressions::NotExpr;
+use datafusion_expr::Operator;
+use datafusion_physical_expr::expressions::{BinaryExpr, IsNullExpr, NotExpr};
+use datafusion_physical_expr::utils::collect_columns;
+use datafusion_physical_expr::{PhysicalExpr, PhysicalExprSimplifier};
 use datafusion_pruning::PruningPredicate;
 use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
 use parquet::basic::Type;
@@ -272,6 +274,11 @@ impl RowGroupAccessPlanFilter {
             parquet_schema,
             row_group_metadatas,
             arrow_schema,
+            // Preserve the existing row-group pruning behavior. This path only
+            // proves whether matching rows may exist, so it uses the
+            // StatisticsConverter default for older parquet-rs files where a
+            // missing null count can mean there are zero nulls.
+            missing_null_counts_as_zero: true,
         };
 
         // try to prune the row groups in a single call
@@ -327,10 +334,33 @@ impl RowGroupAccessPlanFilter {
             return;
         }
 
-        // Use NotExpr to create the inverted predicate
-        let inverted_expr = Arc::new(NotExpr::new(Arc::clone(predicate.orig_expr())));
+        let mut inverted_expr: Arc<dyn PhysicalExpr> =
+            Arc::new(NotExpr::new(Arc::clone(predicate.orig_expr())));
 
-        // Simplify the NOT expression (e.g., NOT(c1 = 0) -> c1 != 0)
+        // Rows where the predicate evaluates to NULL do not pass the filter.
+        // Include NULL checks in the inverted expression so a row group is only
+        // considered fully matched when every referenced column is known non-null.
+        // This is conservative for null-accepting predicates, but fully matched
+        // row groups must not have false positives.
+        let mut columns = collect_columns(predicate.orig_expr())
+            .into_iter()
+            .filter(|column| arrow_schema.field(column.index()).is_nullable())
+            .collect::<Vec<_>>();
+        columns.sort_by(|a, b| {
+            a.index()
+                .cmp(&b.index())
+                .then_with(|| a.name().cmp(b.name()))
+        });
+
+        for column in columns {
+            inverted_expr = Arc::new(BinaryExpr::new(
+                inverted_expr,
+                Operator::Or,
+                Arc::new(IsNullExpr::new(Arc::new(column))),
+            ));
+        }
+
+        // Simplify the inverted expression (e.g., NOT(c1 = 0) -> c1 != 0)
         // before building the pruning predicate
         let simplifier = PhysicalExprSimplifier::new(arrow_schema);
         let Ok(inverted_expr) = simplifier.simplify(inverted_expr) else {
@@ -350,6 +380,11 @@ impl RowGroupAccessPlanFilter {
                 .map(|&i| &groups[i])
                 .collect::<Vec<_>>(),
             arrow_schema,
+            // Fully matched row groups require a stronger proof: every row
+            // must pass the predicate. Missing null counts are unknown here;
+            // treating them as zero can incorrectly mark nullable row groups as
+            // fully matched and make limit pruning unsound.
+            missing_null_counts_as_zero: false,
         };
 
         let Ok(inverted_values) = inverted_predicate.prune(&inverted_pruning_stats)
@@ -582,6 +617,7 @@ struct RowGroupPruningStatistics<'a> {
     parquet_schema: &'a SchemaDescriptor,
     row_group_metadatas: Vec<&'a RowGroupMetaData>,
     arrow_schema: &'a Schema,
+    missing_null_counts_as_zero: bool,
 }
 
 impl<'a> RowGroupPruningStatistics<'a> {
@@ -598,7 +634,8 @@ impl<'a> RowGroupPruningStatistics<'a> {
             &column.name,
             self.arrow_schema,
             self.parquet_schema,
-        )?)
+        )?
+        .with_missing_null_counts_as_zero(self.missing_null_counts_as_zero))
     }
 }
 
@@ -767,6 +804,65 @@ mod tests {
         assert_pruned(row_groups, ExpectedPruning::Some(vec![1]))
     }
 
+    #[test]
+    fn row_group_fully_matched_requires_known_non_null_predicate_columns() {
+        use datafusion_expr::{col, lit};
+
+        let schema = Arc::new(Schema::new(vec![Field::new("c1", DataType::Int32, true)]));
+        let expr = logical2physical(&col("c1").gt(lit(15)), &schema);
+        let pruning_predicate = PruningPredicate::try_new(expr, schema.clone()).unwrap();
+
+        let field = PrimitiveTypeField::new("c1", PhysicalType::INT32);
+        let schema_descr = get_test_schema_descr(vec![field]);
+
+        // All three row groups have non-null values in the predicate range,
+        // so none are pruned. Only the second row group can be proven fully
+        // matched because it is the only one with a known zero null count.
+        let rg_with_null = get_row_group_meta_data(
+            &schema_descr,
+            vec![ParquetStatistics::int32(
+                Some(16),
+                Some(20),
+                None,
+                Some(1),
+                false,
+            )],
+        );
+        let rg_without_null = get_row_group_meta_data(
+            &schema_descr,
+            vec![ParquetStatistics::int32(
+                Some(16),
+                Some(20),
+                None,
+                Some(0),
+                false,
+            )],
+        );
+        let rg_unknown_null_count = get_row_group_meta_data(
+            &schema_descr,
+            vec![ParquetStatistics::int32(
+                Some(16),
+                Some(20),
+                None,
+                None,
+                false,
+            )],
+        );
+
+        let metrics = parquet_file_metrics();
+        let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(3));
+        row_groups.prune_by_statistics(
+            &schema,
+            &schema_descr,
+            &[rg_with_null, rg_without_null, rg_unknown_null_count],
+            &pruning_predicate,
+            &metrics,
+        );
+
+        assert_eq!(row_groups.access_plan.row_group_indexes(), vec![0, 1, 2]);
+        assert_eq!(row_groups.is_fully_matched(), &vec![false, true, false]);
+    }
+
     #[test]
     fn row_group_pruning_predicate_missing_stats() {
         use datafusion_expr::{col, lit};