mohammadnaqvi04 commented on code in PR #25391:
URL: https://github.com/apache/datafusion/pull/25391#discussion_r4321144898
##########
datafusion/optimizer/src/decorrelate_predicate_subquery.rs:
##########
@@ -1063,6 +1089,116 @@ fn build_join(
}))
}
+/// Builds the join for a correlated `EXISTS` subquery whose groupless
+/// aggregate requires count-bug compensation.
+#[expect(clippy::too_many_arguments)]
+fn build_join_with_count_bug(
+ left: &LogicalPlan,
+ sub_query_alias: LogicalPlan,
+ join_filter_opt: Option<Expr>,
+ in_predicate_opt: Option<&Expr>,
+ join_type: JoinType,
+ alias: &str,
+ expr_map: &crate::decorrelate::ExprResultMap,
+ pull_up_having_expr: Option<&Expr>,
+ forces_empty_result: bool,
+) -> Result<LogicalPlan> {
+ if in_predicate_opt.is_some() {
+ return not_impl_err!(
+ "IN/NOT IN count-bug compensation for groupless aggregates is not
implemented"
+ );
+ }
+
+ let joined = LogicalPlanBuilder::from(left.clone())
+ .join_on(sub_query_alias, JoinType::Left, join_filter_opt)?
+ .build()?;
+
+ let left_projection: Vec<Expr> = 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,
+ // (3) the subquery was truncated to zero rows unconditionally
+ // (LIMIT 0), which is false for every outer row regardless of
+ // whether it matched.
+ let exists_expr = if forces_empty_result {
+ lit(false)
+ } else {
+ match having_arm {
+ Some(when_expr) => {
+ when(indicator_col.clone().is_null(), unmatched_having_default)
+ .when(when_expr, lit(false))
+ .otherwise(lit(true))?
+ .rewrite(&mut expr_rewrite)
+ .data()?
+ }
+ None => lit(true),
Review Comment:
Thanks, this is an interesting case. I added a `reads_agg_output` check to
the `compensable` gate so if any join filter column is also a key in the
count-bug expr map, compensation can't tell a HAVING-failure miss from an
absent group, so it falls back to the uncompensated path. Verified against your
repro, now returns only `11`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
