decorrelate_predicate_subquery removes inner LIMIT inside EXISTS even when a HAVING or WHERE above it depends on the row count #26629

Describe the bug

Describe the bug

In datafusion/optimizer/src/decorrelate.rs:276-309, PullUpCorrelatedExpr::f_down reasons that "for EXISTS
only emptiness matters" and deletes any Limit with SkipType::Literal(0) and non-zero fetch:

match (limit.get_skip_type()?, fetch) {
    (SkipType::Literal(0), FetchType::Literal(_)) => {
        self.column_refs_above.pop();
        let mut t = self.f_down((*limit.input).clone())?;
        t.transformed = true;
        Ok(t)
    }
    _ => self.stop_pull_up(plan),
}

Removing LIMIT k preserves emptiness only when the Limit is the outermost operator of the EXISTS
subquery. When the LIMIT sits below an Aggregate (HAVING COUNT(*) > 1) or a Filter
(WHERE sub.v = 200), deleting it changes the rows seen by the enclosing operator and the result is no longer
equivalent.

To Reproduce

CREATE TABLE a(k INT, x INT) AS VALUES (1, 10), (2, 20);
CREATE TABLE b(k INT, v INT) AS VALUES (1, 100), (1, 200), (2, 300);

-- Case 1: LIMIT 1 below HAVING COUNT(*) > 1 (can never be > 1 -> must return 0 rows)
SELECT a.k FROM a
WHERE EXISTS (
    SELECT 1 FROM (SELECT b.k, b.v FROM b WHERE b.k = a.k LIMIT 1) sub
    HAVING COUNT(*) > 1
);

-- Case 2: LIMIT 1 below WHERE sub.v = 200 (first row for k=1 has v=100 -> must return 0 rows)
SELECT a.k FROM a
WHERE EXISTS (
    SELECT 1 FROM (SELECT b.k, b.v FROM b WHERE b.k = a.k LIMIT 1) sub
    WHERE sub.v = 200
);

Expected behavior

• Case 1: expected []; DataFusion 55.0.0 returns [(1,)].
• Case 2: expected []; DataFusion 55.0.0 returns [(1,)].

The optimized plan shows the inner Limit gone while HAVING COUNT(*) > 1 / Filter b.v = 200 remain.

Additional context

Root cause

decorrelate.rs:276-309 applies the "emptiness only" argument without checking the position of the Limit
relative to aggregates and filters. The condition must be: the Limit is the top-level operator of the EXISTS
subquery (or every operator between the Limit and the EXISTS root is row-count-preserving and cannot observe
the difference).

Suggested fix

Only strip LIMIT when it is the root of the EXISTS subquery (possibly under row-preserving
Projection/SubqueryAlias). Otherwise stop pull-up (stop_pull_up) and decorrelate using the
row_number()-based limit decorrelation already implemented for correlated LIMIT (as used for scalar
subqueries). Also handle LIMIT 0 explicitly: EXISTS (... LIMIT 0) is always false and must not return rows.

Willingness to contribute

None