add test

datafusion/physical-optimizer/src/pruning.rs

@@ -1912,7 +1912,7 @@ mod tests {
     use datafusion_common::assert_batches_eq;
     use datafusion_expr::{col, lit};
 
-    use arrow::array::Decimal128Array;
+    use arrow::array::{record_batch, Decimal128Array};
     use arrow::{
         array::{BinaryArray, Int32Array, Int64Array, StringArray, UInt64Array},
         datatypes::TimeUnit,
@@ -4276,6 +4276,40 @@ mod tests {
         // TODO: add other negative test for other case and op
     }
 
+    #[tokio::test]
+    async fn test_rewrite_expr_to_prunable_with_nulls() {
+        let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+
+        let rewriter = PredicateRewriter::new();
+
+        let transform_expr = |expr| {
+            let expr = logical2physical(&expr, &schema);
+            rewriter.rewrite_predicate_to_statistics_predicate(&expr, &schema)
+        };
+
+        // transform an arbitrary valid expression that we know is handled
+        let known_expression = col("a").eq(lit(12));
+        let known_expression_transformed = transform_expr(known_expression.clone());
+
+        // construct a set of statistics containers with nulls
+        let statistics_batch = record_batch!(
+            ("a_min", Int32, [Some(1), Some(0), None]),
+            ("a_max", Int32, [Some(2), Some(1), Some(15)]),
+            ("a_row_count", UInt64, [Some(2), Some(2), Some(2)]),
+            ("a_null_count", UInt64, [Some(0), Some(0), Some(0)])
+        ).unwrap();
+
+        let res = known_expression_transformed.evaluate(&statistics_batch).unwrap();
+
+        let ColumnarValue::Array(matches) = res else { panic!("should have returned an array") };
+
+        let matches = matches.as_any().downcast_ref::<BooleanArray>().expect("matches should be a boolean array");
+        let values = matches.iter().collect::<Vec<_>>();
+        let expected = vec![Some(false), Some(false), Some(true)];
+
+        assert_eq!(values, expected);
+    }
+
     #[test]
     fn prune_with_contained_one_column() {
         let schema = Arc::new(Schema::new(vec![Field::new("s1", DataType::Utf8, true)]));