This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new dd3848c62661 [SPARK-58018][SQL][TEST][FOLLOWUP] Drain slow_udf
scalar-subquery jobs in the AQE cancellation test so they don't flake sibling
tests
dd3848c62661 is described below
commit dd3848c626611653bc67a94e2bb921cecd3997e7
Author: Wenchen Fan <[email protected]>
AuthorDate: Mon Jul 20 22:40:19 2026 +0800
[SPARK-58018][SQL][TEST][FOLLOWUP] Drain slow_udf scalar-subquery jobs in
the AQE cancellation test so they don't flake sibling tests
Followup to https://github.com/apache/spark/pull/57096.
### What changes were proposed in this pull request?
`AdaptiveQueryExecSuite`'s `SPARK-47148: AQE should avoid to submit shuffle
job on cancellation` test registers a `slow_udf` that sleeps for 15s (raised
from 3s in #57096 to deflake the test) and uses it inside scalar subqueries.
Those scalar-subquery jobs on the `df`/`df3` shuffle stages are submitted
asynchronously and run to completion (they are not the stages the test
cancels), so each `slow_udf` invocation keeps a task busy for the full 15s
while holding one of the test session's [...]
`QueryTest.withTempDir` runs a task drain after each test:
```scala
protected override def withTempDir(f: File => Unit): Unit = {
super.withTempDir { dir =>
f(dir)
waitForTasksToFinish() // eventually(timeout(10.seconds)) {
numRunningTasks == 0 }
}
}
```
So if another `withTempDir`-using test is scheduled within ~10s after
`SPARK-47148` runs, that later test can see the leftover `slow_udf` tasks still
occupying slots and fail its own drain with `1 was not equal to 0` — a failure
that has nothing to do with the later test itself. This is a latent cross-test
dependency: `SPARK-47148` leaks slot-holding tasks past its own scope, and the
15s sleep makes the leak long enough to outlive a sibling test's 10s drain
window.
This PR makes `SPARK-47148` wait for its own `slow_udf` scalar-subquery
jobs to finish before returning, so they cannot bleed into a sibling test:
```scala
} finally {
spark.experimental.extraStrategies = Nil
eventually(timeout(60.seconds), interval(500.milliseconds)) {
assert(spark.sparkContext.statusTracker.getActiveJobIds().isEmpty,
"slow_udf scalar-subquery jobs are still running")
}
}
```
It polls `getActiveJobIds` rather than `numRunningTasks`: when the
`finally` runs, the subquery jobs are already *submitted* but their tasks may
not have *started* yet, so a running-task count can race ahead and return
early, whereas the job stays active until it completes. The 60s timeout
comfortably outlasts the 15s sleep (the two subquery jobs serialize onto the 2
task slots, ~30s total).
### Why are the changes needed?
`SPARK-47148` currently leaves long-running `slow_udf` tasks occupying task
slots after it returns. With the 15s sleep this can flake any `withTempDir`
test that happens to run shortly after it, via a task-drain timeout unrelated
to that test's own logic. Confining the cleanup to the test that creates the
leak removes this cross-test flakiness at its source.
### Does this PR introduce _any_ user-facing change?
No. Test-only.
### How was this patch tested?
`build/sbt "sql/testOnly
org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite -- -z
SPARK-47148"` — the test passes, taking ~30s as it now drains its own two 15s
scalar-subquery jobs (per-test timeout is 20 minutes, so no meaningful cost).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57309 from cloud-fan/SPARK-58018-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../execution/adaptive/AdaptiveQueryExecSuite.scala | 18 ++++++++++++++++++
1 file changed, 18 insertions(+)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
index 7446168c84fc..f84f3ad5cb6c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
@@ -20,6 +20,8 @@ package org.apache.spark.sql.execution.adaptive
import java.io.File
import java.net.URI
+import scala.concurrent.duration._
+
import org.apache.logging.log4j.Level
import org.scalatest.PrivateMethodTester
@@ -1056,6 +1058,22 @@ class AdaptiveQueryExecSuite
}
} finally {
spark.experimental.extraStrategies = Nil
+ // `slow_udf` runs inside scalar-subquery jobs on the `df`/`df3`
shuffle stages. These jobs
+ // are submitted asynchronously and run to completion (they are not
the cancelled stages),
+ // so each `slow_udf` invocation sleeps the full 15s while holding one
of the test session's
+ // task slots (`local[2]`) -- and a task can even start running only
after this test body
+ // has already returned. `QueryTest.withTempDir` runs a task drain
(`waitForTasksToFinish`,
+ // a 10s `numRunningTasks <= 0` wait) after each test, so a later
`withTempDir` test
+ // scheduled within that window would see these leftover tasks still
occupying slots and
+ // time out its drain. Wait here, within this test's own scope, for
those jobs to finish so
+ // they cannot bleed into a sibling test. We poll `getActiveJobIds`
(not `numRunningTasks`):
+ // the subquery jobs are already submitted when this block runs even
though their tasks may
+ // not have started yet, so a running-task count can race ahead and
return early, whereas
+ // the job stays active until it completes. The timeout outlasts the
15s sleep.
+ eventually(timeout(60.seconds), interval(500.milliseconds)) {
+ assert(spark.sparkContext.statusTracker.getActiveJobIds().isEmpty,
+ "slow_udf scalar-subquery jobs are still running")
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
