HyukjinKwon commented on code in PR #41701:
URL: https://github.com/apache/spark/pull/41701#discussion_r1251417288
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -60,6 +67,102 @@ case class SessionHolder(userId: String, sessionId: String,
session: SparkSessio
}
}
}
+
+ private[connect] lazy val artifactManager = new
SparkConnectArtifactManager(this)
+
+ /**
+ * Add an artifact to this SparkConnect session.
+ *
+ * @param remoteRelativePath
+ * @param serverLocalStagingPath
+ * @param fragment
+ */
+ private[connect] def addArtifact(
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path,
+ fragment: Option[String]): Unit = {
+ artifactManager.addArtifact(remoteRelativePath, serverLocalStagingPath,
fragment)
+ }
+
+ /**
+ * A [[JobArtifactSet]] for this SparkConnect session.
+ */
+ def connectJobArtifactSet: JobArtifactSet = artifactManager.jobArtifactSet
+
+ /**
+ * A [[ClassLoader]] for jar/class file resources specific to this
SparkConnect session.
+ */
+ def classloader: ClassLoader = artifactManager.classloader
+
+ /**
+ * Expire this session and trigger state cleanup mechanisms.
+ */
+ private[connect] def expireSession(): Unit = {
+ logDebug(s"Expiring session with userId: $userId and sessionId:
$sessionId")
+ artifactManager.cleanUpResources()
+ }
+
+ /**
+ * Execute a block of code using this session's classloader.
+ * @param f
+ * @tparam T
+ */
+ def withContext[T](f: => T): T = {
+ // Needed for deserializing and evaluating the UDF on the driver
+ Utils.withContextClassLoader(classloader) {
+ // Needed for propagating the dependencies to the executors.
+ JobArtifactSet.withActive(connectJobArtifactSet) {
+ f
+ }
+ }
+ }
+
+ /**
+ * Set the session-based Python paths to include in Python UDF.
+ * @param f
+ * @tparam T
+ */
+ def withSessionBasedPythonPaths[T](f: => T): T = {
+ try {
+ session.conf.set(
Review Comment:
yeah I planned to remove this and @vicennial removed it in
https://github.com/apache/spark/pull/41789#discussion_r1247634835. This was
just a hack I added to avoid additional refactoring.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
