Vivek1106-04 opened a new pull request, #58065:
URL: https://github.com/apache/spark/pull/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`:
```scala
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`](https://github.com/apache/spark/blob/ff26d01d4c0130b02e9a448ff167271820823f9c/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala#L1896-L1901)
recognizes only `IntegerLiteral`, and `LimitPushDown` is
[listed](https://github.com/apache/spark/blob/ff26d01d4c0130b02e9a448ff167271820823f9c/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L112)
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:
```sql
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)
--
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]
