commit 264ca4dc320e516bd88679ed2404f1c15d2f24eb
author vicennial <venkata.gudesa@databricks.com> Mon Oct 13 11:01:58 2025 -0400
committer Herman van Hovell <herman@databricks.com> Mon Oct 13 11:01:58 2025 -0400
tree 89bccb1440c2004b3b8c47b4ceb18ec3dafa2ee0
parent 3ac4a48fb8c5e0bdd9e7f0a4b98094531a7807f6

[SPARK-53455][CONNECT] Add `CloneSession` RPC

### What changes were proposed in this pull request?

Adds a new experimental/developer RPC `CloneSession` to the `SparkConnectService`.

  ✅ CLONED (from SparkSession)

  - SessionState - SQL configs, temp views, UDFs, catalog metadata
  - ArtifactManager - JARs, files, classes added to session
  - ManagedJobTags - Job group tags for tracking
  - SharedState (reference) - Metastore, global temp views
  - SparkContext (reference) - Core Spark engine

  ❌ NOT CLONED (SessionHolder - Spark Connect layer)

  - planCache - (Partially analyzed) Logical plans for query optimization
  - operationIds - Currently executing operations
  - errorIdToError - Recent errors for debugging
  - eventManager - Session lifecycle events
  - dataFrameCache - DataFrames for foreachBatch callbacks
  - mlCache - ML models and pipelines
  - listenerCache - Streaming query listeners
  - pipelineExecutions - Active pipeline contexts
  - dataflowGraphRegistry - Registered dataflow graphs
  - streamingForeachBatchRunnerCleanerCache - Python streaming workers
  - pythonAccumulator - Python metrics collection
  - Session timings - Start time, last access, custom timeout

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.
### Why are the changes needed?

Spark Connect introduced the concept of resource isolation (via `ArtifactManager`, which has been ported to classic Spark) and thus, jars/pyfiles/artifacts added to each session are isolated from other sessions.

A slight rough edge is that if a user wishes to fork the state of a session but maintain independence, the only possible way is to create a new session and reupload/reinit all base jars/artifacts/pyfiles, etc.

Support for cloning through the API helps address the rough edge while maintaining all the benefits of session resource isolation.
### Does this PR introduce _any_ user-facing change?

Yes

```python
  spark = SparkSession.builder.remote("sc://localhost").getOrCreate()
  spark.conf.set("my.custom.config", "value")
  spark.addArtifact("/path/to/my.jar")
  spark.sql("CREATE TEMP VIEW my_view AS SELECT 1 AS id")

  # Clone the session
  cloned_spark = spark.cloneSession()

  # The cloned session has all the same state
  assert cloned_spark.conf.get("my.custom.config") == "value"
  assert cloned_spark.sql("SELECT * FROM my_view").collect() == [Row(id=1)]

  # But operations are isolated between sessions
  cloned_spark.sql("DROP VIEW my_view")  # Only affects cloned session
  spark.sql("SELECT * FROM my_view").collect()  # Original still works
```

### How was this patch tested?

New individual unit tests along with new test suites.

### Was this patch authored or co-authored using generative AI tooling?

Co-authored with assistance from Claude Code.

Closes #52200 from vicennial/cloneAPI.

Authored-by: vicennial <venkata.gudesa@databricks.com>
Signed-off-by: Herman van Hovell <herman@databricks.com>