This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-connect-swift.git
The following commit(s) were added to refs/heads/main by this push:
new c627124 [SPARK-58543] Support reattachable execution with
`ReattachExecute` and `ReleaseExecute`
c627124 is described below
commit c627124ccbc05927aa0af4a42924154a41f2e6ce
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Aug 4 07:12:17 2026 -0700
[SPARK-58543] Support reattachable execution with `ReattachExecute` and
`ReleaseExecute`
### 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
Closes #477 from dongjoon-hyun/SPARK-58543.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
Sources/SparkConnect/DataFrame+Actions.swift | 144 +++++++--------
Sources/SparkConnect/DataFrame.swift | 3 +-
.../SparkConnectClient+ReattachableExecute.swift | 196 +++++++++++++++++++++
Sources/SparkConnect/SparkConnectClient.swift | 23 +--
Sources/SparkConnect/SparkConnectError.swift | 5 +-
Sources/SparkConnect/TypeAliases.swift | 3 +
.../ReattachableExecuteTests.swift | 72 ++++++++
7 files changed, 342 insertions(+), 104 deletions(-)
diff --git a/Sources/SparkConnect/DataFrame+Actions.swift
b/Sources/SparkConnect/DataFrame+Actions.swift
index 845685e..89a004e 100644
--- a/Sources/SparkConnect/DataFrame+Actions.swift
+++ b/Sources/SparkConnect/DataFrame+Actions.swift
@@ -36,76 +36,65 @@ extension DataFrame {
/// Execute the plan and try to fill `schema` and `batches`.
private func execute() async throws {
- // `ExecutePlan` can have side effects on the server. Retry only until the
first
- // response arrives because the operation is not started before that.
- let received = Atomic(false)
- try await withRetry(shouldRetry: { !received.load(ordering: .relaxed) &&
RetryPolicy.canRetry($0) }) {
- // Clear all existing batches.
- self.batches.removeAll()
-
- try await withGPRC(retryable: false) { client in
- let service = Spark_Connect_SparkConnectService.Client(wrapping:
client)
- try await
service.executePlan(spark.client.getExecutePlanRequest(plan)) {
- response in
- for try await m in response.messages {
- received.store(true, ordering: .relaxed)
- if m.hasSchema {
- // The original schema should arrive before ArrowBatches
- await self.setSchema(m.schema)
- }
- let ipcStreamBytes = m.arrowBatch.data
- if !ipcStreamBytes.isEmpty && m.arrowBatch.rowCount > 0 {
- let IPC_CONTINUATION_TOKEN = Int32(-1)
- let totalSize = Int64(ipcStreamBytes.count)
- // Schema
- guard totalSize >= 8,
- ipcStreamBytes[0..<4].int32 == IPC_CONTINUATION_TOKEN
- else {
- throw SparkConnectError.InvalidArrowData
- }
- let schemaSize = Int64(ipcStreamBytes[4..<8].int32)
- guard schemaSize >= 0, 8 + schemaSize + 4 <= totalSize else {
- throw SparkConnectError.InvalidArrowData
- }
- let schema = Data(ipcStreamBytes[8..<(8 + schemaSize)])
-
- // Arrow IPC Data
- guard ipcStreamBytes[(8 + schemaSize)..<(8 + schemaSize +
4)].int32
- == IPC_CONTINUATION_TOKEN
- else {
- throw SparkConnectError.InvalidArrowData
- }
- var pos: Int64 = 8 + schemaSize + 4
- guard pos + 4 <= totalSize else {
- throw SparkConnectError.InvalidArrowData
- }
- let dataHeaderSize = Int64(ipcStreamBytes[pos..<(pos + 4)].int32)
- pos += 4
- guard dataHeaderSize >= 0, pos + dataHeaderSize + 8 <= totalSize
else {
- throw SparkConnectError.InvalidArrowData
- }
- let dataHeader = Data(ipcStreamBytes[pos..<(pos +
dataHeaderSize)])
- pos += dataHeaderSize
- let dataBodySize = totalSize - pos - 8
- let dataBody = Data(ipcStreamBytes[pos..<(pos + dataBodySize)])
-
- // Read ArrowBatches
- let reader = ArrowReader()
- let arrowResult = ArrowReader.makeArrowReaderResult()
- if case .failure(let error) = reader.fromMessage(
- schema, dataBody: Data(), result: arrowResult)
- {
- throw error
- }
- if case .failure(let error) = reader.fromMessage(
- dataHeader, dataBody: dataBody, result: arrowResult)
- {
- throw error
- }
- await self.addBatches(arrowResult.batches)
- }
- }
+ // Clear all existing batches.
+ self.batches.removeAll()
+
+ try await
spark.client.executePlanWithReattach(spark.client.getExecutePlanRequest(plan)) {
+ m in
+ if m.hasSchema {
+ // The original schema should arrive before ArrowBatches
+ await self.setSchema(m.schema)
+ }
+ let ipcStreamBytes = m.arrowBatch.data
+ if !ipcStreamBytes.isEmpty && m.arrowBatch.rowCount > 0 {
+ let IPC_CONTINUATION_TOKEN = Int32(-1)
+ let totalSize = Int64(ipcStreamBytes.count)
+ // Schema
+ guard totalSize >= 8,
+ ipcStreamBytes[0..<4].int32 == IPC_CONTINUATION_TOKEN
+ else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ let schemaSize = Int64(ipcStreamBytes[4..<8].int32)
+ guard schemaSize >= 0, 8 + schemaSize + 4 <= totalSize else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ let schema = Data(ipcStreamBytes[8..<(8 + schemaSize)])
+
+ // Arrow IPC Data
+ guard ipcStreamBytes[(8 + schemaSize)..<(8 + schemaSize + 4)].int32
+ == IPC_CONTINUATION_TOKEN
+ else {
+ throw SparkConnectError.InvalidArrowData
}
+ var pos: Int64 = 8 + schemaSize + 4
+ guard pos + 4 <= totalSize else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ let dataHeaderSize = Int64(ipcStreamBytes[pos..<(pos + 4)].int32)
+ pos += 4
+ guard dataHeaderSize >= 0, pos + dataHeaderSize + 8 <= totalSize else {
+ throw SparkConnectError.InvalidArrowData
+ }
+ let dataHeader = Data(ipcStreamBytes[pos..<(pos + dataHeaderSize)])
+ pos += dataHeaderSize
+ let dataBodySize = totalSize - pos - 8
+ let dataBody = Data(ipcStreamBytes[pos..<(pos + dataBodySize)])
+
+ // Read ArrowBatches
+ let reader = ArrowReader()
+ let arrowResult = ArrowReader.makeArrowReaderResult()
+ if case .failure(let error) = reader.fromMessage(
+ schema, dataBody: Data(), result: arrowResult)
+ {
+ throw error
+ }
+ if case .failure(let error) = reader.fromMessage(
+ dataHeader, dataBody: dataBody, result: arrowResult)
+ {
+ throw error
+ }
+ await self.addBatches(arrowResult.batches)
}
}
}
@@ -185,20 +174,9 @@ extension DataFrame {
public func count() async throws -> Int64 {
let counter = Atomic(Int64(0))
- // `ExecutePlan` can have side effects on the server. Retry only until the
first
- // response arrives because the operation is not started before that.
- let received = Atomic(false)
- try await withRetry(shouldRetry: { !received.load(ordering: .relaxed) &&
RetryPolicy.canRetry($0) }) {
- try await withGPRC(retryable: false) { client in
- let service = Spark_Connect_SparkConnectService.Client(wrapping:
client)
- try await
service.executePlan(spark.client.getExecutePlanRequest(plan)) {
- response in
- for try await m in response.messages {
- received.store(true, ordering: .relaxed)
- counter.add(m.arrowBatch.rowCount, ordering: .relaxed)
- }
- }
- }
+ try await
spark.client.executePlanWithReattach(spark.client.getExecutePlanRequest(plan)) {
+ m in
+ counter.add(m.arrowBatch.rowCount, ordering: .relaxed)
}
return counter.load(ordering: .relaxed)
}
diff --git a/Sources/SparkConnect/DataFrame.swift
b/Sources/SparkConnect/DataFrame.swift
index 754f148..26ee5cd 100644
--- a/Sources/SparkConnect/DataFrame.swift
+++ b/Sources/SparkConnect/DataFrame.swift
@@ -305,10 +305,9 @@ public actor DataFrame: Sendable {
}
func withGPRC<Result: Sendable>(
- retryable: Bool = true,
_ f: (GRPCClient<GRPCNIOTransportHTTP2.HTTP2ClientTransport.Posix>) async
throws -> Result
) async throws -> Result {
- try await withRetry(shouldRetry: { retryable && RetryPolicy.canRetry($0)
}) {
+ try await withRetry {
try await withGRPCClient(
transport: .http2NIOPosix(
target: .dns(host: spark.client.host, port: spark.client.port),
diff --git a/Sources/SparkConnect/SparkConnectClient+ReattachableExecute.swift
b/Sources/SparkConnect/SparkConnectClient+ReattachableExecute.swift
new file mode 100644
index 0000000..2fc05a6
--- /dev/null
+++ b/Sources/SparkConnect/SparkConnectClient+ReattachableExecute.swift
@@ -0,0 +1,196 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+import GRPCCore
+import Synchronization
+
+/// The stream progress of a reattachable `ExecutePlan` execution, shared with
the `@Sendable`
+/// response stream handlers.
+private final class ReattachableExecuteState: Sendable {
+ private struct Progress {
+ var lastResponseID: String? = nil
+ var serverSideSessionID: String? = nil
+ var responseCount = 0
+ var resultComplete = false
+ }
+
+ private let progress = Mutex(Progress())
+
+ /// Record a received response and assert that the server side session has
not changed.
+ func record(_ response: ExecutePlanResponse) throws {
+ try progress.withLock { p in
+ if let id = p.serverSideSessionID, id != response.serverSideSessionID {
+ throw SparkConnectError.invalidState(
+ SparkConnectError.Details(
+ message:
+ "Server side session ID changed. Create a new SparkSession to
continue. "
+ + "(Old: \(id), New: \(response.serverSideSessionID))"))
+ }
+ p.serverSideSessionID = response.serverSideSessionID
+ p.lastResponseID = response.responseID
+ p.responseCount += 1
+ if case .resultComplete = response.responseType {
+ p.resultComplete = true
+ }
+ }
+ }
+
+ var lastResponseID: String? { progress.withLock { $0.lastResponseID } }
+ var responseCount: Int { progress.withLock { $0.responseCount } }
+ var resultComplete: Bool { progress.withLock { $0.resultComplete } }
+}
+
+extension SparkConnectClient {
+ /// Return a copy of `request` whose `request_options` asks the server to
keep the execution
+ /// reattachable with `ReattachExecute` after the response stream is broken.
+ static func getReattachableExecutePlanRequest(
+ _ request: ExecutePlanRequest
+ ) -> ExecutePlanRequest {
+ var reattachableRequest = request
+ var option = ExecutePlanRequest.RequestOption()
+ var reattachOptions = ReattachOptions()
+ reattachOptions.reattachable = true
+ option.reattachOptions = reattachOptions
+ reattachableRequest.requestOptions.append(option)
+ return reattachableRequest
+ }
+
+ /// Create a ``ReattachExecuteRequest`` resuming the response stream of
`request` after
+ /// `lastResponseID`, or from the start if nil.
+ static func getReattachExecuteRequest(
+ _ request: ExecutePlanRequest, _ lastResponseID: String?
+ ) -> ReattachExecuteRequest {
+ var reattach = ReattachExecuteRequest()
+ reattach.sessionID = request.sessionID
+ reattach.userContext = request.userContext
+ reattach.operationID = request.operationID
+ reattach.clientType = request.clientType
+ if let lastResponseID {
+ reattach.lastResponseID = lastResponseID
+ }
+ return reattach
+ }
+
+ /// Create a ``ReleaseExecuteRequest`` releasing the whole operation of
`request` on the server.
+ static func getReleaseAllRequest(_ request: ExecutePlanRequest) ->
ReleaseExecuteRequest {
+ var release = ReleaseExecuteRequest()
+ release.sessionID = request.sessionID
+ release.userContext = request.userContext
+ release.operationID = request.operationID
+ release.clientType = request.clientType
+ release.releaseAll = ReleaseExecuteRequest.ReleaseAll()
+ return release
+ }
+
+ /// Execute `request` as a reattachable execution
+ /// like
`org.apache.spark.sql.connect.client.ExecutePlanResponseReattachableIterator`.
+ ///
+ /// The plan is executed with `ReattachOptions.reattachable=true`. When the
response stream is
+ /// broken by a retriable RPC error, or ends without a `ResultComplete`
message, the stream is
+ /// resumed with `ReattachExecute` after the last received response. The
retriable errors are
+ /// governed by ``RetryPolicy/defaultPolicy`` whose backoff is reset
whenever an attempt
+ /// receives a new response. The server side buffer is released with a
`release_all`
+ /// `ReleaseExecute` at the end.
+ /// - Parameters:
+ /// - request: An ``ExecutePlanRequest`` to execute.
+ /// - processResponse: A handler invoked for each ``ExecutePlanResponse``
in order.
+ func executePlanWithReattach(
+ _ request: ExecutePlanRequest,
+ _ processResponse: @Sendable @escaping (ExecutePlanResponse) async throws
-> Void
+ ) async throws {
+ let initialRequest = Self.getReattachableExecutePlanRequest(request)
+ let state = ReattachableExecuteState()
+
+ @Sendable func consume(
+ _ response: GRPCCore.StreamingClientResponse<ExecutePlanResponse>
+ ) async throws {
+ for try await m in response.messages {
+ try state.record(m)
+ try await processResponse(m)
+ }
+ }
+
+ var retryState = RetryPolicyState(.defaultPolicy)
+ var responseCountAtLastError = 0
+ var reattach = false
+ while !state.resultComplete {
+ do {
+ try await withGPRC(retryable: false) { client in
+ let service = SparkConnectService.Client(wrapping: client)
+ if reattach {
+ let reattachRequest =
+ Self.getReattachExecuteRequest(initialRequest,
state.lastResponseID)
+ try await service.reattachExecute(reattachRequest, onResponse:
consume)
+ } else {
+ try await service.executePlan(initialRequest, onResponse: consume)
+ }
+ if state.resultComplete {
+ _ = try await
service.releaseExecute(Self.getReleaseAllRequest(initialRequest))
+ }
+ }
+ // When the stream ended gracefully without `ResultComplete`, the
server has more
+ // responses and the loop continues with `ReattachExecute`.
+ reattach = true
+ } catch {
+ reattach = true
+ let responseCount = state.responseCount
+ if Self.isOperationNotFound(error) {
+ guard responseCount == 0 else {
+ await releaseAll(initialRequest)
+ throw SparkConnectError.invalidState(
+ SparkConnectError.Details(
+ message:
+ "OPERATION_NOT_FOUND on the server but responses were
already received from it."))
+ }
+ // The initial `ExecutePlan` didn't reach the server. Start over
with `ExecutePlan`.
+ reattach = false
+ } else if !RetryPolicy.canRetry(error) {
+ await releaseAll(initialRequest)
+ throw error
+ }
+ // An attempt which received new responses resets the backoff.
+ if responseCount > responseCountAtLastError {
+ responseCountAtLastError = responseCount
+ retryState = RetryPolicyState(.defaultPolicy)
+ }
+ guard !Task.isCancelled, let wait = retryState.nextAttempt(error) else
{
+ await releaseAll(initialRequest)
+ throw error
+ }
+ try await Task.sleep(for: wait)
+ }
+ }
+ }
+
+ /// Inform the server to release the operation of `request` after giving up,
ignoring failures
+ /// because the server releases abandoned executions by itself.
+ private func releaseAll(_ request: ExecutePlanRequest) async {
+ try? await withGPRC(retryable: false) { client in
+ let service = SparkConnectService.Client(wrapping: client)
+ _ = try await service.releaseExecute(Self.getReleaseAllRequest(request))
+ }
+ }
+
+ /// Return true if the error means that the operation of the initial
`ExecutePlan` doesn't
+ /// exist on the server.
+ private static func isOperationNotFound(_ error: Error) -> Bool {
+ guard let error = error as? RPCError else { return false }
+ return error.message.contains("INVALID_HANDLE.OPERATION_NOT_FOUND")
+ || error.message.contains("INVALID_HANDLE.SESSION_NOT_FOUND")
+ }
+}
diff --git a/Sources/SparkConnect/SparkConnectClient.swift
b/Sources/SparkConnect/SparkConnectClient.swift
index 79411ef..8449f4c 100644
--- a/Sources/SparkConnect/SparkConnectClient.swift
+++ b/Sources/SparkConnect/SparkConnectClient.swift
@@ -24,7 +24,6 @@ import Foundation
import GRPCCore
import GRPCNIOTransportHTTP2
import GRPCProtobuf
-import Synchronization
/// Conceptually the remote spark session that communicates with the server
public actor SparkConnectClient {
@@ -858,23 +857,11 @@ public actor SparkConnectClient {
@discardableResult
func execute(_ sessionID: String, _ command: Command) async throws ->
[ExecutePlanResponse] {
- // `ExecutePlan` can have side effects on the server. Retry only until the
first
- // response arrives because the operation is not started before that.
- let received = Atomic(false)
- try await withRetry(shouldRetry: { !received.load(ordering: .relaxed) &&
RetryPolicy.canRetry($0) }) {
- self.result.removeAll()
- try await withGPRC(retryable: false) { client in
- let service = SparkConnectService.Client(wrapping: client)
- var plan = Plan()
- plan.opType = .command(command)
- try await service.executePlan(getExecutePlanRequest(plan)) {
- response in
- for try await m in response.messages {
- received.store(true, ordering: .relaxed)
- await self.addResponse(m)
- }
- }
- }
+ self.result.removeAll()
+ var plan = Plan()
+ plan.opType = .command(command)
+ try await executePlanWithReattach(getExecutePlanRequest(plan)) { m in
+ await self.addResponse(m)
}
return result
}
diff --git a/Sources/SparkConnect/SparkConnectError.swift
b/Sources/SparkConnect/SparkConnectError.swift
index f45bb62..f20459b 100644
--- a/Sources/SparkConnect/SparkConnectError.swift
+++ b/Sources/SparkConnect/SparkConnectError.swift
@@ -55,6 +55,7 @@ public enum SparkConnectError: Error, Sendable, Equatable {
case invalidArgument(Details)
case invalidArrowData(Details)
case invalidSessionID(Details)
+ case invalidState(Details)
case invalidType(Details)
case invalidViewName(Details)
case localRelationTooLarge(Details)
@@ -75,6 +76,7 @@ public enum SparkConnectError: Error, Sendable, Equatable {
public static var InvalidArgument: SparkConnectError {
.invalidArgument(Details()) }
public static var InvalidArrowData: SparkConnectError {
.invalidArrowData(Details()) }
public static var InvalidSessionID: SparkConnectError {
.invalidSessionID(Details()) }
+ public static var InvalidState: SparkConnectError { .invalidState(Details())
}
public static var InvalidType: SparkConnectError { .invalidType(Details()) }
public static var InvalidViewName: SparkConnectError {
.invalidViewName(Details()) }
public static var LocalRelationTooLarge: SparkConnectError {
.localRelationTooLarge(Details()) }
@@ -95,7 +97,7 @@ public enum SparkConnectError: Error, Sendable, Equatable {
case .catalogNotFound(let details), .columnNotFound(let details),
.dataSourceNotFound(let details), .invalidArgument(let details),
.invalidArrowData(let details), .invalidSessionID(let details),
- .invalidType(let details), .invalidViewName(let details),
+ .invalidState(let details), .invalidType(let details),
.invalidViewName(let details),
.localRelationTooLarge(let details), .outputTypeUnspecified(let details),
.parseSyntaxError(let details), .schemaNotFound(let details),
.sessionClosed(let details), .sqlConfNotFound(let details),
@@ -125,6 +127,7 @@ public enum SparkConnectError: Error, Sendable, Equatable {
case .invalidArgument: return "invalidArgument"
case .invalidArrowData: return "invalidArrowData"
case .invalidSessionID: return "invalidSessionID"
+ case .invalidState: return "invalidState"
case .invalidType: return "invalidType"
case .invalidViewName: return "invalidViewName"
case .localRelationTooLarge: return "localRelationTooLarge"
diff --git a/Sources/SparkConnect/TypeAliases.swift
b/Sources/SparkConnect/TypeAliases.swift
index f3027df..1099931 100644
--- a/Sources/SparkConnect/TypeAliases.swift
+++ b/Sources/SparkConnect/TypeAliases.swift
@@ -52,7 +52,10 @@ typealias Plan = Spark_Connect_Plan
typealias Project = Spark_Connect_Project
typealias Range = Spark_Connect_Range
typealias Read = Spark_Connect_Read
+typealias ReattachExecuteRequest = Spark_Connect_ReattachExecuteRequest
+typealias ReattachOptions = Spark_Connect_ReattachOptions
typealias Relation = Spark_Connect_Relation
+typealias ReleaseExecuteRequest = Spark_Connect_ReleaseExecuteRequest
typealias Repartition = Spark_Connect_Repartition
typealias RepartitionByExpression = Spark_Connect_RepartitionByExpression
typealias Sample = Spark_Connect_Sample
diff --git a/Tests/SparkConnectTests/ReattachableExecuteTests.swift
b/Tests/SparkConnectTests/ReattachableExecuteTests.swift
new file mode 100644
index 0000000..41be4db
--- /dev/null
+++ b/Tests/SparkConnectTests/ReattachableExecuteTests.swift
@@ -0,0 +1,72 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+import Testing
+
+@testable import SparkConnect
+
+/// A test suite for the reattachable execution request builders
+@Suite(.serialized)
+struct ReattachableExecuteTests {
+ static func executePlanRequest() -> ExecutePlanRequest {
+ var request = ExecutePlanRequest()
+ request.clientType = "swift"
+ request.userContext = "user".toUserContext
+ request.sessionID = "session-id"
+ request.operationID = "operation-id"
+ return request
+ }
+
+ @Test
+ func reattachableExecutePlanRequest() {
+ let request =
SparkConnectClient.getReattachableExecutePlanRequest(Self.executePlanRequest())
+ #expect(request.requestOptions.count == 1)
+ #expect(request.requestOptions[0].reattachOptions.reattachable)
+ #expect(request.sessionID == "session-id")
+ #expect(request.operationID == "operation-id")
+ }
+
+ @Test
+ func reattachExecuteRequest() {
+ let request =
+ SparkConnectClient.getReattachExecuteRequest(Self.executePlanRequest(),
"response-id")
+ #expect(request.sessionID == "session-id")
+ #expect(request.userContext.userID == "user")
+ #expect(request.operationID == "operation-id")
+ #expect(request.clientType == "swift")
+ #expect(request.hasLastResponseID)
+ #expect(request.lastResponseID == "response-id")
+ }
+
+ @Test
+ func reattachExecuteRequestFromStart() {
+ let request =
SparkConnectClient.getReattachExecuteRequest(Self.executePlanRequest(), nil)
+ #expect(!request.hasLastResponseID)
+ }
+
+ @Test
+ func releaseAllRequest() {
+ let request =
SparkConnectClient.getReleaseAllRequest(Self.executePlanRequest())
+ #expect(request.sessionID == "session-id")
+ #expect(request.userContext.userID == "user")
+ #expect(request.operationID == "operation-id")
+ #expect(request.clientType == "swift")
+ #expect(request.release == .releaseAll(ReleaseExecuteRequest.ReleaseAll()))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
