diff --git a/datafusion/expr/src/expr_schema.rs b/datafusion/expr/src/expr_schema.rs index 8a9d0b8937cda..07fa6852a103b 100644 --- a/datafusion/expr/src/expr_schema.rs +++ b/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::::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] diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index ba4fccfb06703..6bd75c11c82f4 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/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::::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 diff --git a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs index 5beb54b5605f4..0e4836bb6dd07 100644 --- a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs +++ b/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] diff --git a/datafusion/physical-expr/src/planner.rs b/datafusion/physical-expr/src/planner.rs index 09b2e5ce04727..9a0bdc33da8e9 100644 --- a/datafusion/physical-expr/src/planner.rs +++ b/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 { 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(); diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 65760886d1156..e76a3e8147a5e 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -2052,6 +2052,25 @@ SELECT (SELECT max(v) FROM sq_values); ---- 3 +# An ungrouped aggregate produces exactly one row even on empty input, so the +# subquery's result is not nullable and SimplifyExpressions folds IS NULL to +# false. The query result alone cannot show the fold (a non-empty subquery +# never evaluates to NULL at runtime), so pin the simplified plan. +query TT +EXPLAIN SELECT (SELECT count(*) FROM sq_empty) IS NULL; +---- +logical_plan +01)Projection: Boolean(false) AS count(*) IS NULL +02)--EmptyRelation: rows=1 +physical_plan +01)ProjectionExec: expr=[false as count(*) IS NULL] +02)--PlaceholderRowExec + +query B +SELECT (SELECT count(*) FROM sq_empty) IS NULL; +---- +false + # Multiple scalar subqueries, one returns multiple rows → error query error DataFusion error: Execution error: Scalar subquery returned more than one row SELECT (SELECT count(*) FROM sq_empty), (SELECT v FROM sq_values);