[
https://issues.apache.org/jira/browse/SPARK-57956?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-57956:
-----------------------------------
Labels: Correctness CorrectnessBug correctness pull-request-available
(was: Correctness CorrectnessBug correctness)
> 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
> Priority: Major
> Labels: Correctness, CorrectnessBug, correctness,
> pull-request-available
>
> 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]
