You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "vicennial (via GitHub)" <gi...@apache.org> on 2025/09/02 19:52:34 UTC
[PR] [WIP[SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
vicennial opened a new pull request, #52200:
URL: https://github.com/apache/spark/pull/52200
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2352752235
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1139,4 +1187,8 @@ service SparkConnectService {
// FetchErrorDetails retrieves the matched exception with details based on a provided error id.
rpc FetchErrorDetails(FetchErrorDetailsRequest) returns (FetchErrorDetailsResponse) {}
+
+ // Clone a session. Creates a new session that shares the same configuration and state
Review Comment:
Sure
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on PR #52200:
URL: https://github.com/apache/spark/pull/52200#issuecomment-3397888844
Thanks, merging to master!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2388960042
##########
sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/SparkSessionCloneSuite.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.connect
+
+import java.util.UUID
+
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, RemoteSparkSession}
+
+class SparkSessionCloneSuite extends ConnectFunSuite with RemoteSparkSession {
+
+ test("cloneSession() creates independent session with auto-generated ID") {
+ spark.sql("SET spark.test.original = 'value1'")
+
+ val clonedSession = spark.cloneSession()
+
+ // Sessions should have different IDs
+ assert(spark.sessionId !== clonedSession.sessionId)
+
+ // Modify original session setting
+ spark.sql("SET spark.test.original = 'modified_original'")
+
+ // Verify original session has the modified value
+ val originalFinal = spark.sql("SET spark.test.original").collect().head.getString(1)
+ assert(originalFinal === "'modified_original'")
+
+ // Verify cloned session retains the original value
+ val clonedValue = clonedSession.sql("SET spark.test.original").collect().head.getString(1)
+ assert(clonedValue === "'value1'")
+ }
+
+ test("cloneSession(sessionId) creates session with specified ID") {
+ val customSessionId = UUID.randomUUID().toString
+
+ val clonedSession = spark.cloneSession(customSessionId)
+
+ // Verify the cloned session has the specified ID
+ assert(clonedSession.sessionId === customSessionId)
+ assert(spark.sessionId !== clonedSession.sessionId)
+ }
+
+ test("invalid session ID format throws exception") {
Review Comment:
I suppose it is covered by server side tests.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317195804
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectCloneSessionSuite.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import java.util.UUID
+
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.SparkSQLException
+import org.apache.spark.sql.test.SharedSparkSession
+
+class SparkConnectCloneSessionSuite extends SharedSparkSession with BeforeAndAfterEach {
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ SparkConnectService.sessionManager.invalidateAllSessions()
+ }
+
+ test("clone session with invalid target session ID format") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val invalidSessionId = "not-a-valid-uuid"
+
+ // Create the source session first
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(sourceKey, None)
+
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, invalidSessionId, None)
+ }
+
+ assert(ex.getCondition == "INVALID_CLONE_SESSION_REQUEST.TARGET_SESSION_ID_FORMAT")
+ assert(ex.getMessage.contains("Target session ID not-a-valid-uuid"))
+ assert(ex.getMessage.contains("must be an UUID string"))
+ }
+
+ test("clone session with target session ID already closed") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val targetSessionId = UUID.randomUUID.toString
+ val targetKey = SessionKey("testUser", targetSessionId)
+
+ // Create and then close a session to put it in the closed cache
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(sourceKey, None)
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(targetKey, None)
+ SparkConnectService.sessionManager.closeSession(targetKey)
+
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, targetSessionId, None)
+ }
+
+ assert(ex.getCondition ==
+ "INVALID_CLONE_SESSION_REQUEST.TARGET_SESSION_ID_ALREADY_CLOSED")
+ assert(ex.getMessage.contains(s"target session ID $targetSessionId"))
+ assert(ex.getMessage.contains("was previously closed"))
+ }
+
+ test("clone session with target session ID already exists") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val targetSessionId = UUID.randomUUID.toString
+ val targetKey = SessionKey("testUser", targetSessionId)
+
+ // Create both source and target sessions
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(sourceKey, None)
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(targetKey, None)
+
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, targetSessionId, None)
+ }
+
+ assert(ex.getCondition ==
+ "INVALID_CLONE_SESSION_REQUEST.TARGET_SESSION_ID_ALREADY_EXISTS")
+ assert(ex.getMessage.contains(s"target session ID $targetSessionId"))
+ assert(ex.getMessage.contains("already exists"))
+ }
+
+ test("clone session with source session not found") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val targetSessionId = UUID.randomUUID.toString
+
+ // Don't create the source session, so it doesn't exist
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, targetSessionId, None)
+ }
+
+ // Source session errors should remain as standard INVALID_HANDLE errors
+ assert(ex.getCondition == "INVALID_HANDLE.SESSION_NOT_FOUND")
+ assert(ex.getMessage.contains("Session not found"))
+ }
+
+ test("successful clone session creates new session") {
Review Comment:
Should we test session independence?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on PR #52200:
URL: https://github.com/apache/spark/pull/52200#issuecomment-3390258125
The CI is green and ready for merge @hvanhovell
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2388936762
##########
python/pyspark/sql/tests/test_connect_compatibility.py:
##########
@@ -268,7 +268,7 @@ def test_spark_session_compatibility(self):
"registerProgressHandler",
"removeProgressHandler",
}
- expected_missing_classic_methods = set()
+ expected_missing_classic_methods = {"cloneSession"}
Review Comment:
It should also be added to classic.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317151452
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1095,6 +1095,54 @@ message CheckpointCommandResult {
CachedRemoteRelation relation = 1;
}
+message CloneSessionRequest {
+ // (Required)
+ //
+ // The session_id specifies a spark session for a user id (which is specified
+ // by user_context.user_id). The session_id is set by the client to be able to
+ // collate streaming responses from different queries within the dedicated session.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ string session_id = 1;
+
+ // (Optional)
+ //
+ // Server-side generated idempotency key from the previous responses (if any). Server
+ // can use this to validate that the server side session has not changed.
+ optional string client_observed_server_side_session_id = 5;
+
+ // (Required) User context
+ //
+ // user_context.user_id and session_id both identify a unique remote spark session on the
+ // server side.
+ UserContext user_context = 2;
+
+ // Provides optional information about the client sending the request. This field
+ // can be used for language or version specific information and is only intended for
+ // logging purposes and will not be interpreted by the server.
+ optional string client_type = 3;
+
+ // (Optional)
+ // The session_id for the new cloned session. If not provided, a new UUID will be generated.
Review Comment:
Is the server allowed to return a different session id if you provide this id?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317157538
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1139,4 +1187,8 @@ service SparkConnectService {
// FetchErrorDetails retrieves the matched exception with details based on a provided error id.
rpc FetchErrorDetails(FetchErrorDetailsRequest) returns (FetchErrorDetailsResponse) {}
+
+ // Clone a session. Creates a new session that shares the same configuration and state
Review Comment:
Can we make this as descriptive as the Scala API doc?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2352768097
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -684,6 +684,34 @@ class SparkSession private[sql] (
}
}
+ /**
+ * 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 will remain isolated from this session - any subsequent changes to either session's
+ * server-side state will not be reflected in the other.
+ *
+ * @note This creates a new server-side session with a new session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(): SparkSession = {
+ SparkSession.builder().client(client.cloneSession()).create()
+ }
+
+ /**
+ * Create a clone of this Spark Connect session on the server side with a custom session ID.
+ * 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 with
+ * the specified session ID. The returned cloned session will remain isolated from this session.
+ *
+ * @param sessionId The custom session ID to use for the cloned session (must be a valid UUID)
+ * @note This creates a new server-side session with the specified session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(sessionId: String): SparkSession = {
Review Comment:
Developer API makes sesnse here, annotating 👍
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2352713826
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1095,6 +1095,54 @@ message CheckpointCommandResult {
CachedRemoteRelation relation = 1;
}
+message CloneSessionRequest {
+ // (Required)
+ //
+ // The session_id specifies a spark session for a user id (which is specified
+ // by user_context.user_id). The session_id is set by the client to be able to
+ // collate streaming responses from different queries within the dedicated session.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ string session_id = 1;
+
+ // (Optional)
+ //
+ // Server-side generated idempotency key from the previous responses (if any). Server
+ // can use this to validate that the server side session has not changed.
+ optional string client_observed_server_side_session_id = 5;
+
+ // (Required) User context
+ //
+ // user_context.user_id and session_id both identify a unique remote spark session on the
+ // server side.
+ UserContext user_context = 2;
+
+ // Provides optional information about the client sending the request. This field
+ // can be used for language or version specific information and is only intended for
+ // logging purposes and will not be interpreted by the server.
+ optional string client_type = 3;
+
+ // (Optional)
+ // The session_id for the new cloned session. If not provided, a new UUID will be generated.
Review Comment:
No it isn't, it must be an ID that doesn't already exist in.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2359467594
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/SparkConnectCloneSessionSuite.scala:
##########
@@ -0,0 +1,122 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.connect.service
+
+import java.util.UUID
+
+import org.scalatest.BeforeAndAfterEach
+
+import org.apache.spark.SparkSQLException
+import org.apache.spark.sql.test.SharedSparkSession
+
+class SparkConnectCloneSessionSuite extends SharedSparkSession with BeforeAndAfterEach {
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ SparkConnectService.sessionManager.invalidateAllSessions()
+ }
+
+ test("clone session with invalid target session ID format") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val invalidSessionId = "not-a-valid-uuid"
+
+ // Create the source session first
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(sourceKey, None)
+
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, invalidSessionId, None)
+ }
+
+ assert(ex.getCondition == "INVALID_CLONE_SESSION_REQUEST.TARGET_SESSION_ID_FORMAT")
+ assert(ex.getMessage.contains("Target session ID not-a-valid-uuid"))
+ assert(ex.getMessage.contains("must be an UUID string"))
+ }
+
+ test("clone session with target session ID already closed") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val targetSessionId = UUID.randomUUID.toString
+ val targetKey = SessionKey("testUser", targetSessionId)
+
+ // Create and then close a session to put it in the closed cache
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(sourceKey, None)
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(targetKey, None)
+ SparkConnectService.sessionManager.closeSession(targetKey)
+
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, targetSessionId, None)
+ }
+
+ assert(ex.getCondition ==
+ "INVALID_CLONE_SESSION_REQUEST.TARGET_SESSION_ID_ALREADY_CLOSED")
+ assert(ex.getMessage.contains(s"target session ID $targetSessionId"))
+ assert(ex.getMessage.contains("was previously closed"))
+ }
+
+ test("clone session with target session ID already exists") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val targetSessionId = UUID.randomUUID.toString
+ val targetKey = SessionKey("testUser", targetSessionId)
+
+ // Create both source and target sessions
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(sourceKey, None)
+ SparkConnectService.sessionManager.getOrCreateIsolatedSession(targetKey, None)
+
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, targetSessionId, None)
+ }
+
+ assert(ex.getCondition ==
+ "INVALID_CLONE_SESSION_REQUEST.TARGET_SESSION_ID_ALREADY_EXISTS")
+ assert(ex.getMessage.contains(s"target session ID $targetSessionId"))
+ assert(ex.getMessage.contains("already exists"))
+ }
+
+ test("clone session with source session not found") {
+ val sourceKey = SessionKey("testUser", UUID.randomUUID.toString)
+ val targetSessionId = UUID.randomUUID.toString
+
+ // Don't create the source session, so it doesn't exist
+ val ex = intercept[SparkSQLException] {
+ SparkConnectService.sessionManager.cloneSession(sourceKey, targetSessionId, None)
+ }
+
+ // Source session errors should remain as standard INVALID_HANDLE errors
+ assert(ex.getCondition == "INVALID_HANDLE.SESSION_NOT_FOUND")
+ assert(ex.getMessage.contains("Session not found"))
+ }
+
+ test("successful clone session creates new session") {
Review Comment:
added
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317159898
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -684,6 +684,34 @@ class SparkSession private[sql] (
}
}
+ /**
+ * 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 will remain isolated from this session - any subsequent changes to either session's
+ * server-side state will not be reflected in the other.
+ *
+ * @note This creates a new server-side session with a new session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(): SparkSession = {
+ SparkSession.builder().client(client.cloneSession()).create()
+ }
+
+ /**
+ * Create a clone of this Spark Connect session on the server side with a custom session ID.
+ * 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 with
+ * the specified session ID. The returned cloned session will remain isolated from this session.
+ *
+ * @param sessionId The custom session ID to use for the cloned session (must be a valid UUID)
+ * @note This creates a new server-side session with the specified session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(sessionId: String): SparkSession = {
Review Comment:
Should we make this part of the public API?
Please annotate this with `@DeveloperAPI`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317177135
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala:
##########
@@ -114,6 +114,73 @@ class SparkConnectSessionManager extends Logging {
Option(getSession(key, None))
}
+ /**
+ * Clone an existing session with a new session ID.
+ * Creates a new SessionHolder with a cloned SparkSession that shares configuration
+ * and catalog state but has independent caches and runtime state.
+ */
+ private[connect] def cloneSession(
+ sourceKey: SessionKey,
+ newSessionId: String,
+ previouslyObservedSessionId: Option[String]): SessionHolder = {
+
+ // Get source session (must exist)
+ val sourceSessionHolder = getIsolatedSession(sourceKey, None)
+
+ previouslyObservedSessionId.foreach(sessionId =>
+ validateSessionId(sourceKey, sourceSessionHolder.session.sessionUUID, sessionId))
+
+ val newKey = SessionKey(sourceKey.userId, newSessionId)
+
+ // Validate new sessionId for clone operation
+ validateCloneTargetSession(newKey)
+
+ // Create cloned session
+ val clonedSessionHolder = getSession(
+ newKey,
+ Some(() => {
+ val session = sessionStore.get(newKey)
+ if (session == null) {
+ // Clone the underlying SparkSession using cloneSession() which preserves
+ // configuration, catalog, session state, temporary views, and registered functions
+ val clonedSparkSession = sourceSessionHolder.session.cloneSession()
+
+ val newHolder = SessionHolder(newKey.userId, newKey.sessionId, clonedSparkSession)
Review Comment:
Shouldn't we clone more state here? For example MLCache, or DataFrameCache?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2352746546
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1095,6 +1095,54 @@ message CheckpointCommandResult {
CachedRemoteRelation relation = 1;
}
+message CloneSessionRequest {
+ // (Required)
+ //
+ // The session_id specifies a spark session for a user id (which is specified
+ // by user_context.user_id). The session_id is set by the client to be able to
+ // collate streaming responses from different queries within the dedicated session.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ string session_id = 1;
+
+ // (Optional)
+ //
+ // Server-side generated idempotency key from the previous responses (if any). Server
+ // can use this to validate that the server side session has not changed.
+ optional string client_observed_server_side_session_id = 5;
+
+ // (Required) User context
+ //
+ // user_context.user_id and session_id both identify a unique remote spark session on the
+ // server side.
+ UserContext user_context = 2;
+
+ // Provides optional information about the client sending the request. This field
+ // can be used for language or version specific information and is only intended for
+ // logging purposes and will not be interpreted by the server.
+ optional string client_type = 3;
+
+ // (Optional)
+ // The session_id for the new cloned session. If not provided, a new UUID will be generated.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ optional string new_session_id = 4;
+}
+
+// Next ID: 5
+message CloneSessionResponse {
+ // Session id of the original session that was cloned.
Review Comment:
Why not? All of our responses contain the id of the calling session. I don't see a reason to leave it out here
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2352760284
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -684,6 +684,34 @@ class SparkSession private[sql] (
}
}
+ /**
+ * 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 will remain isolated from this session - any subsequent changes to either session's
+ * server-side state will not be reflected in the other.
+ *
+ * @note This creates a new server-side session with a new session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(): SparkSession = {
+ SparkSession.builder().client(client.cloneSession()).create()
+ }
+
+ /**
+ * Create a clone of this Spark Connect session on the server side with a custom session ID.
+ * 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 with
+ * the specified session ID. The returned cloned session will remain isolated from this session.
Review Comment:
ack
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317196567
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -2508,6 +2508,29 @@
],
"sqlState" : "22P03"
},
+ "INVALID_CLONE_SESSION_REQUEST" : {
+ "message" : [
+ "Invalid session clone request."
+ ],
+ "subClass" : {
+ "TARGET_SESSION_ID_ALREADY_CLOSED" : {
+ "message" : [
+ "Cannot clone session to target session ID <targetSessionId> because a session with this ID was previously closed."
+ ]
+ },
+ "TARGET_SESSION_ID_ALREADY_EXISTS" : {
+ "message" : [
+ "Cannot clone session to target session ID <targetSessionId> because a session with this ID already exists."
+ ]
+ },
+ "TARGET_SESSION_ID_FORMAT" : {
+ "message" : [
+ "Target session ID <targetSessionId> for clone operation must be an UUID string of the format '00112233-4455-6677-8899-aabbccddeeff'."
Review Comment:
Isn't there a different error for this as well?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2388947167
##########
sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/SparkSessionCloneSuite.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.connect
+
+import java.util.UUID
+
+import org.apache.spark.sql.connect.test.{ConnectFunSuite, RemoteSparkSession}
+
+class SparkSessionCloneSuite extends ConnectFunSuite with RemoteSparkSession {
+
+ test("cloneSession() creates independent session with auto-generated ID") {
+ spark.sql("SET spark.test.original = 'value1'")
+
+ val clonedSession = spark.cloneSession()
+
+ // Sessions should have different IDs
+ assert(spark.sessionId !== clonedSession.sessionId)
+
+ // Modify original session setting
+ spark.sql("SET spark.test.original = 'modified_original'")
+
+ // Verify original session has the modified value
+ val originalFinal = spark.sql("SET spark.test.original").collect().head.getString(1)
+ assert(originalFinal === "'modified_original'")
+
+ // Verify cloned session retains the original value
+ val clonedValue = clonedSession.sql("SET spark.test.original").collect().head.getString(1)
+ assert(clonedValue === "'value1'")
+ }
+
+ test("cloneSession(sessionId) creates session with specified ID") {
+ val customSessionId = UUID.randomUUID().toString
+
+ val clonedSession = spark.cloneSession(customSessionId)
+
+ // Verify the cloned session has the specified ID
+ assert(clonedSession.sessionId === customSessionId)
+ assert(spark.sessionId !== clonedSession.sessionId)
+ }
+
+ test("invalid session ID format throws exception") {
Review Comment:
Also test when people try to provide an ID that is already taken?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2359470500
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala:
##########
@@ -114,6 +114,73 @@ class SparkConnectSessionManager extends Logging {
Option(getSession(key, None))
}
+ /**
+ * Clone an existing session with a new session ID.
+ * Creates a new SessionHolder with a cloned SparkSession that shares configuration
+ * and catalog state but has independent caches and runtime state.
+ */
+ private[connect] def cloneSession(
+ sourceKey: SessionKey,
+ newSessionId: String,
+ previouslyObservedSessionId: Option[String]): SessionHolder = {
+
+ // Get source session (must exist)
+ val sourceSessionHolder = getIsolatedSession(sourceKey, None)
+
+ previouslyObservedSessionId.foreach(sessionId =>
+ validateSessionId(sourceKey, sourceSessionHolder.session.sessionUUID, sessionId))
+
+ val newKey = SessionKey(sourceKey.userId, newSessionId)
+
+ // Validate new sessionId for clone operation
+ validateCloneTargetSession(newKey)
+
+ // Create cloned session
+ val clonedSessionHolder = getSession(
+ newKey,
+ Some(() => {
+ val session = sessionStore.get(newKey)
+ if (session == null) {
+ // Clone the underlying SparkSession using cloneSession() which preserves
+ // configuration, catalog, session state, temporary views, and registered functions
+ val clonedSparkSession = sourceSessionHolder.session.cloneSession()
+
+ val newHolder = SessionHolder(newKey.userId, newKey.sessionId, clonedSparkSession)
Review Comment:
Intentionally did not clone caches, added some more info to the PR description.
> The clone preserves all SQL/catalog state but creates a fresh runtime environment. An analogy is cloning a database schema/config but not the active connections, caches, or running jobs.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2352713826
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1095,6 +1095,54 @@ message CheckpointCommandResult {
CachedRemoteRelation relation = 1;
}
+message CloneSessionRequest {
+ // (Required)
+ //
+ // The session_id specifies a spark session for a user id (which is specified
+ // by user_context.user_id). The session_id is set by the client to be able to
+ // collate streaming responses from different queries within the dedicated session.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ string session_id = 1;
+
+ // (Optional)
+ //
+ // Server-side generated idempotency key from the previous responses (if any). Server
+ // can use this to validate that the server side session has not changed.
+ optional string client_observed_server_side_session_id = 5;
+
+ // (Required) User context
+ //
+ // user_context.user_id and session_id both identify a unique remote spark session on the
+ // server side.
+ UserContext user_context = 2;
+
+ // Provides optional information about the client sending the request. This field
+ // can be used for language or version specific information and is only intended for
+ // logging purposes and will not be interpreted by the server.
+ optional string client_type = 3;
+
+ // (Optional)
+ // The session_id for the new cloned session. If not provided, a new UUID will be generated.
Review Comment:
No it isn't, it must be an ID that doesn't already exist/already closed
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2388952807
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -684,6 +684,34 @@ class SparkSession private[sql] (
}
}
+ /**
+ * 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 will remain isolated from this session - any subsequent changes to either session's
+ * server-side state will not be reflected in the other.
+ *
+ * @note This creates a new server-side session with a new session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(): SparkSession = {
+ SparkSession.builder().client(client.cloneSession()).create()
+ }
+
+ /**
+ * Create a clone of this Spark Connect session on the server side with a custom session ID.
+ * 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 with
+ * the specified session ID. The returned cloned session will remain isolated from this session.
+ *
+ * @param sessionId The custom session ID to use for the cloned session (must be a valid UUID)
+ * @note This creates a new server-side session with the specified session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(sessionId: String): SparkSession = {
Review Comment:
Please add this method the `org.apache.spark.sql.SparkSession` class as well.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "asf-gitbox-commits (via GitHub)" <gi...@apache.org>.
asf-gitbox-commits closed pull request #52200: [SPARK-53455][CONNECT] Add `CloneSession` RPC
URL: https://github.com/apache/spark/pull/52200
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317163990
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1095,6 +1095,54 @@ message CheckpointCommandResult {
CachedRemoteRelation relation = 1;
}
+message CloneSessionRequest {
+ // (Required)
+ //
+ // The session_id specifies a spark session for a user id (which is specified
+ // by user_context.user_id). The session_id is set by the client to be able to
+ // collate streaming responses from different queries within the dedicated session.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ string session_id = 1;
+
+ // (Optional)
+ //
+ // Server-side generated idempotency key from the previous responses (if any). Server
+ // can use this to validate that the server side session has not changed.
+ optional string client_observed_server_side_session_id = 5;
+
+ // (Required) User context
+ //
+ // user_context.user_id and session_id both identify a unique remote spark session on the
+ // server side.
+ UserContext user_context = 2;
+
+ // Provides optional information about the client sending the request. This field
+ // can be used for language or version specific information and is only intended for
+ // logging purposes and will not be interpreted by the server.
+ optional string client_type = 3;
+
+ // (Optional)
+ // The session_id for the new cloned session. If not provided, a new UUID will be generated.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ optional string new_session_id = 4;
+}
+
+// Next ID: 5
+message CloneSessionResponse {
+ // Session id of the original session that was cloned.
Review Comment:
Do we need the old ids?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2317155729
##########
sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala:
##########
@@ -684,6 +684,34 @@ class SparkSession private[sql] (
}
}
+ /**
+ * 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 will remain isolated from this session - any subsequent changes to either session's
+ * server-side state will not be reflected in the other.
+ *
+ * @note This creates a new server-side session with a new session ID while preserving
+ * the current session's configuration and state.
+ */
+ def cloneSession(): SparkSession = {
+ SparkSession.builder().client(client.cloneSession()).create()
+ }
+
+ /**
+ * Create a clone of this Spark Connect session on the server side with a custom session ID.
+ * 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 with
+ * the specified session ID. The returned cloned session will remain isolated from this session.
Review Comment:
'remain' is a bit weird since the session is new.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2352713826
##########
sql/connect/common/src/main/protobuf/spark/connect/base.proto:
##########
@@ -1095,6 +1095,54 @@ message CheckpointCommandResult {
CachedRemoteRelation relation = 1;
}
+message CloneSessionRequest {
+ // (Required)
+ //
+ // The session_id specifies a spark session for a user id (which is specified
+ // by user_context.user_id). The session_id is set by the client to be able to
+ // collate streaming responses from different queries within the dedicated session.
+ // The id should be an UUID string of the format `00112233-4455-6677-8899-aabbccddeeff`
+ string session_id = 1;
+
+ // (Optional)
+ //
+ // Server-side generated idempotency key from the previous responses (if any). Server
+ // can use this to validate that the server side session has not changed.
+ optional string client_observed_server_side_session_id = 5;
+
+ // (Required) User context
+ //
+ // user_context.user_id and session_id both identify a unique remote spark session on the
+ // server side.
+ UserContext user_context = 2;
+
+ // Provides optional information about the client sending the request. This field
+ // can be used for language or version specific information and is only intended for
+ // logging purposes and will not be interpreted by the server.
+ optional string client_type = 3;
+
+ // (Optional)
+ // The session_id for the new cloned session. If not provided, a new UUID will be generated.
Review Comment:
No it isn't, it must be an ID that doesn't already exist in.
I do realise this case isn't handled, good catch...
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [WIP][SPARK-53455][CONNECT] Add `CloneSession` RPC [spark]
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #52200:
URL: https://github.com/apache/spark/pull/52200#discussion_r2359557874
##########
common/utils/src/main/resources/error/error-conditions.json:
##########
@@ -2508,6 +2508,29 @@
],
"sqlState" : "22P03"
},
+ "INVALID_CLONE_SESSION_REQUEST" : {
+ "message" : [
+ "Invalid session clone request."
+ ],
+ "subClass" : {
+ "TARGET_SESSION_ID_ALREADY_CLOSED" : {
+ "message" : [
+ "Cannot clone session to target session ID <targetSessionId> because a session with this ID was previously closed."
+ ]
+ },
+ "TARGET_SESSION_ID_ALREADY_EXISTS" : {
+ "message" : [
+ "Cannot clone session to target session ID <targetSessionId> because a session with this ID already exists."
+ ]
+ },
+ "TARGET_SESSION_ID_FORMAT" : {
+ "message" : [
+ "Target session ID <targetSessionId> for clone operation must be an UUID string of the format '00112233-4455-6677-8899-aabbccddeeff'."
Review Comment:
yes, but this makes it clear that the format of the target session is invalid, so the exception sent to the server isn't confusing
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org