fix: preserve nullability for non-empty scalar subqueries

datafusion/expr/src/expr_schema.rs

@@ -369,9 +369,7 @@ impl ExprSchemable for Expr {
 
                 Ok(expr_nullable | subquery_nullable)
             }
-            // A scalar subquery may return no rows, in which case it evaluates to NULL
-            // regardless of the nullability of its projected field.
-            Expr::ScalarSubquery(_) => Ok(true),
+            Expr::ScalarSubquery(subquery) => Ok(scalar_subquery_nullable(subquery)),
             Expr::BinaryExpr(BinaryExpr { left, right, .. }) => {
                 Ok(left.nullable(input_schema)? || right.nullable(input_schema)?)
             }
@@ -517,15 +515,15 @@ impl ExprSchemable for Expr {
             | Expr::Exists { .. } => {
                 Ok(Arc::new(Field::new(&schema_name, DataType::Boolean, false)))
             }
-            Expr::ScalarSubquery(subquery) => Ok(Arc::new(
-                subquery
-                    .subquery
-                    .schema()
-                    .field(0)
-                    .as_ref()
-                    .clone()
-                    .with_nullable(true),
-            )),
+            Expr::ScalarSubquery(subquery) => {
+                let field = subquery.subquery.schema().field(0);
+                Ok(Arc::new(
+                    field
+                        .as_ref()
+                        .clone()
+                        .with_nullable(scalar_subquery_nullable(subquery)),
+                ))
+            }
             Expr::BinaryExpr(BinaryExpr { left, right, op }) => {
                 let (left_field, right_field) =
                     (left.to_field(schema)?.1, right.to_field(schema)?.1);
@@ -759,6 +757,15 @@ fn unwrap_certainly_null_expr(expr: &Expr) -> &Expr {
     }
 }
 
+/// Returns whether a scalar subquery may evaluate to NULL.
+///
+/// This is the case if the subquery's projected field is nullable, or if the
+/// subquery may return no rows: a scalar subquery that produces no rows
+/// evaluates to NULL regardless of the nullability of its projected field.
+fn scalar_subquery_nullable(subquery: &Subquery) -> bool {
+    subquery.subquery.schema().field(0).is_nullable() || subquery.subquery.min_rows() == 0
+}
+
 /// Cast subquery in InSubquery/ScalarSubquery to a given type.
 ///
 /// 1. **Projection plan**: If the subquery is a projection (i.e. a SELECT statement with specific
@@ -804,6 +811,7 @@ mod tests {
 
     use super::*;
     use crate::logical_plan::builder::LogicalTableSource;
+    use crate::test::function_stub::count;
     use crate::{
         LogicalPlanBuilder, and, col, in_subquery, lit, not, or,
         out_ref_col_with_metadata, scalar_subquery, when,
@@ -1275,20 +1283,37 @@ mod tests {
     }
 
     #[test]
-    fn scalar_subquery_is_nullable_with_non_nullable_output() {
-        let subquery = LogicalPlanBuilder::empty(false)
+    fn scalar_subquery_nullability_accounts_for_min_rows() {
+        let possibly_empty = LogicalPlanBuilder::empty(false)
             .project(vec![lit(1)])
             .unwrap()
             .build()
             .unwrap();
-        assert!(!subquery.schema().field(0).is_nullable());
+        assert!(!possibly_empty.schema().field(0).is_nullable());
 
-        let expr = scalar_subquery(Arc::new(subquery));
+        let expr = scalar_subquery(Arc::new(possibly_empty));
         assert!(expr.nullable(&MockExprSchema::new()).unwrap());
 
         let field = expr.to_field(&MockExprSchema::new()).unwrap().1;
         assert_eq!(field.data_type(), &DataType::Int32);
         assert!(field.is_nullable());
+
+        let always_one = LogicalPlanBuilder::empty(false)
+            .aggregate(Vec::<Expr>::new(), vec![count(lit(1))])
+            .unwrap()
+            .build()
+            .unwrap();
+        assert!(!always_one.schema().field(0).is_nullable());
+
+        let expr = scalar_subquery(Arc::new(always_one));
+        assert!(!expr.nullable(&MockExprSchema::new()).unwrap());
+        assert!(
+            !expr
+                .to_field(&MockExprSchema::new())
+                .unwrap()
+                .1
+                .is_nullable()
+        );
     }
 
     #[test]

datafusion/expr/src/logical_plan/plan.rs

@@ -1427,7 +1427,9 @@ impl LogicalPlan {
                 })
             }
             LogicalPlan::TableScan(TableScan { fetch, .. }) => *fetch,
-            LogicalPlan::EmptyRelation(_) => Some(0),
+            LogicalPlan::EmptyRelation(EmptyRelation {
+                produce_one_row, ..
+            }) => Some(usize::from(*produce_one_row)),
             LogicalPlan::RecursiveQuery(_) => None,
             LogicalPlan::Subquery(_) => None,
             LogicalPlan::SubqueryAlias(SubqueryAlias { input, .. }) => input.max_rows(),
@@ -1451,6 +1453,82 @@ impl LogicalPlan {
         }
     }
 
+    /// Returns a lower bound on the number of rows that this plan can output.
+    ///
+    /// A return value of `0` means that the plan may produce no rows. A positive
+    /// value guarantees that the plan produces at least that many rows.
+    ///
+    /// See [`Self::max_rows`] for the corresponding upper bound.
+    pub fn min_rows(&self) -> usize {
+        match self {
+            LogicalPlan::Projection(Projection { input, .. })
+            | LogicalPlan::Window(Window { input, .. })
+            | LogicalPlan::Repartition(Repartition { input, .. })
+            | LogicalPlan::SubqueryAlias(SubqueryAlias { input, .. }) => input.min_rows(),
+            LogicalPlan::Filter(_) => 0,
+            LogicalPlan::Aggregate(Aggregate { group_expr, .. }) => {
+                // An ungrouped aggregate always produces one row, even for an
+                // empty input.
+                usize::from(group_expr.is_empty())
+            }
+            LogicalPlan::Sort(Sort { input, fetch, .. }) => fetch
+                .map(|fetch| input.min_rows().min(fetch))
+                .unwrap_or_else(|| input.min_rows()),
+            LogicalPlan::Join(Join {
+                left,
+                right,
+                on,
+                filter,
+                join_type,
+                ..
+            }) => match join_type {
+                JoinType::Inner if on.is_empty() && filter.is_none() => {
+                    left.min_rows().saturating_mul(right.min_rows())
+                }
+                JoinType::Left | JoinType::LeftMark => left.min_rows(),
+                JoinType::Right | JoinType::RightMark => right.min_rows(),
+                JoinType::Full => left.min_rows().max(right.min_rows()),
+                JoinType::Inner
+                | JoinType::LeftSemi
+                | JoinType::RightSemi
+                | JoinType::LeftAnti
+                | JoinType::RightAnti => 0,
+            },
+            LogicalPlan::Union(Union { inputs, .. }) => inputs
+                .iter()
+                .fold(0, |rows, input| rows.saturating_add(input.min_rows())),
+            LogicalPlan::EmptyRelation(EmptyRelation {
+                produce_one_row, ..
+            }) => usize::from(*produce_one_row),
+            LogicalPlan::Subquery(Subquery { subquery, .. }) => subquery.min_rows(),
+            LogicalPlan::Limit(limit) => {
+                match (limit.get_skip_type(), limit.get_fetch_type()) {
+                    (Ok(SkipType::Literal(skip)), Ok(FetchType::Literal(fetch))) => fetch
+                        .map(|fetch| {
+                            limit.input.min_rows().saturating_sub(skip).min(fetch)
+                        })
+                        .unwrap_or_else(|| limit.input.min_rows().saturating_sub(skip)),
+                    _ => 0,
+                }
+            }
+            LogicalPlan::Distinct(
+                Distinct::All(input) | Distinct::On(DistinctOn { input, .. }),
+            ) => usize::from(input.min_rows() > 0),
+            LogicalPlan::Values(values) => values.values.len(),
+            LogicalPlan::TableScan(_)
+            | LogicalPlan::RecursiveQuery(_)
+            | LogicalPlan::Unnest(_)
+            | LogicalPlan::Ddl(_)
+            | LogicalPlan::Explain(_)
+            | LogicalPlan::Analyze(_)
+            | LogicalPlan::Dml(_)
+            | LogicalPlan::Copy(_)
+            | LogicalPlan::DescribeTable(_)
+            | LogicalPlan::Statement(_)
+            | LogicalPlan::Extension(_) => 0,
+        }
+    }
+
     /// Returns the skip (offset) of this plan node, if it has one.
     ///
     /// Only [`LogicalPlan::Limit`] carries a skip value; all other variants
@@ -5716,6 +5794,113 @@ mod tests {
             .unwrap()
     }
 
+    #[test]
+    fn min_rows_is_a_conservative_lower_bound() -> Result<()> {
+        let no_rows = LogicalPlanBuilder::empty(false).build()?;
+        let one_row = LogicalPlanBuilder::empty(true).build()?;
+        assert_eq!(no_rows.min_rows(), 0);
+        assert_eq!(no_rows.max_rows(), Some(0));
+        assert_eq!(one_row.min_rows(), 1);
+        assert_eq!(one_row.max_rows(), Some(1));
+
+        let projection = LogicalPlanBuilder::from(one_row.clone())
+            .project(vec![lit(1)])?
+            .build()?;
+        assert_eq!(projection.min_rows(), 1);
+
+        let filter = LogicalPlanBuilder::from(projection.clone())
+            .filter(lit(true))?
+            .build()?;
+        assert_eq!(filter.min_rows(), 0);
+
+        let aggregate = LogicalPlanBuilder::from(no_rows)
+            .aggregate(Vec::<Expr>::new(), vec![count(lit(1))])?
+            .build()?;
+        assert_eq!(aggregate.min_rows(), 1);
+
+        let offset = LogicalPlanBuilder::from(projection.clone())
+            .limit(1, None)?
+            .build()?;
+        assert_eq!(offset.min_rows(), 0);
+
+        let union = LogicalPlanBuilder::from(projection.clone())
+            .union(projection)?
+            .build()?;
+        assert_eq!(union.min_rows(), 2);
+        assert_eq!(
+            LogicalPlanBuilder::from(union)
+                .distinct()?
+                .build()?
+                .min_rows(),
+            1
+        );
+
+        let values =
+            LogicalPlanBuilder::values(vec![vec![lit(1)], vec![lit(2)]])?.build()?;
+        assert_eq!(values.min_rows(), 2);
+
+        let sort_key = col("column1").sort(true, false);
+        let sort = LogicalPlanBuilder::from(values.clone())
+            .sort(vec![sort_key.clone()])?
+            .build()?;
+        assert_eq!(sort.min_rows(), 2);
+        let sort_with_fetch = LogicalPlanBuilder::from(values.clone())
+            .sort_with_limit(vec![sort_key], Some(1))?
+            .build()?;
+        assert_eq!(sort_with_fetch.min_rows(), 1);
+
+        let limit = LogicalPlanBuilder::from(values.clone())
+            .limit(1, Some(5))?
+            .build()?;
+        assert_eq!(limit.min_rows(), 1);
+
+        let grouped = LogicalPlanBuilder::from(values)
+            .aggregate(vec![col("column1")], vec![count(lit(1))])?
+            .build()?;
+        assert_eq!(grouped.min_rows(), 0);
+
+        let scan = table_scan(Some("employee"), &employee_schema(), None)?.build()?;
+        assert_eq!(scan.min_rows(), 0);
+
+        Ok(())
+    }
+
+    #[test]
+    fn min_rows_of_joins() -> Result<()> {
+        let two_rows = LogicalPlanBuilder::values(vec![vec![lit(1)], vec![lit(2)]])?
+            .alias("l")?
+            .build()?;
+        let one_row = LogicalPlanBuilder::values(vec![vec![lit(1)]])?
+            .alias("r")?
+            .build()?;
+
+        let cross_join = LogicalPlanBuilder::from(two_rows.clone())
+            .cross_join(one_row.clone())?
+            .build()?;
+        assert_eq!(cross_join.min_rows(), 2);
+
+        for (join_type, expected_min_rows) in [
+            // An inner join with a join condition may filter out every row,
+            // while outer joins preserve the rows of the outer side(s).
+            (JoinType::Inner, 0),
+            (JoinType::Left, 2),
+            (JoinType::Right, 1),
+            (JoinType::Full, 2),
+            (JoinType::LeftSemi, 0),
+        ] {
+            let join = LogicalPlanBuilder::from(two_rows.clone())
+                .join_on(
+                    one_row.clone(),
+                    join_type,
+                    [col("l.column1").eq(col("r.column1"))],
+                )?
+                .build()?;
+            assert_eq!(join.min_rows(), expected_min_rows, "{join_type} join");
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_replace_invalid_placeholder() {
         // test empty placeholder

datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs

@@ -3790,17 +3790,25 @@ mod tests {
 
     #[test]
     fn simplify_scalar_subquery_is_null() {
-        let subquery = LogicalPlanBuilder::empty(false)
+        let possibly_empty = LogicalPlanBuilder::empty(false)
             .project(vec![lit(1)])
             .unwrap()
             .build()
             .unwrap();
-        let scalar_subquery = scalar_subquery(Arc::new(subquery));
+        let possibly_empty = scalar_subquery(Arc::new(possibly_empty));
 
         assert_eq!(
-            simplify(scalar_subquery.clone().is_null()),
-            scalar_subquery.is_null()
+            simplify(possibly_empty.clone().is_null()),
+            possibly_empty.is_null()
         );
+
+        let always_one = LogicalPlanBuilder::empty(true)
+            .project(vec![lit(1)])
+            .unwrap()
+            .build()
+            .unwrap();
+        let always_one = scalar_subquery(Arc::new(always_one));
+        assert_eq!(simplify(always_one.is_null()), lit(false));
     }
 
     #[test]

datafusion/physical-expr/src/planner.rs

@@ -538,8 +538,7 @@ pub fn create_physical_expr(
                     let dt = schema.field(0).data_type().clone();
                     Ok(Arc::new(ScalarSubqueryExpr::new(
                         dt,
-                        // A scalar subquery may return no rows and evaluate to NULL.
-                        true,
+                        e.nullable(input_dfschema)?,
                         index,
                         planning_ctx.results().clone(),
                     )))
@@ -744,7 +743,11 @@ pub fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
 mod tests {
     use arrow::array::{ArrayRef, BooleanArray, RecordBatch, StringArray};
     use arrow::datatypes::{DataType, Field};
-    use datafusion_expr::col;
+    use datafusion_common::HashMap;
+    use datafusion_expr::physical_planning_context::{
+        ScalarSubqueryResults, SubqueryIndex,
+    };
+    use datafusion_expr::{LogicalPlanBuilder, col, scalar_subquery};
 
     use super::*;
 
@@ -798,6 +801,35 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn scalar_subquery_nullability_accounts_for_min_rows() -> Result<()> {
+        for (produce_one_row, expected_nullable) in [(false, true), (true, false)] {
+            let plan = LogicalPlanBuilder::empty(produce_one_row)
+                .project(vec![lit(1)])?
+                .build()?;
+            let expr = scalar_subquery(Arc::new(plan));
+            let Expr::ScalarSubquery(subquery) = &expr else {
+                unreachable!()
+            };
+
+            let index = SubqueryIndex::new(0);
+            let planning_ctx = PhysicalPlanningContext::new(
+                HashMap::from([(subquery.clone(), index)]),
+                ScalarSubqueryResults::new(1),
+            );
+            let physical_expr = create_physical_expr(
+                &expr,
+                &DFSchema::empty(),
+                &ExecutionProps::new(),
+                &planning_ctx,
+            )?;
+
+            assert_eq!(physical_expr.nullable(&Schema::empty())?, expected_nullable);
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_cast_lowering_preserves_target_field_metadata() -> Result<()> {
         let schema = test_cast_schema();