fix(physical-plan): preserve Exact(0) in FilterExec for null_count, distinct_count and total_byte_size upon empty input

datafusion/physical-plan/src/filter.rs

@@ -401,8 +401,11 @@ impl FilterExec {
             }
         };
 
-        let total_byte_size =
-            input_total_byte_size.with_estimated_selectivity(selectivity);
+        let total_byte_size = if is_infeasible {
+            Precision::Exact(0)
+        } else {
+            input_total_byte_size.with_estimated_selectivity(selectivity)
+        };
 
         Ok(Statistics {
             num_rows,
@@ -1032,15 +1035,17 @@ fn interval_bound_to_precision(
 }
 
 /// Caps a row-bounded column statistic (a null count or distinct count) at the
-/// filtered row estimate, since a column cannot have more nulls or distinct
-/// values than it has rows. Known counts are demoted to inexact because the
-/// filtered row count is itself an estimate.
+/// filtered row count, since a column cannot have more nulls or distinct values
+/// than it has rows. Known counts are demoted to inexact because a
+/// filter-derived row bound is normally an estimate, the exception being an
+/// exact zero, which proves the column is empty.
 fn cap_at_rows(
     value: Precision<usize>,
     filtered_num_rows: Precision<usize>,
 ) -> Precision<usize> {
     match filtered_num_rows {
         Precision::Absent => value.to_inexact(),
+        Precision::Exact(0) => Precision::Exact(0),
         rows => value.to_inexact().min(&rows),
     }
 }
@@ -2904,24 +2909,37 @@ mod tests {
     #[tokio::test]
     async fn test_filter_statistics_preserves_exactly_empty_input() -> Result<()> {
         // A satisfiable predicate over an exactly empty input: the filter cannot
-        // produce rows, so the whole estimate stays exact.
-        let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+        // produce rows, so the whole estimate stays exact. Column `b` is not
+        // mentioned by the predicate, so its null and distinct counts go through
+        // the generic row cap.
+        let schema = Schema::new(vec![
+            Field::new("a", DataType::Int32, true),
+            Field::new("b", DataType::Int32, true),
+        ]);
         let input_stats = Statistics {
             num_rows: Precision::Exact(0),
             total_byte_size: Precision::Exact(0),
-            column_statistics: vec![ColumnStatistics {
-                null_count: Precision::Exact(0),
-                byte_size: Precision::Exact(0),
-                ..Default::default()
-            }],
+            column_statistics: vec![
+                ColumnStatistics {
+                    null_count: Precision::Exact(0),
+                    byte_size: Precision::Exact(0),
+                    ..Default::default()
+                },
+                ColumnStatistics {
+                    null_count: Precision::Exact(3),
+                    distinct_count: Precision::Exact(7),
+                    byte_size: Precision::Exact(0),
+                    ..Default::default()
+                },
+            ],
         };
         let predicate = Arc::new(BinaryExpr::new(
             Arc::new(Column::new("a", 0)),
             Operator::Gt,
             Arc::new(Literal::new(ScalarValue::Int32(Some(5)))),
         ));
 
-        let input = Arc::new(StatisticsExec::new(input_stats, schema));
+        let input = Arc::new(StatisticsExec::new(input_stats, schema.clone()));
         let filter: Arc<dyn ExecutionPlan> =
             Arc::new(FilterExec::try_new(predicate, input)?);
         let statistics =
@@ -2933,6 +2951,45 @@ mod tests {
             statistics.column_statistics[0].byte_size,
             Precision::Exact(0)
         );
+        assert_eq!(
+            statistics.column_statistics[1].null_count,
+            Precision::Exact(0)
+        );
+        assert_eq!(
+            statistics.column_statistics[1].distinct_count,
+            Precision::Exact(0)
+        );
+
+        // A contradictory predicate (`a = 1 AND a = 2`) proves the same for an
+        // input of any size, so the byte size is exact too.
+        let input = Arc::new(StatisticsExec::new(
+            Statistics {
+                num_rows: Precision::Inexact(1000),
+                total_byte_size: Precision::Inexact(8000),
+                column_statistics: vec![ColumnStatistics::new_unknown(); 2],
+            },
+            schema,
+        ));
+        let contradiction = Arc::new(BinaryExpr::new(
+            Arc::new(BinaryExpr::new(
+                Arc::new(Column::new("a", 0)),
+                Operator::Eq,
+                Arc::new(Literal::new(ScalarValue::Int32(Some(1)))),
+            )),
+            Operator::And,
+            Arc::new(BinaryExpr::new(
+                Arc::new(Column::new("a", 0)),
+                Operator::Eq,
+                Arc::new(Literal::new(ScalarValue::Int32(Some(2)))),
+            )),
+        ));
+        let filter: Arc<dyn ExecutionPlan> =
+            Arc::new(FilterExec::try_new(contradiction, input)?);
+        let statistics =
+            StatisticsContext::new().compute(filter.as_ref(), &StatisticsArgs::new())?;
+
+        assert_eq!(statistics.num_rows, Precision::Exact(0));
+        assert_eq!(statistics.total_byte_size, Precision::Exact(0));
 
         Ok(())
     }