[WIP][SPARK-35414][SQL] Submit broadcast collect job first to avoid broadcast timeout in AQE

[zhongyu09]

What changes were proposed in this pull request?

1. replace executeCollectIterator() by executeCollectIteratorFuture() in SparkPlan.scala to run collect query in async way and return the future of collect result
2. in BroadcastExchangeExec->relationFuture, call executeCollectIteratorFuture() in current thread and get the collectFuture, wait collectFuture in "broadcast-exchange" thread

Why are the changes needed?

#31269 gives a partial fix to SPARK-33933, which is not a perfect solution. This changes can make sure the broadcast collect job is submitted before shuffle map jobs. #31269 ensure the calling of materialize() of BroadcastQueryStage is before ShuffleQueryStage. In BroadcastQueryStage's materialize(), doPrepare() will call relationFuture, which will submit collect job before return the future.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Add UT

[AmplabJenkins]

Can one of the admins verify this patch?

[weixiuli]

Is there any progress for this pr ?

[zhongyu09]

Is there any progress for this pr ?

In https://github.com/zhongyu09/spark/runs/2829717969, unit test for "BroadcastExchange should cancel the job group if timeout" fails. I realize that there could be race condition in this solution and cause hang when doing broadcast.

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!