dongjoon-hyun commented on code in PR #59194:
URL: https://github.com/apache/spark/pull/59194#discussion_r4328844923
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1008,8 +1008,10 @@ object LimitPushDown extends Rule[LogicalPlan] {
val newAgg = EliminateSorts(a.copy(child = LocalLimit(le,
a.child))).asInstanceOf[Aggregate]
Limit(le, p.copy(child = Project(newAgg.aggregateExpressions,
newAgg.child)))
// Merge offset value and limit value into LocalLimit and pushes down
LocalLimit through Offset.
+ // Fold the merged limit so that the LocalLimit can be planned without a
later ConstantFolding
+ // run: BasicOperators only plans LocalLimit(IntegerLiteral, _).
case LocalLimit(le, Offset(oe, grandChild)) =>
- Offset(oe, LocalLimit(Add(le, oe), grandChild))
+ Offset(oe, LocalLimit(ConstantFolding.constantFolding(Add(le, oe)),
grandChild))
Review Comment:
**The analysis check does not cover every `LocalLimit` over `Offset` that
reaches this case, so this sum can still overflow.** The PR description says
the overflow case "is already rejected during analysis with
`SUM_OF_LIMIT_AND_OFFSET_EXCEEDS_MAX_INT`", but CheckAnalysis.scala:921-935
only checks a `LocalLimit` whose direct child is an `Offset` in the analyzed
plan. The optimizer also creates such pairs, which the analyzer never sees: the
`Union` case at L976-977 and `maybePushLocalLimit`'s `case (_, None)` at
L942-944 push the limit onto an `Offset` branch, the semi/anti join case at
L1016-1017 pushes `Literal(1)`, and `PushProjectionThroughLimitAndOffset` or
`RemoveNoopOperators` can move or remove a `Project` between them. For example,
with a table `t` that has 3 rows:
```sql
(SELECT * FROM t OFFSET 1) UNION ALL (SELECT * FROM t) LIMIT 2147483647
```
- With ANSI on (the default), this valid query fails with
`[ARITHMETIC_OVERFLOW]`, although the user wrote no addition.
- With ANSI off, the sum wraps to -2147483648, and `LocalLimitExec` returns
no rows for that branch (limit.scala:200 and :233), so the query silently
returns 3 rows instead of 5.
- With ANSI off, `SELECT * FROM (SELECT * FROM t OFFSET 10) LIMIT
2147483647` still matches `OffsetAndLimit`, because `globalLimit + offset ==
localLimit` (basicLogicalOperators.scala:2041) wraps the same way, and then
fails the `CollectLimitExec` assertion (limit.scala:49) as an INTERNAL_ERROR.
This is not a regression in the default configuration, since
`ConstantFolding` produced the same values before. But with `ConstantFolding`
excluded and ANSI off, the `UNION ALL` query used to fail with `No plan for
LocalLimit (2147483647 + 1)`, and with this PR it returns wrong rows silently.
I traced this by reading the code and did not run it. I know asugranyes pointed
out on #56180 that an `Add` left behind cannot be planned without
`ConstantFolding`, but that only affects sums that overflow, which cannot give
a correct plan here anyway, and keeping the `Add` makes them fail loudly
instead of returning wrong rows. Could you fold only when the sum fits in an
`Int`, e.g.
```scala
val merged = Add(le, oe)
val fits = le.eval().asInstanceOf[Int].toLong + oe.eval().asInstanceOf[Int]
<= Int.MaxValue
Offset(oe, LocalLimit(if (fits) ConstantFolding.constantFolding(merged) else
merged, grandChild))
```
and also correct the overflow sentence in the PR description, with a
follow-up JIRA for the overflow itself?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushdownSuite.scala:
##########
@@ -335,6 +335,16 @@ class LimitPushdownSuite extends PlanTest {
GlobalLimit(1, Offset(2, LocalLimit(3, testRelation))).analyze)
}
+ test("SPARK-57125: Push down limit through Offset folds the merged limit") {
Review Comment:
Minor: This test checks the output of `LimitPushDown` alone, but nothing
checks the user-visible part of the fix: that a `LIMIT`/`OFFSET` query is
planned and returns the right rows when `ConstantFolding` is excluded.
SPARK-59044 fixed the same kind of failure for `EliminateOffsets` and added an
end-to-end test for both planning paths, "SPARK-59044: OFFSET 0 succeeds when
EliminateOffsets is in excludedRules" (SQLQuerySuite.scala:5359 on master). For
example:
```scala
withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
ConstantFolding.ruleName) {
checkAnswer(sql("SELECT id FROM range(10) LIMIT 1 OFFSET 2"), Row(2))
checkAnswer(sql("SELECT count(*) FROM (SELECT id FROM range(10) LIMIT 2
OFFSET 2)"), Row(2))
}
```
By my reading, both queries fail on master with `No plan for LocalLimit
(...)` and pass with this PR. The source needs more rows than the limit plus
the offset; otherwise `EliminateLimits` removes the limit before this rule
runs. Could you add such a test next to the SPARK-59044 one, after a rebase?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1008,8 +1008,10 @@ object LimitPushDown extends Rule[LogicalPlan] {
val newAgg = EliminateSorts(a.copy(child = LocalLimit(le,
a.child))).asInstanceOf[Aggregate]
Limit(le, p.copy(child = Project(newAgg.aggregateExpressions,
newAgg.child)))
// Merge offset value and limit value into LocalLimit and pushes down
LocalLimit through Offset.
+ // Fold the merged limit so that the LocalLimit can be planned without a
later ConstantFolding
+ // run: BasicOperators only plans LocalLimit(IntegerLiteral, _).
case LocalLimit(le, Offset(oe, grandChild)) =>
Review Comment:
**Evaluating the sum here, before `EliminateOffsets` and `EliminateLimits`
run, turns a query that returns an empty result on master into an error.** In
the operator optimization batch, this rule (L112) runs before
`EliminateOffsets` (L122), `EliminateLimits` (L123) and `ConstantFolding`
(L139), so on master the unfolded `Add` could be pruned with its subtree before
anything evaluated it. Now it is evaluated right here. For example, with ANSI
on (the default) and a table `t`:
```sql
SELECT * FROM (SELECT * FROM (SELECT * FROM t OFFSET 10) LIMIT 2147483647)
LIMIT 0
```
The analysis check does not apply, because a `Project` sits between the
inner `LIMIT` and `OFFSET`. On master, this rule leaves `LocalLimit(2147483647
+ 10, t)` unevaluated, and `EliminateLimits` then replaces the whole plan with
an empty `LocalRelation` (`case GlobalLimit(IntegerLiteral(0), child)` at
L3116), so the query returns no rows. With this PR, L1014 evaluates `2147483647
+ 10` first, and the query fails with `[ARITHMETIC_OVERFLOW]`.
`spark.range(100).offset(10).select("id").limit(Int.MaxValue).limit(0).collect()`
behaves the same, and so does an outer `OFFSET 2147483647` instead of `LIMIT
0`. This is an edge case, but it contradicts "No user-facing change". I traced
this by reading the code and did not run it. Could you fold only when the sum
fits in an `Int`, as suggested on L1014, so that this case behaves as on master?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1008,8 +1008,10 @@ object LimitPushDown extends Rule[LogicalPlan] {
val newAgg = EliminateSorts(a.copy(child = LocalLimit(le,
a.child))).asInstanceOf[Aggregate]
Limit(le, p.copy(child = Project(newAgg.aggregateExpressions,
newAgg.child)))
// Merge offset value and limit value into LocalLimit and pushes down
LocalLimit through Offset.
+ // Fold the merged limit so that the LocalLimit can be planned without a
later ConstantFolding
+ // run: BasicOperators only plans LocalLimit(IntegerLiteral, _).
Review Comment:
Minor: About the PR description. The "Does this PR introduce _any_
user-facing change?" section answers "No", but the template counts bug fixes
and other behavior changes as user-facing. With `ConstantFolding` excluded,
`SELECT id FROM range(10) LIMIT 1 OFFSET 2` used to fail with `No plan for
LocalLimit (1 + 2)` and now returns `2`, and SPARK-59044 answered "Yes" for the
same kind of change. Also, "`ConstantFolding` folds the `Add` anyway" does not
mean that the default-configuration plans stay the same: the code that runs
between this rule and `ConstantFolding` in the same iteration (e.g.
`maybePushLocalLimit` and `LimitPushDownThroughWindow`, which match
`IntegerLiteral` limits) now sees the literal one iteration earlier, so some
optimized plans can differ, although the results do not. Could you answer "Yes"
there, with the before and after behavior?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -1008,8 +1008,10 @@ object LimitPushDown extends Rule[LogicalPlan] {
val newAgg = EliminateSorts(a.copy(child = LocalLimit(le,
a.child))).asInstanceOf[Aggregate]
Limit(le, p.copy(child = Project(newAgg.aggregateExpressions,
newAgg.child)))
// Merge offset value and limit value into LocalLimit and pushes down
LocalLimit through Offset.
+ // Fold the merged limit so that the LocalLimit can be planned without a
later ConstantFolding
+ // run: BasicOperators only plans LocalLimit(IntegerLiteral, _).
case LocalLimit(le, Offset(oe, grandChild)) =>
- Offset(oe, LocalLimit(Add(le, oe), grandChild))
+ Offset(oe, LocalLimit(ConstantFolding.constantFolding(Add(le, oe)),
grandChild))
Review Comment:
Minor: About a backport. `ConstantFolding.constantFolding` became
`private[sql]` in SPARK-52875, so this line compiles on master, branch-4.x,
branch-4.2 and branch-4.1. On branch-4.0 it is still `private def
constantFolding(` (expressions.scala:74), so a cherry-pick there would not
compile, and branch-4.0 has no `PushProjectionThroughLimitAndOffsetSuite`. If
this fix should also go to branch-4.0, could you note in the JIRA that the
backport needs a separate patch (e.g. evaluating the sum directly, as
`EliminateLimits` does at L3119-3124)?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushdownSuite.scala:
##########
@@ -335,6 +335,16 @@ class LimitPushdownSuite extends PlanTest {
GlobalLimit(1, Offset(2, LocalLimit(3, testRelation))).analyze)
}
+ test("SPARK-57125: Push down limit through Offset folds the merged limit") {
+ // The merged LocalLimit must be folded without a later ConstantFolding
run.
+ object OptimizeOnlyLimitPushDown extends RuleExecutor[LogicalPlan] {
Review Comment:
Nit: This local executor is left over from the first commit, where it served
two cases. With one case left, it wraps a single rule, and
`comparePlans(LimitPushDown(testRelation.offset(2).limit(1).analyze), ...)`
does the same, as other optimizer suites do (e.g.
InferFiltersFromConstraintsSuite.scala:398). The input and the expected plan
are also the same as in the test above, so only the missing `ConstantFolding`
differs. Separately, "How was this patch tested?" could also mention the
updated `PushProjectionThroughLimitAndOffsetSuite` expectations, which fail
without the fix too. Could you call the rule directly here, unless you prefer
the `RuleExecutor` style of this suite?
--
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]
