MaxGekk commented on code in PR #59043:
URL: https://github.com/apache/spark/pull/59043#discussion_r4176910102
##########
sql/core/src/test/scala/org/apache/spark/sql/test/SharedSparkSession.scala:
##########
@@ -26,27 +26,30 @@ import org.apache.spark.sql.classic
trait SharedSparkSession extends QueryTest with classic.SparkSessionBinder {
- // Runs func (which must trigger exactly one SQL execution) and returns the
SQL metrics of that
- // execution as a map keyed by (planNodeId, planNodeName, metricName) ->
metricValue.
- def runAndFetchMetrics(func: => Unit): Map[(Long, String, String), String] =
{
+ // Runs func (which must trigger exactly one SQL execution), waits until the
status store has
+ // computed the SQL metrics of that execution and returns its execution id.
+ def runAndWaitForExecution(func: => Unit): Long = {
val statusStore = spark.sharedState.statusStore
- val oldCount = statusStore.executionsList().size
+ // The listener updates the status store asynchronously, so the latest
execution in the store
+ // may still be one from before func. Execution ids only grow, so wait for
an execution with a
+ // larger id than any existing one. Execution counts cannot be used for
this, as old
+ // executions are evicted once spark.sql.ui.retainedExecutions is reached.
+ val lastExecId =
statusStore.executionsList().lastOption.map(_.executionId).getOrElse(-1L)
Review Comment:
The baseline is read from the status store, which the listener bus fills
asynchronously. If a previous test's execution hasn't reached the store yet, it
can land after this line with an id above `lastExecId` and its metrics
computed, and the helper would return it instead of the execution `func` ran -
the same failure, just rarer. Draining the bus first makes the baseline cover
every execution already started:
```suggestion
spark.sparkContext.listenerBus.waitUntilEmpty()
val lastExecId =
statusStore.executionsList().lastOption.map(_.executionId).getOrElse(-1L)
```
`waitUntilEmpty()` is `private[spark]`, so it's reachable from
`org.apache.spark.sql.test`, and the SPARK-36030 write test in
`SQLAppStatusListenerSuite` already calls it.
--
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]
