Add base session created in SparkConnectService #52895
Conversation
garlandz-db
force-pushed
the
isolated_root_session
branch
from
c2df6a1 to
88e521e
Compare
.../server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala
Show resolved
Hide resolved
| */ | ||
| def initializeBaseSession(sc: SparkContext): Unit = synchronized { | ||
| if (baseSession.isEmpty) { | ||
| baseSession = Some(SparkSession.builder().sparkContext(sc).getOrCreate()) |
There was a problem hiding this comment.
Please call newSession() on this session. The session returned by getOrCreate() is either an existing session, or will be accessible to others. It can be tampered with, and we should avoid that.
| return | ||
| } | ||
|
|
||
| sessionManager.initializeBaseSession(sc) |
There was a problem hiding this comment.
I think this is fine for now, at some point we should consider making the initialisation logic of connect less singleton heavy so we can pass the SparkContext as a constructor parameter.
There was a problem hiding this comment.
but session manager is a class object. what difference does it make for SparkConnectService to be a class object too
| * Initialize the base SparkSession from the provided SparkContext. | ||
| * This should be called once during SparkConnectService startup. | ||
| */ | ||
| def initializeBaseSession(sc: SparkContext): Unit = synchronized { |
There was a problem hiding this comment.
You can drop synchronized here...
|
|
||
| private def newIsolatedSession(): SparkSession = { | ||
| val active = SparkSession.active | ||
| if (active.sparkContext.isStopped) { |
There was a problem hiding this comment.
@garlandz-db can you figure out why this branch is here. We may have to recreate the session if this is an actual problem...
There was a problem hiding this comment.
the original pr: #43701 by Kent Yao. if the spark context is stopped then active.newSession() would throw an exception
org.apache.spark.SparkException: com.google.common.util.concurrent.UncheckedExecutionException: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
my guess: spark cluster prob isnt useful but technically you can call spark connect apis still. so we can create a valid spark session and continue handling the rpc.
however our fix is tangential to that error. we do not need to use active in this case.
garlandz-db
force-pushed
the
isolated_root_session
branch
from
88e521e to
6f70e4f
Compare
.../server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala
Outdated
Show resolved
Hide resolved
…/service/SparkConnectSessionManager.scala
2 participants
What changes were proposed in this pull request?
This PR makes SparkConnectService rely on its own SparkSession that is private and only intended for copying session configs to create new Sessions
Why are the changes needed?
The default session can get cleaned up in which case the SparkConnectService cannot recover as session creation fails on subsequent rpcs
Does this PR introduce any user-facing change?
No
How was this patch tested?
Added basic testing
Was this patch authored or co-authored using generative AI tooling?
Yes