yadavay-amzn commented on code in PR #55839:
URL: https://github.com/apache/spark/pull/55839#discussion_r3583708552
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -352,6 +356,88 @@ class AdaptiveQueryExecSuite
}
}
+ test("empty materialized stage short-circuits AQE through sort wrappers") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "10") {
+ val jobEndEvents = new mutable.ArrayBuffer[SparkListenerJobEnd]
+ val listener = new SparkListener {
+ override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit =
jobEndEvents.synchronized {
+ jobEndEvents += jobEnd
+ }
+ }
+ spark.sparkContext.addSparkListener(listener)
+ try {
+ val left = spark.range(0, 1, 1, 1).where("id <
0").select($"id".as("k"))
+ val right = spark.range(0, 200, 1, 20).as[Long].map { id =>
+ Thread.sleep(200)
+ id
+ }.select($"value".as("k"))
+ val df = left.join(right, Seq("k"))
+
+ checkAnswer(df, Seq.empty)
+
+ val finalPlan = df.queryExecution.executedPlan
+ .asInstanceOf[AdaptiveSparkPlanExec].executedPlan
+ assert(collect(finalPlan) { case s: SortMergeJoinExec => s }.isEmpty)
+
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ assert(jobEndEvents.synchronized {
+ jobEndEvents.exists {
+ case SparkListenerJobEnd(_, _, JobFailed(e)) =>
+ e.getMessage.toLowerCase(Locale.ROOT).contains("cancel")
+ case _ => false
+ }
+ })
+ } finally {
+ spark.sparkContext.removeSparkListener(listener)
+ }
+ }
+ }
+
+ test("empty filtered global aggregate stage is not treated as non-empty") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+ withTempView("empty_rows") {
+ spark.range(1).where("id < 0").createOrReplaceTempView("empty_rows")
+
+ val df = sql(
+ "SELECT * FROM testData LEFT ANTI JOIN " +
+ "(SELECT count(*) c FROM empty_rows HAVING c < 0) r")
+
+ checkAnswer(df, testData.collect().toSeq)
Review Comment:
Non-blocking: this test doesn't actually guard the invariant it names.
`testData LEFT ANTI JOIN (SELECT count(*) c FROM empty_rows HAVING c < 0)`
returns all of `testData` in both worlds - with the correct `+1` guard the
subquery is 1 row -> HAVING filters -> 0 rows -> anti-join keeps all left; but
if the guard were removed and the agg wrongly marked empty, LeftAnti with an
empty right also returns all left. Same output, and there's no plan assertion -
so it passes even if the `BigInt(1)` guard in `getEstimatedRowCount` is
deleted. The invariant is currently protected only by the pre-existing
SPARK-35442 test, which doesn't traverse the new wrapper recursion.
Suggest a direct check that observes the single row through the new path,
e.g.:
```scala
val df = sql("SELECT count(*) AS c FROM (SELECT * FROM empty_rows ORDER BY
a)")
checkAnswer(df, Row(0))
assert(!stripAQEPlan(df.queryExecution.executedPlan).isInstanceOf[EmptyRelationExec])
```
--
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]
