[SPARK-44421][SPARK-44423][CONNECT] Reattachable execution in Spark Connect

[juliuszsompolski]

What changes were proposed in this pull request?

Implement server support and scala client for reattachable execution with two new RPCs:

• ReattachExecute
• ReleaseExecute

With reattachable execution, if the response stream of ExecutePlan RPC gets broken, it can be reattached to with ReattachExecute RPC. The server now holds a bit of a backtrack buffer, in case the client coming back with reattach after error did not receive the latest sent responses, and have to backtrack. The client can sends ReleaseExecute responses to let the server know what has already been consumed and can be freed.

Server: ExecuteResponseObserver

If ExecuteHolder is reattachable, ExecuteResponseSender is modified to not throw away responses that were returned to ExecuteGrpcResponseSender immediately, but keep a buffer of spark.connect.execute.reattachable.observerRetryBufferSize (default 1 MB) responses behind the last returned response. This facilitates returning consumer being able to backtrack in case the last returned response did not successfully arrive at the client.

Server: ExecuteGrpcResponseSender

If ExecuteHolder is reattachable, ExecuteGrpcResponseSender is modified to gracefully end the RPC response stream after a given time (spark.connect.execute.reattachable.senderMaxStreamDuration, default 5 minutes) or size of responses (spark.connect.execute.reattachable.senderMaxStreamSize, default 1 GB). After that the response is completed, and client needs to come back with a ReattachExecute RPC to continue. Clients know whether it's the real end of the results, because ExecuteThreadRunner now puts a marker ResultComplete message at the real end of the stream.

Cutting the stream manually in a graceful way like that can help prevent errors related to long open streams. It also helps with long held idle stream that may get broken because of idleness. The client is able to handle these errors, but this also provides a cleaner way to minimize the number of network errors.

ExecuteGrpcResponseSender also implements flow control of responses sent to the GRPC stream. Normally, grpcResponseObserver.onNext would never block. This doesn't however mean that it would send all responses immediately. The responses could be queued if the client does not keep up receiving and processing them. This queue can grow ubounded, and we don't know what was actually sent, so we don't know how much of a buffer ExecuteResponseObserver really needs to keep. Flow control utilizes functions of ServerCallStreamObserver, which provides means to check if the stream is actually ready to send data, and callbacks to notify when it becomes ready. ExecuteGrpcResponseSender will wait with grpcResponseObserver.onNext until it becomes ready.

This also helps with clients that would abandon their result stream. Now we will block from sending the responses, and time out the RPC after senderMaxStreamDuration, instead of pushing the responses to a GRPC outbound queue that is outside our control. The TODO server mechanism would then remove the execution with no RPC attached. Note: in such a situation, responses will still pile up in ExecuteResponseObserver; there is unfortunately no flow control of the Spark task results being pushed to it.

Client: ExecutePlanResponseReattachableIterator

Client wraps the whole implementation inside ExecutePlanResponseReattachableIterator. It's a wrapper that starts with an ExecutePlan result iterator, but if this iterator either finishes without a ResultComplete message, or an intermittent network error occurs, it can acquire a new iterator to continue where it left off: by tracking the last received response, it can request a new iterator from ReattachExecute that starts after that.

The client sends ReleaseExecute RPCs asynchronously to release the responses it already consumed (potentially faster than the ExecuteResponseObserver would release from its backtrack buffer). The client doesn't need to check their result or block on them. If something fails, the executions will get cleaned by the TODO cleanup mechanism in the server.

TODO as followup

• What remains TODO for a followup PR is server mechanism for dealing with executions that have been abandoned - the client did not come back with ReatachExecute, and didn't free the execution with ReleaseExecute. This will be done by an operation manager that monitors all active executions, and interrupts and removes ones that were not released, but don't have any client RPC attached to them for a while.
• Python client implementation is a followup.

Why are the changes needed?

Robust handling of long running queries in face of intermittent network glitches.

Does this PR introduce any user-facing change?

The re-attachable execution mechanism is not exposed in any public API. Several internal configs have been added:

Connect server configs:

• spark.connect.execute.reattachable.senderMaxStreamDuration
• spark.connect.execute.reattachable.senderMaxStreamSize
• spark.connect.execute.reattachable.observerRetryBufferSize

Connect scala client configuration element: useReattachableExecute

How was this patch tested?

Existing E2E suite utilize re-attachable execution. The tests have been configure to exercise reattaching by using an extremely low senderMaxStreamDuration and senderMaxStreamSize.

[HyukjinKwon]

Should it be CloseableIterator instead?

[juliuszsompolski]

The regular iterator returned from regular executePlan is not a closeable iterator, and grpc-java doesn't seem to play nicely with closeable iterators in general: grpc/grpc-java#2409
I think the best I can do in Java is make sure that internal users in the client consume it... and over places that give control over it to user (like Dataset.toLocalIterator) there's no good control.
If an iterator is left open and idle:

• the server will close the RPC stream after 5 minutes (STREAM_MAX_TIME)
• then the server will tear down executions without attached RPC stream after another 5 minutes (TODO)

[HyukjinKwon]

qq why do we do this with StreamObserver? can't we just retry with retry you defined below?

[juliuszsompolski]

We want the client to call the releaseExecute async and don't block from continuing with the results on getting an ack that it executed. retry would be a blocking iteration... unless I mixed something up and I could use it for async as well.

[grundprinzip]

Did a first pass on the server side.

[grundprinzip]

Completed is a weird term here. But I don't have better suggestions immediately either.

[juliuszsompolski]

the completed here comes from server doing an onComplete on the stream... couldn't find a better word either.

[grundprinzip]

Is the total response size only update when we remove the items?

[juliuszsompolski]

This is indeed calculated on the fly when removed, and iterates back from the returned response until retryBufferSize is kept. This could be smarter, we could keep some "second finger" on the oldest buffered, and then cache the total between that and the highest consumed (our "first finger"), but it does actually get a bit more complicated when after a reattach the client backtracks (so our "first finger" moves back)

[juliuszsompolski]

Given that it's a total of 1 MB of responses, even if they were tiny (100 bytes?), it's an iteration over at most thousands of elements, and usually a few elements, so I thought it's not worth the complication for the optimization.

[HyukjinKwon]

https://github.com/juliuszsompolski/apache-spark/actions/runs/5720841151/job/15504640463

This fails. Since I am also actively working on this item together, I will just merge with ignoring this for now. Should be easy to fix. I will leave this to you @juliuszsompolski

[HyukjinKwon]

Merged to master and branch-3.5.

[ueshin]

Hi, this seems to break 3.5 build.

[error] /home/runner/work/spark/spark/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala:328:9: not found: value logDebug
[error]         logDebug(s"Session not found: ($userId, $sessionId)")
[error]         ^
[error] one error found

[HyukjinKwon]

Sure, made a followup: #42254

[juliuszsompolski]

We call hasNext above to assure that it's not empty at this point.

…

On Tue, Aug 1, 2023, 09:36 Hyukjin Kwon ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala <#42228 (comment)>: > + // hasNext will trigger reattach in case the stream completed without responseComplete + if (!hasNext()) { + throw new java.util.NoSuchElementException() + } + + // Get next response, possibly triggering reattach in case of stream error. + var firstTry = true + val ret = retry { + if (firstTry) { + // on first try, we use the existing iterator. + firstTry = false + } else { + // on retry, the iterator is borked, so we need a new one + iterator = rawBlockingStub.reattachExecute(createReattachExecuteRequest()) + } + iterator.next() qq: can this iterator empty? — Reply to this email directly, view it on GitHub <#42228 (review)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AF64GG7M7ZLJRJY53M6GDOLXTCWZPANCNFSM6AAAAAA25LFB6E> . You are receiving this because you were mentioned.Message ID: ***@***.***>