diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 0c37f00b64355..3a8fa59770ae5 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -74,6 +74,16 @@ 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. + /// + /// `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 { @@ -95,6 +105,7 @@ impl PullUpCorrelatedExpr { collected_count_expr_map: HashMap::new(), pull_up_having_expr: None, pulled_up_scalar_agg: false, + correlated_filters: Vec::new(), } } @@ -184,6 +195,11 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { .all(|&e| can_pullup_over_aggregation(e)); let (mut join_filters, subquery_filters) = find_join_exprs(subquery_filter_exprs)?; + for expr in &join_filters { + if !self.correlated_filters.contains(expr) { + self.correlated_filters.push(expr.clone()); + } + } if let Some(in_predicate) = &self.in_predicate_opt { // in_predicate may be already included in the join filters, remove it from the join filters first. join_filters = remove_duplicated_filter(join_filters, in_predicate)?; diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 0ad8b44c40def..df200b19e7348 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -29,16 +29,15 @@ 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::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; @@ -129,8 +128,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, + )?; + cur_input = plan; + other_exprs.push(expr); + } + expr => other_exprs.push(expr), + }, } } // The subquery expression is embedded within another expression @@ -180,7 +194,11 @@ fn rewrite_inner_subqueries( subquery: Subquery { subquery, .. }, negated, }) => match mark_join(&cur_input, &subquery, None, negated, alias)? { - Some((plan, exists_expr)) => { + Some(MarkJoin { + plan, + mark: exists_expr, + .. + }) => { cur_input = plan; Ok(Transformed::yes(exists_expr)) } @@ -207,6 +225,7 @@ fn rewrite_inner_subqueries( Ok(Expr::eq(*expr.clone(), output_expr)) })?; mark_join(&cur_input, &subquery, Some(&in_predicate), negated, alias)? + .map(|join| (join.plan, join.mark)) }; match rewritten { Some((plan, exists_expr)) => { @@ -222,6 +241,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, @@ -229,32 +263,67 @@ fn in_subquery_value_mark_join( negated: bool, alias: &Arc, ) -> Result> { + // An outer reference in the value belongs to an enclosing subquery. It + // cannot be resolved in the joins this builds, and `build_join` would read + // it as a constant, so leave the predicate for the enclosing rule. + if expr.contains_outer() { + return Ok(None); + } + let output_expr = subquery .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(left, subquery, Some(&in_predicate), false, alias)? else { return Ok(None); }; - // 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)? - else { - return Ok(None); - }; - let Some((final_plan, subquery_non_empty)) = - mark_join(&null_plan, subquery, None, false, alias)? + // 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 }, + ))); + } + + // The value is not matched here, so the answer is UNKNOWN when the + // subquery gives a NULL, and also when the value is NULL and the subquery + // gives any row at all. One mark join answers both: it keeps a subquery row + // in the scope of the outer row when the subquery value is NULL, or when + // the outer value is NULL and the row is in scope at all. + // + // For an outer value that is not NULL the mark reads "the subquery gives a + // NULL". For an outer value that is NULL it reads "the subquery gives a + // row", which is the weaker fact that this case needs, and which the first + // reading implies. `IS NULL` is two-valued on both sides, so neither test + // adds an UNKNOWN of its own. + let unknown_alias = alias.next("__correlated_sq"); + let subquery_value = Expr::Column(create_col_from_scalar_expr( + &output_expr, + unknown_alias.clone(), + )?); + let Some(MarkJoin { + plan: final_plan, + mark: unknown, + .. + }) = mark_join_with_alias( + &matched_plan, + subquery, + None, + Some(subquery_value.is_null().or(expr.is_null())), + false, + &unknown_alias, + )? else { return Ok(None); }; - let unknown = subquery_has_null.or(expr.is_null().and(subquery_non_empty)); let result = when(matched, lit(true)) .when(unknown, lit(ScalarValue::Boolean(None))) .otherwise(lit(false))?; @@ -365,13 +434,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(), + None, join_type, - subquery_alias, - ) + &subquery_alias, + )? + .map(|join| join.plan)) } /// This is used to handle the case when the subquery is embedded in a more complex boolean @@ -395,58 +466,220 @@ fn mark_join( in_predicate_opt: Option<&Expr>, negated: bool, alias_generator: &Arc, -) -> Result> { +) -> Result> { let alias = alias_generator.next("__correlated_sq"); + mark_join_with_alias(left, subquery, in_predicate_opt, None, negated, &alias) +} - let exists_col = Expr::Column(Column::new(Some(alias.clone()), "mark")); +/// Same as [`mark_join`], but the caller owns the alias, so it can give an +/// `extra_predicate` that names a column of the aliased subquery. +fn mark_join_with_alias( + left: &LogicalPlan, + subquery: &LogicalPlan, + in_predicate_opt: Option<&Expr>, + extra_predicate: Option, + negated: bool, + alias: &str, +) -> Result> { + let exists_col = Expr::Column(Column::new(Some(alias.to_string()), "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)), - ) + Ok(build_join( + left, + subquery, + in_predicate_opt, + extra_predicate, + JoinType::LeftMark, + alias, + )? + .map(|join| MarkJoin { + plan: join.plan, + mark: exists_expr, + three_valued_exact: join.mark_is_three_valued_exact, + })) +} + +/// 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 a key can be NULL in scope, 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, +} + +/// The join keys of the join that replaces an `IN` or `NOT IN` predicate. +struct JoinKeys { + /// The equalities that the hash join can use as keys. The `IN` predicate + /// is the first one when its value holds a column; the others are the + /// correlation. + equijoin_keys: Vec<(Expr, Expr)>, + /// The part of the join filter that the split could not turn into keys. + residual_filter: Option, + /// True if the `IN` value or the subquery column it is compared with can + /// be NULL inside the scope of an outer row, see + /// [`key_may_be_null_in_scope`]. 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. + value_may_be_null: bool, +} + +impl JoinKeys { + /// Splits `join_filter` and asks the two sides of the `IN` predicate for + /// their nullability in scope. `scope_filters` are the correlated + /// conjuncts that the subquery applies to its own rows, see + /// [`PullUpCorrelatedExpr::correlated_filters`]. + fn new( + in_value: &InValue, + join_filter: &Expr, + left_schema: &DFSchema, + right_schema: &DFSchema, + scope_filters: &[Expr], + ) -> Result { + let (equijoin_keys, residual_filter) = split_eq_and_noneq_join_predicate( + join_filter.clone(), + left_schema, + right_schema, + )?; + Ok(Self { + equijoin_keys, + residual_filter, + value_may_be_null: in_value.may_be_null_in_scope( + left_schema, + right_schema, + scope_filters, + )?, + }) + } +} + +/// 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. + 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, + scope_filters, + ) || key_may_be_null_in_scope( + self.subquery_column.nullable(right_schema)?, + &self.output_expr, + scope_filters, + )) + } } -/// Check if join keys in the join filter may contain NULL values +/// Can this join key be NULL for a row inside the scope of an outer row? /// -/// 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( - join_filter: &Expr, - left_schema: &DFSchemaRef, - right_schema: &DFSchemaRef, -) -> 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); - } - // Check in right schema - if let Ok(field) = right_schema.field_from_column(&col) - && field.as_ref().is_nullable() - { - return Ok(true); +/// `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. +fn key_may_be_null_in_scope(nullable: bool, key: &Expr, scope_filters: &[Expr]) -> bool { + if !nullable { + return false; + } + let key = strip_casts(key); + !scope_filters + .iter() + .any(|filter| filter_rejects_null(filter, key)) +} + +/// Is `filter` never TRUE when `key` is NULL? +fn filter_rejects_null(filter: &Expr, key: &Expr) -> bool { + match filter { + Expr::BinaryExpr(BinaryExpr { left, op, right }) => { + matches!( + op, + Operator::Eq + | Operator::NotEq + | Operator::Lt + | Operator::LtEq + | Operator::Gt + | Operator::GtEq + ) && (strip_casts(left) == key || strip_casts(right) == key) } + Expr::IsNotNull(expr) => strip_casts(expr) == key, + _ => 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 +} + +/// 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>, + extra_predicate: Option, join_type: JoinType, - alias: String, -) -> Result> { + alias: &str, +) -> Result> { let mut pull_up = PullUpCorrelatedExpr::new() .with_in_predicate_opt(in_predicate_opt.cloned()) .with_exists_sub_query(in_predicate_opt.is_none()); @@ -468,47 +701,27 @@ 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 whose join filter - // is nothing but that predicate, recorded together with the subquery column - // it is compared against and a name for the column it can be projected as. - // Correlated subqueries are excluded on purpose: their correlation predicate - // is a second join key, and null-aware hash joins accept only a single key. - let mut in_value_expr = None; - - let mut join_filter = match (join_filter_opt, in_predicate_opt.cloned()) { - ( - Some(join_filter), - Some(Expr::BinaryExpr(BinaryExpr { - left, - op: Operator::Eq, - right, - })), - ) => { - let right_col = create_col_from_scalar_expr(&right, alias)?; - let in_predicate = Expr::eq(left.deref().clone(), Expr::Column(right_col)); - in_predicate.and(join_filter) - } - (Some(join_filter), _) => join_filter, - ( - _, - 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(); - in_value_expr = Some((value.clone(), right_col.clone(), value_name)); - - Expr::eq(value, Expr::Column(right_col)) + // 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(), + }) } - (None, None) => lit(true), - _ => return Ok(None), + Some(_) => return Ok(None), + None => None, }; // ` IN/NOT IN ()`: the outer value expression holds no @@ -519,37 +732,117 @@ fn build_join( // very NULLs that make `NOT IN` UNKNOWN, and a join without equi-join keys // is planned as a nested loop join, which has no null-aware implementation. // Projecting the constant as a column of the outer side turns the predicate - // into a real equi-join key so the null-aware hash join handles it. + // into a real equi-join key so the null-aware hash join handles it. A + // correlated subquery gets the same projection: its correlation is then a + // second key, which the null-aware mark join below accepts. let mut projected_left = None; - 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 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() + && matches!(join_type, JoinType::LeftAnti | JoinType::LeftMark) + && in_value.may_be_null_in_scope( + left.schema(), + sub_query_alias.schema(), + &pull_up.correlated_filters, + )? => + { + // 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()?, - ); - // `in_value_expr` is only set when the `IN` equality is the whole join - // filter, so it can simply be rebuilt against the projected column. - join_filter = Expr::eq(Expr::Column(value_col), Expr::Column(right_col)); - } + other => other, + }; + // The columns of the outer plan, which an anti join keeps as they are. + let outer_columns = left.schema().columns(); let left = projected_left.as_ref().unwrap_or(left); + let join_filter = match (&in_value, join_filter_opt) { + (Some(in_value), Some(correlation)) => { + Expr::eq(in_value.value.clone(), in_value.subquery_column.clone()) + .and(correlation) + } + (Some(in_value), None) => { + Expr::eq(in_value.value.clone(), in_value.subquery_column.clone()) + } + (None, Some(correlation)) => correlation, + (None, None) => lit(true), + }; + + // `lit(true)` is the filter of a join that has no predicate of its own, and + // `true AND p` is only noise in the plan. + let join_filter = match extra_predicate { + Some(extra) if join_filter == lit(true) => extra, + Some(extra) => join_filter.and(extra), + None => join_filter, + }; + + // The keys of the join that replaces an `IN` or `NOT IN` predicate. An + // `EXISTS` has no value to compare, so its join never needs null-aware + // semantics. + let join_keys = match &in_value { + Some(in_value) + if matches!(join_type, JoinType::LeftMark | JoinType::LeftAnti) => + { + Some(JoinKeys::new( + in_value, + &join_filter, + left.schema(), + sub_query_alias.schema(), + &pull_up.correlated_filters, + )?) + } + _ => None, + }; + + // A `NOT IN` in a filter builds a `LeftAnti` join, and needs null-aware + // semantics when the value can be NULL in scope. The null-aware `LeftAnti` + // executor takes one key only (see `NullAwareMode::try_new`), and no hash + // join can mark the UNKNOWN rows of a residual filter + // (https://github.com/apache/datafusion/issues/25336). So: + // + // * A residual filter: give up here. The caller then materializes the + // UNKNOWN rows with more joins, see `in_subquery_value_mark_join`. + // * More than one key: the null-aware `LeftMark` executor takes any number + // of keys, the others being the scope of the outer row. Build that join + // instead and keep the rows whose mark is FALSE, which is `NOT IN` under + // three-valued logic. + // * One key: the null-aware `LeftAnti` join below. + let anti_join_as_mark = match &join_keys { + Some(keys) if join_type == JoinType::LeftAnti && keys.value_may_be_null => { + if keys.residual_filter.is_some() { + return Ok(None); + } + keys.equijoin_keys.len() > 1 + } + _ => false, + }; + let join_type = if anti_join_as_mark { + JoinType::LeftMark + } else { + join_type + }; + if matches!(join_type, JoinType::LeftMark | JoinType::RightMark) { let right_schema = sub_query_alias.schema(); @@ -584,30 +877,18 @@ 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 - }; - - // 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, - left.schema(), - right_projected.schema(), - )?; + // Only a filter that the hash join can turn into keys gives an exact + // mark: a residual predicate leaves the UNKNOWN rows unmarked. Such a + // mark is exact under SQL three-valued logic once it is null-aware + // when a key can be NULL in scope, which lets a projected `IN` use this + // join on its own. A residual filter keeps the join a plain mark join, + // and the caller materializes the UNKNOWN rows with more joins. + let (null_aware, mark_is_three_valued_exact) = match &join_keys { + Some(keys) if keys.residual_filter.is_none() => { + (keys.value_may_be_null, true) + } + _ => (false, false), + }; let new_plan = LogicalPlanBuilder::from(left.clone()) .join_detailed_with_options( @@ -620,25 +901,36 @@ fn build_join( )? .build()?; + // `NOT IN` keeps the rows whose mark is FALSE. `NOT mark` is TRUE for + // those rows only, and the projection removes the mark column again. + let new_plan = if anti_join_as_mark { + let mark = Expr::Column(Column::new(Some(alias.to_string()), "mark")); + LogicalPlanBuilder::from(new_plan) + .filter(not(mark))? + .project(outer_columns.into_iter().map(Expr::from))? + .build()? + } else { + new_plan + }; + debug!( "predicate subquery optimized:\n{}", new_plan.display_indent() ); - return Ok(Some(new_plan)); + return Ok(Some(BuiltJoin { + plan: new_plan, + mark_is_three_valued_exact, + })); } - // Determine if this should be a null-aware anti join - // Null-aware semantics are only needed for NOT IN subqueries, not NOT EXISTS: - // - NOT IN: Uses three-valued logic, requires null-aware handling - // - 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())?; + // Null-aware semantics are only needed for a `NOT IN` anti join, which + // follows three-valued logic. `NOT EXISTS` and `IN` are two-valued, and + // `join_keys` is `None` for them. The join here has one key and no + // residual filter: the other shapes were handled above. + let null_aware = join_keys + .as_ref() + .is_some_and(|keys| keys.value_may_be_null); // join our sub query into the main plan let new_plan = if null_aware { @@ -662,7 +954,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 +1041,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 @@ -1349,20 +1653,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] + 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, + @" + Projection: CASE WHEN __correlated_sq_1.mark THEN Boolean(true) WHEN __correlated_sq_2.mark THEN Boolean(NULL) ELSE Boolean(false) END AS is_present [is_present:Boolean;N] + LeftMark Join: Filter: test.a > __correlated_sq_2.a AND (__correlated_sq_2.c IS NULL OR test.c IS NULL) [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, __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] - SubqueryAlias: __correlated_sq_3 [c:UInt32] + Projection: __correlated_sq_2.c, __correlated_sq_2.a [c:UInt32, a:UInt32] + SubqueryAlias: __correlated_sq_2 [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] " @@ -1499,7 +1910,7 @@ mod tests { /// correlation predicate is a second equi-join key, and null-aware hash /// joins accept only one. #[test] - fn constant_not_in_correlated_subquery_is_not_rewritten() -> Result<()> { + fn constant_not_in_correlated_subquery_becomes_a_mark_join() -> Result<()> { let outer_scan = nullable_scalar_mark_scan("outer_t")?; let inner_scan = nullable_scalar_mark_scan("inner_t")?; @@ -1518,12 +1929,16 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" - LeftAnti Join: Filter: Int32(3) = __correlated_sq_1.id AND outer_t.grp = __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N] - TableScan: outer_t [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] + @" + Projection: outer_t.id, outer_t.grp [id:Int32;N, grp:Int32;N] + Filter: NOT __correlated_sq_1.mark [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32, mark:Boolean;N] + LeftMark Join: Filter: __correlated_sq_1_value = __correlated_sq_1.id AND outer_t.grp = __correlated_sq_1.grp null_aware [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32, mark:Boolean;N] + Projection: outer_t.id, outer_t.grp, Int32(3) AS __correlated_sq_1_value [id:Int32;N, grp:Int32;N, __correlated_sq_1_value:Int32] + 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] " ) } @@ -1584,6 +1999,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/joins.slt b/datafusion/sqllogictest/test_files/joins.slt index d2453af739c07..da95e7084e0e0 100644 --- a/datafusion/sqllogictest/test_files/joins.slt +++ b/datafusion/sqllogictest/test_files/joins.slt @@ -1988,11 +1988,18 @@ where join_t1.t1_id + 12 not in (select join_t2.t2_id + 1 from join_t2 where join_t1.t1_int > 0) ---- logical_plan -01)LeftAnti Join: CAST(join_t1.t1_id AS Int64) + Int64(12) = __correlated_sq_1.join_t2.t2_id + Int64(1) Filter: join_t1.t1_int > UInt32(0) null_aware -02)--TableScan: join_t1 projection=[t1_id, t1_name, t1_int] -03)--SubqueryAlias: __correlated_sq_1 -04)----Projection: CAST(join_t2.t2_id AS Int64) + Int64(1) -05)------TableScan: join_t2 projection=[t2_id] +01)Projection: join_t1.t1_id, join_t1.t1_name, join_t1.t1_int +02)--Filter: __correlated_sq_3.mark IS DISTINCT FROM Boolean(true) OR __correlated_sq_2.mark IS NOT DISTINCT FROM Boolean(true) +03)----LeftMark Join: Filter: join_t1.t1_int > UInt32(0) AND (__correlated_sq_3.join_t2.t2_id + Int64(1) IS NULL OR CAST(join_t1.t1_id AS Int64) + Int64(12) IS NULL) +04)------Filter: __correlated_sq_2.mark IS DISTINCT FROM Boolean(true) +05)--------LeftMark Join: CAST(join_t1.t1_id AS Int64) + Int64(12) = __correlated_sq_2.join_t2.t2_id + Int64(1) Filter: join_t1.t1_int > UInt32(0) +06)----------TableScan: join_t1 projection=[t1_id, t1_name, t1_int] +07)----------SubqueryAlias: __correlated_sq_2 +08)------------Projection: CAST(join_t2.t2_id AS Int64) + Int64(1) +09)--------------TableScan: join_t2 projection=[t2_id] +10)------SubqueryAlias: __correlated_sq_3 +11)--------Projection: CAST(join_t2.t2_id AS Int64) + Int64(1) +12)----------TableScan: join_t2 projection=[t2_id] # In subquery to join with outer filter diff --git a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt index 8023684ac3ee0..4644a3d3b6022 100644 --- a/datafusion/sqllogictest/test_files/null_aware_anti_join.slt +++ b/datafusion/sqllogictest/test_files/null_aware_anti_join.slt @@ -641,19 +641,23 @@ ORDER BY 1; statement ok DROP TABLE naconst_clash; -# A constant value expression with a non-equality correlation leaves the -# null-aware join without any equi-join key. Only `HashJoinExec` implements -# null-aware semantics and it needs a key, so the planner reports the gap -# instead of falling back to a nested loop join that ignores the NULLs and -# silently returns wrong results. +# A constant value expression with a non-equality correlation leaves a residual +# join filter, and no hash join can mark the UNKNOWN rows of a residual filter. +# The `NOT IN` does not become an anti join. It becomes the mark joins that +# materialize its three-valued result, the same plan as for a projected `IN`, +# and a filter on that result. For `id = 1` the correlated subquery result is +# `{NULL}`, so `3 NOT IN (...)` is UNKNOWN and the row is dropped. For `id = 2` +# the result is empty and the row is kept. statement ok CREATE TABLE naconst_corr_t1(id INT, g INT) AS VALUES (1, 1), (2, 2); statement ok CREATE TABLE naconst_corr_t2(id INT, g INT) AS VALUES (1, 1), (NULL, 2); -query error DataFusion error: Error during planning: null_aware LeftAnti join requires equi\-join keys, but the join has none +query I SELECT id FROM naconst_corr_t1 WHERE 3 NOT IN (SELECT id FROM naconst_corr_t2 WHERE naconst_corr_t2.g > naconst_corr_t1.g); +---- +2 statement ok DROP TABLE naconst_corr_t1; 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..eba525f538ddb 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -97,3 +97,520 @@ 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, __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 m +02)--LeftMark Join: Filter: __correlated_sq_2.z < n1.z AND (__correlated_sq_2.id IS NULL OR n1.id IS NULL) +03)----LeftMark Join: n1.id = __correlated_sq_1.id Filter: __correlated_sq_1.z < n1.z +04)------TableScan: n1 projection=[id, z] +05)------SubqueryAlias: __correlated_sq_1 +06)--------TableScan: n2 projection=[id, z] +07)----SubqueryAlias: __correlated_sq_2 +08)------TableScan: n2 projection=[id, z] +physical_plan +01)ProjectionExec: expr=[id@0 as id, mark@1 IS NOT DISTINCT FROM true OR mark@2 IS NOT DISTINCT FROM true AND mark@1 IS DISTINCT FROM true AND NULL as m] +02)--NestedLoopJoinExec: join_type=RightMark, filter=z@3 < z@1 AND (id@2 IS NULL OR id@0 IS NULL), projection=[id@0, mark@2, mark@3] +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)----RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +05)------HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(id@0, id@0)], filter=z@1 < z@0 +06)--------DataSourceExec: partitions=1, partition_sizes=[1] +07)--------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 join has two keys, the value +# and the correlation, and the null-aware `LeftAnti` executor takes one key +# only. The `NOT IN` becomes a null-aware mark join, which takes any number of +# keys, and a filter on the mark. `main` keeps the `k = 1` row, which 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)Projection: t1.k +02)--Filter: NOT __correlated_sq_1.mark +03)----LeftMark Join: nullif(CAST(t1.k AS Int64), Int64(1)) = __correlated_sq_1.t2.k + Int64(10), t1.k = __correlated_sq_1.k null_aware +04)------TableScan: t1 projection=[k] +05)------SubqueryAlias: __correlated_sq_1 +06)--------Projection: CAST(t2.k AS Int64) + Int64(10), t2.k +07)----------TableScan: t2 projection=[k] +physical_plan +01)FilterExec: NOT mark@1, projection=[k@0] +02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +03)----HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(nullif(t1.k,Int64(1))@1, t2.k + Int64(10)@0), (k@0, k@1)], projection=[k@0, mark@2], null_aware +04)------ProjectionExec: expr=[k@0 as k, nullif(CAST(k@0 AS Int64), 1) as nullif(t1.k,Int64(1))] +05)--------DataSourceExec: partitions=1, partition_sizes=[1] +06)------ProjectionExec: expr=[CAST(k@0 AS Int64) + 10 as t2.k + Int64(10), k@0 as k] +07)--------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. No hash +# join can mark the UNKNOWN rows of a residual filter +# (https://github.com/apache/datafusion/issues/25336), so a `NOT IN` whose +# value can be NULL does not become an anti join. It becomes the mark joins +# that materialize its three-valued result, the same plan as for a projected +# `IN`, and a filter on that result. 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)--Filter: __correlated_sq_3.mark IS DISTINCT FROM Boolean(true) OR __correlated_sq_2.mark IS NOT DISTINCT FROM Boolean(true) +03)----Projection: ra.k, __correlated_sq_2.mark, __correlated_sq_3.mark +04)------LeftMark Join: Filter: __correlated_sq_3.z > ra.z AND nullif(CAST(ra.k AS Int64), Int64(1)) IS NULL +05)--------Filter: __correlated_sq_2.mark IS DISTINCT FROM Boolean(true) +06)----------LeftMark Join: nullif(CAST(ra.k AS Int64), Int64(1)) = __correlated_sq_2.rb.k Filter: __correlated_sq_2.z > ra.z +07)------------TableScan: ra projection=[k, z] +08)------------SubqueryAlias: __correlated_sq_2 +09)--------------Projection: CAST(rb.k AS Int64), rb.z +10)----------------TableScan: rb projection=[k, z] +11)--------SubqueryAlias: __correlated_sq_3 +12)----------TableScan: rb projection=[z] +physical_plan +01)FilterExec: mark@2 IS DISTINCT FROM true OR mark@1 IS NOT DISTINCT FROM true, projection=[k@0] +02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +03)----NestedLoopJoinExec: join_type=LeftMark, filter=z@2 > z@1 AND nullif(CAST(k@0 AS Int64), 1) IS NULL, projection=[k@0, mark@2, mark@3] +04)------CoalescePartitionsExec +05)--------FilterExec: mark@2 IS DISTINCT FROM true +06)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1 +07)------------HashJoinExec: mode=CollectLeft, join_type=RightMark, on=[(rb.k@0, nullif(ra.k,Int64(1))@2)], filter=z@1 > z@0, projection=[k@0, z@1, mark@3] +08)--------------ProjectionExec: expr=[CAST(k@0 AS Int64) as rb.k, z@1 as z] +09)----------------DataSourceExec: partitions=1, partition_sizes=[1] +10)--------------ProjectionExec: expr=[k@0 as k, z@1 as z, nullif(CAST(k@0 AS Int64), 1) as nullif(ra.k,Int64(1))] +11)----------------DataSourceExec: partitions=1, partition_sizes=[1] +12)------DataSourceExec: partitions=1, partition_sizes=[1] + +# The correlation names no column of the subquery, so it stays as a residual +# filter and the subquery is either the whole of `ic` or empty. A NULL in +# `ic.id` then makes the answer UNKNOWN only for the rows whose correlation +# holds. `main` builds a null-aware anti join that does not apply the residual +# when it looks for a NULL, sees one in `ic`, and drops every row. This is the +# shape whose plan `joins.slt` pins. +statement ok +CREATE TABLE oc(id INT, g INT) AS VALUES (1, 1), (2, 0), (NULL, 0); + +statement ok +CREATE TABLE ic(id INT) AS VALUES (5), (NULL); + +# `g > 0` holds for `id = 1` only, so its subquery is `{5, NULL}` and +# `1 NOT IN {5, NULL}` is UNKNOWN. The other two rows have an empty subquery, +# and ` NOT IN ()` is TRUE, the NULL row included. +query I rowsort +SELECT id FROM oc WHERE oc.id NOT IN (SELECT ic.id FROM ic WHERE oc.g > 0); +---- +2 +NULL + +statement ok +DROP TABLE oc; + +statement ok +DROP TABLE ic; + +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 is a different problem. The +# grand-total row of `ROLLUP` exists for every outer row, so a miss is UNKNOWN +# here, not FALSE. The pull up moves the filter above the aggregate, which +# loses that row; the same plan gives a wrong `EXISTS` too. That is a bug in +# the pull up, https://github.com/apache/datafusion/issues/25519, and is not +# changed here: 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;