Josh Rosen created SPARK-58385:
----------------------------------
Summary: LimitPushDown incorrectly removes a GlobalLimit whose
limit expression is not a literal
Key: SPARK-58385
URL: https://issues.apache.org/jira/browse/SPARK-58385
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.0.0
Reporter: Josh Rosen
This is a report of a correctness bug present in Spark 2.0.0 and all later
versions.
[{{LimitPushDown.maybePushLocalLimit}}|https://github.com/apache/spark/blob/5ce51a624421acceec378e0170da24a8fc2a96eb/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L916]
decides whether a pushed-down limit may replace a child's existing limit by
inspecting {{{}plan.maxRowsPerPartition{}}}:
{code:java}
private def maybePushLocalLimit(limitExp: Expression, plan: LogicalPlan):
LogicalPlan = {
(limitExp, plan.maxRowsPerPartition) match {
case (IntegerLiteral(newLimit), Some(childMaxRows)) if newLimit <
childMaxRows =>
// If the child has a cap on max rows per partition and the cap is
larger than
// the new limit, put a new LocalLimit there.
LocalLimit(limitExp, stripGlobalLimitIfPresent(plan))
case (_, None) =>
// If the child has no cap, put the new LocalLimit.
LocalLimit(limitExp, stripGlobalLimitIfPresent(plan))
case _ =>
// Otherwise, don't put a new LocalLimit.
plan
}
}{code}
* The first arm strips a child {{GlobalLimit}} only under a guard
({{{}newLimit < childMaxRows{}}}).
* The {{(_, None)}} arm strips unconditionally, reading {{None}} as "the child
has no cap" — but {{None}} also means "the cap is statically unknown".
*
[{{GlobalLimit.maxRows}}|https://github.com/apache/spark/blob/5ce51a624421acceec378e0170da24a8fc2a96eb/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala#L1896-L1901]
/ {{maxRowsPerPartition}} recognize only {{{}IntegerLiteral{}}}, and a limit
written as a foldable expression ({{{}LIMIT 2+3{}}}) is still {{Add(2, 3)}}
when {{LimitPushDown}} runs: it is
[listed|https://github.com/apache/spark/blob/5ce51a624421acceec378e0170da24a8fc2a96eb/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L112]
before {{ConstantFolding}} in the same operator-optimization rule set, so on
the first fixed-point pass the expression is unfolded, {{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/generated SQL that renders arithmetic into the query
text, e.g. a pagination template producing {{{}LIMIT 20*5{}}}.
{*}Repro{*}:
{code:java}
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);
-- outputs: 10 WRONG (expected: 5){code}
This returns an incorrect answer because the larger outer limit strips the
nested global limit. If we manually fold the inner limit to {{5}} .
{*}Fix{*}:
Treat {{None}} as "unknown", not "uncapped": in the {{(_, None)}} arm, never
strip a {{GlobalLimit}} child (push the {{LocalLimit}} beneath it, or leave the
plan unchanged). Alternatively (or additionally), make {{GlobalLimit.maxRows}}
evaluate foldable limit expressions so the guarded arm applies. Stripping is
only sound when the child's cap is known and the guard {{newLimit <
childMaxRows}} holds, as the literal arm already implements.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
