diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs index 777e8350782a..bf7976ca8999 100644 --- a/datafusion/expr/src/utils.rs +++ b/datafusion/expr/src/utils.rs @@ -1574,6 +1574,38 @@ pub fn collect_subquery_cols( }) } +/// Determine the subquery-side expressions a join filter compares, +/// stopping at the first subquery-referencing sub-expression rather than +/// descending to the bare columns inside of it. +/// +/// `t1.a = t2.b` yields `t2.b`, matching [`collect_subquery_cols`]. But +/// `t1.a = CAST(t2.b AS INT)` yields `CAST(t2.b AS INT)` whole, not `t2.b`. +/// The two differ whenever the correlated predicate wraps the subquery's +/// column in an expression, in which case we need to evaluate the full +/// expression before grouping, so that two column values which compare +/// equal after the wrapping expression end up in the same group. +pub fn collect_subquery_join_exprs( + exprs: &[Expr], + subquery_schema: &DFSchema, +) -> Result> { + let mut result: Vec = vec![]; + for expr in exprs { + expr.apply(|e| { + if e.column_refs() + .into_iter() + .all(|col| subquery_schema.has_column(col)) + && !e.column_refs().is_empty() + && !result.contains(e) + { + result.push(e.clone()); + return Ok(TreeNodeRecursion::Jump); + } + Ok(TreeNodeRecursion::Continue) + })?; + } + Ok(result) +} + #[cfg(test)] mod tests { use super::*; diff --git a/datafusion/optimizer/src/decorrelate.rs b/datafusion/optimizer/src/decorrelate.rs index 0a1db9b2128c..7dcf8fde935e 100644 --- a/datafusion/optimizer/src/decorrelate.rs +++ b/datafusion/optimizer/src/decorrelate.rs @@ -426,6 +426,50 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { None }; + if self.exists_sub_query + && self.pull_up_having_expr.is_some() + && let Some(expr) = conjunction(subquery_filters.clone()) + { + let unqualified_expr = expr + .transform_up(|e| { + if let Expr::Column(Column { name, .. }) = &e { + Ok(Transformed::yes(Expr::Column( + Column::new_unqualified(name), + ))) + } else { + Ok(Transformed::no(e)) + } + }) + .data()?; + let combined = match self.pull_up_having_expr.take() { + Some(existing) => existing.and(unqualified_expr), + None => unqualified_expr, + }; + self.pull_up_having_expr = Some(combined); + let new_plan = + LogicalPlanBuilder::from((*plan_filter.input).clone()).build()?; + let mut carried_correlated_cols = correlated_subquery_cols; + if let Some(existing) = + self.correlated_subquery_cols_map.get(&*plan_filter.input) + { + carried_correlated_cols.extend(existing.iter().cloned()); + } + self.correlated_subquery_cols_map + .insert(new_plan.clone(), carried_correlated_cols); + if !expr_result_map_for_count_bug.is_empty() { + self.collected_count_expr_map + .insert(new_plan.clone(), expr_result_map_for_count_bug); + } else if let Some(input_map) = self + .collected_count_expr_map + .get(&*plan_filter.input) + .cloned() + { + self.collected_count_expr_map + .insert(new_plan.clone(), input_map); + } + return Ok(Transformed::yes(new_plan)); + } + match (&pull_up_expr_opt, &self.pull_up_having_expr) { (Some(_), Some(_)) => { // Error path @@ -464,15 +508,18 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { &mut local_correlated_cols, ); // add missing columns to Projection - let mut missing_exprs = - self.collect_missing_exprs(&projection.expr, &local_correlated_cols)?; + let mut missing_exprs = self.collect_missing_exprs( + &projection.expr, + &local_correlated_cols, + projection.input.schema(), + )?; let mut expr_result_map_for_count_bug = HashMap::new(); if let Some(expr_result_map) = self.collected_count_expr_map.get(&*projection.input) { proj_exprs_evaluation_result_on_empty_batch( - &projection.expr, + &missing_exprs, projection.input.schema(), expr_result_map, &mut expr_result_map_for_count_bug, @@ -511,7 +558,6 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { &self.correlated_subquery_cols_map, &mut local_correlated_cols, ); - // A grouping set that omits a column the pull up adds either // keeps its sets as they are (every set already groups by the // column), or gains the column in every set, `ROLLUP(i.k)` @@ -544,6 +590,7 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { self.collect_missing_exprs( &aggregate.group_expr, &local_correlated_cols, + aggregate.input.schema(), )? } GroupingSetsPullUp::Unsafe => { @@ -558,6 +605,7 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { self.collect_missing_exprs( &aggregate.group_expr, &local_correlated_cols, + aggregate.input.schema(), )? } } @@ -566,6 +614,7 @@ impl TreeNodeRewriter for PullUpCorrelatedExpr { self.collect_missing_exprs( &aggregate.group_expr, &local_correlated_cols, + aggregate.input.schema(), )? }; @@ -800,9 +849,10 @@ impl PullUpCorrelatedExpr { } fn collect_missing_exprs( - &self, + &mut self, exprs: &[Expr], correlated_subquery_cols: &BTreeSet, + _subquery_schema: &DFSchemaRef, ) -> Result> { let mut missing_exprs = Vec::with_capacity(exprs.len() + correlated_subquery_cols.len()); @@ -813,19 +863,22 @@ impl PullUpCorrelatedExpr { } for col in correlated_subquery_cols.iter() { let col_expr = Expr::Column(col.clone()); - if !missing_exprs.contains(&col_expr) { + if !collides_with_existing(&missing_exprs, &col_expr) { missing_exprs.push(col_expr) } } if let Some(pull_up_having) = &self.pull_up_having_expr { - let filter_apply_columns = pull_up_having.column_refs(); - for col in filter_apply_columns { - // add to missing_exprs if not already there - let contains = missing_exprs + for col in pull_up_having.column_refs() { + let col_expr = Expr::Column(col.clone()); + // `col` is unqualified but the projection may already have + // it as `agg.c`. Check by bare name so that doesn't get + // added twice. + let already_present = missing_exprs .iter() - .any(|expr| matches!(expr, Expr::Column(c) if c == col)); - if !contains { - missing_exprs.push(Expr::Column(col.clone())) + .any(|expr| matches!(expr, Expr::Column(c) if c.name == col.name)) + || collides_with_existing(&missing_exprs, &col_expr); + if !already_present { + missing_exprs.push(col_expr) } } } @@ -925,6 +978,16 @@ fn partitions_by_all(expr: &Expr, cols: &BTreeSet) -> bool { }) } +/// True if `candidate` collides with an existing entry's schema name, +/// e.g. a bare column already wrapped in an equally-named `CAST`. +/// Colliding means `LogicalPlanBuilder::project` would reject it as a duplicate. +fn collides_with_existing(exprs: &[Expr], candidate: &Expr) -> bool { + let candidate_name = candidate.schema_name().to_string(); + exprs + .iter() + .any(|expr| expr.schema_name().to_string() == candidate_name) +} + fn can_pullup_over_aggregation(expr: &Expr) -> bool { if let Expr::BinaryExpr(BinaryExpr { left, @@ -979,7 +1042,7 @@ fn remove_duplicated_filter( Expr::BinaryExpr(b) => b.op.swap() == Some(b.op), _ => true, }, - "remove_duplicated_filter: in_predicate must use a commutative operator" + "in_predicate must use a commutative operator" ); Ok(filters @@ -1105,9 +1168,14 @@ fn filter_exprs_evaluation_result_on_empty_batch( let simplifier = ExprSimplifier::new(info); let result_expr = simplifier.simplify(result_expr)?; match &result_expr { - // evaluate to false or null on empty batch, no need to pull up Expr::Literal(ScalarValue::Null, _) - | Expr::Literal(ScalarValue::Boolean(Some(false)), _) => None, + | Expr::Literal(ScalarValue::Boolean(None), _) + | Expr::Literal(ScalarValue::Boolean(Some(false)), _) => { + for (name, exprs) in input_expr_result_map_for_count_bug { + expr_result_map_for_count_bug.insert(name.clone(), exprs.clone()); + } + None + } // evaluate to true on empty batch, need to pull up the expr Expr::Literal(ScalarValue::Boolean(Some(true)), _) => { for (name, exprs) in input_expr_result_map_for_count_bug { diff --git a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs index 99b57a5494aa..9c20108da261 100644 --- a/datafusion/optimizer/src/decorrelate_predicate_subquery.rs +++ b/datafusion/optimizer/src/decorrelate_predicate_subquery.rs @@ -20,14 +20,17 @@ use std::collections::BTreeSet; use std::ops::Deref; use std::sync::Arc; -use crate::decorrelate::PullUpCorrelatedExpr; +use crate::analyzer::type_coercion::TypeCoercionRewriter; +use crate::decorrelate::{PullUpCorrelatedExpr, UN_MATCHED_ROW_INDICATOR}; use crate::extract_equijoin_predicate::split_eq_and_noneq_join_predicate; use crate::optimizer::ApplyOrder; use crate::utils::replace_qualified_name; use crate::{OptimizerConfig, OptimizerRule}; use datafusion_common::alias::AliasGenerator; -use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; +use datafusion_common::tree_node::{ + Transformed, TransformedResult, TreeNode, TreeNodeRecursion, +}; use datafusion_common::{ Column, DFSchema, NullEquality, Result, ScalarValue, assert_or_internal_err, internal_datafusion_err, internal_err, not_impl_err, plan_err, @@ -760,6 +763,30 @@ fn with_null_aware_value_keys( )) } +fn subquery_operands_are_bare_columns( + join_filters: &[Expr], + subquery_cols: &BTreeSet, +) -> bool { + join_filters.iter().all(|filter| { + let mut bare = true; + filter + .apply(|e| { + let refs = e.column_refs(); + if !refs.is_empty() && refs.iter().all(|c| subquery_cols.contains(*c)) { + bare = matches!(e, Expr::Column(_)); + return Ok(if bare { + TreeNodeRecursion::Jump + } else { + TreeNodeRecursion::Stop + }); + } + Ok(TreeNodeRecursion::Continue) + }) + .expect("infallible"); + bare + }) +} + /// The outcome of [`build_join`]. struct BuiltJoin { /// The outer plan with the subquery joined into it. @@ -777,11 +804,48 @@ fn build_join( alias: &str, needs_null_aware_mark: bool, ) -> Result> { - let mut pull_up = PullUpCorrelatedExpr::new() - .with_in_predicate_opt(in_predicate_opt.cloned()) - .with_exists_sub_query(in_predicate_opt.is_none()); - - let new_plan = subquery.clone().rewrite(&mut pull_up).data()?; + let exists_sub_query = in_predicate_opt.is_none(); + let rewrite = + |handle_count_bug: bool| -> Result<(LogicalPlan, PullUpCorrelatedExpr)> { + let mut pull_up = PullUpCorrelatedExpr::new() + .with_in_predicate_opt(in_predicate_opt.cloned()) + .with_exists_sub_query(exists_sub_query) + .with_need_handle_count_bug(handle_count_bug); + let new_plan = subquery.clone().rewrite(&mut pull_up).data()?; + Ok((new_plan, pull_up)) + }; + let (mut new_plan, mut pull_up) = rewrite(exists_sub_query)?; + + let mut count_bug_compensation = pull_up + .collected_count_expr_map + .get(&new_plan) + .filter(|m| !m.is_empty()) + .cloned(); + // A HAVING-failure miss looks identical to an absent group once compensated, so bail if the join filter reads the aggregate's own output. + let reads_agg_output = count_bug_compensation.as_ref().is_some_and(|expr_map| { + pull_up + .join_filters + .iter() + .flat_map(|f| f.column_refs()) + .any(|c| expr_map.contains_key(&c.name)) + }); + let subquery_cols: BTreeSet = pull_up + .correlated_subquery_cols_map + .values() + .flatten() + .cloned() + .collect(); + let keys_are_bare = count_bug_compensation.is_none() + || subquery_operands_are_bare_columns(&pull_up.join_filters, &subquery_cols); + // Fall back to the uncompensated rewrite whenever compensation can't safely apply, so a HAVING never gets silently dropped. + let compensable = pull_up.can_pull_up + && !reads_agg_output + && keys_are_bare + && (count_bug_compensation.is_some() || pull_up.pull_up_having_expr.is_none()); + if exists_sub_query && !compensable { + (new_plan, pull_up) = rewrite(false)?; + count_bug_compensation = None; + } if !pull_up.can_pull_up { return Ok(None); } @@ -801,6 +865,26 @@ fn build_join( replace_qualified_name(filter, &all_correlated_cols, alias).map(Some) })?; + if let Some(expr_map) = count_bug_compensation + && !expr_map.is_empty() + { + return build_join_with_count_bug( + left, + sub_query_alias, + join_filter_opt, + join_type, + alias, + &expr_map, + pull_up.pull_up_having_expr.as_ref(), + ) + .map(|plan| { + Some(BuiltJoin { + plan, + mark_is_three_valued_exact: false, + }) + }); + } + // 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. A // multi-column `(a, b) IN (SELECT x, y ...)` has one pair per tuple @@ -1059,6 +1143,116 @@ fn build_join( })) } +/// Builds the join for a correlated `EXISTS` subquery whose groupless +/// aggregate requires count-bug compensation. +fn build_join_with_count_bug( + left: &LogicalPlan, + sub_query_alias: LogicalPlan, + join_filter_opt: Option, + join_type: JoinType, + alias: &str, + expr_map: &crate::decorrelate::ExprResultMap, + pull_up_having_expr: Option<&Expr>, +) -> Result { + let pull_up_having_expr = pull_up_having_expr + .map(|f| { + f.clone() + .transform_up(|e| match e { + Expr::Column(c) => Ok(Transformed::yes(Expr::Column(Column::new( + Some(alias), + c.name, + )))), + _ => Ok(Transformed::no(e)), + }) + .data() + }) + .transpose()?; + let pull_up_having_expr = pull_up_having_expr.as_ref(); + + let joined = LogicalPlanBuilder::from(left.clone()) + .join_on(sub_query_alias, JoinType::Left, join_filter_opt)? + .build()?; + + let left_projection: Vec = left + .schema() + .columns() + .into_iter() + .map(Expr::from) + .collect(); + + let indicator_col = Expr::Column(Column::new(Some(alias), UN_MATCHED_ROW_INDICATOR)); + + let having_arm = pull_up_having_expr.map(|f| f.clone().is_not_true()); + + let mut expr_rewrite = TypeCoercionRewriter::new(joined.schema()); + + // The subquery's HAVING clause, evaluated against an unmatched row's + // default aggregate values (e.g. count(*) defaults to 0, sum(x) to + // NULL), is usually `true`, but not once a second filter is combined + // into it, so it cannot be assumed. + let unmatched_having_default = match pull_up_having_expr { + Some(f) => f + .clone() + .transform_up(|e| { + if let Expr::Column(Column { name, .. }) = &e + && let Some(default_value) = expr_map.get(name) + { + return Ok(Transformed::yes(default_value.clone())); + } + Ok(Transformed::no(e)) + }) + .data() + .and_then(|simplified| simplified.rewrite(&mut expr_rewrite).data())?, + None => lit(true), + }; + + // EXISTS is true by default (the groupless aggregate always + // produces a row), unless one of three cases holds: + // (1) the outer row joined against an inner group whose HAVING + // predicate failed, + // (2) an unmatched outer row's own default aggregate values fail + // that same HAVING clause. + let exists_expr = match having_arm { + Some(when_expr) => when( + indicator_col.clone().is_null(), + unmatched_having_default.is_true(), + ) + .when(when_expr, lit(false)) + .otherwise(lit(true))? + .rewrite(&mut expr_rewrite) + .data()?, + None => lit(true), + }; + + let new_plan = match join_type { + JoinType::LeftMark | JoinType::RightMark => { + let mut proj_exprs = left_projection; + proj_exprs.push(exists_expr.alias_qualified(Some(alias), "mark")); + LogicalPlanBuilder::from(joined) + .project(proj_exprs)? + .build()? + } + JoinType::LeftAnti => LogicalPlanBuilder::from(joined) + .filter(not(exists_expr))? + .project(left_projection)? + .build()?, + JoinType::LeftSemi => LogicalPlanBuilder::from(joined) + .filter(exists_expr)? + .project(left_projection)? + .build()?, + _ => { + return internal_err!("unsupported join type {join_type:?}"); + } + }; + + debug!( + "predicate subquery (count bug) optimized:\n{}", + new_plan.display_indent() + ); + + Ok(new_plan) +} + #[derive(Debug)] struct SubqueryInfo { query: Subquery, @@ -1111,6 +1305,7 @@ mod tests { TableScanBuilder, and, binary_expr, col, cube, grouping_set, out_ref_col, rollup, table_scan, }; + use datafusion_functions_aggregate::count::count_udaf; macro_rules! assert_optimized_plan_equal { ( @@ -1438,7 +1633,7 @@ mod tests { SubqueryAlias: __correlated_sq_2 [o_custkey:Int64] Projection: orders.o_custkey [o_custkey:Int64] TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N] - " + " ) } @@ -1666,7 +1861,7 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" + @" Projection: customer.c_custkey [c_custkey:Int64] LeftSemi Join: Filter: customer.c_custkey = __correlated_sq_1.o_custkey AND (customer.c_custkey = __correlated_sq_1.o_custkey OR __correlated_sq_1.o_orderkey = Int32(1)) [c_custkey:Int64, c_name:Utf8] TableScan: customer [c_custkey:Int64, c_name:Utf8] @@ -2722,7 +2917,7 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" + @" Projection: customer.c_custkey [c_custkey:Int64] LeftSemi Join: Filter: customer.c_custkey = __correlated_sq_1.o_custkey OR __correlated_sq_1.o_orderkey = Int32(1) [c_custkey:Int64, c_name:Utf8] TableScan: customer [c_custkey:Int64, c_name:Utf8] @@ -2855,6 +3050,37 @@ mod tests { ) } + /// Test for correlated exists subquery filter with disjunction and count bug + #[test] + fn exists_subquery_disjunction_with_count_bug() -> Result<()> { + let sq = Arc::new( + LogicalPlanBuilder::from(test_table_scan_with_name("sq")?) + .filter(out_ref_col(DataType::UInt32, "test.a").eq(col("sq.a")))? + .aggregate(Vec::::new(), vec![count_udaf().call(vec![])])? + .build()?, + ); + + let plan = LogicalPlanBuilder::from(test_table_scan()?) + .filter(exists(sq).or(col("test.c").eq(lit(1))))? + .project(vec![col("test.b")])? + .build()?; + + assert_optimized_plan_equal!( + plan, + @r" + Projection: test.b [b:UInt32] + Projection: test.a, test.b, test.c [a:UInt32, b:UInt32, c:UInt32] + Filter: __correlated_sq_1.mark OR test.c = Int32(1) [a:UInt32, b:UInt32, c:UInt32, mark:Boolean] + Projection: test.a, test.b, test.c, Boolean(true) AS mark [a:UInt32, b:UInt32, c:UInt32, mark:Boolean] + Left Join: Filter: test.a = __correlated_sq_1.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, __always_true:Boolean;N, count():Int64;N] + TableScan: test [a:UInt32, b:UInt32, c:UInt32] + SubqueryAlias: __correlated_sq_1 [a:UInt32, __always_true:Boolean, count():Int64] + Aggregate: groupBy=[[sq.a, Boolean(true) AS __always_true]], aggr=[[count()]] [a:UInt32, __always_true:Boolean, count():Int64] + TableScan: sq [a:UInt32, b:UInt32, c:UInt32] + " + ) + } + /// Test for correlated EXISTS subquery filter #[test] fn exists_subquery_correlated() -> Result<()> { @@ -2987,7 +3213,7 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" + @" Projection: test.b [b:UInt32] LeftSemi Join: Filter: UInt32(1) + __correlated_sq_1.a > test.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32] TableScan: test [a:UInt32, b:UInt32, c:UInt32] @@ -3046,7 +3272,7 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" + @" Projection: test.b [b:UInt32] LeftSemi Join: Filter: UInt32(1) + __correlated_sq_1.a > test.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32] TableScan: test [a:UInt32, b:UInt32, c:UInt32] @@ -3077,7 +3303,7 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" + @" Projection: test.b [b:UInt32] LeftSemi Join: Filter: UInt32(1) + __correlated_sq_1.a > test.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32] TableScan: test [a:UInt32, b:UInt32, c:UInt32] @@ -3108,7 +3334,7 @@ mod tests { assert_optimized_plan_equal!( plan, - @r" + @" Projection: test.b [b:UInt32] LeftSemi Join: Filter: UInt32(1) + __correlated_sq_1.a > test.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32] TableScan: test [a:UInt32, b:UInt32, c:UInt32] diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 39b3961c8245..3cd3bfbf193b 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/datafusion/sqllogictest/test_files/subquery.slt @@ -1294,6 +1294,19 @@ select t1.t1_int from t1 where ( 2 4 +# correlated_scalar_subquery_having_default_not_coerced_to_match_stacked_filter +query II rowsort +SELECT t1_id, ( + SELECT c FROM ( + SELECT count(*) AS c FROM t2 WHERE t2.t2_id = t1.t1_id HAVING count(*) < 10 + ) s WHERE c > 5 +) AS v FROM t1 +---- +11 NULL +22 NULL +33 NULL +44 NULL + #correlated_scalar_subquery_count_agg_in_case_when query TT explain @@ -1857,6 +1870,275 @@ item1 1970-01-01T00:00:03 75 statement ok drop table source_table; + + +# correlated_exists_groupless_count_agg_count_bug +query II rowsort +SELECT t1_id, t1_int FROM t1 WHERE EXISTS (SELECT count(*) FROM t2 WHERE t1.t1_id = t2.t2_id) +---- +11 1 +22 2 +33 3 +44 4 + +# correlated_exists_groupless_count_agg_count_bug_having +query TT +explain SELECT t1_id, t1_int FROM t1 WHERE EXISTS (SELECT count(*) FROM t2 WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0) +---- +logical_plan +01)Projection: t1.t1_id, t1.t1_int +02)--Filter: CASE WHEN __correlated_sq_1.__always_true IS NULL THEN Boolean(true) WHEN __correlated_sq_1.count(Int64(1)) = Int64(0) IS NOT TRUE THEN Boolean(false) ELSE Boolean(true) END +03)----Projection: t1.t1_id, t1.t1_int, __correlated_sq_1.count(Int64(1)), __correlated_sq_1.__always_true +04)------Left Join: t1.t1_id = __correlated_sq_1.t2_id +05)--------TableScan: t1 projection=[t1_id, t1_int] +06)--------SubqueryAlias: __correlated_sq_1 +07)----------Projection: t2.t2_id, count(Int64(1)), Boolean(true) AS __always_true +08)------------Aggregate: groupBy=[[t2.t2_id]], aggr=[[count(Int64(1))]] +09)--------------TableScan: t2 projection=[t2_id] + +query II +SELECT t1_id, t1_int FROM t1 WHERE EXISTS (SELECT count(*) FROM t2 WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0) +---- +33 3 + +# correlated_exists_groupless_compound_having_select_dummy_column +query II +SELECT t1_id, t1_int FROM t1 WHERE EXISTS (SELECT 1 FROM t2 WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0 AND count(*) < 100) +---- +33 3 + +# correlated_not_exists_groupless_compound_having_select_dummy_column +query II rowsort +SELECT t1_id, t1_int FROM t1 WHERE NOT EXISTS (SELECT 1 FROM t2 WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0 AND count(*) < 100) +---- +11 1 +22 2 +44 4 + +# correlated_exists_groupless_having_reads_outer_column_falls_back +query I +SELECT t1_id FROM t1 WHERE EXISTS ( + SELECT count(*) FROM t2 WHERE t2.t2_id = t1.t1_id HAVING count(*) = t1.t1_int +) +---- +11 + +# correlated_exists_groupless_having_with_stacked_filter_via_cte +query II +SELECT t1_id, t1_int FROM t1 WHERE EXISTS ( + WITH agg AS (SELECT count(*) AS c FROM t2 WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0) + SELECT c FROM agg WHERE c >= 0 +) +---- +33 3 + +# correlated_not_exists_groupless_having_with_stacked_filter_via_cte +query II rowsort +SELECT t1_id, t1_int FROM t1 WHERE NOT EXISTS ( + WITH agg AS (SELECT count(*) AS c FROM t2 WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0) + SELECT c FROM agg WHERE c >= 0 +) +---- +11 1 +22 2 +44 4 + +# correlated_exists_groupless_having_with_stacked_filter_on_different_aggregate +query II +SELECT t1_id, t1_int FROM t1 WHERE EXISTS ( + WITH agg AS (SELECT count(*) AS c, sum(t2_int) AS s FROM t2 WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0) + SELECT c FROM agg WHERE s > 5 +) +---- + +# correlated_not_exists_groupless_count_agg_count_bug +query II +SELECT t1_id, t1_int FROM t1 WHERE NOT EXISTS (SELECT count(*) FROM t2 WHERE t1.t1_id = t2.t2_id) +---- + +# correlated_exists_groupless_null_default_agg_always_true +query II rowsort +SELECT t1_id, t1_int FROM t1 WHERE EXISTS (SELECT sum(t2_int) FROM t2 WHERE t1.t1_id = t2.t2_id) +---- +11 1 +22 2 +33 3 +44 4 + +# correlated_not_exists_groupless_null_default_agg_always_true +query II +SELECT t1_id, t1_int FROM t1 WHERE NOT EXISTS (SELECT sum(t2_int) FROM t2 WHERE t1.t1_id = t2.t2_id) +---- + +# correlated_not_exists_groupless_null_default_having_is_null_not_false +query I rowsort +SELECT t1_id FROM t1 WHERE NOT EXISTS ( + SELECT c FROM ( + SELECT count(*) AS c, sum(t2_int) AS s FROM t2 + WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0 + ) agg WHERE s > 5 +) +---- +11 +22 +33 +44 + +# correlated_not_exists_groupless_null_default_having_is_null_not_false_embedded_in_or +query I rowsort +SELECT t1_id FROM t1 WHERE t1_id > 100 OR NOT EXISTS ( + SELECT c FROM ( + SELECT count(*) AS c, sum(t2_int) AS s FROM t2 + WHERE t1.t1_id = t2.t2_id HAVING count(*) = 0 + ) agg WHERE s > 5 +) +---- +11 +22 +33 +44 + +# correlated_exists_groupless_count_agg_count_bug_embedded_in_or +query I rowsort +select t1_id from t1 +where t1_id > 40 or exists (select count(*) from t2 where t1.t1_id = t2.t2_id having count(*) = 0) +---- +33 +44 + +# correlated_not_exists_groupless_count_agg_count_bug_embedded_in_or +query I rowsort +select t1_id from t1 +where t1_id > 40 or not exists (select count(*) from t2 where t1.t1_id = t2.t2_id having count(*) = 0) +---- +11 +22 +44 + +# correlated_exists_groupless_null_default_agg_count_bug_embedded_in_or +query I rowsort +select t1_id from t1 +where t1_id > 40 or exists (select sum(t2_int) from t2 where t1.t1_id = t2.t2_id having sum(t2_int) is null) +---- +33 +44 + +# correlated_not_exists_groupless_null_default_agg_count_bug_embedded_in_or +query I rowsort +select t1_id from t1 +where t1_id > 40 or not exists (select sum(t2_int) from t2 where t1.t1_id = t2.t2_id having sum(t2_int) is null) +---- +11 +22 +44 + +statement ok +CREATE TABLE dedup_outer(k INT) AS VALUES (1), (2); + +statement ok +CREATE TABLE dedup_inner(k TEXT, v INT) AS VALUES ('1', 10), ('1', 20), ('01', 30), ('01', 40); + +# TODO(#26014): should match DuckDB (1, 2), not main's answer. +# correlated_exists_groupless_count_agg_duplicate_match_dedup +query I rowsort +SELECT o.k FROM dedup_outer o WHERE EXISTS ( + SELECT count(*) FROM dedup_inner i WHERE CAST(i.k AS INT) = o.k +) +---- +1 + +# TODO(#26014): should match DuckDB, not main's answer. +# correlated_not_exists_groupless_count_agg_duplicate_match_dedup +query I rowsort +SELECT o.k FROM dedup_outer o WHERE NOT EXISTS ( + SELECT count(*) FROM dedup_inner i WHERE CAST(i.k AS INT) = o.k HAVING count(*) = 0 +) +---- +1 +2 + +statement ok +DROP TABLE dedup_outer; + +statement ok +DROP TABLE dedup_inner; + +statement ok +CREATE TABLE cte_dedup_outer(k INT) AS VALUES (1), (2), (3); + +statement ok +CREATE TABLE cte_dedup_inner(k TEXT, v INT) AS VALUES ('1', 10), ('01', 20), ('2', 30); + +# TODO(#26014): should match DuckDB, not main's answer. +# correlated_exists_groupless_count_agg_duplicate_match_dedup_via_cte +query I +SELECT o.k FROM cte_dedup_outer o WHERE EXISTS ( + WITH agg AS (SELECT count(*) AS c FROM cte_dedup_inner i WHERE CAST(i.k AS INT) = o.k) + SELECT c FROM agg WHERE c = 2 +) +---- + +# TODO(#26014): should match DuckDB, not main's answer. +# correlated_not_exists_groupless_count_agg_duplicate_match_dedup_via_cte +query I rowsort +SELECT o.k FROM cte_dedup_outer o WHERE NOT EXISTS ( + WITH agg AS (SELECT count(*) AS c FROM cte_dedup_inner i WHERE CAST(i.k AS INT) = o.k) + SELECT c FROM agg WHERE c = 2 +) +---- +1 +2 +3 + +statement ok +DROP TABLE cte_dedup_outer; + +statement ok +DROP TABLE cte_dedup_inner; + +statement ok +CREATE TABLE limit_zero_outer(k INT) AS VALUES (1); + +statement ok +CREATE TABLE limit_zero_inner(k INT, v INT) AS VALUES (1, NULL); + +# correlated_exists_groupless_count_agg_limit_zero +query I +SELECT o.k FROM limit_zero_outer o WHERE EXISTS ( + SELECT count(*) FROM limit_zero_inner i WHERE i.k = o.k LIMIT 0 +) +---- + +# correlated_not_exists_groupless_count_agg_limit_zero +query I +SELECT o.k FROM limit_zero_outer o WHERE NOT EXISTS ( + SELECT count(*) FROM limit_zero_inner i WHERE i.k = o.k LIMIT 0 +) +---- +1 + +# correlated_in_groupless_max_agg_is_not_count_bug_compensated +query I +SELECT t1_id FROM t1 +WHERE t1_int + 2 IN (SELECT max(t2_int) FROM t2 WHERE t2.t2_id = t1.t1_id) +---- +11 + +# correlated_not_in_groupless_count_agg_is_not_count_bug_compensated +query I rowsort +SELECT t1_id FROM t1 +WHERE t1_int NOT IN (SELECT count(*) FROM t2 WHERE t2.t2_id = t1.t1_id) +---- +22 +33 +44 + +statement ok +DROP TABLE limit_zero_outer; + +statement ok +DROP TABLE limit_zero_inner; + statement count 0 drop table t1;