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 044cc6c [SPARK-58610] Support `cloneSession` in `SparkSession`
044cc6c is described below
commit 044cc6c0934ff34e2595c16ce939fd503a5cac42
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Aug 5 22:59:53 2026 -0700
[SPARK-58610] Support `cloneSession` in `SparkSession`
### What changes were proposed in this pull request?
This PR aims to support `cloneSession` API in `SparkSession` by using the
`CloneSession` RPC.
```swift
// Clone with a server-generated session ID
let cloned = try await spark.cloneSession()
// Clone with a custom session ID (must be a valid UUID)
let cloned = try await
spark.cloneSession("00112233-4455-6677-8899-aabbccddeeff")
```
- `SparkSession.cloneSession(_ newSessionID: String? = nil)` follows the
PySpark API shape (`cloneSession(new_session_id=None)`) and covers both Scala
overloads (`cloneSession()` / `cloneSession(sessionId:)`) via the optional
parameter.
- `SparkConnectClient.cloneSession` issues the `CloneSession` RPC and
validates that the returned session ID matches the requested one when provided,
like the Scala and Python clients.
- The existing `newSession()` semantics are unchanged, consistent with the
Scala client where `newSession()` (fresh empty session) and `cloneSession()`
(server-side state clone) are distinct APIs.
### Why are the changes needed?
The `CloneSession` RPC was added via
[SPARK-53455](https://github.com/apache/spark/commit/264ca4dc320c9de52ce53fdbdca8146ea9d1a5b6)
and is available since [Apache Spark 4.1.0
(2025-12-11)](https://github.com/apache/spark/releases/tag/v4.1.0). This allows
users to create a new independent session that inherits the original session's
server-side state (SQL configurations, temporary views, registered functions,
catalog state), which `newSession()` cannot provide.
### Does this PR introduce _any_ user-facing change?
Yes, this adds a new API, `SparkSession.cloneSession`. For old servers
without `CloneSession` RPC support (Apache Spark 4.0.x), the server error is
propagated like the Scala and Python clients.
### How was this patch tested?
Pass the CIs with the newly added integration tests, `cloneSession` and
`cloneSessionWithSessionID`, which verify that the cloned session inherits SQL
configurations and temporary views and that subsequent changes to either
session are isolated from the other. The tests are guarded to run on Apache
Spark 4.1+ servers, and were verified manually against a local Apache Spark
4.2.0 Connect server.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #485 from dongjoon-hyun/SPARK-58610.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../Documentation.docc/SparkSession.md | 1 +
Sources/SparkConnect/SparkConnectClient.swift | 22 ++++++++++++
Sources/SparkConnect/SparkSession.swift | 30 ++++++++++++++--
Sources/SparkConnect/TypeAliases.swift | 2 ++
Tests/SparkConnectTests/SparkSessionTests.swift | 40 ++++++++++++++++++++++
5 files changed, 93 insertions(+), 2 deletions(-)
diff --git a/Sources/SparkConnect/Documentation.docc/SparkSession.md
b/Sources/SparkConnect/Documentation.docc/SparkSession.md
index 914a686..6ffa6b9 100644
--- a/Sources/SparkConnect/Documentation.docc/SparkSession.md
+++ b/Sources/SparkConnect/Documentation.docc/SparkSession.md
@@ -35,6 +35,7 @@ let csvDf = spark.read.csv("path/to/file.csv")
- ``builder``
- ``newSession()``
+- ``cloneSession(_:)``
- ``stop()``
### Session Information
diff --git a/Sources/SparkConnect/SparkConnectClient.swift
b/Sources/SparkConnect/SparkConnectClient.swift
index bff6214..0ff8e46 100644
--- a/Sources/SparkConnect/SparkConnectClient.swift
+++ b/Sources/SparkConnect/SparkConnectClient.swift
@@ -997,6 +997,28 @@ public actor SparkConnectClient {
}
}
+ /// Clone the current session on the server side.
+ /// - Parameter newSessionID: An optional session ID for the cloned session
(must be a valid
+ /// UUID). If not provided, the server generates a new UUID.
+ /// - Returns: A ``CloneSessionResponse`` instance containing the new
session ID.
+ func cloneSession(_ newSessionID: String? = nil) async throws ->
CloneSessionResponse {
+ var request = CloneSessionRequest()
+ request.sessionID = self.sessionID!
+ request.userContext = self.userContext
+ request.clientType = self.clientType
+ if let newSessionID {
+ request.newSessionID = newSessionID
+ }
+ return try await withGPRC { client in
+ let service = SparkConnectService.Client(wrapping: client)
+ let response = try await service.cloneSession(request)
+ if let newSessionID, response.newSessionID != newSessionID {
+ throw SparkConnectError.InvalidArgument
+ }
+ return response
+ }
+ }
+
/// Parse a DDL string to ``Spark_Connect_DataType`` instance.
/// - Parameter ddlString: A string to parse.
/// - Returns: A ``Spark_Connect_DataType`` instance.
diff --git a/Sources/SparkConnect/SparkSession.swift
b/Sources/SparkConnect/SparkSession.swift
index aeaf480..907dd91 100644
--- a/Sources/SparkConnect/SparkSession.swift
+++ b/Sources/SparkConnect/SparkSession.swift
@@ -36,6 +36,9 @@ public actor SparkSession {
let client: SparkConnectClient
+ /// The connection string used to create this session.
+ let connection: String
+
/// Runtime configuration interface for Spark.
public let conf: RuntimeConf
@@ -44,10 +47,15 @@ public actor SparkSession {
/// Create a session that uses the specified connection string and userID.
/// - Parameters:
/// - connection: a string in a patter, `sc://{host}:{port}`
- init(_ connection: String) throws {
+ /// - sessionID: an optional session ID string. If given, this takes
precedence over the
+ /// `session_id` parameter in `connection`.
+ init(_ connection: String, _ sessionID: String? = nil) throws {
+ self.connection = connection
self.client = try SparkConnectClient(remote: connection)
// Since `Session ID` belongs to `SparkSession`, we handle this here.
- if connection.contains(regexSessionID) {
+ if let sessionID {
+ self.sessionID = sessionID
+ } else if connection.contains(regexSessionID) {
self.sessionID = connection.firstMatch(of:
regexSessionID)!.1.lowercased()
} else {
self.sessionID = UUID().uuidString.lowercased()
@@ -65,6 +73,24 @@ public actor SparkSession {
return try await SparkSession.builder.create()
}
+ /// Create a clone of this Spark Connect session on the server side. The
server-side session is
+ /// cloned with all its current state (SQL configurations, temporary views,
registered functions,
+ /// catalog state) copied over to a new independent session. The returned
cloned session is
+ /// isolated from this session - any subsequent changes to either session's
server-side state
+ /// will not be reflected in the other.
+ ///
+ /// This requires Apache Spark 4.1.0 or above Spark Connect servers.
+ /// - Parameter newSessionID: An optional session ID for the cloned session
(must be a valid
+ /// UUID). If not provided, the server generates a new UUID.
+ /// - Returns: A new ``SparkSession`` instance with the cloned session.
+ public func cloneSession(_ newSessionID: String? = nil) async throws ->
SparkSession {
+ let response = try await client.cloneSession(newSessionID)
+ let session = try SparkSession(self.connection, response.newSessionID)
+ let versionResponse = try await session.client.connect(session.sessionID)
+ await session.setVersion(versionResponse.sparkVersion.version)
+ return session
+ }
+
func setVersion(_ version: String) {
self.version = version
}
diff --git a/Sources/SparkConnect/TypeAliases.swift
b/Sources/SparkConnect/TypeAliases.swift
index 712ba2c..f7f4d83 100644
--- a/Sources/SparkConnect/TypeAliases.swift
+++ b/Sources/SparkConnect/TypeAliases.swift
@@ -20,6 +20,8 @@ typealias ActionType = Spark_Connect_MergeAction.ActionType
typealias Aggregate = Spark_Connect_Aggregate
typealias AnalyzePlanRequest = Spark_Connect_AnalyzePlanRequest
typealias AnalyzePlanResponse = Spark_Connect_AnalyzePlanResponse
+typealias CloneSessionRequest = Spark_Connect_CloneSessionRequest
+typealias CloneSessionResponse = Spark_Connect_CloneSessionResponse
typealias CollectMetrics = Spark_Connect_CollectMetrics
typealias Command = Spark_Connect_Command
typealias ConfigRequest = Spark_Connect_ConfigRequest
diff --git a/Tests/SparkConnectTests/SparkSessionTests.swift
b/Tests/SparkConnectTests/SparkSessionTests.swift
index e514750..ac68fc2 100644
--- a/Tests/SparkConnectTests/SparkSessionTests.swift
+++ b/Tests/SparkConnectTests/SparkSessionTests.swift
@@ -57,6 +57,46 @@ struct SparkSessionTests {
await newSpark.stop()
}
+ @Test
+ func cloneSession() async throws {
+ await SparkSession.builder.clear()
+ let spark = try await SparkSession.builder.getOrCreate()
+ if await spark.version >= "4.1" {
+ let viewName = "VIEW_" + UUID().uuidString.replacingOccurrences(of: "-",
with: "")
+ try await spark.conf.set("spark.test.cloneSession", "original")
+ try await spark.range(3).createTempView(viewName)
+
+ let cloned = try await spark.cloneSession()
+ #expect(cloned != spark)
+ #expect(cloned.sessionID != spark.sessionID)
+ // The cloned session inherits the original session's state.
+ #expect(try await cloned.conf.get("spark.test.cloneSession") ==
"original")
+ #expect(try await cloned.table(viewName).count() == 3)
+
+ // Subsequent changes are isolated between the two sessions.
+ try await cloned.conf.set("spark.test.cloneSession", "cloned")
+ #expect(try await spark.conf.get("spark.test.cloneSession") ==
"original")
+ #expect(try await spark.catalog.dropTempView(viewName))
+ #expect(try await cloned.table(viewName).count() == 3)
+ await cloned.stop()
+ }
+ await spark.stop()
+ }
+
+ @Test
+ func cloneSessionWithSessionID() async throws {
+ await SparkSession.builder.clear()
+ let spark = try await SparkSession.builder.getOrCreate()
+ if await spark.version >= "4.1" {
+ let newSessionID = UUID().uuidString.lowercased()
+ let cloned = try await spark.cloneSession(newSessionID)
+ #expect(cloned.sessionID == newSessionID)
+ #expect(try await cloned.range(1).count() == 1)
+ await cloned.stop()
+ }
+ await spark.stop()
+ }
+
@Test
func sessionID() async throws {
await SparkSession.builder.clear()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
