[raghavyadav01]

Which issue does this PR close?

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

Rationale for this change

PullUpCorrelatedExpr::f_down reasoned that "for EXISTS only emptiness matters" and dropped any
Limit with skip=0 and a non-zero fetch out of an EXISTS subquery.

That holds for the emptiness of the Limit's own output. What EXISTS reads is whether the
whole subquery is empty, so the argument breaks as soon as a node above the Limit can be
empty for a reason of its own. On the subquery.slt tables:

SELECT t1_id FROM t1 WHERE EXISTS (
    SELECT 1 FROM (SELECT t2_id FROM t2 WHERE t2_int = t1_int LIMIT 1) s
    HAVING count(*) > 1
);

count(*) over the single row LIMIT 1 keeps can never exceed 1, whichever row that is, so the
HAVING holds for nothing and the query returns no rows. Dropping the Limit lets the aggregate
count the whole scan, and on main this returns 33. A Filter above the Limit goes wrong the
same way.

What changes are included in this PR?

PullUpCorrelatedExpr now tracks, per node on the path from the subquery's root, whether the
emptiness of that node's output follows from the emptiness of its input alone, and only drops the
Limit when that holds of every node above it. The entry is pushed and popped in lockstep with
the existing column_refs_above.

The condition is emptiness, not row count, which matters for how little this rejects. A
Projection, a SubqueryAlias, a Sort, a Window and a Repartition each output one row per
input row. A DISTINCT and a grouped Aggregate output rows exactly when their input has rows,
and an Aggregate without a grouping always outputs its one row — so no Aggregate can be
emptied by the rows a Limit held back. These all keep decorrelating:

EXISTS (SELECT count(*) FROM (SELECT t2_id FROM t2 WHERE t2_id = t1_id LIMIT 1) s)
EXISTS (SELECT DISTINCT s.t2_id FROM (SELECT t2_id FROM t2 WHERE t2_id = t1_id LIMIT 1) s)
EXISTS (SELECT s.t2_id FROM (SELECT t2_id FROM t2 WHERE t2_id = t1_id LIMIT 1) s GROUP BY s.t2_id)
EXISTS (SELECT 1 FROM t2 WHERE t2_id = t1_id LIMIT 1)

Filter is what the predicate leaves out, together with the nodes that can empty a non-empty
input: Join, Union, Unnest. Anything not listed is treated as unsafe, so a new LogicalPlan
variant does not silently become droppable.

LIMIT 0 keeps its existing rewrite to an empty relation, which is sound whatever sits above it,
since both sides are always empty.

The trade-off worth reviewing

Where the Limit now has to stay, the subquery does not decorrelate and the query fails to plan
with Physical plan does not support logical expression Exists(..).

That is the trade subquery.slt already records a few lines above these new tests, for
EXISTS (SELECT * FROM t2 WHERE t2_id = t1_id offset 1), asserted to error "rather than
returning wrong rows" — and main does the same for the equivalent correlated IN subquery with
a LIMIT. So this puts the two broken shapes into an existing, already-asserted bucket rather
than inventing a new behaviour. Happy to go further if you would rather these decorrelate properly
via the row_number() route the issue mentions; that looked like a much larger change and I did
not want to bundle it.

What is the testing strategy for this PR?

datafusion/sqllogictest/test_files/subquery.slt, in the existing EXISTS-with-Limit/offset
block: the two shapes that returned wrong answers, and the four above that must keep working.
The three new assertions all fail on main — the explain shows main dropping the Limit
into a LeftSemi join, and both error assertions fail because main returns rows.

The rule's own unit tests in decorrelate_predicate_subquery.rs pin the decorrelated plan shape
for a Limit under a Projection and under an Aggregate, and that the HAVING and Filter
shapes are left untransformed.

• cargo test -p datafusion-optimizer: 951 pass.
• Full sqllogictests: fails in the same 236 places before and after this change, identical sets.
• cargo clippy --all-targets --workspace --features avro,integration-tests,extended_tests -- -D warnings: clean.
• cargo fmt --all --check, typos, and RUSTDOCFLAGS="-D warnings" cargo doc --document-private-items: clean.

🤖 Generated with Claude Code

[raghavyadav01]

Rebased past #26551 — no conflict; that change is in the Filter arm of PullUpCorrelatedExpr and this one is in Limit. Optimizer tests pass, and the one subquery_sort.slt failure reproduces identically on untouched main.

@jayzhan211 would you have time to look? You merged #26551 and #25781 in this area. The change is narrow: EXISTS currently drops an inner LIMIT unconditionally, which is wrong when a node above the LIMIT can empty the subquery on its own — subquery.slt already asserts the offset 1 case should fail rather than return wrong rows, and this extends that reasoning to fetch.