[SPARK-39328][SQL][TESTS] Fix flaky test SPARK-37753: Inhibit broadcast in left outer join when there are many empty partitions on outer/left side

### What changes were proposed in this pull request?

Improve test `SPARK-37753: Inhibit broadcast in left outer join when there are many empty partitions on outer/left side` of `AdaptiveQueryExecSuite`

### Why are the changes needed?

This test appears to always succeed in the Apache GitHub Action runner environment, But some environments, test does not seem to proceed as intended.

On my environment:
`4.18.0-553.8.1.el8_10.x86_64`
`Intel(R) Xeon(R) Silver 4210 CPU  2.20GHz`
`64G Mem`
And ran test in master branch following the guide of official documentation
```
./build/sbt
testOnly org.apache.spark.sql.execution.adaptive.AdaptiveQueryExecSuite
...
- SPARK-37753: Inhibit broadcast in left outer join when there are many empty partitions on outer/left side *** FAILED ***
  The code passed to eventually never returned normally. Attempted 25 times over 15.040156205999999 seconds. Last failure message:
```
even increasing the test's timeout to 1500 seconds results to failure after lots of retries.
```
SPARK-37753: Inhibit broadcast in left outer join when there are many empty partitions on outer/left side *** FAILED ***
  The code passed to failAfter did not complete within 20 minutes. (AdaptiveQueryExecSuite.scala:743)
```

---

The test says
```scala
    // if the right side is completed first and the left side is still being executed,
    // the right side does not know whether there are many empty partitions on the left side,
    // so there is no demote, and then the right side is broadcast in the planning stage.
    // so retry several times here to avoid unit test failure.
    eventually(timeout(15.seconds), interval(500.milliseconds)) {
...
```
It seems test failure occurs with very high probability by loading the ‘right side’ completes first.

While the reason is unclear, I believe it would be better to regulate the subquery loading speed in a predictable manner via applying simple udf rather than retrying until both sides load in the desired order.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Rerun the test.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #52388 from Last-remote11/SPARK-39328.

Authored-by: cafri.sun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit e7b7acf)
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: cafri.sun

sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala

@@ -22,7 +22,6 @@ import java.net.URI
 
 import org.apache.logging.log4j.Level
 import org.scalatest.PrivateMethodTester
-import org.scalatest.time.SpanSugar._
 
 import org.apache.spark.SparkException
 import org.apache.spark.rdd.RDD
@@ -747,16 +746,22 @@ class AdaptiveQueryExecSuite
     // if the right side is completed first and the left side is still being executed,
     // the right side does not know whether there are many empty partitions on the left side,
     // so there is no demote, and then the right side is broadcast in the planning stage.
-    // so retry several times here to avoid unit test failure.
-    eventually(timeout(15.seconds), interval(500.milliseconds)) {
+    // so apply `slow_udf` to delay right side to avoid unit test failure.
+    withUserDefinedFunction("slow_udf" -> true) {
+      spark.udf.register("slow_udf", (x: Int) => {
+        Thread.sleep(300)
+        x
+      })
+
       withSQLConf(
         SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
         SQLConf.NON_EMPTY_PARTITION_RATIO_FOR_BROADCAST_JOIN.key -> "0.5") {
         // `testData` is small enough to be broadcast but has empty partition ratio over the config.
         withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "200") {
           val (plan, adaptivePlan) = runAdaptiveAndVerifyResult(
             "SELECT * FROM (select * from testData where value = '1') td" +
-              " left outer join testData2 ON key = a")
+              " left outer join (select slow_udf(a) as a, b from testData2) as td2" +
+              " ON td.key = td2.a")
           val smj = findTopLevelSortMergeJoin(plan)
           assert(smj.size == 1)
           val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)