Yuan Yusi created SPARK-57956:
---------------------------------
Summary: AQE can remove global LIMIT during optimization
Key: SPARK-57956
URL: https://issues.apache.org/jira/browse/SPARK-57956
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.1.2
Environment: Apache Spark 4.1.2
PySpark 4.1.2
Java 17
Reproduced on macOS system.
Reporter: Yuan Yusi
h1. AQE can remove global LIMIT and return too many rows
h2. What is the issue?
AQE can drop a root global {{LIMIT}} during adaptive re-optimization.
The following query ends with {{{}LIMIT 1{}}}, so it should return at most one
row. With AQE enabled on Spark 4.1.2, it can return two rows.
{code:sql}
WITH
top AS (
SELECT id % 1 AS k
FROM range(0, 2, 1, 1)
OFFSET 1
),
b AS (
SELECT *
FROM VALUES (0), (0) AS b(k)
)
SELECT 1 AS c
FROM top LEFT JOIN b USING (k)
LIMIT 1;
{code}
This is a correctness issue, as AQE changes the SQL result cardinality.
h2. Reproduction
Enable AQE in Spark, i.e.
{code:scala}
spark.conf.set("spark.sql.adaptive.enabled", "true")
{code}
Then run the SQL query above.
Record the observed result row count and compare it to the expected row count.
h2. Expected behavior
The query has a root {{{}LIMIT 1{}}}, so every execution should return exactly
one row.
h2. Actual behavior
AQE returns two rows for the failing executions:
{noformat}
(c=1)
(c=1)
{noformat}
The final AQE plan does not contain the root {{{}CollectLimit 1{}}}.
Initial plan:
{noformat}
CollectLimit 1
+- Project
+- SortMergeJoin ..., LeftOuter
{noformat}
Final AQE plan:
{noformat}
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
ResultQueryStage
+- Project
+- BroadcastHashJoin ..., LeftOuter
{noformat}
The final plan returns both joined rows.
h2. Control run
If AQE is disabled, the query is correct.
If AQE is enabled but AQE's {{EliminateLimits}} runtime optimizer rule is
excluded, the bug also disappears. For example:
{code:scala}
spark.conf.set(
"spark.sql.adaptive.optimizer.excludedRules",
"org.apache.spark.sql.catalyst.optimizer.EliminateLimits")
{code}
This strongly points to AQE limit elimination as the failure path.
h2. Suspected cause
The failure appears to be caused by an unsafe interaction between:
* {{LogicalQueryStage.computeStats()}}
* {{LogicalQueryStage.maxRows}}
* AQE {{EliminateLimits}}
In Spark 4.1.2:
{code:scala}
Batch("Eliminate Limits", fixedPoint, EliminateLimits)
{code}
{{EliminateLimits}} removes a {{GlobalLimit}} when:
{code:scala}
child.maxRows.exists(_ <= limit)
{code}
{{LogicalQueryStage.maxRows}} is derived from runtime {{{}stats.rowCount{}}}.
However, {{LogicalQueryStage.computeStats()}} can use the first nested
{{QueryStageExec}} statistics even when there are physical operators above that
stage. The Spark source already has a TODO noting this is inaccurate:
{noformat}
this is not accurate when there is other physical nodes above QueryStageExec
{noformat}
In this query, the left side of the join has one row after {{{}OFFSET 1{}}},
but the left outer join can output two rows because the right side has two
matching rows. AQE can incorrectly treat the child under the root limit as
having {{{}maxRows = 1{}}}, then remove {{{}LIMIT 1{}}}, allowing two rows to
escape.
h2. How long has this likely been broken?
I ran a source-level {{git bisect}} between {{v3.2.4}} and {{v3.3.0}} using the
predicate "does {{AQEOptimizer.scala}} contain AQE {{{}EliminateLimits{}}}?".
The first bad source commit is:
{noformat}
SPARK-36424: Support eliminate limits in AQE Optimizer
Commit: bb6f65acca2918a0ceb13b612d210f1b46fa1add
PR: https://github.com/apache/spark/pull/33651
{noformat}
Parent commit:
{noformat}
e051a540a10cdda42dc86a6195c0357aea8900e4
{noformat}
h2. Proposed fix
Conservative option:
Do not derive {{LogicalQueryStage.maxRows}} from runtime row count when the
{{LogicalQueryStage}} wraps a non-direct physical subtree.
For example:
{code:scala}
override def maxRows: Option[Long] = {
if (isDirectStage) {
stats.rowCount.map(_.min(Long.MaxValue).toLong)
} else {
logicalPlan.maxRows
}
}
{code}
Safer option:
{code:scala}
override def maxRows: Option[Long] = {
if (isDirectStage) {
stats.rowCount.map(_.min(Long.MaxValue).toLong)
} else {
None
}
}
{code}
Alternative:
Disable AQE {{EliminateLimits}} for plans containing non-direct
{{LogicalQueryStage}} nodes until runtime row-count bounds are known to be safe
for the whole subtree.
h2. Suggested regression test
Add an AQE SQL correctness test asserting the row count is exactly one:
{code:scala}
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "1048576") {
val rows = sql("""
WITH
top AS (
SELECT id % 1 AS k
FROM range(0, 2, 1, 1)
OFFSET 1
),
b AS (
SELECT *
FROM VALUES (0), (0) AS b(k)
)
SELECT 1 AS c
FROM top LEFT JOIN b USING (k)
LIMIT 1
""").collect()
assert(rows.length == 1)
}
{code}
The exact returned row is not important because there is no final ordering. The
cardinality must be exactly one.
h2. Environment
* Apache Spark 4.1.2
* PySpark 4.1.2
* Java 17
* {{local[2]}}
Reproduced on macOS system.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
