Prune columns are all null in ParquetExec by row_counts , handle IS NOT NULL

datafusion/core/src/datasource/physical_plan/parquet/row_groups.rs

@@ -342,8 +342,10 @@ impl<'a> PruningStatistics for RowGroupPruningStatistics<'a> {
         scalar.to_array().ok()
     }
 
-    fn row_counts(&self, _column: &Column) -> Option<ArrayRef> {
-        None
+    fn row_counts(&self, column: &Column) -> Option<ArrayRef> {
+        let (c, _) = self.column(&column.name)?;
+        let scalar = ScalarValue::UInt64(Some(c.num_values() as u64));
+        scalar.to_array().ok()
     }
 
     fn contained(
@@ -1026,15 +1028,17 @@ mod tests {
         column_statistics: Vec<ParquetStatistics>,
     ) -> RowGroupMetaData {
         let mut columns = vec![];
+        let number_row = 1000;
         for (i, s) in column_statistics.iter().enumerate() {
             let column = ColumnChunkMetaData::builder(schema_descr.column(i))
                 .set_statistics(s.clone())
+                .set_num_values(number_row)
                 .build()
                 .unwrap();
             columns.push(column);
         }
         RowGroupMetaData::builder(schema_descr.clone())
-            .set_num_rows(1000)
+            .set_num_rows(number_row)
             .set_total_byte_size(2000)
             .set_column_metadata(columns)
             .build()

datafusion/core/tests/parquet/mod.rs

@@ -28,6 +28,7 @@ use arrow::{
     record_batch::RecordBatch,
     util::pretty::pretty_format_batches,
 };
+use arrow_array::new_null_array;
 use chrono::{Datelike, Duration, TimeDelta};
 use datafusion::{
     datasource::{physical_plan::ParquetExec, provider_as_source, TableProvider},
@@ -75,6 +76,7 @@ enum Scenario {
     DecimalLargePrecisionBloomFilter,
     ByteArray,
     PeriodsInColumnNames,
+    AllNullValues,
 }
 
 enum Unit {
@@ -630,6 +632,27 @@ fn make_names_batch(name: &str, service_name_values: Vec<&str>) -> RecordBatch {
     RecordBatch::try_new(schema, vec![Arc::new(name), Arc::new(service_name)]).unwrap()
 }
 
+/// Return record batch with i8, i16, i32, and i64 sequences with all Null values
+fn make_all_null_values() -> RecordBatch {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("i8", DataType::Int8, true),
+        Field::new("i16", DataType::Int16, true),
+        Field::new("i32", DataType::Int32, true),
+        Field::new("i64", DataType::Int64, true),
+    ]));
+
+    RecordBatch::try_new(
+        schema,
+        vec![
+            new_null_array(&DataType::Int8, 5),
+            new_null_array(&DataType::Int16, 5),
+            new_null_array(&DataType::Int32, 5),
+            new_null_array(&DataType::Int64, 5),
+        ],
+    )
+    .unwrap()
+}
+
 fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
     match scenario {
         Scenario::Timestamps => {
@@ -799,6 +822,13 @@ fn create_data_batch(scenario: Scenario) -> Vec<RecordBatch> {
                 ),
             ]
         }
+        Scenario::AllNullValues => {
+            vec![
+                make_all_null_values(),
+                make_int_batches(1, 6),
+                make_all_null_values(),
+            ]
+        }
     }
 }
 

datafusion/core/tests/parquet/row_group_pruning.rs

@@ -1264,3 +1264,37 @@ async fn prune_periods_in_column_names() {
         .test_row_group_prune()
         .await;
 }
+
+#[tokio::test]
+async fn test_row_group_all_null_values() {
+    // Tree row groups:
+    // 1. all Null values
+    // 2. values from 1 to 5
+    // 3. all Null values
+
+    // After pruning, only row group 2 should be selected
+    RowGroupPruningTest::new()
+        .with_scenario(Scenario::AllNullValues)
+        .with_query("SELECT * FROM t WHERE \"i8\" <= 5")
+        .with_expected_errors(Some(0))
+        .with_matched_by_stats(Some(1))
+        .with_pruned_by_stats(Some(2))
+        .with_expected_rows(5)
+        .with_matched_by_bloom_filter(Some(0))
+        .with_pruned_by_bloom_filter(Some(0))
+        .test_row_group_prune()
+        .await;
+
+    // After pruning, only row group 1,3 should be selected
+    RowGroupPruningTest::new()
+        .with_scenario(Scenario::AllNullValues)
+        .with_query("SELECT * FROM t WHERE \"i8\" is Null")
+        .with_expected_errors(Some(0))
+        .with_matched_by_stats(Some(2))
+        .with_pruned_by_stats(Some(1))
+        .with_expected_rows(10)
+        .with_matched_by_bloom_filter(Some(0))
+        .with_pruned_by_bloom_filter(Some(0))
+        .test_row_group_prune()
+        .await;
+}