From 95c11814f2ef445b22c9a77b37699694fe609f9f Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:03:33 -0500 Subject: [PATCH 01/11] test: keep an un-decorrelated IN subquery in the extracted-alias regression test The regression test for the extracted-alias generator relied on an IN subquery in the SELECT list surviving until ExtractLeafExpressions runs. IN subqueries in projections are now decorrelated, so make the subquery correlated with a LIMIT, which DecorrelatePredicateSubquery cannot pull up. The generator still starts at 2 inside the subquery, so the test guards the same behaviour. Co-Authored-By: Claude Fable 5.1 --- .../test_files/projection_pushdown.slt | 58 +++++++++---------- 1 file changed, 29 insertions(+), 29 deletions(-) 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; From 7aae42b3fc24f5d6a6d7ff6822d17b7f742044cb Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:08:47 -0500 Subject: [PATCH 02/11] perf: use one null-aware mark join for hashable IN subqueries in projections DecorrelatePredicateSubquery materialized every projected IN and NOT IN subquery with three mark joins. Two of them have no join predicate for an uncorrelated subquery and plan as nested-loop mark joins over outer x inner rows. A single LeftMark join already carries SQL three-valued logic when its filter is hashable-only: the join is null-aware when the keys may be NULL, and exact otherwise. Use that join alone in the hashable case and keep the three-join materialization only when a residual non-equality filter remains. Co-Authored-By: Christian McArthur Co-Authored-By: Claude Fable 5.1 --- .../src/decorrelate_predicate_subquery.rs | 196 +++++++++++++++--- 1 file changed, 170 insertions(+), 26 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 0ad8b44c40def..2bb6b9de80bb5 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -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,14 +424,52 @@ 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, + }, + ), ) } @@ -440,13 +506,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()); @@ -596,10 +671,12 @@ fn build_join( 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. + // Put null-aware semantics into the nullable mark column when the + // predicate can be implemented by hash keys. 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 = join_type == JoinType::LeftMark && in_predicate_opt.is_some() && mark_filter_is_hashable_only @@ -625,7 +702,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 @@ -662,7 +742,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)] @@ -1346,25 +1429,86 @@ 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] - 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: sq.c [c: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] + 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] " ) } From 75e22e9f2c04cc63a1b2546c1ba1a1a27b7151cd Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:09:59 -0500 Subject: [PATCH 03/11] test: cover mark-join plan shapes and NULL semantics for projected IN subqueries Add plan guards for the two paths of a projected IN subquery: one null-aware LeftMark join per subquery when the join filter is hashable only, and the three-join materialization when a non-equality correlation stays as a residual join filter. Add result tests for IN, NOT IN, EXISTS, an aggregate probe, a CASE wrapper and a COALESCE wrapper over tables that hold NULL keys. The expected values agree with DuckDB 1.5.2 and PostgreSQL 17. Co-Authored-By: Claude Fable 5.1 --- .../test_files/subquery_projection.slt | 154 ++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/datafusion/sqllogictest/test_files/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index ab7c80d9de165..10af36d16f206 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -97,3 +97,157 @@ 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 OR n1.id IS NULL AND __correlated_sq_3.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_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, mark@1 IS NOT DISTINCT FROM true OR (mark@2 OR id@0 IS NULL AND mark@3) IS NOT DISTINCT FROM true AND mark@1 IS DISTINCT FROM true AND NULL 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; From 0f4fa7e5cb14bb39645770d21bc1a32698b6a9a3 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Tue, 15 Sep 2026 17:10:52 -0500 Subject: [PATCH 04/11] fix: derive mark-join null-awareness from the join key expressions `join_keys_may_be_null` decided whether a `LeftMark` or `LeftAnti` join must be null-aware from the nullability of the columns that the join filter refers to. A join key is an expression, not only a column, and 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. With such a key the join was not null-aware, so the mark column was `false` where SQL asks for NULL. Over a table with a column `id INT NOT NULL` that holds 1, 2 and 4, and a subquery that gives 1 and 2: - `NULLIF(id, 1) IN (SELECT id FROM r3)` gave `false` for `id = 1`. The correct answer is UNKNOWN. - `TRY_CAST(s AS INT) IN (SELECT id FROM r3)` gave `false` where the text is not a number. The correct answer is UNKNOWN. The `NOT IN` filter path has the same gap, which is older than the projection path. `SELECT id FROM nn WHERE NULLIF(id, 1) NOT IN (SELECT id FROM r3)` returned the row for `id = 1`, which UNKNOWN must remove. The helper now takes the equijoin keys and the residual filter from `split_eq_and_noneq_join_predicate` and asks each key expression for its nullability against the schema of its own side. A cast around a key, such as the `CAST(id AS Int64)` that type coercion adds, keeps the nullability of the expression in it. For a residual filter there is no key expression, so the helper keeps the older column test there and is never less conservative than before. The `LeftMark` path already split the filter to find out whether it is hashable only. It now reuses that split instead of splitting twice. The projection path thus stays one null-aware mark join, and remains as fast as before. Co-Authored-By: Claude Fable 5.1 --- .../src/decorrelate_predicate_subquery.rs | 184 ++++++++++++++---- .../test_files/subquery_projection.slt | 85 ++++++++ 2 files changed, 230 insertions(+), 39 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 2bb6b9de80bb5..306e9a118552d 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; @@ -473,19 +473,45 @@ fn mark_join_detailed( ) } -/// Check if join keys in the join filter may contain NULL values +/// Check if the join keys can be NULL. /// -/// 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. +/// 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. +/// +/// 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 { @@ -659,32 +685,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 + }; + + // 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. 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 = join_type == JoinType::LeftMark - && in_predicate_opt.is_some() - && mark_filter_is_hashable_only - && join_keys_may_be_null( - &join_filter, + // 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( @@ -714,11 +747,23 @@ 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(), + )?; + 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 { @@ -829,6 +874,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 @@ -1513,6 +1567,58 @@ mod tests { ) } + /// 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] + " + ) + } + #[test] fn unsupported_correlated_in_projection_is_left_unchanged() -> Result<()> { let subquery = Arc::new( diff --git a/datafusion/sqllogictest/test_files/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index 10af36d16f206..59e608e3a1604 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -251,3 +251,88 @@ 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 + +statement ok +DROP TABLE nn; + +statement ok +DROP TABLE r3; + +statement ok +DROP TABLE r3n; From a30d591f8db2e219ec79643f9318169aeb568953 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 16 Sep 2026 09:25:14 -0500 Subject: [PATCH 05/11] fix: keep the column null test for a correlated NOT IN with two keys The previous commit made a `LeftAnti` join for `NOT IN` null-aware when a key expression can be NULL. Most scalar functions report a nullable result, so a key such as `upper(s)` over a `NOT NULL` column made the join null-aware. A correlated `NOT IN` has two keys (the value and the correlation), and a null-aware `LeftAnti` hash join supports one key only. Thus this query failed to plan: SELECT * FROM t1 WHERE upper(t1.s) NOT IN (SELECT t2.s FROM t2 WHERE t2.k = t1.k); Error during planning: null_aware LeftAnti joins only support single column join key, got 2 columns For a `LeftAnti` join with more than one key, use the column test on the whole join filter again, as on `main`. The expression test stays for one key and for all mark joins. Also add tests for an empty subquery with a `NULLIF` key on the single mark join path, and for the correlated `NOT IN` above. Co-Authored-By: Claude Opus 5 --- .../src/decorrelate_predicate_subquery.rs | 29 ++++++-- .../test_files/subquery_projection.slt | 71 +++++++++++++++++++ 2 files changed, 94 insertions(+), 6 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 306e9a118552d..c171a0d3fc805 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -755,12 +755,29 @@ fn build_join( left.schema(), sub_query_alias.schema(), )?; - join_keys_may_be_null( - &equijoin_keys, - residual_filter.as_ref(), - left.schema(), - sub_query_alias.schema(), - )? + if equijoin_keys.len() > 1 { + // A null-aware `LeftAnti` hash join supports one key only. A + // correlated `NOT IN` has two or more keys (the value and the + // correlation), so keep the column test on the whole filter here. + // A function key such as `upper(s)` over a non-nullable column + // then does not make the join null-aware and fail to plan. 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 }; diff --git a/datafusion/sqllogictest/test_files/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index 59e608e3a1604..abab73f68d63c 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -328,6 +328,77 @@ 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; + statement ok DROP TABLE nn; From 7679b231f6b1af0f75f78e4b5504212f59e480d1 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Fri, 18 Sep 2026 10:52:01 -0500 Subject: [PATCH 06/11] fix: give `join_keys_may_be_null` its new argument list at the constant `IN` call `join_keys_may_be_null` takes the equi-join keys and the residual filter since the previous commit. The call that the constant-value projection of #25348 makes still passes the earlier three arguments, so `datafusion-optimizer` does not build. The value expression of a constant `IN` holds no column, so the equality is not an equi-join key. There is no key expression to ask for its nullability, and the column test on the whole filter is the only test available there. The call thus passes no keys and the whole filter as the residual, which is what the earlier three-argument version did. Co-Authored-By: Claude Opus 5 --- .../optimizer/src/decorrelate_predicate_subquery.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index c171a0d3fc805..359841a55fe9c 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -625,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. From e501c78c943243c66799b89e33259e837c3b5bf1 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Fri, 18 Sep 2026 10:52:01 -0500 Subject: [PATCH 07/11] test: regenerate the stale plan of a projected `IN` with a non-equality correlation The expected plan holds the simplified boolean form of the `CASE` that `in_subquery_value_mark_join` builds. The tree gives the `CASE` itself. The result rows do not change, and the query below the `EXPLAIN` checks them. Co-Authored-By: Claude Opus 5 --- datafusion/sqllogictest/test_files/subquery_projection.slt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/datafusion/sqllogictest/test_files/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index abab73f68d63c..255ac44545cf8 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -141,7 +141,7 @@ 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 OR n1.id IS NULL AND __correlated_sq_3.mark) IS NOT DISTINCT FROM Boolean(true) AND __correlated_sq_1.mark IS DISTINCT FROM Boolean(true) AND Boolean(NULL) AS m +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 @@ -155,7 +155,7 @@ logical_plan 12)----SubqueryAlias: __correlated_sq_3 13)------TableScan: n2 projection=[z] physical_plan -01)ProjectionExec: expr=[id@0 as id, mark@1 IS NOT DISTINCT FROM true OR (mark@2 OR id@0 IS NULL AND mark@3) IS NOT DISTINCT FROM true AND mark@1 IS DISTINCT FROM true AND NULL as m] +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 From 3af87370c1c4202a37d304984d307ae6e17c4924 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Fri, 18 Sep 2026 12:11:55 -0500 Subject: [PATCH 08/11] fix: keep the column null test for a `NOT IN` anti join with a residual filter A null-aware `LeftAnti` join does not apply the residual join filter when it decides whether a NULL makes the result UNKNOWN (https://github.com/apache/datafusion/issues/25336). It looks only at whether the probe side holds any row. For a correlated `NOT IN` with a non-equality correlation this is wrong in one direction: a build row whose key is NULL is dropped although the correlated subquery result for that row is empty, and `NULL NOT IN ()` is TRUE. The column test that this branch replaces never reached that state, because a key expression such as `NULLIF(k, 1)` over `NOT NULL` columns holds no nullable column. The expression test does reach it, so it is a new wrong result: ```sql CREATE TABLE ra(k INT NOT NULL, z INT NOT NULL) AS VALUES (1, 10), (2, 20); CREATE TABLE rb(k INT NOT NULL, z INT NOT NULL) AS VALUES (5, 50); SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z < ra.z); -- correct, and `main`: 1, 2. before this commit: 2 ``` The expression test thus applies only when the split gives one key and no residual. A residual now keeps the column test, as more than one key already does. Every result of this branch is then the same as `main`'s or better. Co-Authored-By: Claude Opus 5 --- .../src/decorrelate_predicate_subquery.rs | 24 ++++++++++---- .../test_files/subquery_projection.slt | 33 +++++++++++++++++++ 2 files changed, 50 insertions(+), 7 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 359841a55fe9c..8f34fb267b9e8 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -763,13 +763,23 @@ fn build_join( left.schema(), sub_query_alias.schema(), )?; - if equijoin_keys.len() > 1 { - // A null-aware `LeftAnti` hash join supports one key only. A - // correlated `NOT IN` has two or more keys (the value and the - // correlation), so keep the column test on the whole filter here. - // A function key such as `upper(s)` over a non-nullable column - // then does not make the join null-aware and fail to plan. The - // column test misses a NULL that only the key expression makes, + 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( diff --git a/datafusion/sqllogictest/test_files/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index 255ac44545cf8..e071259660179 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -399,6 +399,39 @@ 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; From 61c6d51d42ceb4d5ed2b4d49b14e35ca30ced3d1 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco Date: Sat, 19 Sep 2026 11:46:58 -0500 Subject: [PATCH 09/11] fix: do not build a null-aware join when the correlation repeats the `IN` predicate `x IN (SELECT y FROM .. WHERE y = x)` writes the `IN` equality a second time, as a correlated filter. `remove_duplicated_filter` drops that filter, because `build_join` adds the same equality back as the join filter. The scope of the subquery is then gone, and the join looked uncorrelated and null-aware, so a miss gave UNKNOWN. 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}`. `PullUpCorrelatedExpr` now reports the shape, and the mark join and the anti join stay plain for it. A node above the dropped filter can still put a NULL into the value column, and then the result is UNKNOWN again. An outer join, a union and a grouping set do this, so the flag is cleared when the pull up passes one of them. `f_up` walks the subquery bottom up, so every such node above the filter is reached after it. This also fixes https://github.com/apache/datafusion/issues/25480, which is the same root cause in a `WHERE ... NOT IN` clause. Co-Authored-By: Claude Opus 5 --- datafusion/optimizer/src/decorrelate.rs | 47 ++++++++++++++++++- .../src/decorrelate_predicate_subquery.rs | 12 ++++- 2 files changed, 56 insertions(+), 3 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 0c37f00b64355..0e54b6f1d790e 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -34,8 +34,8 @@ use datafusion_expr::utils::{ collect_subquery_cols, conjunction, find_join_exprs, split_conjunction, }; use datafusion_expr::{ - BinaryExpr, Cast, EmptyRelation, Expr, ExprSchemable, FetchType, LogicalPlan, - LogicalPlanBuilder, Operator, expr, lit, + BinaryExpr, Cast, EmptyRelation, Expr, ExprSchemable, FetchType, JoinType, + LogicalPlan, LogicalPlanBuilder, Operator, expr, lit, }; /// This struct rewrite the sub query plan by pull up the correlated @@ -74,6 +74,21 @@ 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, + /// The subquery writes the `IN` predicate a second time, as a correlated + /// filter: `x IN (SELECT y FROM .. WHERE y = x)`. + /// + /// [`remove_duplicated_filter`] drops that filter, because the caller adds + /// the same equality back as the join filter. The `IN` is then never + /// UNKNOWN: every row in the scope of an outer row satisfies `y = x`, so + /// `y` is not NULL there and the subquery result is either empty or `{x}`. + /// A null-aware join reports UNKNOWN for a miss, so the caller must not + /// build one for this shape. See + /// . + /// + /// A node above that filter can still put a NULL into the value column, so + /// the flag is cleared again when the pull up passes such a node. See + /// [`plan_may_add_null_rows`]. + pub in_predicate_is_correlation: bool, } impl Default for PullUpCorrelatedExpr { @@ -95,6 +110,7 @@ impl PullUpCorrelatedExpr { collected_count_expr_map: HashMap::new(), pull_up_having_expr: None, pulled_up_scalar_agg: false, + in_predicate_is_correlation: false, } } @@ -173,6 +189,12 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { } fn f_up(&mut self, plan: LogicalPlan) -> Result> { + // `f_up` walks the subquery bottom up, so a node that this reaches + // after the `Filter` sits above it and can undo what the filter + // promised about the value column. + if self.in_predicate_is_correlation && plan_may_add_null_rows(&plan) { + self.in_predicate_is_correlation = false; + } let subquery_schema = plan.schema(); match &plan { LogicalPlan::Filter(plan_filter) => { @@ -186,7 +208,9 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { find_join_exprs(subquery_filter_exprs)?; 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. + let filter_count = join_filters.len(); join_filters = remove_duplicated_filter(join_filters, in_predicate)?; + self.in_predicate_is_correlation |= join_filters.len() < filter_count; } let correlated_subquery_cols = collect_subquery_cols(&join_filters, subquery_schema)?; @@ -480,6 +504,25 @@ fn collect_local_correlated_cols( } } +/// Can this plan node give a row whose value column is NULL although a filter +/// below it kept only rows where that column equals the `IN` value? +/// +/// See [`PullUpCorrelatedExpr::in_predicate_is_correlation`]. +fn plan_may_add_null_rows(plan: &LogicalPlan) -> bool { + match plan { + // An outer join gives a NULL for every column of an unmatched side. + LogicalPlan::Join(join) => join.join_type != JoinType::Inner, + // A filter in one branch of a union says nothing about the others. + LogicalPlan::Union(_) => true, + // A grouping set gives a NULL for each grouping column it rolls up. + LogicalPlan::Aggregate(aggregate) => aggregate + .group_expr + .iter() + .any(|expr| matches!(expr, Expr::GroupingSet(_))), + _ => false, + } +} + fn remove_duplicated_filter( filters: Vec, in_predicate: &Expr, diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 8f34fb267b9e8..02579cba390b4 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -718,6 +718,10 @@ fn build_join( // filters stay on the legacy path because hash join execution cannot // mark UNKNOWN candidates for residual predicates. let null_aware = match hashable_only_split { + // The subquery repeats the `IN` predicate as its correlation, so + // its result is never UNKNOWN and a plain mark join is exact. See + // `PullUpCorrelatedExpr::in_predicate_is_correlation`. + Some(_) if pull_up.in_predicate_is_correlation => false, Some((equijoin_keys, residual_filter)) => join_keys_may_be_null( equijoin_keys, residual_filter.as_ref(), @@ -757,7 +761,13 @@ fn build_join( // // 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() { + // + // A subquery that repeats the `IN` predicate as its correlation is never + // UNKNOWN either, see `PullUpCorrelatedExpr::in_predicate_is_correlation`. + let null_aware = if join_type == JoinType::LeftAnti + && in_predicate_opt.is_some() + && !pull_up.in_predicate_is_correlation + { let (equijoin_keys, residual_filter) = split_eq_and_noneq_join_predicate( join_filter.clone(), left.schema(), From 7bb69a59193862045ca265d7d82fef319aa48b77 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco Date: Sat, 19 Sep 2026 11:47:07 -0500 Subject: [PATCH 10/11] test: cover a correlation that repeats the `IN` predicate The new `subquery_projection.slt` section holds the three queries from the review: the `IN` whose correlation is the `IN` predicate, the same with a second correlation, and the `WHERE ... NOT IN` form. It also pins the plan, which no longer says `null_aware`, and the `ROLLUP` shape, which still does because the grouping set puts a NULL back into the value column. Every expected value agrees with DuckDB 1.5.2. Two unit tests cover the mark join and the anti join. Co-Authored-By: Claude Opus 5 --- .../src/decorrelate_predicate_subquery.rs | 57 ++++++++++ .../test_files/subquery_projection.slt | 106 ++++++++++++++++++ 2 files changed, 163 insertions(+) diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 02579cba390b4..fcab913b0c66f 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -1879,6 +1879,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/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index e071259660179..d72d237a6dd5f 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -440,3 +440,109 @@ 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 + +# A node above the dropped filter can put a NULL back into the value column, and +# the result is UNKNOWN again. A grouping set rolls `ci.k` up to a NULL row, so +# this join stays null-aware. +query TT +EXPLAIN SELECT co.id, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLLUP(ci.k)) AS m FROM co; +---- +logical_plan +01)Projection: co.id, __correlated_sq_1.mark AS m +02)--LeftMark Join: co.k = __correlated_sq_1.k null_aware +03)----TableScan: co projection=[id, k] +04)----SubqueryAlias: __correlated_sq_1 +05)------Projection: ci.k +06)--------Aggregate: groupBy=[[ROLLUP (ci.k)]], aggr=[[]] +07)----------TableScan: ci projection=[k] +physical_plan +01)ProjectionExec: expr=[id@0 as id, mark@1 as m] +02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(k@1, k@0)], projection=[id@0, mark@2], null_aware +03)----DataSourceExec: partitions=1, partition_sizes=[1] +04)----ProjectionExec: expr=[k@0 as k] +05)------AggregateExec: mode=FinalPartitioned, gby=[k@0 as k, __grouping_id@1 as __grouping_id], aggr=[] +06)--------RepartitionExec: partitioning=Hash([k@0, __grouping_id@1], 4), input_partitions=1 +07)----------AggregateExec: mode=Partial, gby=[(NULL as k), (k@0 as k)], aggr=[] +08)------------DataSourceExec: partitions=1, partition_sizes=[1] + +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 NULL +9 9 NULL +5 NULL NULL + +statement ok +DROP TABLE co; + +statement ok +DROP TABLE ci; From fb08efa197aed66a861947b19adbc992600cba0c Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco Date: Sat, 19 Sep 2026 18:07:55 -0500 Subject: [PATCH 11/11] refactor: derive null-aware joins from the `IN` value in scope The rule decided null-awareness with three special cases: a flag that `PullUpCorrelatedExpr` set when it dropped a duplicated correlation and cleared again for a list of plan nodes, a switch to a weaker column test when the null-aware `LeftAnti` executor could not take the join, and an empty key list that selected that weaker test at the constant `IN` call. Replace them with one question. `IN` can only be UNKNOWN when its value or the subquery output can be NULL inside the scope of an outer row. A correlated conjunct such as `y = x` is never TRUE for a NULL, so a key it compares cannot be NULL in scope. `PullUpCorrelatedExpr` now records every correlated conjunct before it drops the duplicated one, and `JoinKeys` asks the two sides of the `IN` predicate for their nullability against those conjuncts. A correlation key that is NULL only empties the scope, so it never makes the join null-aware. The executor limits are then applied in one place. A `NOT IN` whose value can be NULL and whose join has more than one key becomes a null-aware `LeftMark` join, which takes any number of keys, plus a filter on the mark. A `NOT IN` with a residual filter becomes the mark joins that materialize its three-valued result, the same plan as a projected `IN`. Both shapes failed to plan or gave wrong results before. The constant value projection now applies to correlated subqueries too, since the mark join accepts the correlation as a second key. The guard for a grouping set above the dropped filter is gone with the flag. That query is wrong on `main` for `EXISTS` as well, because the pull up moves the filter above the aggregate; that is https://github.com/apache/datafusion/issues/25519. Co-Authored-By: Claude Fable 5.1 --- datafusion/optimizer/src/decorrelate.rs | 61 +- .../src/decorrelate_predicate_subquery.rs | 599 ++++++++++-------- datafusion/sqllogictest/test_files/joins.slt | 21 +- .../test_files/null_aware_anti_join.slt | 16 +- .../test_files/subquery_projection.slt | 123 ++-- 5 files changed, 470 insertions(+), 350 deletions(-) diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 0e54b6f1d790e..d1c0b5d47d54e 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -34,8 +34,8 @@ use datafusion_expr::utils::{ collect_subquery_cols, conjunction, find_join_exprs, split_conjunction, }; use datafusion_expr::{ - BinaryExpr, Cast, EmptyRelation, Expr, ExprSchemable, FetchType, JoinType, - LogicalPlan, LogicalPlanBuilder, Operator, expr, lit, + BinaryExpr, Cast, EmptyRelation, Expr, ExprSchemable, FetchType, LogicalPlan, + LogicalPlanBuilder, Operator, expr, lit, }; /// This struct rewrite the sub query plan by pull up the correlated @@ -74,21 +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, - /// The subquery writes the `IN` predicate a second time, as a correlated - /// filter: `x IN (SELECT y FROM .. WHERE y = x)`. + /// Every correlated conjunct that a `Filter` of the subquery applies, + /// before [`remove_duplicated_filter`] drops the ones that the `IN` + /// predicate already covers. /// - /// [`remove_duplicated_filter`] drops that filter, because the caller adds - /// the same equality back as the join filter. The `IN` is then never - /// UNKNOWN: every row in the scope of an outer row satisfies `y = x`, so - /// `y` is not NULL there and the subquery result is either empty or `{x}`. - /// A null-aware join reports UNKNOWN for a miss, so the caller must not - /// build one for this shape. See - /// . - /// - /// A node above that filter can still put a NULL into the value column, so - /// the flag is cleared again when the pull up passes such a node. See - /// [`plan_may_add_null_rows`]. - pub in_predicate_is_correlation: bool, + /// `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 { @@ -110,7 +105,7 @@ impl PullUpCorrelatedExpr { collected_count_expr_map: HashMap::new(), pull_up_having_expr: None, pulled_up_scalar_agg: false, - in_predicate_is_correlation: false, + correlated_filters: Vec::new(), } } @@ -189,12 +184,6 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { } fn f_up(&mut self, plan: LogicalPlan) -> Result> { - // `f_up` walks the subquery bottom up, so a node that this reaches - // after the `Filter` sits above it and can undo what the filter - // promised about the value column. - if self.in_predicate_is_correlation && plan_may_add_null_rows(&plan) { - self.in_predicate_is_correlation = false; - } let subquery_schema = plan.schema(); match &plan { LogicalPlan::Filter(plan_filter) => { @@ -206,11 +195,14 @@ 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. - let filter_count = join_filters.len(); join_filters = remove_duplicated_filter(join_filters, in_predicate)?; - self.in_predicate_is_correlation |= join_filters.len() < filter_count; } let correlated_subquery_cols = collect_subquery_cols(&join_filters, subquery_schema)?; @@ -504,25 +496,6 @@ fn collect_local_correlated_cols( } } -/// Can this plan node give a row whose value column is NULL although a filter -/// below it kept only rows where that column equals the `IN` value? -/// -/// See [`PullUpCorrelatedExpr::in_predicate_is_correlation`]. -fn plan_may_add_null_rows(plan: &LogicalPlan) -> bool { - match plan { - // An outer join gives a NULL for every column of an unmatched side. - LogicalPlan::Join(join) => join.join_type != JoinType::Inner, - // A filter in one branch of a union says nothing about the others. - LogicalPlan::Union(_) => true, - // A grouping set gives a NULL for each grouping column it rolls up. - LogicalPlan::Aggregate(aggregate) => aggregate - .group_expr - .iter() - .any(|expr| matches!(expr, Expr::GroupingSet(_))), - _ => false, - } -} - fn remove_duplicated_filter( filters: Vec, in_predicate: &Expr, diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index fcab913b0c66f..303d02c8b68ce 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -29,8 +29,7 @@ use crate::{OptimizerConfig, OptimizerRule}; use datafusion_common::alias::AliasGenerator; use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; use datafusion_common::{ - Column, DFSchema, 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; @@ -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)) => { @@ -252,7 +271,7 @@ fn in_subquery_value_mark_join( plan: matched_plan, mark: matched, three_valued_exact, - }) = mark_join_detailed(left, subquery, Some(&in_predicate), false, alias)? + }) = mark_join(left, subquery, Some(&in_predicate), false, alias)? else { return Ok(None); }; @@ -270,13 +289,19 @@ fn in_subquery_value_mark_join( 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)? + let Some(MarkJoin { + plan: null_plan, + mark: 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)? + let Some(MarkJoin { + plan: final_plan, + mark: subquery_non_empty, + .. + }) = mark_join(&null_plan, subquery, None, false, alias)? else { return Ok(None); }; @@ -397,7 +422,7 @@ fn build_join_top( subquery, in_predicate_opt.as_ref(), join_type, - subquery_alias, + &subquery_alias, )? .map(|join| join.plan)) } @@ -423,10 +448,20 @@ fn mark_join( in_predicate_opt: Option<&Expr>, negated: bool, alias_generator: &Arc, -) -> Result> { +) -> 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( - mark_join_detailed(left, subquery, in_predicate_opt, negated, alias_generator)? - .map(|mark_join| (mark_join.plan, mark_join.mark)), + 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, + }, + ), ) } @@ -441,7 +476,7 @@ struct MarkJoin { /// /// 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 + /// 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 @@ -449,87 +484,149 @@ struct MarkJoin { 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"); +/// 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, +} - let exists_col = Expr::Column(Column::new(Some(alias.clone()), "mark")); - let exists_expr = if negated { !exists_col } else { exists_col }; +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, + )?, + }) + } +} - Ok( - 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, - }, - ), - ) +/// 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 the join keys can be NULL. +/// Can this join key be NULL for a row inside the scope of an outer row? /// -/// 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. +/// `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 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 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 +/// (). /// -/// 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( - equijoin_keys: &[(Expr, Expr)], - residual: Option<&Expr>, - left_schema: &DFSchema, - right_schema: &DFSchema, -) -> Result { - 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(residual, &mut columns)?; +/// 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)) +} - // 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); +/// 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`]. @@ -546,7 +643,7 @@ fn build_join( subquery: &LogicalPlan, in_predicate_opt: Option<&Expr>, join_type: JoinType, - alias: String, + alias: &str, ) -> Result> { let mut pull_up = PullUpCorrelatedExpr::new() .with_in_predicate_opt(in_predicate_opt.cloned()) @@ -569,47 +666,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 @@ -620,45 +697,109 @@ 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) - // 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. - 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), + }; + + // 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(); @@ -693,42 +834,17 @@ fn build_join( sub_query_alias.clone() }; - 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 - }; - // 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 { - // The subquery repeats the `IN` predicate as its correlation, so - // its result is never UNKNOWN and a plain mark join is exact. See - // `PullUpCorrelatedExpr::in_predicate_is_correlation`. - Some(_) if pull_up.in_predicate_is_correlation => false, - Some((equijoin_keys, residual_filter)) => join_keys_may_be_null( - equijoin_keys, - residual_filter.as_ref(), - left.schema(), - right_projected.schema(), - )?, - None => false, + // 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()) @@ -742,6 +858,18 @@ 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() @@ -749,66 +877,17 @@ fn build_join( return Ok(Some(BuiltJoin { plan: new_plan, - mark_is_three_valued_exact: mark_filter_is_hashable_only, + 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 no join key can be NULL on either side, we don't need - // null-aware semantics because NULLs cannot exist in the keys. - // - // A subquery that repeats the `IN` predicate as its correlation is never - // UNKNOWN either, see `PullUpCorrelatedExpr::in_predicate_is_correlation`. - let null_aware = if join_type == JoinType::LeftAnti - && in_predicate_opt.is_some() - && !pull_up.in_predicate_is_correlation - { - 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 - }; + // 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 { @@ -1794,7 +1873,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")?; @@ -1813,12 +1892,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] " ) } diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt index d2453af739c07..c412f62baa289 100644 --- a/datafusion/sqllogictest/test_files/joins.slt +++ b/datafusion/sqllogictest/test_files/joins.slt @@ -1988,11 +1988,22 @@ 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: NOT CASE WHEN __correlated_sq_2.mark THEN Boolean(true) WHEN __correlated_sq_3.mark OR CAST(join_t1.t1_id AS Int64) + Int64(12) IS NULL AND __correlated_sq_4.mark THEN Boolean(NULL) ELSE Boolean(false) END +03)----LeftMark Join: Filter: join_t1.t1_int > UInt32(0) +04)------LeftMark Join: Filter: join_t1.t1_int > UInt32(0) +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)------------Filter: CAST(join_t2.t2_id AS Int64) + Int64(1) IS NULL +13)--------------TableScan: join_t2 projection=[t2_id] +14)------SubqueryAlias: __correlated_sq_4 +15)--------Projection: CAST(join_t2.t2_id AS Int64) + Int64(1) +16)----------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/subquery_projection.slt b/datafusion/sqllogictest/test_files/subquery_projection.slt index d72d237a6dd5f..06a4e18d58d68 100644 --- a/datafusion/sqllogictest/test_files/subquery_projection.slt +++ b/datafusion/sqllogictest/test_files/subquery_projection.slt @@ -384,27 +384,50 @@ 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 +# 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); ---- -1 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. 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. +# 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); @@ -418,14 +441,43 @@ SELECT k FROM ra WHERE NULLIF(ra.k, 1) NOT IN (SELECT rb.k FROM rb WHERE rb.z < 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. +# 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); ---- -1 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: NOT CASE WHEN __correlated_sq_2.mark THEN Boolean(true) WHEN __correlated_sq_3.mark OR nullif(CAST(ra.k AS Int64), Int64(1)) IS NULL AND __correlated_sq_4.mark THEN Boolean(NULL) ELSE Boolean(false) END +03)----Projection: ra.k, __correlated_sq_2.mark, __correlated_sq_3.mark, __correlated_sq_4.mark +04)------LeftMark Join: Filter: __correlated_sq_4.z > ra.z +05)--------LeftMark Join: Filter: __correlated_sq_3.z > ra.z +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)----------EmptyRelation: rows=0 +12)--------SubqueryAlias: __correlated_sq_4 +13)----------TableScan: rb projection=[z] +physical_plan +01)FilterExec: NOT CASE WHEN mark@1 THEN true WHEN mark@2 OR nullif(CAST(k@0 AS Int64), 1) IS NULL AND mark@3 THEN NULL ELSE false END, projection=[k@0] +02)--NestedLoopJoinExec: join_type=RightMark, filter=z@1 > z@0, projection=[k@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)------EmptyExec +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] + statement ok DROP TABLE ra; @@ -506,30 +558,27 @@ NULL 3 9 9 5 NULL -# A node above the dropped filter can put a NULL back into the value column, and -# the result is UNKNOWN again. A grouping set rolls `ci.k` up to a NULL row, so -# this join stays null-aware. -query TT -EXPLAIN SELECT co.id, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLLUP(ci.k)) AS m FROM co; +# 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; ---- -logical_plan -01)Projection: co.id, __correlated_sq_1.mark AS m -02)--LeftMark Join: co.k = __correlated_sq_1.k null_aware -03)----TableScan: co projection=[id, k] -04)----SubqueryAlias: __correlated_sq_1 -05)------Projection: ci.k -06)--------Aggregate: groupBy=[[ROLLUP (ci.k)]], aggr=[[]] -07)----------TableScan: ci projection=[k] -physical_plan -01)ProjectionExec: expr=[id@0 as id, mark@1 as m] -02)--HashJoinExec: mode=CollectLeft, join_type=LeftMark, on=[(k@1, k@0)], projection=[id@0, mark@2], null_aware -03)----DataSourceExec: partitions=1, partition_sizes=[1] -04)----ProjectionExec: expr=[k@0 as k] -05)------AggregateExec: mode=FinalPartitioned, gby=[k@0 as k, __grouping_id@1 as __grouping_id], aggr=[] -06)--------RepartitionExec: partitioning=Hash([k@0, __grouping_id@1], 4), input_partitions=1 -07)----------AggregateExec: mode=Partial, gby=[(NULL as k), (k@0 as k)], aggr=[] -08)------------DataSourceExec: partitions=1, partition_sizes=[1] +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; ---- @@ -537,9 +586,9 @@ SELECT co.id, co.k, co.k IN (SELECT ci.k FROM ci WHERE ci.k = co.k GROUP BY ROLL 2 1 true NULL 1 true 2 2 true -NULL 3 NULL -9 9 NULL -5 NULL NULL +NULL 3 false +9 9 false +5 NULL false statement ok DROP TABLE co;