fix: correlated exists/not exists subqueries hit the count bug for groupless aggregates

[mohammadnaqvi04]

Which issue does this PR close?

• Part of Wrong results: correlated EXISTS/IN subqueries over a groupless aggregate get the wrong outer rows #24960.

Rationale for this change

See #24960.

This is a split of #25008, which originally covered EXISTS/NOT EXISTS/IN/NOT IN together. That PR has been split into two smaller PRs between this one for EXISTS/NOT EXISTS and a (to come) follow-up stacked on top of it for IN/NOT IN.

What changes are included in this PR?

Correlated EXISTS/NOT EXISTS subqueries had count-bug compensation available but never turned it on.

• Turn on count-bug compensation for correlated EXISTS/NOT EXISTS subqueries.
• When compensation is needed, decorrelate into a LEFT JOIN and substitute the correct value instead of dropping the row.
• Make unsupported join shapes fail loudly instead of silently falling back to incorrect behavior.

IN/NOT IN compensation is out of scope for this PR and returns an explicit "not implemented" error. This will be covered by the follow-up PR.

What is the testing strategy for this PR?

Added twenty sqllogictest cases in subquery.slt.

Ran all datafusion, datafusion-cli, and datafusion-sqllogictest suites, and ./dev/rust_lint.sh.

Also differentially fuzz-tested against DuckDB with a standalone harness, and did several adversarial review passes with Claude to look for gaps beyond the fuzzer's generated shapes.

Are there any user-facing changes?

None, except in the sense that this fixes a user-visible bug.

AI Usage

AI was used during development of this PR, primarily to help review the test suite and verify test cases against the implemented fix.

[mohammadnaqvi04]

This decorrelates a groupless-aggregate subquery to a LEFT JOIN, grouped by the correlated column, with the unmatched-row NULL replaced by the aggregate's empty-input default (e.g. count(*) → 0, sum(x) -> NULL).
For

SELECT t1.t1_int FROM t1 WHERE EXISTS (
    SELECT count(*) FROM t2 WHERE t1.t1_int = t2.t2_int
);

this is equivalent to

SELECT t1.t1_int
FROM t1
LEFT JOIN (SELECT t2_int, count(*) AS cnt FROM t2 GROUP BY t2_int) AS sq
  ON t1.t1_int = sq.t2_int;

[mohammadnaqvi04]

LIMIT 0 on the subquery collapses it to an EmptyRelation before the count-bug join ever sees it, so without this flag, join-compensation has no way of knowing that the subquery is unconditionally empty and defaults every row to matched. For

SELECT t1.t1_int FROM t1 WHERE EXISTS (
    SELECT count(*) FROM t2 WHERE t1.t1_int = t2.t2_int LIMIT 0
);

t1_int should never appear in the result, since LIMIT 0 empties the subquery regardless of whether t2 has a matching row. Setting forces_empty_result is what makes build_join_with_count_bug return false unconditionally instead of falling back to its usual "matched" default.

[mohammadnaqvi04]

@AdamGS just a heads up, I rebased this on top of #25348 since it touched build_join too. No logic changes on my end. One test assertion needed a --complete regen after the rebase. Ready for another look whenever you get a chance.

[mohammadnaqvi04]

@jayzhan211 tagging you as well, since I noticed you were active in this area. Would appreciate a look when you have time.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 84.03042% with 42 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.75%. Comparing base (a23b89f) to head (aa5a002).
⚠️ Report is 3 commits behind head on main.

Files with missing lines	Patch %	Lines
...on/optimizer/src/decorrelate_predicate_subquery.rs	81.61%	4 Missing and 21 partials ⚠️
datafusion/optimizer/src/decorrelate.rs	85.18%	10 Missing and 6 partials ⚠️
datafusion/expr/src/utils.rs	94.73%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25391      +/-   ##
==========================================
- Coverage   82.75%   82.75%   -0.01%     
==========================================
  Files        1147     1147              
  Lines      449881   450105     +224     
  Branches   449881   450105     +224     
==========================================
+ Hits       372306   372471     +165     
- Misses      54905    54938      +33     
- Partials    22670    22696      +26     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[jayzhan211]

Thanks @mohammadnaqvi04 , here are some suggestions

[jayzhan211]

with_need_handle_count_bug(true) also applies to IN/NOT IN, so every correlated IN over a groupless aggregate now hits the not_impl_err at :1107. That includes max/sum, which main answers correctly.

-- main and DuckDB: 11; this PR: "IN/NOT IN count-bug compensation ... is not implemented"
SELECT t1_id FROM t1
WHERE t1_int + 2 IN (SELECT max(t2_int) FROM t2 WHERE t2.t2_id = t1.t1_id);
-- main and DuckDB: 22, 33, 44; this PR: same error
SELECT t1_id FROM t1
WHERE t1_int NOT IN (SELECT count(*) FROM t2 WHERE t2.t2_id = t1.t1_id);

Keep IN on the old path until the follow-up, drop the not_impl_err branch, and add both queries to subquery.slt

[jayzhan211]

With count-bug handling on, this arm moves a HAVING that is true on empty input out of the plan and into pull_up_having_expr. build_join only applies it on the compensated path. A DISTINCT, GROUP BY or join above the aggregate doesn't carry collected_count_expr_map, so the plan becomes a plain semi join and the HAVING is lost.

-- DuckDB: 33; main: (empty, count bug); this PR: 11, 22, 44 (count(*) = 0 is gone)
SELECT t1_id FROM t1 WHERE EXISTS (
  SELECT DISTINCT c FROM (
    SELECT count(*) AS c FROM t2 WHERE t2.t2_id = t1.t1_id HAVING count(*) = 0
  ) s
);

When compensation can't be used, rerun the rewrite without it (main's behavior). rewrite(exists) also keeps IN on the old path:

let exists = 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)
            .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)?;
let mut count_bug_compensation = pull_up
    .collected_count_expr_map
    .get(&new_plan)
    .filter(|m| !m.is_empty())
    .cloned();
// Only the compensated join applies a HAVING moved into
// `pull_up_having_expr`, so it needs a count map at the root.
let compensable = pull_up.can_pull_up
    && (count_bug_compensation.is_some() || pull_up.pull_up_having_expr.is_none());
if exists && !compensable {
    (new_plan, pull_up) = rewrite(false)?;
    count_bug_compensation = None;
}
if !pull_up.can_pull_up {
    return Ok(None);
}

[jayzhan211]

A HAVING that reads an outer column becomes a join conjunct, so a group that exists but fails it is also a LEFT JOIN miss. The CASE treats every miss as "no group", and with no HAVING left, exists_expr is lit(true).

-- DuckDB and main: 11; this PR: 11, 22, 33, 44 (NOT EXISTS returns nothing)
SELECT t1_id FROM t1 WHERE EXISTS (
  SELECT count(*) FROM t2 WHERE t2.t2_id = t1.t1_id HAVING count(*) = t1.t1_int
);

Add this to the compensable check in build_join (with the fallback suggested at decorrelate.rs:419):

// A miss is only "no group" if the join compares nothing but group keys.
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))
});

[jayzhan211]

The unmatched arm returns the HAVING evaluated on the defaults as-is, which is NULL when it reads sum/max. LeftAnti keeps a row only when NOT exists_expr is true, so a NULL drops it; a mark join's NOT mark is NULL too. NOT EXISTS loses rows whose subquery is empty.

-- DuckDB and main: 11, 22, 33, 44; this PR: 11, 22, 44 (same with `t1_id > 100 OR NOT EXISTS`)
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
);

[jayzhan211]

ScalarSubqueryToJoin also runs this branch, and it returns the empty-input default for unmatched rows without checking pull_up_having_expr. When a stacked filter that the default fails is merged in, the scalar subquery returns the default instead of NULL.

-- DuckDB and main: v is NULL for every row; this PR: v = 0 for t1_id = 33
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;

[jayzhan211]

Suggested change

	if self.pull_up_having_expr.is_some()
	if self.exists_sub_query
	&& self.pull_up_having_expr.is_some()

[jayzhan211]

The wrapped-key alias runs in every pull-up (plain EXISTS, IN, scalar), not just the compensated aggregate. Once the CAST is aliased, the bare column is no longer projected even if another conjunct still reads it. The alias is also unqualified and named only after col.name, so two scalar subqueries on the same column collide.

CREATE TABLE o5(a BIGINT, c INT) AS VALUES (1, 0), (2, 5);
CREATE TABLE i5(b INT) AS VALUES (1), (2);
-- main: 1; this PR: Schema error: No field named __correlated_sq_1.b
SELECT a FROM o5 WHERE EXISTS (SELECT 1 FROM i5 WHERE o5.a = i5.b AND o5.c < i5.b);

CREATE TABLE o6(x BIGINT) AS VALUES (1), (1);
CREATE TABLE i6(id INT, k INT) AS VALUES (1, 1);
-- main: (1, 1, 1) twice; this PR: Ambiguous reference to unqualified field __correlated_group_expr_k
SELECT x,
  (SELECT count(*) FROM i6 WHERE i6.k = o6.x),
  (SELECT sum(id) FROM i6 WHERE i6.k = o6.x)
FROM o6;

Suggest dropping the alias rewrite (collect_subquery_join_exprs, correlated_col_aliases, collides_with_existing) and falling back to the old rewrite when a correlated key is wrapped. The CAST-dedup cases then keep main's answer, which can be a follow-up:

/// Whether every subquery-side operand of `join_filters` is a bare column, so a
/// LEFT JOIN against the subquery grouped by those columns matches at most one
/// group per outer row.
fn subquery_operands_are_bare_columns(
    join_filters: &[Expr],
    subquery_cols: &BTreeSet<Column>,
) -> 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
    })
}

Then in build_join, require it whenever a count map is present:

let subquery_cols: BTreeSet<Column> = 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);

[jayzhan211]

Suggested change

	.with_need_handle_count_bug(true);
	.with_need_handle_count_bug(in_predicate_opt.is_none());

[jayzhan211]

Suggested change

	when(indicator_col.clone().is_null(), unmatched_having_default)
	when(indicator_col.clone().is_null(), unmatched_having_default.is_true())