[SPARK-34064][SQL] Cancel the running broadcast sub-jobs when SQL statement is cancelled

[LantaoJin]

What changes were proposed in this pull request?

The refactor PR for #31119.
In this PR, we add a local property key spark.statement.id which set by STS. STS will set statementId to this local property,
in broadcast exchange, the runId reads the value from this property if defined, or uses a random UUID.

Why are the changes needed?

When broadcasting a table takes too long and the SQL statement is cancelled. However, the background Spark job is still running and it wastes resources.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Manually test.
Since broadcasting a table is too fast to cancel in UT, but it is very easy to verify manually:

1. Start a Spark thrift-server with less resource (or set executor memory to a very big value) in YARN.
2. When the driver is running but no executors are launched, submit a SQL which will broadcast tables from beeline.
3. Cancel the SQL in beeline

[LantaoJin]

cc @cloud-fan @yaooqinn

[yaooqinn]

https://github.com/apache/spark/pull/31227/files#diff-1e596e9e9bd50e3075847a66f8fecee741289b2b0478f9099c3bf2b680ebaa35R749-R755

This looks like a silent API behavior change.

[LantaoJin]

I will add this to description

[cloud-fan]

This change is a bit too low-level and may have a big impact. Can we do something in SQLExecution.withNewExecutionId like

val originalJobGroupId = ...
try {
  val actualGroupId = ... // read using the special key
  sc.setJonGroupId(actualGroupId)
  ...
} finally {
  // set back to the original id
}

We also need to update the broadcast exchange to read the special local property.

[SparkQA]

Test build #134186 has finished for PR 31227 at commit 79e4bfb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[LantaoJin]

I have a little bit confuse, why not just add a local property such as spark.statement.id.
In STS, we set the statementId to the thread local statementId and groupId.
In broadcast exchange:
We get runId from Option(statementId).map(UUID.fromString).getOrElse(RandomUUID)
For STS case, statementId = groupId = runId
For non-STS case, statementId is undefined, runId = groupId = randomUUID

[cloud-fan]

ah that sounds simpler

[cloud-fan]

We can define it in sql/core module, as it's only used there and the STS module. Probably in SQLExecution object.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38780/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38780/

[SparkQA]

Test build #134195 has finished for PR 31227 at commit 1a751ba.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

@LantaoJin any updates?

[SparkQA]

Test build #137564 has finished for PR 31227 at commit 1a751ba.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[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!