[SPARK-58385][SQL] Do not strip a GlobalLimit whose cap is not statically known in LimitPushDown - #58065

What changes were proposed in this pull request?

LimitPushDown.maybePushLocalLimit decides whether a pushed-down limit may replace a child's existing limit by inspecting plan.maxRowsPerPartition:

private def maybePushLocalLimit(limitExp: Expression, plan: LogicalPlan): LogicalPlan = {
  (limitExp, plan.maxRowsPerPartition) match {
    case (IntegerLiteral(newLimit), Some(childMaxRows)) if newLimit < childMaxRows =>
      LocalLimit(limitExp, stripGlobalLimitIfPresent(plan))

    case (_, None) =>
      LocalLimit(limitExp, stripGlobalLimitIfPresent(plan))   // <- here

    case _ =>
      plan
  }
}

The (_, None) arm read None as "the child has no cap" and stripped a child GlobalLimit unconditionally. But None also means "the cap is not statically known".

This PR changes that arm to push the new LocalLimit without stripping the child GlobalLimit. Stripping stays in the literal arm, where the cap is known and the newLimit < childMaxRows guard holds.

Note that a GlobalLimit with a literal limit always reports Some(cap) and so can only reach the literal arm. The stripGlobalLimitIfPresent call removed here was therefore reachable only for the incorrect case, and no correct plan change is lost.

Why are the changes needed?

This is a correctness bug present since Spark 2.0.0.

GlobalLimit.maxRows recognizes only IntegerLiteral, and LimitPushDown is listed before ConstantFolding in the same operator-optimization rule set. So a limit written as a foldable expression (LIMIT 2+3) is still Add(2, 3) on the first fixed-point pass, maxRowsPerPartition returns None, and the rule deletes the semantic GlobalLimit — replacing a global row cap with a per-partition one.

Foldable non-literal LIMIT expressions are valid SQL: checkLimitLikeClause accepts any foldable integer expression. They arise naturally from templated or generated SQL that renders arithmetic into the query text, e.g. a pagination template producing LIMIT 20*5.

Repro:

CREATE TABLE la USING parquet AS SELECT 1 AS id;
CREATE TABLE lb USING parquet AS SELECT id FROM range(0, 40, 1, 4);

SELECT count(*) FROM (SELECT * FROM la CROSS JOIN (SELECT * FROM lb LIMIT 2+3) LIMIT 100);
-- before: 10   (wrong)
-- after:   5

The larger outer limit strips the nested global limit during its pushdown, so the inner LIMIT 5 becomes a per-partition cap over lb's 4 partitions. Manually folding the inner limit to 5 produces the correct answer, which is what makes this specific to the unfolded form.

This is related to SPARK-57956 in that both are about incorrect limit stripping, but the underlying causes differ: SPARK-57956 was LogicalQueryStage.maxRows surfacing a cost estimate as a hard bound, this one is None being read as "uncapped" rather than "unknown".

Does this PR introduce any user-facing change?

Yes, it fixes the wrong results shown above. Queries whose plans contain a GlobalLimit with a foldable non-literal limit expression underneath a pushed-down limit now return the correct number of rows.

There is no change for limits written as plain literals, which is the overwhelmingly common case: those take the literal arm, whose behavior is untouched.

One secondary effect worth flagging: for a foldable non-literal child limit the rule now keeps the GlobalLimit instead of removing it, so the plan retains one more node than before. Results are correct in both directions of that change — the previous plan was simply wrong. Once ConstantFolding folds the expression on a later pass, the surrounding limit rules simplify the plan as usual.

The alternative fix suggested on the JIRA — teaching GlobalLimit.maxRows to evaluate foldable limit expressions — would additionally let the guarded literal arm apply here and recover the stripping. That changes maxRows semantics for every consumer of the API, so it is deliberately left out of this correctness fix and can be pursued separately as an optimization.

How was this patch tested?

New tests, each verified to fail without the fix:

• LimitPushdownSuite, two plan tests covering the join and Union pushdown paths, asserting the child GlobalLimit survives. Without the fix the optimized plan drops GlobalLimit 2 and keeps only LocalLimit 2.
• SQLQuerySuite, an end-to-end test using the repro above. Without the fix it returns 10 instead of 5.

Existing regression suites, all green:

• catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.* — 1409 tests
• sql/testOnly org.apache.spark.sql.SQLQuerySuite org.apache.spark.sql.JoinSuite org.apache.spark.sql.DataFrameSetOperationsSuite org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite — 510 tests

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5)