dongjoon-hyun opened a new pull request, #477:
URL: https://github.com/apache/spark-connect-swift/pull/477
### What changes were proposed in this pull request?
This PR supports reattachable execution in `SparkConnectClient` like the
Scala side
`org.apache.spark.sql.connect.client.ExecutePlanResponseReattachableIterator`
and the Python
side
`pyspark.sql.connect.client.reattach.ExecutePlanResponseReattachableIterator`.
1. A new `executePlanWithReattach` helper executes `ExecutePlan` with
`ReattachOptions.reattachable=true` and tracks `response_id`,
`server_side_session_id`, and
`ResultComplete` of the response stream.
2. When the response stream is broken by a retriable RPC error (governed by
the existing
`RetryPolicy.defaultPolicy` with exponential backoff, reset whenever an
attempt receives a
new response), or ends gracefully without a `ResultComplete` message, the
stream is resumed
with `ReattachExecute` after the last received response on a fresh
channel.
3. When `ReattachExecute` fails with `INVALID_HANDLE.OPERATION_NOT_FOUND` or
`INVALID_HANDLE.SESSION_NOT_FOUND` before receiving any response, the
initial `ExecutePlan`
didn't reach the server and is started over. After receiving responses,
it is surfaced as
the new `SparkConnectError.invalidState`, which is also thrown when the
server side session
ID changes in the middle of a stream.
4. The server side buffer is released with a `release_all` `ReleaseExecute`
when
`ResultComplete` arrives, and in a best-effort way when the execution is
abandoned after an
error. Incremental `release_until` calls are a server-side optimization
hint (the server
auto-releases the sent responses beyond
`spark.connect.execute.reattachable.observerRetryBufferSize`)
and are left as a follow-up.
5. All three `ExecutePlan` consumption paths (`DataFrame.execute`,
`DataFrame.count`, and
`SparkConnectClient.execute`) use the new helper, superseding the
temporary
retry-until-the-first-response pattern introduced by SPARK-58541.
### Why are the changes needed?
`ExecutePlan` was a single non-resumable server stream. Once the first
response arrived, any
broken connection failed the whole query without a recovery path, and the
temporary retry of
SPARK-58541 could not help because re-executing the plan can have side
effects. Reattachable
execution is the mechanism the Scala and Python clients use by default
(Spark 3.5+ servers),
and this PR brings the Swift client to parity.
### Does this PR introduce _any_ user-facing change?
Yes, executions now survive transient response stream breaks by resuming
after the last
received response instead of failing, and the server releases its response
buffers explicitly
at the end of each execution. Successful executions return the same results
as before.
### How was this patch tested?
Pass the CIs with the newly added `ReattachableExecuteTests` test suite
covering the
reattachable `ExecutePlan`, `ReattachExecute`, and `ReleaseExecute` request
builders. All
existing test suites pass with a live Spark Connect server 4.2.0, now
exercising every
`ExecutePlan` through the reattachable path. Note that a real mid-stream
disconnection is not
reproducible reliably in the integration tests (the server ends idle
reattachable streams only
after `spark.connect.execute.reattachable.senderMaxStreamDuration`, 2
minutes by default), so
the reattach path is verified indirectly through the graceful stream-end
handling.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
