diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 0ad8b44c40def..8f34fb267b9e8 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -29,7 +29,7 @@ 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, + Column, DFSchema, ExprSchema, NullEquality, Result, ScalarValue, assert_or_internal_err, plan_err, }; use datafusion_expr::expr::{Exists, InSubquery}; @@ -37,8 +37,8 @@ use datafusion_expr::expr_rewriter::create_col_from_scalar_expr; 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; @@ -222,6 +222,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. A residual non-equality filter stays on the join in the +/// other case. The mark column then 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, @@ -233,12 +248,24 @@ 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)? + let Some(MarkJoin { + plan: matched_plan, + mark: matched, + three_valued_exact, + }) = mark_join_detailed(left, subquery, Some(&in_predicate), false, alias)? 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())? @@ -365,13 +392,14 @@ 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, - ) + )? + .map(|join| join.plan)) } /// This is used to handle the case when the subquery is embedded in a more complex boolean @@ -396,30 +424,94 @@ fn mark_join( negated: bool, alias_generator: &Arc, ) -> Result> { + Ok( + mark_join_detailed(left, subquery, in_predicate_opt, negated, alias_generator)? + .map(|mark_join| (mark_join.plan, mark_join.mark)), + ) +} + +/// 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. + /// + /// This holds when the join filter is hashable only, that is when it is a + /// conjunction of equalities that the hash join can use as join keys. The + /// join is then null-aware if the keys may be NULL, which marks the + /// UNKNOWN rows NULL, and a plain mark is exact if no key can be NULL. + /// + /// A residual non-equality filter breaks this, because hash join execution + /// cannot mark UNKNOWN candidates for a residual predicate. + three_valued_exact: bool, +} + +/// Same as [`mark_join`], but also reports what the mark column can promise. +fn mark_join_detailed( + left: &LogicalPlan, + subquery: &LogicalPlan, + in_predicate_opt: Option<&Expr>, + negated: bool, + alias_generator: &Arc, +) -> Result> { let alias = alias_generator.next("__correlated_sq"); let exists_col = Expr::Column(Column::new(Some(alias.clone()), "mark")); let exists_expr = if negated { !exists_col } else { exists_col }; Ok( - build_join(left, subquery, in_predicate_opt, JoinType::LeftMark, alias)? - .map(|plan| (plan, exists_expr)), + build_join(left, subquery, in_predicate_opt, JoinType::LeftMark, alias)?.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 +/// Check if the join keys can be NULL. +/// +/// A null-aware join is more expensive than the plain join. It is necessary +/// only when a join key can be NULL, because only then can the join find an +/// UNKNOWN result. The caller gives the join keys that +/// [`split_eq_and_noneq_join_predicate`] found, plus the residual filter that +/// the split could not turn into keys. /// -/// 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. +/// The keys are full expressions, not only columns. An expression can be NULL +/// although all of its columns are not nullable. Examples are `NULLIF(id, 1)`, +/// `TRY_CAST(s AS INT)` and a `CASE` expression with no `ELSE` branch. Thus +/// this function asks each key expression for its nullability against the +/// schema of its own side. A key that a cast wraps, such as +/// `CAST(id AS Int64)`, keeps the nullability of the expression in it. +/// +/// The hash join cannot use the residual filter as keys, so there is no key +/// expression to ask. For the residual this function keeps the older and less +/// exact test: it reports true if the residual refers to any nullable column. +/// The result for a filter with no equality pair is thus never less +/// conservative than before. fn join_keys_may_be_null( - join_filter: &Expr, - left_schema: &DFSchemaRef, - right_schema: &DFSchemaRef, + equijoin_keys: &[(Expr, Expr)], + residual: Option<&Expr>, + left_schema: &DFSchema, + right_schema: &DFSchema, ) -> Result { - // Extract columns from the join filter + for (left_key, right_key) in equijoin_keys { + if left_key.nullable(left_schema)? || right_key.nullable(right_schema)? { + return Ok(true); + } + } + + let Some(residual) = residual else { + return Ok(false); + }; + + // Extract columns from the residual filter let mut columns = std::collections::HashSet::new(); - expr_to_columns(join_filter, &mut columns)?; + expr_to_columns(residual, &mut columns)?; // Check if any column is nullable for col in columns { @@ -440,13 +532,22 @@ fn join_keys_may_be_null( Ok(false) } +/// 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, -) -> 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()); @@ -524,7 +625,15 @@ fn build_join( if let Some((value, right_col, mut value_name)) = in_value_expr && value.column_refs().is_empty() && matches!(join_type, JoinType::LeftAnti | JoinType::LeftMark) - && join_keys_may_be_null(&join_filter, left.schema(), sub_query_alias.schema())? + // The value expression holds no column, so the `IN` equality is not an + // equi-join key. There is thus no key expression to ask, and the column + // test on the whole filter is the only test available here. + && join_keys_may_be_null( + &[], + Some(&join_filter), + left.schema(), + sub_query_alias.schema(), + )? { // The projected column is unqualified, so a left field that already has // this name — however unlikely — would make the reference ambiguous. @@ -584,30 +693,39 @@ fn build_join( sub_query_alias.clone() }; - let mark_filter_is_hashable_only = - if join_type == JoinType::LeftMark && in_predicate_opt.is_some() { - let (_, residual_filter) = split_eq_and_noneq_join_predicate( - join_filter.clone(), - left.schema(), - right_projected.schema(), - )?; - residual_filter.is_none() - } else { - false - }; + let mark_split = if join_type == JoinType::LeftMark && in_predicate_opt.is_some() + { + Some(split_eq_and_noneq_join_predicate( + join_filter.clone(), + left.schema(), + right_projected.schema(), + )?) + } else { + None + }; - // For scalar NOT IN mark joins, propagate null-aware semantics into the - // nullable mark column when the predicate can be implemented by hash keys. - // Non-equality correlated filters stay on the legacy path because hash join - // execution cannot mark UNKNOWN candidates for residual predicates. - let null_aware = join_type == JoinType::LeftMark - && in_predicate_opt.is_some() - && mark_filter_is_hashable_only - && join_keys_may_be_null( - &join_filter, + // Only a filter that the hash join can turn into keys gives an exact + // mark. A residual predicate leaves the UNKNOWN rows unmarked. + let hashable_only_split = mark_split + .as_ref() + .filter(|(_, residual_filter)| residual_filter.is_none()); + let mark_filter_is_hashable_only = hashable_only_split.is_some(); + + // Put null-aware semantics into the nullable mark column when the + // predicate can be implemented by hash keys and a key can be NULL. The + // mark column is then exact under SQL three-valued logic, which lets a + // projected `IN` use this join on its own. Non-equality correlated + // filters stay on the legacy path because hash join execution cannot + // mark UNKNOWN candidates for residual predicates. + let null_aware = match hashable_only_split { + Some((equijoin_keys, residual_filter)) => join_keys_may_be_null( + equijoin_keys, + residual_filter.as_ref(), left.schema(), right_projected.schema(), - )?; + )?, + None => false, + }; let new_plan = LogicalPlanBuilder::from(left.clone()) .join_detailed_with_options( @@ -625,7 +743,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: mark_filter_is_hashable_only, + })); } // Determine if this should be a null-aware anti join @@ -634,11 +755,50 @@ fn build_join( // - NOT EXISTS: Uses two-valued logic, regular anti join is correct // We can distinguish them: NOT IN has in_predicate_opt, NOT EXISTS does not // - // Additionally, if the join keys are non-nullable on both sides, we don't need - // null-aware semantics because NULLs cannot exist in the data. - let null_aware = join_type == JoinType::LeftAnti - && in_predicate_opt.is_some() - && join_keys_may_be_null(&join_filter, left.schema(), sub_query_alias.schema())?; + // Additionally, if no join key can be NULL on either side, we don't need + // null-aware semantics because NULLs cannot exist in the keys. + let null_aware = if join_type == JoinType::LeftAnti && in_predicate_opt.is_some() { + let (equijoin_keys, residual_filter) = split_eq_and_noneq_join_predicate( + join_filter.clone(), + left.schema(), + sub_query_alias.schema(), + )?; + if equijoin_keys.len() > 1 || residual_filter.is_some() { + // Keep the column test on the whole filter for these two shapes. + // + // More than one key: a null-aware `LeftAnti` hash join supports one + // key only. A correlated `NOT IN` has two or more keys (the value + // and the correlation), and a key expression that the column test + // misses would make the join null-aware and fail to plan. + // + // A residual filter: the null-aware `LeftAnti` executor does not + // apply the residual when it decides whether a NULL makes the + // result UNKNOWN (https://github.com/apache/datafusion/issues/25336). + // It would thus drop a row whose correlated subquery result is + // empty, and ` NOT IN ()` is TRUE. The column test + // keeps such a join out of the null-aware path, exactly as on + // `main`. + // + // The column test misses a NULL that only the key expression makes, + // as in `NULLIF(id, 1)`: see + // https://github.com/apache/datafusion/issues/25347. + join_keys_may_be_null( + &[], + Some(&join_filter), + left.schema(), + sub_query_alias.schema(), + )? + } else { + join_keys_may_be_null( + &equijoin_keys, + residual_filter.as_ref(), + left.schema(), + sub_query_alias.schema(), + )? + } + } else { + false + }; // join our sub query into the main plan let new_plan = if null_aware { @@ -662,7 +822,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)] @@ -746,6 +909,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 @@ -1346,23 +1518,136 @@ mod tests { ])? .build()?; + assert_optimized_plan_equal!( + plan, + @r" + 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 keeps the three-join materialization, + /// because the mark column of the join is not exact in that case. + #[test] + fn in_subquery_in_projection_with_residual_filter() -> Result<()> { + let subquery = Arc::new( + LogicalPlanBuilder::from(test_table_scan_with_name("sq")?) + .filter(out_ref_col(DataType::UInt32, "test.a").gt(col("sq.a")))? + .project(vec![col("sq.c")])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(test_table_scan()?) + .project(vec![in_subquery(col("c"), subquery).alias("is_present")])? + .build()?; + 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] + LeftMark Join: Filter: test.a > __correlated_sq_3.a [a:UInt32, b:UInt32, c:UInt32, mark:Boolean;N, mark:Boolean;N, mark:Boolean;N] + LeftMark Join: Filter: test.a > __correlated_sq_2.a [a:UInt32, b:UInt32, c:UInt32, mark:Boolean;N, mark:Boolean;N] + LeftMark Join: Filter: test.c = __correlated_sq_1.c AND test.a > __correlated_sq_1.a [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] + Projection: __correlated_sq_1.c, __correlated_sq_1.a [c:UInt32, a:UInt32] + SubqueryAlias: __correlated_sq_1 [c:UInt32, a:UInt32] + Projection: sq.c, sq.a [c:UInt32, a:UInt32] + TableScan: sq [a:UInt32, b:UInt32, c:UInt32] + Projection: __correlated_sq_2.a [a:UInt32] + SubqueryAlias: __correlated_sq_2 [c:UInt32, a:UInt32] + Filter: sq.c IS NULL [c:UInt32, a:UInt32] + Projection: sq.c, sq.a [c:UInt32, a: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_3.a [a:UInt32] + SubqueryAlias: __correlated_sq_3 [c:UInt32, a:UInt32] + Projection: sq.c, sq.a [c:UInt32, a:UInt32] + TableScan: sq [a:UInt32, b:UInt32, c:UInt32] + " + ) + } + + /// `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] " 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_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index ab7c80d9de165..e071259660179 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -97,3 +97,346 @@ 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, so this query keeps +# the three-join materialization. +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, CASE WHEN __correlated_sq_1.mark THEN Boolean(true) WHEN __correlated_sq_2.mark OR n1.id IS NULL AND __correlated_sq_3.mark THEN Boolean(NULL) ELSE Boolean(false) END AS m +02)--LeftMark Join: Filter: __correlated_sq_3.z < n1.z +03)----LeftMark Join: Filter: __correlated_sq_2.z < n1.z +04)------LeftMark Join: n1.id = __correlated_sq_1.id Filter: __correlated_sq_1.z < n1.z +05)--------TableScan: n1 projection=[id, z] +06)--------SubqueryAlias: __correlated_sq_1 +07)----------TableScan: n2 projection=[id, z] +08)------SubqueryAlias: __correlated_sq_2 +09)--------Projection: n2.z +10)----------Filter: n2.id IS NULL +11)------------TableScan: n2 projection=[id, z] +12)----SubqueryAlias: __correlated_sq_3 +13)------TableScan: n2 projection=[z] +physical_plan +01)ProjectionExec: expr=[id@0 as id, CASE WHEN mark@1 THEN true WHEN mark@2 OR id@0 IS NULL AND mark@3 THEN NULL ELSE false END as m] +02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0, projection=[id@0, mark@2, mark@3, mark@4] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)----NestedLoopJoinExec: join_type=RightMark, filter=z@1 < z@0 +05)------FilterExec: id@0 IS NULL, projection=[z@1] +06)--------DataSourceExec: partitions=1, partition_sizes=[1] +07)------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +08)--------HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], filter=z@1 < z@0 +09)----------DataSourceExec: partitions=1, partition_sizes=[1] +10)----------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 null-aware `LeftAnti` hash join supports one key +# only, so a function key over non-nullable columns 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 two key join is not +# null-aware, so this row is wrong. See +# 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); +---- +1 +2 + +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` executor does not apply the residual when it decides +# whether a NULL makes the result UNKNOWN, so a function key over non-nullable +# columns must not make this join null-aware either. 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. The correct result is `2` only. The plain anti join also gives `1`, which +# is the gap that https://github.com/apache/datafusion/issues/25336 closes. +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 + +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;