You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/21 08:01:30 UTC
(datafusion) branch gh-readonly-queue/main/pr-24534-584c30d28f8cd5792cc530b151db812230c1c3f3 created (now 425ebe7110)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24534-584c30d28f8cd5792cc530b151db812230c1c3f3
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 425ebe7110 fix: preserve nullability for non-empty scalar subqueries (#24534)
This branch includes the following new commits:
new 425ebe7110 fix: preserve nullability for non-empty scalar subqueries (#24534)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: preserve nullability for non-empty scalar subqueries (#24534)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24534-584c30d28f8cd5792cc530b151db812230c1c3f3
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 425ebe7110b51f481fc43318b480d26eba94a9c8
Author: Fredrik Fornwall <fr...@fornwall.net>
AuthorDate: Fri Aug 21 08:01:07 2026 +0000
fix: preserve nullability for non-empty scalar subqueries (#24534)
## Which issue does this PR close?
- Follow-up to apache/datafusion#24516, which fixed
apache/datafusion#24513. The follow-up was suggested during review of
that PR.
## Rationale for this change
apache/datafusion#24516 conservatively made every scalar subquery
nullable because an empty result evaluates to `NULL`. Plans guaranteed
to return at least one row, such as ungrouped aggregates, cannot produce
that empty-result `NULL`; marking them nullable loses schema precision
and prevents valid expression simplification.
## What changes are included in this PR?
- Adds a conservative `LogicalPlan::min_rows()` lower bound.
- Uses the lower bound in logical schema derivation. Physical
scalar-subquery planning now defers to `Expr::nullable` instead of
hardcoding `nullable = true`, so the logical rule is the single source
of truth.
- Makes `max_rows()` account for the one-row form of `EmptyRelation`,
keeping both bounds consistent.
## Are these changes tested?
Yes. Unit tests cover row bounds, logical schema derivation, optimizer
simplification, and physical expression nullability for both possibly
empty and guaranteed non-empty scalar subqueries. A sqllogictest pins
the user-visible effect: `EXPLAIN` shows that `(SELECT count(*) FROM
empty_table) IS NULL` now folds to `false`, eliminating the subquery
from the plan. (Query results alone cannot show the change: a guaranteed
non-empty scalar subquery never evaluates to `NULL` at runtime.)
As an ablation check, restoring the conservative always-nullable
behavior makes the logical, optimizer, physical, and sqllogictest tests
fail.
The required all-feature Clippy check and extended workspace test suite
pass.
## Are there any user-facing changes?
Yes. Scalar subqueries proven to return at least one row now preserve
the nullability of their projected field, allowing valid simplification.
Potentially empty scalar subqueries remain nullable.
`LogicalPlan::min_rows()` is a new public API.
---
AI usage: Implemented with Codex; reviewed and test coverage extended
with Claude Code. I reviewed the code and made modifications where
appropriate.
---------
Signed-off-by: Fredrik Fornwall <fr...@fornwall.net>
---
datafusion/expr/src/expr_schema.rs | 57 +++++--
datafusion/expr/src/logical_plan/plan.rs | 187 ++++++++++++++++++++-
.../src/simplify_expressions/expr_simplifier.rs | 16 +-
datafusion/physical-expr/src/planner.rs | 38 ++++-
datafusion/sqllogictest/test_files/subquery.slt | 19 +++
5 files changed, 293 insertions(+), 24 deletions(-)
diff --git a/datafusion/expr/src/expr_schema.rs b/datafusion/expr/src/expr_schema.rs
index 8a9d0b8937..07fa6852a1 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::<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]
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index ba4fccfb06..6bd75c11c8 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::<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
diff --git a/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs b/datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs
index 5beb54b560..0e4836bb6d 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 09b2e5ce04..9a0bdc33da 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<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();
diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt
index 65760886d1..e76a3e8147 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);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org