[SPARK-43755][CONNECT] Move execution out of SparkExecutePlanStreamHandler and to a different thread #41315

juliuszsompolski · 2023-05-25T14:43:16Z

What changes were proposed in this pull request?

Move code related to execution from being done directly in the GRPC callback in SparkConnextStreamHandler, to it's own classes.
ExecutionHolder (renamed from ExecuteHolder) launches the execution in it's own thread using ExecuteThreadRunner
The execution pushes responses via ExecuteResponseObserver (running in the execution thread)
ExecuteResponseObserver notifies ExecuteGrpcResponseSender (running in the rpc handler thread) to send the responses.
The actual execution code is refactored into SparkConnectPlanExecution

This allows to improve query interruption, by making interrupt method interrupt the execution thread. This makes interrupt work also when no Spark Jobs are running.

The refactoring further opens the possibilities of detaching query execution from a single RPC execution. Right now ExecutionHolder is waiting for the execution thread to finish, and ExecutePlanResponseObserver is forwarding the responses directly to the RPC observer.

In a followup, we can design different modes of execution, e.g.
ExecutePlanResponseObserver buffering the responses. A client which lost connection could then reconnect and ask for the stream to be retransmitted.

ExecutionHolder returning the operationId to the client directly, and then client requesting results in separate RPCs, with more control over the response stream, instead of having it just pushed to it.

Why are the changes needed?

Improve the working of interrupt
Refactoring that opens up possibilities of detaching query execution from a single RPC.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing Spark Connect CI covers the execution.

juliuszsompolski · 2023-05-25T17:44:57Z

@hvanhovell @grundprinzip @LuciferYang

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

...rver/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseObserver.scala

juliuszsompolski · 2023-05-25T17:50:03Z

...server/src/main/scala/org/apache/spark/sql/connect/execution/SparkConnectPlanExecution.scala

This class is no-changes verbatim moved out of SparkConnectStreamHandler.

...or/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/executionUtils.scala

juliuszsompolski · 2023-05-26T16:52:18Z

[info] *** 1 TEST FAILED ***
[error] Failed: Total 3649, Failed 1, Errors 0, Passed 3648, Ignored 10, Canceled 2
[error] Failed tests:
[error] 	org.apache.spark.storage.BlockManagerProactiveReplicationSuite

unrelated flake.
Will wait with retriggering CI for review comments.

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

...server/src/main/scala/org/apache/spark/sql/connect/execution/SparkConnectPlanExecution.scala

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

hvanhovell

A few nit picks, but looks good overall.

grundprinzip

Looks good. Couple of nits that are mostly around naming and documentation. My only real comment would be to iron out what the public interface is going to be. Please make sure that all other classes are package private.

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

grundprinzip · 2023-05-26T14:35:06Z

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

Is there a particular reason for the nesting? Why not create an ExecutionThread that inherits from Thread instead of creating a thread that keeps a reference to the outside that is called from within?

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

...server/src/main/scala/org/apache/spark/sql/connect/execution/SparkConnectPlanExecution.scala

...ector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

...server/src/main/scala/org/apache/spark/sql/connect/execution/SparkConnectPlanExecution.scala

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala

juliuszsompolski · 2023-07-07T15:09:02Z

Ignore for now - ExecutePlanResponseSender is unfinished, didn't go through nit comments yet.

grundprinzip · 2023-07-10T18:50:38Z

...rver/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseObserver.scala

+    if (lastIndex.nonEmpty) {
+      throw new IllegalStateException("Stream onError can't be called after stream completed")
+    }
+    error = Some(t)


And we were just talking about this. It's great that we cache this here for an explicit error response to pikcup.

Adding to what we were talking:

This picks up errors of ExecutePlan, but it wouldn't be hard to extend this execution mechanism to other RPCs.

However, this will pick up errors on the execution thread only. If errors happen on the grpc thread (these would be more internal errors like unable to start execution thread. So those would need a different mechanism.

TBH, if I were to tabula-rasa design it, I would not use GRPC errors for application errors (like errors for Spark execution), but have these returned as an explicit message type, and reserve GRPC onError for server errors. That would free us from GRPC onError size limitations that we now need such workarounds for. That would make it easier to distinguish network/framework errors from user errors, which would make it easier to establish retry policies, or to keep stats on user errors vs. system errors... But such change couldn't be done backwards compatible at this point. Maybe for Spark 4.0?

The gRPC errors are not just for internal system errors. The state of the stream is undefined after an error has been thrown and I think its fair to close it with onError. Using the response trailers for exceptions that contain an 8MB query plan and description is not great.

My current thinking is to leverage the reattach RPC to fetch the last error and get a hifi response message with all of the metadata of the error.

...server/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseSender.scala

grundprinzip

First pass on the new files

...rver/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseObserver.scala

grundprinzip · 2023-07-12T04:08:41Z

...rver/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseObserver.scala

+  }
+
+  def completed(): Boolean = synchronized {
+    lastIndex.isDefined


why not make lastIndex a flag completed instead

ditto lastIndex != index in the responses list.

...rver/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseObserver.scala

...server/src/main/scala/org/apache/spark/sql/connect/execution/ExecutePlanResponseSender.scala

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteRunner.scala

rangadi · 2023-07-13T18:32:56Z

...nnect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala

+        case e: Throwable =>
+          logDebug(s"Exception in execute: $e")
+          // Always cancel all remaining execution after error.
+          executeHolder.sessionHolder.session.sparkContext.cancelJobsWithTag(executeHolder.jobTag)


Does this wait for cancellation to succeed? What happens if this throws?
I see currently only way is to do session.interruptAll().
Wondering if we plan to provide per-query API to cancel and if that would have any contract regarding cancellation.

This doesn't wait and doesn't throw, it's just sends an async ping to DAGScheduler to cancel whatever is left running. See SparkContext.cancelJobsWithTag -> DAGScheduler.cancelJobsWithTag.
This is "to be safe and clean everything up" like in https://github.com/apache/spark/blob/master/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala#L248 in Thriftserver

I plan to add two more types of cancelation:

per-query (requires operation_id, to be introduces in a next PR that also deals with query reattach)

per user settable tag (very similar to SparkContext.addJobTag / cancelJobsWithTag, but on a Spark Connet SparkSession

juliuszsompolski · 2023-07-14T07:44:09Z

Resolved (trivial) merge conflict.
Previous CI run had 1 flaky test:

2023-07-14T00:42:34.1310770Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m*** 1 TEST FAILED ***�[0m�[0m
2023-07-14T00:42:34.1515215Z �[0m[�[0m�[31merror�[0m] �[0m�[0mFailed: Total 9369, Failed 1, Errors 0, Passed 9368, Ignored 27�[0m
2023-07-14T00:42:34.1681702Z �[0m[�[0m�[31merror�[0m] �[0m�[0mFailed tests:�[0m
2023-07-14T00:42:34.1682533Z �[0m[�[0m�[31merror�[0m] �[0m�[0m	org.apache.spark.sql.execution.streaming.MicroBatchExecutionSuite�[0m

...ector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala

…onnect/service/SessionHolder.scala

juliuszsompolski · 2023-07-14T10:32:25Z

CI before the lint change was clean except for lint:
https://github.com/juliuszsompolski/apache-spark/actions/runs/5551730507

xuanyuanking · 2023-07-14T16:57:24Z

Thanks! Merged to master.

…ndler and to a different thread ### What changes were proposed in this pull request? * Move code related to execution from being done directly in the GRPC callback in SparkConnextStreamHandler, to it's own classes. * `ExecutionHolder` (renamed from `ExecuteHolder`) launches the execution in it's own thread using `ExecuteThreadRunner` * The execution pushes responses via `ExecuteResponseObserver` (running in the execution thread) * `ExecuteResponseObserver` notifies `ExecuteGrpcResponseSender` (running in the rpc handler thread) to send the responses. * The actual execution code is refactored into `SparkConnectPlanExecution` This allows to improve query interruption, by making `interrupt` method interrupt the execution thread. This makes `interrupt` work also when no Spark Jobs are running. The refactoring further opens the possibilities of detaching query execution from a single RPC execution. Right now `ExecutionHolder` is waiting for the execution thread to finish, and `ExecutePlanResponseObserver` is forwarding the responses directly to the RPC observer. In a followup, we can design different modes of execution, e.g. `ExecutePlanResponseObserver` buffering the responses. A client which lost connection could then reconnect and ask for the stream to be retransmitted. * `ExecutionHolder` returning the operationId to the client directly, and then client requesting results in separate RPCs, with more control over the response stream, instead of having it just pushed to it. ### Why are the changes needed? * Improve the working of `interrupt` * Refactoring that opens up possibilities of detaching query execution from a single RPC. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing Spark Connect CI covers the execution. Closes apache#41315 from juliuszsompolski/sc-execute-thread. Lead-authored-by: Juliusz Sompolski <julek@databricks.com> Co-authored-by: Hyukjin Kwon <gurwls223@gmail.com> Signed-off-by: Yuanjian Li <yuanjian.li@databricks.com>

github-actions bot added CONNECT CORE SQL labels May 25, 2023

juliuszsompolski changed the title ~~[SPARK-43755]~~ [SPARK-43755][CONNECT] Move execution out of SparkExecutePlanStreamHandler and to a different thread May 25, 2023

juliuszsompolski commented May 25, 2023

View reviewed changes

hvanhovell reviewed Jun 1, 2023

View reviewed changes

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala Outdated Show resolved Hide resolved

hvanhovell reviewed Jun 1, 2023

View reviewed changes

...server/src/main/scala/org/apache/spark/sql/connect/execution/SparkConnectPlanExecution.scala Outdated Show resolved Hide resolved

hvanhovell reviewed Jun 1, 2023

View reviewed changes

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala Outdated Show resolved Hide resolved

hvanhovell reviewed Jun 1, 2023

View reviewed changes

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala Outdated Show resolved Hide resolved

hvanhovell reviewed Jun 1, 2023

View reviewed changes

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala Outdated Show resolved Hide resolved

hvanhovell reviewed Jun 1, 2023

View reviewed changes

...tor/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecutionHolder.scala Outdated Show resolved Hide resolved

hvanhovell reviewed Jun 1, 2023

View reviewed changes

juliuszsompolski mentioned this pull request Jun 14, 2023

[SPARK-43923][CONNECT] Post listenerBus events during ExecutePlanRequest #41443

Closed

grundprinzip reviewed Jul 3, 2023

View reviewed changes

grundprinzip mentioned this pull request Jul 5, 2023

[SPARK-43879][CONNECT] Decouple handle command and send response on server side #41527

Closed

juliuszsompolski force-pushed the sc-execute-thread branch from 93ba7c2 to b3dd570 Compare July 6, 2023 17:00

juliuszsompolski added 3 commits July 7, 2023 00:30

move execution

3d5ee30

fix compile

2ba6d26

renames

bbf8761

juliuszsompolski force-pushed the sc-execute-thread branch from b3dd570 to bbf8761 Compare July 7, 2023 13:31

github-actions bot removed the CORE label Jul 7, 2023

dump

1a8db60

todo - deadlocks and needs debug

bbba4ca

grundprinzip reviewed Jul 10, 2023

View reviewed changes

juliuszsompolski added 2 commits July 11, 2023 13:24

passes tests, needs cleanup, needs a bit of thought about error handling

e4e3615

separate ExecuteHolder / ExecuteRunner; todo comments and docs

6f39e7b

grundprinzip reviewed Jul 12, 2023

View reviewed changes

juliuszsompolski added 7 commits July 12, 2023 11:35

fix hang and cancellation

20cb3dd

docs and comments

bcea9d4

nits

0f88963

scalastyle

2e6effa

change job tag pattern

e98db60

self review nits and cleanup

fee8838

add error class to documentation

0757c1c

github-actions bot added the DOCS label Jul 13, 2023

typo

a0d0792

rangadi reviewed Jul 13, 2023

View reviewed changes

juliuszsompolski added 3 commits July 14, 2023 00:33

Merge branch 'master' into sc-execute-thread

428dd77

fix OPERATION_CANCELED error class

9104557

Merge branch 'master' into sc-execute-thread

aa4093c

HyukjinKwon approved these changes Jul 14, 2023

View reviewed changes

HyukjinKwon reviewed Jul 14, 2023

View reviewed changes

...ector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala Outdated Show resolved Hide resolved

grundprinzip approved these changes Jul 14, 2023

View reviewed changes

Update connector/connect/server/src/main/scala/org/apache/spark/sql/c…

c5008b5

…onnect/service/SessionHolder.scala

xuanyuanking closed this in 18d0a27 Jul 14, 2023

This was referenced Jul 18, 2023

[SPARK-43755][CONNECT][MINOR] Open AdaptiveSparkPlanHelper.allChildren instead of using copy in MetricGenerator #42060

Closed

[SPARK-44422][CONNECT] Spark Connect fine grained interrupt #42009

Closed

dongjoon-hyun mentioned this pull request Aug 27, 2025

[SPARK-53339][CONNECT] Fix an issue which occurs when an operation in pending state is interrupted #52083

Open

[SPARK-43755][CONNECT] Move execution out of SparkExecutePlanStreamHandler and to a different thread #41315

[SPARK-43755][CONNECT] Move execution out of SparkExecutePlanStreamHandler and to a different thread #41315

Uh oh!

Conversation

juliuszsompolski commented May 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

juliuszsompolski commented May 25, 2023

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

juliuszsompolski commented May 26, 2023

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hvanhovell left a comment

Choose a reason for hiding this comment

Uh oh!

grundprinzip left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

juliuszsompolski commented Jul 7, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

grundprinzip left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

juliuszsompolski Jul 13, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

juliuszsompolski commented Jul 14, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

juliuszsompolski commented Jul 14, 2023

Uh oh!

juliuszsompolski commented May 25, 2023 •

edited

Loading

juliuszsompolski Jul 13, 2023 •

edited

Loading

juliuszsompolski commented Jul 14, 2023 •

edited

Loading