From 5d819f9c1c8bd2df438487b7e09fd8230663966c Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 27 Sep 2026 23:11:56 -0500 Subject: [PATCH 1/3] perf: use one null-aware mark join for hashable IN subqueries in projections A projected `IN` / `NOT IN` subquery used three mark joins to materialize its three-valued result. When the mark column of the first join is already exact, that join alone is the full rewrite: - no side of the `IN` predicate can be NULL in the scope of an outer row, or - the join is null-aware and the `IN` equality is its first key. The null-aware hash join also applies a residual non-equality filter when it decides whether a NULL makes the mark UNKNOWN (#25339). Key nullability is read from the key expressions, not only their columns, so `NULLIF(id, 1)` or `TRY_CAST(s AS INT)` over a non-nullable column is treated as nullable. A correlated conjunct that rejects NULL on a key, such as a correlation that repeats the `IN` predicate, keeps that key out of the null-aware decision (#25480). A `NOT IN` filter whose `IN` equality cannot be the first key of a null-aware anti join now falls back to the mark-join materialization instead of planning a wrong or unsupported anti join. Co-Authored-By: Claude Opus 5.5 --- datafusion/optimizer/src/decorrelate.rs | 18 + .../src/decorrelate_predicate_subquery.rs | 676 ++++++++++++++---- .../test_files/projection_pushdown.slt | 58 +- .../sqllogictest/test_files/subquery.slt | 15 +- .../test_files/subquery_projection.slt | 495 +++++++++++++ 5 files changed, 1107 insertions(+), 155 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 4f89532e50525..3f2a78c456aa4 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -76,6 +76,18 @@ pub struct PullUpCorrelatedExpr { /// whether we have converted a scalar aggregation into a group aggregation. When unnesting /// lateral joins, we need to produce a left outer join in such cases. pub pulled_up_scalar_agg: bool, + /// Every correlated conjunct that a `Filter` of the subquery applies, + /// before `remove_duplicated_filter` drops the ones that the `IN` + /// predicate already covers. The outer references are kept, so that a + /// subquery column and an outer column with the same qualified name stay + /// apart. + /// + /// `join_filters` holds only the conjuncts that the join still needs. + /// This list is what the subquery enforces on its own rows. The caller + /// uses it to tell if a join key can be NULL inside the scope of an outer + /// row: `x IN (SELECT y FROM .. WHERE y = x)` keeps every NULL `y` out of + /// its result, although `join_filters` no longer says so. + pub correlated_filters: Vec, } impl Default for PullUpCorrelatedExpr { @@ -97,6 +109,7 @@ impl PullUpCorrelatedExpr { collected_count_expr_map: HashMap::new(), pull_up_having_expr: None, pulled_up_scalar_agg: false, + correlated_filters: Vec::new(), } } @@ -195,6 +208,11 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { .iter() .filter(|e| e.contains_outer()) .all(|&e| can_pullup_over_aggregation(e)); + for expr in &subquery_filter_exprs { + if expr.contains_outer() && !self.correlated_filters.contains(expr) { + self.correlated_filters.push((*expr).clone()); + } + } let (mut join_filters, subquery_filters) = find_join_exprs(subquery_filter_exprs)?; if let Some(in_predicate) = &self.in_predicate_opt { diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index cd5ca84f66b9e..8bee46a197543 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -21,6 +21,7 @@ use std::ops::Deref; use std::sync::Arc; use crate::decorrelate::PullUpCorrelatedExpr; +use crate::extract_equijoin_predicate::split_eq_and_noneq_join_predicate; use crate::optimizer::ApplyOrder; use crate::utils::replace_qualified_name; use crate::{OptimizerConfig, OptimizerRule}; @@ -28,16 +29,17 @@ use crate::{OptimizerConfig, OptimizerRule}; use datafusion_common::alias::AliasGenerator; use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; use datafusion_common::{ - Column, DFSchemaRef, ExprSchema, NullEquality, Result, ScalarValue, - assert_or_internal_err, plan_err, + Column, DFSchema, NullEquality, Result, ScalarValue, assert_or_internal_err, plan_err, }; use datafusion_expr::expr::{Exists, InSubquery}; -use datafusion_expr::expr_rewriter::create_col_from_scalar_expr; +use datafusion_expr::expr_rewriter::{ + create_col_from_scalar_expr, strip_outer_reference, +}; use datafusion_expr::logical_plan::{JoinType, Subquery}; use datafusion_expr::utils::{conjunction, expr_to_columns, split_conjunction_owned}; use datafusion_expr::{ - BinaryExpr, Expr, Filter, LogicalPlan, LogicalPlanBuilder, Operator, exists, - in_subquery, lit, not, not_exists, not_in_subquery, when, + BinaryExpr, Expr, ExprSchemable, Filter, LogicalPlan, LogicalPlanBuilder, Operator, + exists, in_subquery, lit, not, not_exists, not_in_subquery, when, }; use log::debug; @@ -128,8 +130,23 @@ impl OptimizerRule for DecorrelatePredicateSubquery { match build_join_top(&subquery, &cur_input, config.alias_generator())? { Some(plan) => cur_input = plan, - // If the subquery can not be converted to a Join, reconstruct the subquery expression and add it to the Filter - None => other_exprs.push(subquery.expr()), + // The subquery cannot become a semi or anti join. A + // `NOT IN` may still become mark joins that + // materialize its three-valued result, see + // `in_subquery_value_mark_join`. Any other subquery + // expression goes back into the filter as it is. + None => match subquery.expr() { + expr @ Expr::InSubquery(InSubquery { + negated: true, .. + }) => { + let (plan, expr) = rewrite_inner_subqueries( + cur_input, expr, config, true, true, + )?; + cur_input = plan; + other_exprs.push(expr); + } + expr => other_exprs.push(expr), + }, } } // The subquery expression is embedded within another expression @@ -236,7 +253,11 @@ fn rewrite_inner_subqueries( alias, needs_null_aware_mark, )? { - Some((plan, exists_expr)) => { + Some(MarkJoin { + plan, + mark: exists_expr, + .. + }) => { cur_input = plan; Ok(Transformed::yes(exists_expr)) } @@ -270,6 +291,7 @@ fn rewrite_inner_subqueries( alias, needs_null_aware_mark, )? + .map(|join| (join.plan, join.mark)) }; match rewritten { Some((plan, exists_expr)) => { @@ -285,6 +307,21 @@ fn rewrite_inner_subqueries( Ok((cur_input, expr_without_subqueries.data)) } +/// Rewrites an `IN` subquery that gives a value, for example in a SELECT list. +/// The value follows SQL three-valued logic: TRUE for a match, FALSE for a miss +/// and NULL (UNKNOWN) when the answer depends on a NULL. +/// +/// There are two paths: +/// +/// * One mark join. When the mark column is already exact under three-valued +/// logic (see [`MarkJoin::three_valued_exact`]), the mark column is the +/// answer and this single join is the full rewrite. This is the usual case. +/// * Three mark joins. In the other case the mark column only tells TRUE from +/// not-TRUE, so the UNKNOWN cases must be materialized: one more join tells +/// if the subquery gives a NULL, and one more tells if the subquery gives +/// any row. A `CASE` expression puts the three marks together. The two extra +/// joins have no join predicate, so use them only when the first path +/// cannot apply. fn in_subquery_value_mark_join( left: &LogicalPlan, subquery: &LogicalPlan, @@ -296,23 +333,41 @@ fn in_subquery_value_mark_join( .head_output_expr()? .map_or(plan_err!("single expression required."), Ok)?; let in_predicate = Expr::eq(expr.clone(), output_expr.clone()); - let Some((matched_plan, matched)) = - mark_join(left, subquery, Some(&in_predicate), false, alias, true)? + let Some(MarkJoin { + plan: matched_plan, + mark: matched, + three_valued_exact, + }) = mark_join(left, subquery, Some(&in_predicate), false, alias, true)? else { return Ok(None); }; + // The mark column is the full answer when it is exact. Negation does not + // change that, because NOT UNKNOWN is UNKNOWN. + if three_valued_exact { + return Ok(Some(( + matched_plan, + if negated { not(matched) } else { matched }, + ))); + } + // SQL IN needs three facts per outer row to distinguish FALSE from UNKNOWN. let null_subquery = LogicalPlanBuilder::from(subquery.clone()) .filter(output_expr.is_null())? .build()?; - let Some((null_plan, subquery_has_null)) = - mark_join(&matched_plan, &null_subquery, None, false, alias, true)? + let Some(MarkJoin { + plan: null_plan, + mark: subquery_has_null, + .. + }) = mark_join(&matched_plan, &null_subquery, None, false, alias, true)? else { return Ok(None); }; - let Some((final_plan, subquery_non_empty)) = - mark_join(&null_plan, subquery, None, false, alias, true)? + let Some(MarkJoin { + plan: final_plan, + mark: subquery_non_empty, + .. + }) = mark_join(&null_plan, subquery, None, false, alias, true)? else { return Ok(None); }; @@ -428,14 +483,15 @@ fn build_join_top( }; let subquery = query_info.query.subquery.as_ref(); let subquery_alias = alias.next("__correlated_sq"); - build_join( + Ok(build_join( left, subquery, in_predicate_opt.as_ref(), join_type, - subquery_alias, + &subquery_alias, true, - ) + )? + .map(|join| join.plan)) } /// This is used to handle the case when the subquery is embedded in a more complex boolean @@ -460,7 +516,7 @@ fn mark_join( negated: bool, alias_generator: &Arc, needs_null_aware_mark: bool, -) -> Result> { +) -> Result> { let alias = alias_generator.next("__correlated_sq"); let exists_col = Expr::Column(Column::new(Some(alias.clone()), "mark")); @@ -471,43 +527,181 @@ fn mark_join( subquery, in_predicate_opt, JoinType::LeftMark, - alias, + &alias, needs_null_aware_mark, )? - .map(|plan| (plan, exists_expr))) + .map(|join| MarkJoin { + plan: join.plan, + mark: exists_expr, + three_valued_exact: join.mark_is_three_valued_exact, + })) } -/// Check if join keys in the join filter may contain NULL values -/// -/// Returns true if any join key column is nullable on either side. -/// This is used to optimize null-aware anti joins: if all join keys are non-nullable, -/// we can use a regular anti join instead of the more expensive null-aware variant. -fn join_keys_may_be_null( +/// A [`JoinType::LeftMark`] join that replaces a subquery predicate. +struct MarkJoin { + /// The outer plan with the subquery joined into it. + plan: LogicalPlan, + /// Reads the mark column of the join, negated if the caller asked for it. + mark: Expr, + /// True when the mark column already gives SQL three-valued `IN` + /// semantics: TRUE for a match, FALSE for a miss and NULL for UNKNOWN. + /// + /// A plain mark is exact when neither side of the `IN` predicate can be + /// NULL in scope. Otherwise the join must be null-aware, and the `IN` + /// equality must be its first key, which is where the null-aware hash join + /// reads the `IN` value. That join also applies a residual non-equality + /// filter when it decides whether a NULL makes the mark UNKNOWN. + /// + /// In any other case the caller materializes the UNKNOWN rows with more + /// joins, see [`in_subquery_value_mark_join`]. + three_valued_exact: bool, +} + +/// Is the `IN` equality the first equality of `join_filter` that the hash +/// join can use as a key? A null-aware hash join reads its first key as the +/// `IN` value and the other keys as the correlation. +fn value_is_first_key( + in_value: &InValue, join_filter: &Expr, - left_schema: &DFSchemaRef, - right_schema: &DFSchemaRef, + left_schema: &DFSchema, + right_schema: &DFSchema, ) -> Result { - // Extract columns from the join filter - let mut columns = std::collections::HashSet::new(); - expr_to_columns(join_filter, &mut columns)?; - - // Check if any column is nullable - for col in columns { - // Check in left schema - if let Ok(field) = left_schema.field_from_column(&col) - && field.as_ref().is_nullable() - { - return Ok(true); + let (equijoin_keys, _) = split_eq_and_noneq_join_predicate( + join_filter.clone(), + left_schema, + right_schema, + )?; + Ok(equijoin_keys.first().is_some_and(|(value, column)| { + value == &in_value.value && column == &in_value.subquery_column + })) +} + +/// The two sides of an `IN` predicate, `value IN (SELECT output_expr ..)`. +struct InValue { + /// The value from the outer plan, as the join filter refers to it. This is + /// a projected column when `value_as_written` is a constant. + value: Expr, + /// The subquery output, as the join filter refers to it: a column of the + /// aliased subquery. + subquery_column: Expr, + /// The value as the query writes it. + value_as_written: Expr, + /// The subquery output expression as the query writes it. + output_expr: Expr, +} + +impl InValue { + /// Can either side be NULL for a row inside the scope of an outer row? + /// See [`key_may_be_null_in_scope`]. The correlated filters name the + /// expressions as the query writes them, so the test matches on those. + /// + /// Only then can `IN` be UNKNOWN, so only then does the join need + /// null-aware semantics. A correlation key that is NULL just empties the + /// scope, which makes `IN` FALSE. + fn may_be_null_in_scope( + &self, + left_schema: &DFSchema, + right_schema: &DFSchema, + scope_filters: &[Expr], + ) -> Result { + Ok(key_may_be_null_in_scope( + self.value.nullable(left_schema)?, + &self.value_as_written, + true, + scope_filters, + ) || key_may_be_null_in_scope( + self.subquery_column.nullable(right_schema)?, + &self.output_expr, + false, + scope_filters, + )) + } +} + +/// Can this join key be NULL for a row inside the scope of an outer row? +/// +/// `nullable` is what the schema says about the key. The key is a full +/// expression, not only a column. An expression can be NULL although none of +/// its columns is nullable: `NULLIF(id, 1)`, `TRY_CAST(s AS INT)`, a `CASE` +/// with no `ELSE` branch, or a scalar function that does not declare its +/// nullability. So the caller asks the key expression itself, against the +/// schema of its own side. +/// +/// The subquery can still keep every NULL out of its result. A correlated +/// conjunct such as `y = x`, `y > x` or `y IS NOT NULL` is never TRUE for a +/// NULL `y`, so no row with a NULL `y` is in the scope of any outer row, and +/// an outer row with a NULL `x` has an empty scope. Neither can make `IN` +/// UNKNOWN, so such a key does not need null-aware semantics. The usual shape +/// is a correlation that repeats the `IN` predicate, `x IN (SELECT y FROM t +/// WHERE y = x)`: the pull up drops that conjunct from the join filter, but +/// it still bounds the subquery result +/// (). +/// +/// This is a sufficient test, not an exact one. A conjunct counts only if it +/// is a comparison or an `IS NOT NULL` on the key expression itself, casts +/// aside. Any other conjunct is assumed to let a NULL through, which keeps +/// the join null-aware. +/// +/// `key_is_outer` tells on which side of the join the key is. The scope +/// filters keep their outer references, and a side of a conjunct matches an +/// outer key only if it names outer columns alone. A subquery column can have +/// the same qualified name as an outer column (`FROM t AS a` inside a query +/// over `a`), and it must not be read as the outer key. +fn key_may_be_null_in_scope( + nullable: bool, + key: &Expr, + key_is_outer: bool, + scope_filters: &[Expr], +) -> bool { + if !nullable { + return false; + } + let key = strip_casts(key); + !scope_filters + .iter() + .any(|filter| filter_rejects_null(filter, key, key_is_outer)) +} + +/// Is `filter` never TRUE when `key` is NULL? See +/// [`key_may_be_null_in_scope`] for `key_is_outer`. +fn filter_rejects_null(filter: &Expr, key: &Expr, key_is_outer: bool) -> bool { + let is_key = |side: &Expr| { + let side = strip_casts(side); + if key_is_outer { + side.contains_outer() + && side.column_refs().is_empty() + && &strip_outer_reference(side.clone()) == key + } else { + !side.contains_outer() && side == key } - // Check in right schema - if let Ok(field) = right_schema.field_from_column(&col) - && field.as_ref().is_nullable() - { - return Ok(true); + }; + match filter { + Expr::BinaryExpr(BinaryExpr { left, op, right }) => { + matches!( + op, + Operator::Eq + | Operator::NotEq + | Operator::Lt + | Operator::LtEq + | Operator::Gt + | Operator::GtEq + ) && (is_key(left) || is_key(right)) } + Expr::IsNotNull(expr) => is_key(expr), + _ => false, } +} - Ok(false) +/// `CAST(NULL)` is NULL and `CAST(x)` is not NULL for a non-null `x`, so a +/// conjunct on `x` says the same about `CAST(x)`, and the other way round. +/// Type coercion adds such casts on one side only. `TRY_CAST` can make a NULL +/// from a value, so it is not unwrapped. +fn strip_casts(expr: &Expr) -> &Expr { + let mut expr = expr; + while let Expr::Cast(cast) = expr { + expr = cast.expr.as_ref(); + } + expr } /// Combines the `IN`/`NOT IN` equality with the correlation predicates, if any. @@ -522,14 +716,23 @@ fn in_predicate_first(in_predicate: Expr, correlation: Option) -> Expr { } } +/// The outcome of [`build_join`]. +struct BuiltJoin { + /// The outer plan with the subquery joined into it. + plan: LogicalPlan, + /// See [`MarkJoin::three_valued_exact`]. This is always false unless the + /// join is a [`JoinType::LeftMark`] join built for an `IN` predicate. + mark_is_three_valued_exact: bool, +} + fn build_join( left: &LogicalPlan, subquery: &LogicalPlan, in_predicate_opt: Option<&Expr>, join_type: JoinType, - alias: String, + alias: &str, needs_null_aware_mark: bool, -) -> Result> { +) -> Result> { let mut pull_up = PullUpCorrelatedExpr::new() .with_in_predicate_opt(in_predicate_opt.cloned()) .with_exists_sub_query(in_predicate_opt.is_none()); @@ -551,54 +754,54 @@ fn build_join( // alias the join filter let join_filter_opt = conjunction(pull_up.join_filters) .map_or(Ok(None), |filter| { - replace_qualified_name(filter, &all_correlated_cols, &alias).map(Some) + replace_qualified_name(filter, &all_correlated_cols, alias).map(Some) })?; - // The outer value expression of an `IN`/`NOT IN` predicate, recorded together - // with the subquery column it is compared against, a name for the column it - // can be projected as, and the correlation predicates that share the join - // filter with it. - let mut in_value_expr = None; - - let mut join_filter = match (join_filter_opt, in_predicate_opt.cloned()) { - ( - correlation_opt, - Some(Expr::BinaryExpr(BinaryExpr { - left, - op: Operator::Eq, - right, - })), - ) => { - let value_name = format!("{alias}_value"); - let right_col = create_col_from_scalar_expr(&right, alias)?; - let value = left.deref().clone(); - let in_predicate = Expr::eq(value.clone(), Expr::Column(right_col.clone())); - let join_filter = in_predicate_first(in_predicate, correlation_opt.clone()); - in_value_expr = Some((value, right_col, value_name, correlation_opt)); - - join_filter + // The two sides of the `IN` predicate: the value from the outer plan and + // the subquery output it is compared with, renamed to the alias. + let in_value = match in_predicate_opt { + Some(Expr::BinaryExpr(BinaryExpr { + left, + op: Operator::Eq, + right, + })) => { + let right_col = create_col_from_scalar_expr(right, alias.to_string())?; + Some(InValue { + value: left.deref().clone(), + subquery_column: Expr::Column(right_col), + value_as_written: left.deref().clone(), + output_expr: right.deref().clone(), + }) } - (Some(join_filter), _) => join_filter, - (None, None) => lit(true), - _ => return Ok(None), + Some(_) => return Ok(None), + None => None, + }; + + // True if the `IN` value or the subquery column it is compared with can be + // NULL inside the scope of an outer row. `EXISTS` has no value to compare. + let value_may_be_null = match &in_value { + Some(in_value) => in_value.may_be_null_in_scope( + left.schema(), + sub_query_alias.schema(), + &pull_up.correlated_filters, + )?, + None => false, }; // Whether this join needs null-aware (`NOT IN` three-valued) semantics. // Decided once: the constant projection below, the mark branch and the anti // branch all depend on the same answer. `NOT EXISTS` is two-valued and has - // no `in_predicate_opt`, so it never qualifies; non-nullable keys on both - // sides mean no NULL can reach the join in the first place. + // no `IN` value, so it never qualifies. // - // The left projection below only adds a non-nullable literal column and the - // right projection only drops unreferenced columns, so neither changes the - // nullability this looks at. - let null_aware = in_predicate_opt.is_some() + // The left projection below only adds a column for a constant value and + // the right projection only drops unreferenced columns, so neither changes + // the nullability this looks at. + let null_aware = value_may_be_null && match join_type { JoinType::LeftAnti => true, JoinType::LeftMark => needs_null_aware_mark, _ => false, - } - && join_keys_may_be_null(&join_filter, left.schema(), sub_query_alias.schema())?; + }; // ` IN/NOT IN ()`: the outer value expression holds no // column reference, so ` = __correlated_sq.col` is not a valid @@ -614,34 +817,67 @@ fn build_join( // into a real equi-join key, which fixes all three. It only pays for itself // on a join that ends up null-aware. let mut projected_left = None; - if let Some((value, right_col, mut value_name, correlation_opt)) = in_value_expr - && value.column_refs().is_empty() - && null_aware - { - // The projected column is unqualified, so a left field that already has - // this name — however unlikely — would make the reference ambiguous. - let left_schema = left.schema(); - while left_schema.fields().iter().any(|f| f.name() == &value_name) { - value_name.push('_'); + let in_value = match in_value { + Some(in_value) if in_value.value.column_refs().is_empty() && null_aware => { + // The projected column is unqualified, so a left field that already + // has this name — however unlikely — would make the reference + // ambiguous. + let mut value_name = format!("{alias}_value"); + let left_schema = left.schema(); + while left_schema.fields().iter().any(|f| f.name() == &value_name) { + value_name.push('_'); + } + let value_col = Column::new_unqualified(value_name); + let projections = left_schema + .columns() + .into_iter() + .map(Expr::from) + .chain(std::iter::once(in_value.value.alias(value_col.name()))) + .collect::>(); + projected_left = Some( + LogicalPlanBuilder::from(left.clone()) + .project(projections)? + .build()?, + ); + Some(InValue { + value: Expr::Column(value_col), + ..in_value + }) } - let value_col = Column::new_unqualified(value_name); - let projections = left_schema - .columns() - .into_iter() - .map(Expr::from) - .chain(std::iter::once(value.alias(value_col.name()))) - .collect::>(); - projected_left = Some( - LogicalPlanBuilder::from(left.clone()) - .project(projections)? - .build()?, - ); - // Rebuild the `IN` equality against the projected column. - let in_predicate = Expr::eq(Expr::Column(value_col), Expr::Column(right_col)); - join_filter = in_predicate_first(in_predicate, correlation_opt); - } + other => other, + }; let left = projected_left.as_ref().unwrap_or(left); + let join_filter = match (&in_value, join_filter_opt) { + (Some(in_value), correlation_opt) => in_predicate_first( + Expr::eq(in_value.value.clone(), in_value.subquery_column.clone()), + correlation_opt, + ), + (None, Some(correlation)) => correlation, + (None, None) => lit(true), + }; + + // Only a null-aware join reads the key order. + let value_is_first_key = match &in_value { + Some(in_value) if null_aware => value_is_first_key( + in_value, + &join_filter, + left.schema(), + sub_query_alias.schema(), + )?, + _ => false, + }; + + // A `NOT IN` in a filter builds a `LeftAnti` join. The null-aware hash + // join reads its first key as the `NOT IN` value and the other keys as the + // scope of the outer row (see `HashJoinExec::null_aware`). If the `IN` + // equality did not become that first key, give up here. The caller then + // materializes the UNKNOWN rows with more joins, see + // `in_subquery_value_mark_join`. + if join_type == JoinType::LeftAnti && null_aware && !value_is_first_key { + return Ok(None); + } + if matches!(join_type, JoinType::LeftMark | JoinType::RightMark) { let right_schema = sub_query_alias.schema(); @@ -680,6 +916,9 @@ fn build_join( // nullable mark column. A non-equality correlation stays behind as a // join filter, which the hash join also applies when it decides // whether a NULL makes the mark UNKNOWN. + let mark_is_three_valued_exact = in_value.is_some() + && (!value_may_be_null || (null_aware && value_is_first_key)); + let new_plan = LogicalPlanBuilder::from(left.clone()) .join_detailed_with_options( right_projected, @@ -696,7 +935,10 @@ fn build_join( new_plan.display_indent() ); - return Ok(Some(new_plan)); + return Ok(Some(BuiltJoin { + plan: new_plan, + mark_is_three_valued_exact, + })); } // join our sub query into the main plan @@ -721,7 +963,10 @@ fn build_join( "predicate subquery optimized:\n{}", new_plan.display_indent() ); - Ok(Some(new_plan)) + Ok(Some(BuiltJoin { + plan: new_plan, + mark_is_three_valued_exact: false, + })) } #[derive(Debug)] @@ -807,6 +1052,15 @@ mod tests { table_scan(Some(name), &schema, None)?.build() } + /// `CASE WHEN test.c = 1 THEN NULL ELSE test.c END`: an expression that can + /// be NULL although `test.c` is not nullable. `NULLIF(c, 1)` and + /// `TRY_CAST(c AS INT)` have the same shape, but the optimizer crate cannot + /// depend on the function crates. + fn nullable_key_expr() -> Result { + when(col("test.c").eq(lit(1u32)), lit(ScalarValue::UInt32(None))) + .otherwise(col("test.c")) + } + fn has_null_aware_left_mark_join(plan: &LogicalPlan) -> bool { if let LogicalPlan::Join(join) = plan && join.join_type == JoinType::LeftMark @@ -817,6 +1071,18 @@ mod tests { plan.inputs().into_iter().any(has_null_aware_left_mark_join) } + fn has_non_null_aware_left_mark_join(plan: &LogicalPlan) -> bool { + if let LogicalPlan::Join(join) = plan + && join.join_type == JoinType::LeftMark + { + return !join.null_aware; + } + + plan.inputs() + .into_iter() + .any(has_non_null_aware_left_mark_join) + } + fn optimize_with_decorrelate(plan: LogicalPlan) -> Result { let optimizer = crate::Optimizer::with_rules(vec![Arc::new( DecorrelatePredicateSubquery::new(), @@ -1530,20 +1796,127 @@ mod tests { assert_optimized_plan_equal!( plan, @r" - Projection: CASE WHEN __correlated_sq_1.mark THEN Boolean(true) WHEN __correlated_sq_2.mark OR test.c IS NULL AND __correlated_sq_3.mark THEN Boolean(NULL) ELSE Boolean(false) END AS is_present [is_present:Boolean;N] - LeftMark Join: Filter: Boolean(true) [a:UInt32, b:UInt32, c:UInt32, mark:Boolean;N, mark:Boolean;N, mark:Boolean;N] - LeftMark Join: Filter: Boolean(true) [a:UInt32, b:UInt32, c:UInt32, mark:Boolean;N, mark:Boolean;N] - LeftMark Join: Filter: test.c = __correlated_sq_1.c [a:UInt32, b:UInt32, c:UInt32, mark:Boolean;N] - TableScan: test [a:UInt32, b:UInt32, c:UInt32] - Projection: __correlated_sq_1.c [c:UInt32] - SubqueryAlias: __correlated_sq_1 [c:UInt32] - Projection: sq.c [c:UInt32] - TableScan: sq [a:UInt32, b:UInt32, c:UInt32] - SubqueryAlias: __correlated_sq_2 [c:UInt32] - Filter: sq.c IS NULL [c:UInt32] - Projection: sq.c [c:UInt32] - TableScan: sq [a:UInt32, b:UInt32, c:UInt32] - SubqueryAlias: __correlated_sq_3 [c:UInt32] + Projection: __correlated_sq_1.mark AS is_present [is_present:Boolean;N] + LeftMark Join: Filter: test.c = __correlated_sq_1.c [a:UInt32, b:UInt32, c:UInt32, mark:Boolean;N] + TableScan: test [a:UInt32, b:UInt32, c:UInt32] + Projection: __correlated_sq_1.c [c:UInt32] + SubqueryAlias: __correlated_sq_1 [c:UInt32] + Projection: sq.c [c:UInt32] + TableScan: sq [a:UInt32, b:UInt32, c:UInt32] + " + ) + } + + /// A residual non-equality correlation stays on the one mark join. The + /// null-aware hash join applies it when it decides whether a NULL makes + /// the mark UNKNOWN, so the mark is still exact. + #[test] + fn in_subquery_in_projection_with_residual_filter() -> Result<()> { + let subquery = Arc::new( + LogicalPlanBuilder::from(nullable_scalar_mark_scan("inner_t")?) + .filter( + out_ref_col(DataType::Int32, "outer_t.grp").gt(col("inner_t.grp")), + )? + .project(vec![col("inner_t.id")])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(nullable_scalar_mark_scan("outer_t")?) + .project(vec![ + in_subquery(col("outer_t.id"), subquery).alias("is_present"), + ])? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + Projection: __correlated_sq_1.mark AS is_present [is_present:Boolean;N] + LeftMark Join: Filter: outer_t.id = __correlated_sq_1.id AND outer_t.grp > __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N, mark:Boolean;N] + TableScan: outer_t [id:Int32;N, grp:Int32;N] + Projection: __correlated_sq_1.id, __correlated_sq_1.grp [id:Int32;N, grp:Int32;N] + SubqueryAlias: __correlated_sq_1 [id:Int32;N, grp:Int32;N] + Projection: inner_t.id, inner_t.grp [id:Int32;N, grp:Int32;N] + TableScan: inner_t [id:Int32;N, grp:Int32;N] + " + ) + } + + /// `NOT IN` reads the same mark column, negated. The keys are nullable here, + /// so the join is null-aware and the mark is NULL for the UNKNOWN rows. + #[test] + fn not_in_subquery_in_projection() -> Result<()> { + let subquery = Arc::new( + LogicalPlanBuilder::from(nullable_scalar_mark_scan("inner_t")?) + .project(vec![col("inner_t.id")])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(nullable_scalar_mark_scan("outer_t")?) + .project(vec![ + not_in_subquery(col("outer_t.id"), subquery).alias("is_absent"), + ])? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + Projection: NOT __correlated_sq_1.mark AS is_absent [is_absent:Boolean;N] + LeftMark Join: Filter: outer_t.id = __correlated_sq_1.id null_aware [id:Int32;N, grp:Int32;N, mark:Boolean;N] + TableScan: outer_t [id:Int32;N, grp:Int32;N] + Projection: __correlated_sq_1.id [id:Int32;N] + SubqueryAlias: __correlated_sq_1 [id:Int32;N] + Projection: inner_t.id [id:Int32;N] + TableScan: inner_t [id:Int32;N, grp:Int32;N] + " + ) + } + + /// A key expression can be NULL although none of its columns is nullable. + /// The mark join must then be null-aware, so the mark is NULL for the rows + /// that give UNKNOWN. + #[test] + fn in_subquery_in_projection_with_nullable_key_expr() -> Result<()> { + let plan = LogicalPlanBuilder::from(test_table_scan()?) + .project(vec![ + in_subquery(nullable_key_expr()?, test_subquery_with_name("sq")?) + .alias("is_present"), + ])? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + Projection: __correlated_sq_1.mark AS is_present [is_present:Boolean;N] + LeftMark Join: Filter: CASE WHEN test.c = UInt32(1) THEN UInt32(NULL) ELSE test.c END = __correlated_sq_1.c null_aware [a:UInt32, b:UInt32, c:UInt32, mark:Boolean;N] + TableScan: test [a:UInt32, b:UInt32, c:UInt32] + Projection: __correlated_sq_1.c [c:UInt32] + SubqueryAlias: __correlated_sq_1 [c:UInt32] + Projection: sq.c [c:UInt32] + TableScan: sq [a:UInt32, b:UInt32, c:UInt32] + " + ) + } + + /// The `NOT IN` filter path builds a `LeftAnti` join. It reads the key + /// nullability the same way, so a nullable key expression over columns that + /// are not nullable also makes that join null-aware. + #[test] + fn not_in_subquery_filter_with_nullable_key_expr() -> Result<()> { + let plan = LogicalPlanBuilder::from(test_table_scan()?) + .filter(not_in_subquery( + nullable_key_expr()?, + test_subquery_with_name("sq")?, + ))? + .project(vec![col("test.b")])? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + Projection: test.b [b:UInt32] + LeftAnti Join: Filter: CASE WHEN test.c = UInt32(1) THEN UInt32(NULL) ELSE test.c END = __correlated_sq_1.c null_aware [a:UInt32, b:UInt32, c:UInt32] + TableScan: test [a:UInt32, b:UInt32, c:UInt32] + SubqueryAlias: __correlated_sq_1 [c:UInt32] Projection: sq.c [c:UInt32] TableScan: sq [a:UInt32, b:UInt32, c:UInt32] " @@ -1704,6 +2077,63 @@ mod tests { Ok(()) } + /// A correlation that repeats the `IN` predicate keeps every NULL out of + /// the subquery result, so the mark join must not be null-aware. + #[test] + fn mark_join_for_in_predicate_correlation_is_not_null_aware() -> Result<()> { + let outer_scan = nullable_scalar_mark_scan("outer_t")?; + let inner_scan = nullable_scalar_mark_scan("inner_t")?; + + let subquery = Arc::new( + LogicalPlanBuilder::from(inner_scan) + .filter(out_ref_col(DataType::Int32, "outer_t.id").eq(col("inner_t.id")))? + .project(vec![col("inner_t.id")])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(outer_scan) + .filter(in_subquery(col("outer_t.id"), subquery).is_null())? + .build()?; + + let optimized = optimize_with_decorrelate(plan)?; + assert!( + has_non_null_aware_left_mark_join(&optimized), + "{}", + optimized.display_indent_schema() + ); + + Ok(()) + } + + /// The same for the `LeftAnti` join that a `NOT IN` filter builds. + #[test] + fn anti_join_for_in_predicate_correlation_is_not_null_aware() -> Result<()> { + let outer_scan = nullable_scalar_mark_scan("outer_t")?; + let inner_scan = nullable_scalar_mark_scan("inner_t")?; + + let subquery = Arc::new( + LogicalPlanBuilder::from(inner_scan) + .filter(out_ref_col(DataType::Int32, "outer_t.id").eq(col("inner_t.id")))? + .project(vec![col("inner_t.id")])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(outer_scan) + .filter(not_in_subquery(col("outer_t.id"), subquery))? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + LeftAnti Join: Filter: outer_t.id = __correlated_sq_1.id [id:Int32;N, grp:Int32;N] + TableScan: outer_t [id:Int32;N, grp:Int32;N] + SubqueryAlias: __correlated_sq_1 [id:Int32;N] + Projection: inner_t.id [id:Int32;N] + TableScan: inner_t [id:Int32;N, grp:Int32;N] + " + ) + } + #[test] fn in_subquery_both_side_expr() -> Result<()> { let table_scan = test_table_scan()?; diff --git a/datafusion/sqllogictest/test_files/projection_pushdown.slt b/datafusion/sqllogictest/test_files/projection_pushdown.slt index c92c95fdfbc59..f0b8b9eeae73b 100644 --- a/datafusion/sqllogictest/test_files/projection_pushdown.slt +++ b/datafusion/sqllogictest/test_files/projection_pushdown.slt @@ -2281,10 +2281,11 @@ SET datafusion.execution.target_partitions = 4; # # Each ingredient below is load-bearing: # -# * The `IN ()` sits in the SELECT list, not in a WHERE clause, so -# `decorrelate_predicate_subquery` (which runs earlier) leaves it alone and it -# is still a subquery expression by the time extraction runs. A subquery in -# WHERE would be flattened into the main plan, where the old scan could see it. +# * The subquery must still be a subquery expression when extraction runs. +# `decorrelate_predicate_subquery` runs earlier and now flattens `IN` and +# `EXISTS` in a SELECT list too, so a plain uncorrelated `IN` is gone before +# extraction. This subquery is correlated and has a `LIMIT`, which that rule +# cannot pull up, so the rule leaves it alone. # * The alias inside the subquery is literally `__datafusion_extracted_1`. Rename # it to anything outside the reserved prefix and there is nothing to collide # with -- the query then passes with or without the fix and guards nothing. @@ -2295,11 +2296,12 @@ SET datafusion.execution.target_partitions = 4; # Without the fix, extraction reuses `__datafusion_extracted_1` and planning # aborts with: Optimizer rule 'push_down_leaf_projections' failed Schema error: # Schema contains duplicate unqualified field name __datafusion_extracted_1. -# With the fix, generated aliases remain distinct, as the plan below shows. +# With the fix, the generator starts at 2, so the alias made inside the subquery +# is `__datafusion_extracted_2` and the two names stay distinct. # -# Keep this as `EXPLAIN` under `logical_plan_only`: the logical plan exposes both -# the collision-free extracted aliases and the mark joins used to preserve the -# three-valued semantics of `IN` in a projection. +# Keep this as `EXPLAIN` under `logical_plan_only`: an `InSubquery` expression +# that survives decorrelation has no physical plan, so only the logical plan can +# show the collision-free extracted aliases. ##################### statement ok @@ -2313,30 +2315,28 @@ SELECT SELECT id FROM ( SELECT id, s['label'] AS __datafusion_extracted_1 - FROM simple_struct - WHERE s['value'] > 120 + FROM simple_struct inner_t + WHERE s['value'] > 120 AND inner_t.id = outer_t.id ) WHERE __datafusion_extracted_1 <> 'delta' + LIMIT 1 ) AS has_matching_label -FROM simple_struct; ----- -logical_plan -01)Projection: simple_struct.id, __correlated_sq_1.mark IS NOT DISTINCT FROM Boolean(true) OR __correlated_sq_2.mark IS NOT DISTINCT FROM Boolean(true) AND __correlated_sq_1.mark IS DISTINCT FROM Boolean(true) AND Boolean(NULL) AS has_matching_label -02)--LeftMark Join: -03)----LeftMark Join: -04)------LeftMark Join: simple_struct.id = __correlated_sq_1.id -05)--------TableScan: simple_struct projection=[id] -06)--------SubqueryAlias: __correlated_sq_1 -07)----------Projection: simple_struct.id -08)------------Filter: __datafusion_extracted_4 > Int64(120) AND __datafusion_extracted_1 != Utf8("delta") -09)--------------Projection: get_field(simple_struct.s, Utf8("value")) AS __datafusion_extracted_4, simple_struct.id, get_field(simple_struct.s, Utf8("label")) AS __datafusion_extracted_1 -10)----------------TableScan: simple_struct projection=[id, s], partial_filters=[get_field(simple_struct.s, Utf8("value")) > Int64(120)] -11)------EmptyRelation: rows=0 -12)----SubqueryAlias: __correlated_sq_3 -13)------Projection: simple_struct.id -14)--------Filter: __datafusion_extracted_6 > Int64(120) AND __datafusion_extracted_1 != Utf8("delta") -15)----------Projection: get_field(simple_struct.s, Utf8("value")) AS __datafusion_extracted_6, simple_struct.id, get_field(simple_struct.s, Utf8("label")) AS __datafusion_extracted_1 -16)------------TableScan: simple_struct projection=[id, s], partial_filters=[Boolean(true), get_field(simple_struct.s, Utf8("value")) > Int64(120)] +FROM simple_struct outer_t; +---- +logical_plan +01)Projection: outer_t.id, outer_t.id IN () AS has_matching_label +02)--Subquery: +03)----Projection: inner_t.id +04)------Projection: inner_t.id, __datafusion_extracted_1 +05)--------SubqueryAlias: inner_t +06)----------Projection: simple_struct.id, simple_struct.s, __datafusion_extracted_1 +07)------------Limit: skip=0, fetch=1 +08)--------------Filter: __datafusion_extracted_2 > Int64(120) AND __datafusion_extracted_1 != Utf8("delta") +09)----------------Filter: simple_struct.id = outer_ref(outer_t.id) +10)------------------Projection: get_field(simple_struct.s, Utf8("value")) AS __datafusion_extracted_2, get_field(simple_struct.s, Utf8("label")) AS __datafusion_extracted_1, simple_struct.id, simple_struct.s +11)--------------------TableScan: simple_struct, partial_filters=[simple_struct.id = outer_ref(outer_t.id), get_field(simple_struct.s, Utf8("value")) > Int64(120)] +12)--SubqueryAlias: outer_t +13)----TableScan: simple_struct projection=[id] statement ok set datafusion.explain.logical_plan_only = false; diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index d27c244e0d4df..170c6232af50b 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -2791,9 +2791,18 @@ SELECT gs_outer.k, EXISTS (SELECT 1 FROM gs_inner WHERE gs_inner.k = gs_outer.k statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists SELECT gs_outer.k, EXISTS (SELECT 1 FROM gs_inner WHERE gs_inner.k = gs_outer.k GROUP BY ROLLUP(gs_inner.j)) FROM gs_outer; -# The `IN` form of the same subquery. -statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression InSubquery -SELECT gs_outer.k, gs_outer.k IN (SELECT gs_inner.k FROM gs_inner WHERE gs_inner.k = gs_outer.k GROUP BY ROLLUP(gs_inner.k)) FROM gs_outer; +# The `IN` form of the same subquery decorrelates: the correlation repeats the +# `IN` predicate, so the pull up drops it and adds no column to the grouping +# set. The grand-total row is a NULL for every outer row, so a miss is UNKNOWN. +# Pinned to the current wrong result: `k = 4` and `k = NULL` should be NULL. +query IB +SELECT gs_outer.k, gs_outer.k IN (SELECT gs_inner.k FROM gs_inner WHERE gs_inner.k = gs_outer.k GROUP BY ROLLUP(gs_inner.k)) FROM gs_outer ORDER BY gs_outer.k; +---- +1 true +2 true +4 false +5 true +NULL false # `NOT EXISTS` too. statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists diff --git a/datafusion/sqllogictest/test_files/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index ab7c80d9de165..ad30a307aab0a 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -97,3 +97,498 @@ FROM outer_values o; 3 NULL 4 true 5 NULL + +# Plan shapes and NULL semantics of a projected IN subquery. +# +# `n1.id` holds a NULL, `n2.id` holds a NULL, and `n3.id` holds none. The mark +# column of a LeftMark join carries the three-valued result on its own when the +# join filter is hashable only, so one join per subquery is enough. + +statement ok +CREATE TABLE n1(id INT, z INT) AS VALUES (1, 10), (2, 20), (NULL, 30), (4, 40); + +statement ok +CREATE TABLE n2(id INT, z INT) AS VALUES (1, 5), (NULL, 50); + +statement ok +CREATE TABLE n3(id INT) AS VALUES (1), (2); + +# One hash mark join per subquery. There is no materialization join, so no +# nested loop join over outer x inner rows. +query TT +EXPLAIN SELECT id, id IN (SELECT id FROM n3) AS m3, id IN (SELECT id FROM n2) AS m2 FROM n1; +---- +logical_plan +01)Projection: n1.id, __correlated_sq_1.mark AS m3, __correlated_sq_2.mark AS m2 +02)--LeftMark Join: n1.id = __correlated_sq_2.id null_aware +03)----LeftMark Join: n1.id = __correlated_sq_1.id null_aware +04)------TableScan: n1 projection=[id] +05)------SubqueryAlias: __correlated_sq_1 +06)--------TableScan: n3 projection=[id] +07)----SubqueryAlias: __correlated_sq_2 +08)------TableScan: n2 projection=[id] +physical_plan +01)ProjectionExec: expr=[id@0 as id, mark@1 as m3, mark@2 as m2] +02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], null_aware +03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], null_aware +04)------DataSourceExec: partitions=1, partition_sizes=[1] +05)------DataSourceExec: partitions=1, partition_sizes=[1] +06)----DataSourceExec: partitions=1, partition_sizes=[1] + +# A non-equality correlation stays a residual join filter. The null-aware hash +# join applies it when it decides whether a NULL makes the mark UNKNOWN, so one +# join is still enough. +query TT +EXPLAIN SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1; +---- +logical_plan +01)Projection: n1.id, __correlated_sq_1.mark AS m +02)--LeftMark Join: n1.id = __correlated_sq_1.id Filter: __correlated_sq_1.z < n1.z null_aware +03)----TableScan: n1 projection=[id, z] +04)----SubqueryAlias: __correlated_sq_1 +05)------TableScan: n2 projection=[id, z] +physical_plan +01)ProjectionExec: expr=[id@0 as id, mark@1 as m] +02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], filter=z@1 < z@0, projection=[id@0, mark@2], null_aware +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)----DataSourceExec: partitions=1, partition_sizes=[1] + +query IB rowsort +SELECT id, id IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) AS m FROM n1; +---- +1 true +2 false +4 false +NULL NULL + +query IBBB rowsort +SELECT + id, + id IN (SELECT id FROM n3) AS m3, + EXISTS (SELECT 1 FROM n2 WHERE n2.id = n1.id) AS e2, + id NOT IN (SELECT id FROM n3 WHERE n3.id > 1) AS nn3 +FROM n1; +---- +1 true true true +2 true false false +4 false false true +NULL NULL false NULL + +query IB rowsort +SELECT id, id IN (SELECT id FROM n2) AS m FROM n1; +---- +1 true +2 NULL +4 NULL +NULL NULL + +query IB rowsort +SELECT id, id NOT IN (SELECT id FROM n2) AS m FROM n1; +---- +1 false +2 NULL +4 NULL +NULL NULL + +query IB rowsort +SELECT id, id IN (SELECT id FROM n3) AS m FROM n1; +---- +1 true +2 true +4 false +NULL NULL + +query IT rowsort +SELECT id, CASE WHEN NOT (id IN (SELECT id FROM n2)) THEN 'a' ELSE 'b' END AS c FROM n1; +---- +1 b +2 b +4 b +NULL b + +query IB rowsort +SELECT z, sum(id) IN (SELECT id FROM n3) AS m FROM n1 GROUP BY z; +---- +10 true +20 true +30 NULL +40 false + +query IB rowsort +SELECT id, COALESCE((id IN (SELECT id FROM n3))::boolean, false) AS matched FROM n1; +---- +1 true +2 true +4 false +NULL false + +query IT rowsort +SELECT id, CASE WHEN id NOT IN (SELECT n2.id FROM n2 WHERE n2.z < n1.z) THEN 'a' ELSE 'b' END AS c FROM n1; +---- +1 b +2 a +4 a +NULL b + +statement ok +DROP TABLE n1; + +statement ok +DROP TABLE n2; + +statement ok +DROP TABLE n3; + +# Nullable key expressions over non-nullable columns. +# +# `nn.id` and `nn.s` are not nullable, but a key expression over them can still +# be NULL. `NULLIF(id, 1)` is NULL for `id = 1`, and `TRY_CAST(s AS INT)` is +# NULL when the text is not a number. The join must be null-aware for these +# keys, so the mark is NULL and `IN` gives UNKNOWN. + +statement ok +CREATE TABLE nn(id INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, '1'), (2, 'x'), (4, '4'); + +statement ok +CREATE TABLE r3(id INT NOT NULL) AS VALUES (1), (2); + +statement ok +CREATE TABLE r3n(id INT NOT NULL) AS VALUES (1), (2), (5); + +# The nullable key expression keeps the plan at one null-aware mark join. +query TT +EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn; +---- +logical_plan +01)Projection: nn.id, __correlated_sq_1.mark AS m +02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) = __correlated_sq_1.r3.id null_aware +03)----TableScan: nn projection=[id] +04)----SubqueryAlias: __correlated_sq_1 +05)------Projection: CAST(r3.id AS Int64) +06)--------TableScan: r3 projection=[id] +physical_plan +01)ProjectionExec: expr=[id@0 as id, mark@1 as m] +02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(nullif(nn.id,Int64(1))@1, r3.id@0)], projection=[id@0, mark@2], null_aware +03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as nullif(nn.id,Int64(1))] +04)------DataSourceExec: partitions=1, partition_sizes=[1] +05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r3.id] +06)------DataSourceExec: partitions=1, partition_sizes=[1] + +# `NULLIF(id, 1)` is NULL for `id = 1`, and `r3` has no NULL, so the answer is +# UNKNOWN for that row. +query IB rowsort +SELECT id, NULLIF(id, 1) IN (SELECT id FROM r3) AS m FROM nn; +---- +1 NULL +2 true +4 false + +# `TRY_CAST('x' AS INT)` is NULL, so the answer is UNKNOWN for that row. +query IB rowsort +SELECT id, TRY_CAST(s AS INT) IN (SELECT id FROM r3) AS m FROM nn; +---- +1 true +2 NULL +4 false + +# The same on the subquery side: the output of the subquery holds a NULL, so a +# row with no match is UNKNOWN. +query IB rowsort +SELECT id, id IN (SELECT NULLIF(id, 5) FROM r3n) AS m FROM nn; +---- +1 true +2 true +4 NULL + +# `NOT IN` reads the same mark column, negated. +query IB rowsort +SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r3) AS m FROM nn; +---- +1 NULL +2 false +4 true + +# The `NOT IN` filter path builds a LeftAnti join and reads the key nullability +# the same way. UNKNOWN does not pass a filter, so `id = 1` drops out. +query I rowsort +SELECT id FROM nn WHERE NULLIF(id, 1) NOT IN (SELECT id FROM r3); +---- +4 + +# An empty subquery gives `false` also for a NULL key, and the plan stays one +# null-aware mark join. +statement ok +CREATE TABLE r_empty(id INT NOT NULL) AS SELECT * FROM r3 WHERE false; + +query TT +EXPLAIN SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn; +---- +logical_plan +01)Projection: nn.id, __correlated_sq_1.mark AS m +02)--LeftMark Join: nullif(CAST(nn.id AS Int64), Int64(1)) = __correlated_sq_1.r_empty.id null_aware +03)----TableScan: nn projection=[id] +04)----SubqueryAlias: __correlated_sq_1 +05)------Projection: CAST(r_empty.id AS Int64) +06)--------TableScan: r_empty projection=[id] +physical_plan +01)ProjectionExec: expr=[id@0 as id, mark@1 as m] +02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(nullif(nn.id,Int64(1))@1, r_empty.id@0)], projection=[id@0, mark@2], null_aware +03)----ProjectionExec: expr=[id@0 as id, nullif(CAST(id@0 AS Int64), 1) as nullif(nn.id,Int64(1))] +04)------DataSourceExec: partitions=1, partition_sizes=[1] +05)----ProjectionExec: expr=[CAST(id@0 AS Int64) as r_empty.id] +06)------DataSourceExec: partitions=1, partition_sizes=[0] + +query IB rowsort +SELECT id, NULLIF(id, 1) IN (SELECT id FROM r_empty) AS m FROM nn; +---- +1 false +2 false +4 false + +query IB rowsort +SELECT id, NULLIF(id, 1) NOT IN (SELECT id FROM r_empty) AS m FROM nn; +---- +1 true +2 true +4 true + +statement ok +DROP TABLE r_empty; + +# A correlated `NOT IN` filter builds a `LeftAnti` join with two keys: the +# value and the correlation. A function key over non-nullable columns that +# cannot be NULL must not make this join null-aware. +statement ok +CREATE TABLE t1(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'a'), (2, 'b'); + +statement ok +CREATE TABLE t2(k INT NOT NULL, s VARCHAR NOT NULL) AS VALUES (1, 'B'), (2, 'B'); + +query IT rowsort +SELECT * FROM t1 WHERE upper(t1.s) NOT IN (SELECT t2.s FROM t2 WHERE t2.k = t1.k); +---- +1 a + +# `NULLIF(k, 1)` is NULL for `k = 1`, and that group of `t2` is not empty, so +# the correct result has no row for `k = 1`. The join has two keys, the value +# first and the correlation second, and is null-aware because the value +# expression can be NULL. `main` tests the columns of the key instead, which +# are not nullable, and keeps the `k = 1` row. That is +# https://github.com/apache/datafusion/issues/25347. +query I rowsort +SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM t2 WHERE t2.k = t1.k); +---- +2 + +query TT +EXPLAIN SELECT k FROM t1 WHERE NULLIF(t1.k, 1) NOT IN (SELECT t2.k + 10 FROM t2 WHERE t2.k = t1.k); +---- +logical_plan +01)LeftAnti Join: nullif(CAST(t1.k AS Int64), Int64(1)) = __correlated_sq_1.t2.k + Int64(10), t1.k = __correlated_sq_1.k null_aware +02)--TableScan: t1 projection=[k] +03)--SubqueryAlias: __correlated_sq_1 +04)----Projection: CAST(t2.k AS Int64) + Int64(10), t2.k +05)------TableScan: t2 projection=[k] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(nullif(t1.k,Int64(1))@1, t2.k + Int64(10)@0), (k@0, k@1)], projection=[k@0], null_aware +02)--ProjectionExec: expr=[k@0 as k, nullif(CAST(k@0 AS Int64), 1) as nullif(t1.k,Int64(1))] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)--ProjectionExec: expr=[CAST(k@0 AS Int64) + 10 as t2.k + Int64(10), k@0 as k] +05)----DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +DROP TABLE t1; + +statement ok +DROP TABLE t2; + +# A non-equality correlation leaves one key and a residual join filter. The +# null-aware `LeftAnti` hash join applies the residual filter when it looks +# for UNKNOWN rows. For `k = 1` the key is NULL and the correlated subquery +# result is empty, and `NULL NOT IN ()` is TRUE. Both rows are +# correct. +statement ok +CREATE TABLE ra(k INT NOT NULL, z INT NOT NULL) AS VALUES (1, 10), (2, 20); + +statement ok +CREATE TABLE rb(k INT NOT NULL, z INT NOT NULL) AS VALUES (5, 50); + +query I rowsort +SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z < ra.z); +---- +1 +2 + +# The same shape where the correlated subquery result is not empty for the NULL +# key: `NULL NOT IN ({5})` is UNKNOWN, so `2` is the only row. `main` builds a +# plain anti join here and also keeps `1`. +query I rowsort +SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z > ra.z); +---- +2 + +query TT +EXPLAIN SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z > ra.z); +---- +logical_plan +01)Projection: ra.k +02)--LeftAnti Join: nullif(CAST(ra.k AS Int64), Int64(1)) = __correlated_sq_1.rb.k Filter: __correlated_sq_1.z > ra.z null_aware +03)----TableScan: ra projection=[k, z] +04)----SubqueryAlias: __correlated_sq_1 +05)------Projection: CAST(rb.k AS Int64), rb.z +06)--------TableScan: rb projection=[k, z] +physical_plan +01)HashJoinExec: mode=CollectLeft, join_type=LeftAnti, on=[(nullif(ra.k,Int64(1))@2, rb.k@0)], filter=z@1 > z@0, projection=[k@0], null_aware +02)--ProjectionExec: expr=[k@0 as k, z@1 as z, nullif(CAST(k@0 AS Int64), 1) as nullif(ra.k,Int64(1))] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)--ProjectionExec: expr=[CAST(k@0 AS Int64) as rb.k, z@1 as z] +05)----DataSourceExec: partitions=1, partition_sizes=[1] + +statement ok +DROP TABLE ra; + +statement ok +DROP TABLE rb; + +statement ok +DROP TABLE nn; + +statement ok +DROP TABLE r3; + +statement ok +DROP TABLE r3n; + +# A correlation that repeats the `IN` predicate. +# +# `x IN (SELECT y FROM .. WHERE y = x)` writes the `IN` equality a second time. +# The decorrelation drops the duplicate and adds the same equality back as the +# join filter, so the scope of the subquery is gone by the time the join is +# built. The result of this shape is never UNKNOWN: every row the subquery +# keeps for an outer row satisfies `y = x`, so `y` is not NULL there and the +# subquery result is either empty or `{x}`. The join must therefore not be +# null-aware. All results below agree with DuckDB 1.5.2. + +statement ok +CREATE TABLE co(id INT, k INT) AS VALUES (1,1),(2,1),(NULL,1),(2,2),(NULL,3),(5,NULL),(9,9); + +statement ok +CREATE TABLE ci(id INT, k INT) AS VALUES (1,1),(NULL,2),(7,1),(3,NULL); + +# The mark join carries the whole result and is not null-aware. +query TT +EXPLAIN SELECT co.id, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k) AS m FROM co; +---- +logical_plan +01)Projection: co.id, __correlated_sq_1.mark AS m +02)--LeftMark Join: co.k = __correlated_sq_1.k +03)----TableScan: co projection=[id, k] +04)----SubqueryAlias: __correlated_sq_1 +05)------TableScan: ci projection=[k] +physical_plan +01)ProjectionExec: expr=[id@0 as id, mark@1 as m] +02)--HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(k@0, k@1)], projection=[id@0, mark@2] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)----DataSourceExec: partitions=1, partition_sizes=[1] + +query IIB +SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k) AS m FROM co ORDER BY k, id; +---- +1 1 true +2 1 true +NULL 1 true +2 2 true +NULL 3 false +9 9 false +5 NULL false + +# The `IN` value carries a second correlation, which stays a join key. +query IIB +SELECT co.id, co.k, co.id IN (SELECT ci.id FROM ci WHERE ci.id = co.id AND ci.k = co.k) AS m FROM co ORDER BY k, id; +---- +1 1 true +2 1 false +NULL 1 false +2 2 false +NULL 3 false +9 9 false +5 NULL false + +# The same shape in a `WHERE` clause builds a `LeftAnti` join, which must not be +# null-aware either. `main` gives no row here, which is +# https://github.com/apache/datafusion/issues/25480. +query II +SELECT co.id, co.k FROM co WHERE co.k NOT IN (SELECT ci.k FROM ci WHERE ci.k = co.k) ORDER BY k, id; +---- +NULL 3 +9 9 +5 NULL + +# The same shape with an expression as the value. The subquery output is then a +# cast, and the join keys refer to it by its column name. The correlation still +# names the expression as the query writes it, and the join is not null-aware. +query IIB +SELECT co.id, co.k, (co.k + 0) IN (SELECT ci.k FROM ci WHERE ci.k = co.k + 0) AS m FROM co ORDER BY k, id; +---- +1 1 true +2 1 true +NULL 1 true +2 2 true +NULL 3 false +9 9 false +5 NULL false + +# A grouping set above the correlated filter puts a NULL back. The grand-total +# row of `ROLLUP` exists for every outer row, so a miss is UNKNOWN here, not +# FALSE. The filter repeats the `IN` predicate, so the pull up drops it and the +# aggregate needs no correlated column. Pinned to the current wrong result: the +# three rows for `k = 3`, `k = 9` and `k = NULL` should be NULL. +query IIB +SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLLUP(ci.k)) AS m FROM co ORDER BY k, id; +---- +1 1 true +2 1 true +NULL 1 true +2 2 true +NULL 3 false +9 9 false +5 NULL false + +statement ok +DROP TABLE co; + +statement ok +DROP TABLE ci; + +# A subquery column with the same qualified name as the outer `IN` value. +# +# Inside the subquery, `a` is `sh_t`, so `a.x = b.k` bounds `sh_t.x`, not the +# outer `a.x`. It does not keep a NULL outer value out of the scope, and the +# join must stay null-aware. The subquery is `{7}` for every outer row, so the +# NULL row is UNKNOWN. Results verified with DuckDB 1.5.2. +statement ok +CREATE TABLE sh_o(x INT, k INT) AS VALUES (NULL, 1), (7, 1), (8, 1); + +statement ok +CREATE TABLE sh_t(x INT, y INT NOT NULL) AS VALUES (1, 7); + +query IB +SELECT a.x, a.x IN (SELECT a.y FROM sh_t AS a WHERE a.x = b.k) AS r +FROM sh_o AS a, (SELECT 1 AS k) AS b ORDER BY 1; +---- +7 true +8 false +NULL NULL + +# The `NOT IN` filter drops the UNKNOWN row. +query I +SELECT a.x FROM sh_o AS a, (SELECT 1 AS k) AS b +WHERE a.x NOT IN (SELECT a.y FROM sh_t AS a WHERE a.x = b.k); +---- +8 + +statement ok +DROP TABLE sh_o; + +statement ok +DROP TABLE sh_t; From 6298bd4dc936f17301651b242c32bd55cc026519 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 27 Sep 2026 23:12:22 -0500 Subject: [PATCH 2/3] fix: forget the NULL-rejecting correlations above a node that can add a NULL A correlated filter that rejects NULL on a key only bounds the subquery result if nothing above it can put a NULL back. An outer join, a union or a grouping set can. With `ROLLUP`, the grand-total row is a NULL for every outer row, so a miss is UNKNOWN, but the join was planned without null-aware semantics and gave FALSE: SELECT k, k IN (SELECT i.k FROM i WHERE i.k = o.k GROUP BY ROLLUP(i.k)) FROM o; The pull up now clears the recorded filters when it passes such a node. Co-Authored-By: Claude Opus 5.5 --- datafusion/optimizer/src/decorrelate.rs | 32 +++++++++++++++- .../src/decorrelate_predicate_subquery.rs | 21 ++++++++++ .../sqllogictest/test_files/subquery.slt | 8 ++-- .../test_files/subquery_projection.slt | 38 ++++++++++++++++--- 4 files changed, 89 insertions(+), 10 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 3f2a78c456aa4..50c34d439f6fe 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -30,7 +30,7 @@ use datafusion_common::{ Column, DFSchemaRef, HashMap, Result, ScalarValue, assert_or_internal_err, plan_err, }; use datafusion_expr::expr::{Alias, GroupingSet}; -use datafusion_expr::logical_plan::Join; +use datafusion_expr::logical_plan::{Join, JoinType}; use datafusion_expr::simplify::SimplifyContext; use datafusion_expr::utils::{ collect_subquery_cols, conjunction, find_join_exprs, split_conjunction, @@ -87,6 +87,9 @@ pub struct PullUpCorrelatedExpr { /// uses it to tell if a join key can be NULL inside the scope of an outer /// row: `x IN (SELECT y FROM .. WHERE y = x)` keeps every NULL `y` out of /// its result, although `join_filters` no longer says so. + /// + /// The list is cleared when the pull up passes a node that can put a NULL + /// back into such a column: an outer join, a union or a grouping set. pub correlated_filters: Vec, } @@ -199,6 +202,12 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { } fn f_up(&mut self, plan: LogicalPlan) -> Result> { + // A node that null-extends or regroups rows can put a NULL back into + // a column that a correlated filter below it rejected, so the filters + // recorded so far no longer bound what the subquery returns. + if !self.correlated_filters.is_empty() && may_reintroduce_nulls(&plan) { + self.correlated_filters.clear(); + } let subquery_schema = plan.schema(); match &plan { LogicalPlan::Filter(plan_filter) => { @@ -566,6 +575,27 @@ impl PullUpCorrelatedExpr { } } +/// Whether `plan` can output a NULL in a column that a filter below it keeps +/// free of NULLs: an outer join fills the columns of its unmatched side with +/// NULL, a union adds the rows of its other inputs, and a grouping set fills +/// the columns it leaves out with NULL, such as the grand total row of `ROLLUP`. +fn may_reintroduce_nulls(plan: &LogicalPlan) -> bool { + match plan { + LogicalPlan::Join(join) => { + matches!( + join.join_type, + JoinType::Left | JoinType::Right | JoinType::Full + ) + } + LogicalPlan::Union(_) => true, + LogicalPlan::Aggregate(aggregate) => aggregate + .group_expr + .iter() + .any(|expr| matches!(expr, Expr::GroupingSet(_))), + _ => false, + } +} + /// Whether every input of `join` that holds outer references is a side whose /// rows the join preserves, so a correlated filter below it can be pulled above /// the join without changing the result. diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 8bee46a197543..e2f4aa40b519e 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -2105,6 +2105,27 @@ mod tests { Ok(()) } + /// A grouping set above the correlation puts a NULL back into the key: the + /// grand-total row of `ROLLUP` is a NULL for every outer row. The + /// correlation then no longer bounds the subquery result, and the mark + /// join stays null-aware. + #[test] + fn mark_join_for_in_predicate_correlation_below_rollup_is_null_aware() -> Result<()> { + let subquery = correlated_grouping_set_subquery("sq", rollup(vec![col("sq.c")]))?; + let plan = LogicalPlanBuilder::from(test_table_scan()?) + .project(vec![in_subquery(col("test.c"), subquery).alias("m")])? + .build()?; + + let optimized = optimize_with_decorrelate(plan)?; + assert!( + has_null_aware_left_mark_join(&optimized), + "{}", + optimized.display_indent_schema() + ); + + Ok(()) + } + /// The same for the `LeftAnti` join that a `NOT IN` filter builds. #[test] fn anti_join_for_in_predicate_correlation_is_not_null_aware() -> Result<()> { diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 170c6232af50b..7c5c0b76bc0e2 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -2793,16 +2793,16 @@ SELECT gs_outer.k, EXISTS (SELECT 1 FROM gs_inner WHERE gs_inner.k = gs_outer.k # The `IN` form of the same subquery decorrelates: the correlation repeats the # `IN` predicate, so the pull up drops it and adds no column to the grouping -# set. The grand-total row is a NULL for every outer row, so a miss is UNKNOWN. -# Pinned to the current wrong result: `k = 4` and `k = NULL` should be NULL. +# set. The grand-total row is a NULL for every outer row, so a miss is UNKNOWN, +# and the null-aware mark join gives NULL for it. query IB SELECT gs_outer.k, gs_outer.k IN (SELECT gs_inner.k FROM gs_inner WHERE gs_inner.k = gs_outer.k GROUP BY ROLLUP(gs_inner.k)) FROM gs_outer ORDER BY gs_outer.k; ---- 1 true 2 true -4 false +4 NULL 5 true -NULL false +NULL NULL # `NOT EXISTS` too. statement error DataFusion error: This feature is not implemented: Physical plan does not support logical expression Exists diff --git a/datafusion/sqllogictest/test_files/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index ad30a307aab0a..f86de505fe342 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -541,8 +541,9 @@ NULL 3 false # A grouping set above the correlated filter puts a NULL back. The grand-total # row of `ROLLUP` exists for every outer row, so a miss is UNKNOWN here, not # FALSE. The filter repeats the `IN` predicate, so the pull up drops it and the -# aggregate needs no correlated column. Pinned to the current wrong result: the -# three rows for `k = 3`, `k = 9` and `k = NULL` should be NULL. +# aggregate needs no correlated column. The filter below the aggregate then +# does not bound the subquery result, and the join stays null-aware. Expected +# results verified with DuckDB. query IIB SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLLUP(ci.k)) AS m FROM co ORDER BY k, id; ---- @@ -550,9 +551,36 @@ SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLL 2 1 true NULL 1 true 2 2 true -NULL 3 false -9 9 false -5 NULL false +NULL 3 NULL +9 9 NULL +5 NULL NULL + +query TT +EXPLAIN SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLLUP(ci.k)) AS m FROM co; +---- +logical_plan +01)Projection: co.id, co.k, __correlated_sq_1.mark AS m +02)--LeftMark Join: co.k = __correlated_sq_1.k null_aware +03)----TableScan: co projection=[id, k] +04)----SubqueryAlias: __correlated_sq_1 +05)------Projection: ci.k +06)--------Aggregate: groupBy=[[ROLLUP (ci.k)]], aggr=[[]] +07)----------TableScan: ci projection=[k] +physical_plan +01)ProjectionExec: expr=[id@0 as id, k@1 as k, mark@2 as m] +02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(k@1, k@0)], null_aware +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)----ProjectionExec: expr=[k@0 as k] +05)------AggregateExec: mode=FinalPartitioned, gby=[k@0 as k, __grouping_id@1 as __grouping_id], aggr=[] +06)--------RepartitionExec: partitioning=Hash([k@0, __grouping_id@1], 4), input_partitions=1 +07)----------AggregateExec: mode=Partial, gby=[(NULL as k), (k@0 as k)], aggr=[] +08)------------DataSourceExec: partitions=1, partition_sizes=[1] + +# The `NOT IN` filter form of the same subquery. Every row is either FALSE or +# UNKNOWN, so no row is kept. +query II +SELECT co.id, co.k FROM co WHERE co.k NOT IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLLUP(ci.k)) ORDER BY k, id; +---- statement ok DROP TABLE co; From 210d4e51978b827f1ac80083317037097fffc294 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 27 Sep 2026 23:26:20 -0500 Subject: [PATCH 3/3] docs: q07 of projection_subquery is no longer the nested-loop control The correlation on `<` now stays a residual filter of the one null-aware mark join, so the query no longer keeps the nested-loop plan. Co-Authored-By: Claude Opus 5.5 --- benchmarks/sql_benchmarks/README.md | 2 +- .../projection_subquery/projection_subquery.suite | 2 +- .../sql_benchmarks/projection_subquery/queries/q07.sql | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/benchmarks/sql_benchmarks/README.md b/benchmarks/sql_benchmarks/README.md index 8047c030927cc..8b68311531c9c 100644 --- a/benchmarks/sql_benchmarks/README.md +++ b/benchmarks/sql_benchmarks/README.md @@ -48,7 +48,7 @@ in the community: | `predicate_eval` | Conjunctive (AND) filter-evaluation micro-benchmarks; each subgroup is a different predicate pattern, to test how an adaptive predicate-ordering system behaves across them ([#11262](https://github.com/apache/datafusion/issues/11262)). Subgroups (`--subgroup`): `costsel`, `cost`, `selectivity`, `cardinality`, `width`, `scale`, `neutral`, `correlation`, `drift`, `nulls`. The suite sets no engine config of its own, so by default it measures DataFusion's built-in left-deep `AND` short-circuit; point it at a system under test by exporting that system's own DataFusion setting (the harness builds its `SessionConfig` with `SessionConfig::from_env`). Every query is a `count(*)`, and the counts are checked in under `predicate_eval/results/`, so `--result-mode validate` also checks that a reordering under test still returns the same rows; the checked-in counts were persisted at the suite defaults (`PRED_ROWS=1000000`, `PRED_FILL=30`), so validation assumes those (the `scale` and `width` subgroups pin their own values per query and validate at any setting). | | `parquet_row_filter_skip` | Micro-benchmark for the per-row-group fully-matched RowFilter skip on Parquet scans ([#23696](https://github.com/apache/datafusion/issues/23696)). Subgroups (`--subgroup`): `skip` (clustered key, most row groups fully matched by statistics so the per-row filter is skipped), `control` (scrambled key, no row group is ever fully matched). Size the data with `PRED_ROWS` and the row-group size with `RG_SIZE`. | | `spill_views` | Sort and `GROUP BY` queries that spill `StringView` and `BinaryView` columns read from Parquet ([#23564](https://github.com/apache/datafusion/issues/23564)). Each query sets its own memory limit and `target_partitions = 4`, so it spills in any environment. Subgroups (`--subgroup`): `repeated` (sort with a low-cardinality payload that shares one dictionary buffer, where spill compaction can remove data), `distinct` (sort and `GROUP BY` on all-distinct strings, where compaction cannot remove data). Set the memory limit of each subgroup with `SPILL_VIEWS_LIMIT_REPEATED` (default `40M`) and `SPILL_VIEWS_LIMIT_DISTINCT` (default `96M`), for example to let both sides of a comparison finish when one of them runs out of memory at the default. The queries must still spill, so check a raised limit with `EXPLAIN ANALYZE` and keep it well below the working set: `q05` stops spilling at about `192M`. Give a whole number of K, M or G units, because the run asserts the limit that the memory pool reports back. | -| `projection_subquery` | `IN`, `NOT IN` and `EXISTS` subqueries that sit in the `SELECT` list instead of a filter ([#25341](https://github.com/apache/datafusion/issues/25341)). Every query projects the boolean subquery result and aggregates it, so the measured cost is the decorrelation plan and not the size of the output. `q07` correlates on `<` instead of `=`, so it keeps the nested-loop plan and is the control that must not change. Size the two inline tables with `PSQ_ROWS` (default `30000`). The counts are checked in under `projection_subquery/results/`, so `--result-mode validate` also proves that a change to the decorrelation still returns the same rows; the checked-in counts were persisted at the default `PSQ_ROWS`, so validation assumes that value. | +| `projection_subquery` | `IN`, `NOT IN` and `EXISTS` subqueries that sit in the `SELECT` list instead of a filter ([#25341](https://github.com/apache/datafusion/issues/25341)). Every query projects the boolean subquery result and aggregates it, so the measured cost is the decorrelation plan and not the size of the output. `q07` correlates on `<` instead of `=`, so the correlation stays a residual filter of the mark join. Size the two inline tables with `PSQ_ROWS` (default `30000`). The counts are checked in under `projection_subquery/results/`, so `--result-mode validate` also proves that a change to the decorrelation still returns the same rows; the checked-in counts were persisted at the default `PSQ_ROWS`, so validation assumes that value. | # Running Benchmarks diff --git a/benchmarks/sql_benchmarks/projection_subquery/projection_subquery.suite b/benchmarks/sql_benchmarks/projection_subquery/projection_subquery.suite index f5f419b50aa1e..e384036197d6d 100644 --- a/benchmarks/sql_benchmarks/projection_subquery/projection_subquery.suite +++ b/benchmarks/sql_benchmarks/projection_subquery/projection_subquery.suite @@ -1,4 +1,4 @@ -description = "IN / NOT IN / EXISTS subqueries that appear in the SELECT list instead of a filter (apache/datafusion#25341). Each query projects the boolean result of a subquery and aggregates it, so the measured work is the decorrelation plan and not the output size. The decorrelation turns one projected IN into mark joins; when the join predicate is not hashable, or the correlation is not an equality, the plan keeps a nested-loop mark join and the cost grows with outer rows times inner rows. Query 07 correlates on `<` and stays on that path, so it is the control that must not change. The two tables are built inline by the load SQL and are sized with PSQ_ROWS." +description = "IN / NOT IN / EXISTS subqueries that appear in the SELECT list instead of a filter (apache/datafusion#25341). Each query projects the boolean result of a subquery and aggregates it, so the measured work is the decorrelation plan and not the output size. The decorrelation turns one projected IN into mark joins. Query 07 correlates on `<`, so the correlation stays a residual filter of the mark join. The two tables are built inline by the load SQL and are sized with PSQ_ROWS." query_pattern = "q{QUERY_ID_PADDED}.benchmark" diff --git a/benchmarks/sql_benchmarks/projection_subquery/queries/q07.sql b/benchmarks/sql_benchmarks/projection_subquery/queries/q07.sql index 54299f0cbdaf4..4069da66a90a2 100644 --- a/benchmarks/sql_benchmarks/projection_subquery/queries/q07.sql +++ b/benchmarks/sql_benchmarks/projection_subquery/queries/q07.sql @@ -1,6 +1,6 @@ --- Shape: IN correlated on `<` instead of `=`. There is no equality to hash on, --- so this query keeps the nested-loop mark joins. It is the control: its cost --- must stay the same when the hashable cases get faster. +-- Shape: IN correlated on `<` instead of `=`. The `IN` equality is the hash key +-- and the `<` correlation stays a residual filter of the one null-aware mark +-- join, which applies it when it decides whether a NULL makes the mark UNKNOWN. SELECT count(*) FILTER (WHERE m) AS true_count, count(*) FILTER (WHERE m IS NULL) AS null_count