cloud-fan commented on code in PR #57347:
URL: https://github.com/apache/spark/pull/57347#discussion_r3647260135
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/LogicalQueryStage.scala:
##########
@@ -71,7 +71,19 @@ case class LogicalQueryStage(
physicalStats.getOrElse(logicalPlan.stats)
}
- override def maxRows: Option[Long] =
stats.rowCount.map(_.min(Long.MaxValue).toLong)
+ override def maxRows: Option[Long] = {
+ // A query stage's runtime `rowCount` is an exact, valid upper bound only
once the stage
+ // is materialized. Before materialization, `stats` falls back to the
logical plan's cost
+ // estimate, which can under-count (e.g. return 0) and must not be treated
as a hard maximum
+ // - otherwise rules such as EliminateLimits can wrongly drop a LIMIT and
change the result
+ // cardinality (SPARK-57956). When the stage is not materialized, fall
back to the logical
+ // plan's `maxRows`, which is always a sound upper bound.
+ if (isMaterialized) {
Review Comment:
Could we require `stats.isRuntime` here as well? `isMaterialized` describes
stage state, but `computeStats()` can still fall back to `logicalPlan.stats`
when its physical-stage lookup yields no statistics; in that case this would
again promote an estimate to a hard upper bound. DBR guards on both
`stats.isRuntime` and `isMaterialized`, and otherwise takes the conservative
fallback.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -3380,6 +3380,26 @@ class AdaptiveQueryExecSuite
}
}
+ test("SPARK-57956: AQE must not eliminate a global LIMIT over a
row-increasing outer join") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "1048576") {
+ // The re-optimization that can wrongly drop the root LIMIT depends on
whether the
+ // OFFSET-side query stage is materialized when EliminateLimits runs,
which is timing
+ // dependent. Repeat the query so the regression is caught
deterministically.
Review Comment:
Please make this regression deterministic by directly constructing an
unmaterialized `LogicalQueryStage` with an underestimated `stats.rowCount`,
then asserting that `maxRows` remains structural/unknown and `EliminateLimits`
retains the limit. Repeating the end-to-end query 20 times only increases the
chance of hitting the bad scheduling state, so the test can still pass with the
broken implementation.
--
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]
