Improve NDV propagation through statistics merge and projection

- Statistics merge: use max as conservative lower bound instead of
  discarding NDV (duplicates may exist across partitions)
- Projection: preserve NDV for single-column expressions as upper bound

Author: asolimando

datafusion/common/src/stats.rs

@@ -637,7 +637,24 @@ impl Statistics {
             col_stats.max_value = col_stats.max_value.max(&item_col_stats.max_value);
             col_stats.min_value = col_stats.min_value.min(&item_col_stats.min_value);
             col_stats.sum_value = col_stats.sum_value.add(&item_col_stats.sum_value);
-            col_stats.distinct_count = Precision::Absent;
+            // Use max as a conservative lower bound for distinct count
+            // (can't accurately merge NDV since duplicates may exist across partitions)
+            col_stats.distinct_count =
+                match (&col_stats.distinct_count, &item_col_stats.distinct_count) {
+                    (Precision::Exact(a), Precision::Exact(b))
+                    | (Precision::Inexact(a), Precision::Exact(b))
+                    | (Precision::Exact(a), Precision::Inexact(b))
+                    | (Precision::Inexact(a), Precision::Inexact(b)) => {
+                        Precision::Inexact(if a >= b { *a } else { *b })
+                    }
+                    (Precision::Exact(v), Precision::Absent)
+                    | (Precision::Inexact(v), Precision::Absent)
+                    | (Precision::Absent, Precision::Exact(v))
+                    | (Precision::Absent, Precision::Inexact(v)) => {
+                        Precision::Inexact(*v)
+                    }
+                    (Precision::Absent, Precision::Absent) => Precision::Absent,
+                };
             col_stats.byte_size = col_stats.byte_size.add(&item_col_stats.byte_size);
         }
 
@@ -1357,8 +1374,8 @@ mod tests {
             col_stats.max_value,
             Precision::Exact(ScalarValue::Int32(Some(20)))
         );
-        // Distinct count should be Absent after merge
-        assert_eq!(col_stats.distinct_count, Precision::Absent);
+        // Distinct count should be Inexact(max) after merge as a conservative lower bound
+        assert_eq!(col_stats.distinct_count, Precision::Inexact(7));
     }
 
     #[test]

datafusion/physical-expr/src/projection.rs

@@ -711,9 +711,25 @@ impl ProjectionExprs {
                     }
                 }
             } else {
-                // TODO stats: estimate more statistics from expressions
-                // (expressions should compute their statistics themselves)
-                ColumnStatistics::new_unknown()
+                // TODO: expressions should compute their own statistics
+                //
+                // For now, try to preserve NDV if the expression references a
+                // single column (as a conservative upper bound).
+                // More accurate NDV propagation would require tracking injectivity
+                // of functions (e.g., `a + 1` preserves NDV exactly, `ABS(a)` may
+                // reduce it, `a % 10` bounds it to 10)
+                let columns = collect_columns(expr);
+                if columns.len() == 1 {
+                    let col_idx = columns.iter().next().unwrap().index();
+                    ColumnStatistics {
+                        distinct_count: stats.column_statistics[col_idx]
+                            .distinct_count
+                            .to_inexact(),
+                        ..ColumnStatistics::new_unknown()
+                    }
+                } else {
+                    ColumnStatistics::new_unknown()
+                }
             };
             column_statistics.push(col_stats);
         }
@@ -2718,10 +2734,11 @@ pub(crate) mod tests {
         // Should have 2 column statistics
         assert_eq!(output_stats.column_statistics.len(), 2);
 
-        // First column (expression) should have unknown statistics
+        // First column (expression `col0 + 1`) preserves NDV from the single
+        // referenced column as a conservative upper bound (marked Inexact)
         assert_eq!(
             output_stats.column_statistics[0].distinct_count,
-            Precision::Absent
+            Precision::Inexact(5)
         );
         assert_eq!(
             output_stats.column_statistics[0].max_value,