cloud-fan commented on code in PR #59952:
URL: https://github.com/apache/spark/pull/59952#discussion_r4432081257
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -526,6 +551,34 @@ private[sql] class DefaultCatalogManager(
}
v1SessionCatalog.setCurrentDatabase(conf.defaultDatabase)
}
+
+ // Unregister every catalog loaded by this manager, and close the ones that
are `AutoCloseable`.
+ // `defaultSessionCatalog` is not closed since it is owned by the caller,
not by this manager.
+ // Closing is terminal: loading a catalog afterwards fails instead of
creating an instance
+ // that would never be closed.
+ //
+ // SPARK-56939: catalog implementations are closed outside of the manager
lock so that a
+ // catalog calling back into the manager while closing cannot deadlock.
+ override private[sql] def close(): Unit = {
+ val loaded = synchronized {
+ closed = true
+ val entries = catalogs.toSeq
+ catalogs.clear()
+ entries
+ }
+ loaded.foreach {
+ case (name, closeable: AutoCloseable) =>
+ try {
+ closeable.close()
+ } catch {
+ // A catalog loaded by a session classloader can fail with a
`LinkageError` while
+ // closing. It must not prevent the other catalogs from being closed.
+ case e @ (NonFatal(_) | _: LinkageError) =>
Review Comment:
**Nit (P3):** `NonFatal` does not match `InterruptedException`. When the
releasing thread is interrupted, or a catalog's `close()` rethrows one, it
escapes this loop. The remaining catalogs, already cleared from the map, are
never closed. `SessionHolder.close()` then throws before
`cleanupPythonWorkerLogs()` and `postClosed()`, and the session manager never
records the session as closed. That contradicts both the new `CatalogPlugin`
Javadoc ("Exceptions it throws are logged and otherwise ignored") and the
comment above this catch. To keep both promises, catch `InterruptedException`
here too, continue with the other catalogs, and restore the thread's interrupt
status once the loop finishes. Alternatively, narrow the Javadoc to non-fatal
exceptions.
**Verification:**
- **Regression:** With one catalog whose close() throws InterruptedException
loaded alongside another closeable catalog, closing the catalog manager closes
the other catalog, returns normally, and leaves the calling thread's interrupt
status set.
- **Inspection:** The CatalogPlugin Javadoc describes exactly the close()
failures that the catalog manager logs and ignores.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala:
##########
@@ -408,36 +408,46 @@ case class SessionHolder(userId: String, sessionId:
String, session: SparkSessio
// intended use is that the client session will get closed when it's
really not used anymore,
// or that it expires due to inactivity, in which case there should be no
races.
- // Clean up all artifacts.
- // Note: there can be concurrent AddArtifact calls still adding something.
- artifactManager.close()
-
- // Clean up running streaming queries.
- // Note: there can be concurrent streaming queries being started.
- SparkConnectService.streamingSessionManager.cleanupRunningQueries(this,
blocking = true)
- streamingForeachBatchRunnerCleanerCache.cleanUpAll() // Clean up any
streaming workers.
- removeAllListeners() // removes all listener and stop python listener
processes if necessary.
- // Stops all pipeline execution and clears the pipeline execution cache
- removeAllPipelineExecutions()
-
- // Clean up dataflow graphs
- dataflowGraphRegistry.dropAllDataflowGraphs()
-
- // if there is a server side listener, clean up related resources
- if (streamingServersideListenerHolder.isServerSideListenerRegistered) {
- streamingServersideListenerHolder.cleanUp()
- }
-
- // Clean up all executions.
- // After closedTimeMs is defined, SessionHolder.addExecuteHolder() will
not allow new executions
- // to be added for this session anymore. Because both
SessionHolder.addExecuteHolder() and
- // SparkConnectExecutionManager.removeAllExecutionsForSession() are
executed under
- // executionsLock, this guarantees that removeAllExecutionsForSession
triggered here will
- // remove all executions and no new executions will be added in the
meanwhile.
-
SparkConnectService.executionManager.removeAllExecutionsForSession(this.key)
+ Utils.tryWithSafeFinally {
+ // Clean up running streaming queries.
+ // Note: there can be concurrent streaming queries being started.
+ SparkConnectService.streamingSessionManager.cleanupRunningQueries(this,
blocking = true)
+ streamingForeachBatchRunnerCleanerCache.cleanUpAll() // Clean up any
streaming workers.
+ removeAllListeners() // removes all listener and stop python listener
processes if necessary.
+ // Stops all pipeline execution and clears the pipeline execution cache
+ removeAllPipelineExecutions()
+
+ // Clean up dataflow graphs
+ dataflowGraphRegistry.dropAllDataflowGraphs()
+
+ // if there is a server side listener, clean up related resources
+ if (streamingServersideListenerHolder.isServerSideListenerRegistered) {
+ streamingServersideListenerHolder.cleanUp()
+ }
- // Clean up ML cache (only if ML models were created)
- mlCache.close()
+ // Clean up all executions.
+ // After closedTimeMs is defined, SessionHolder.addExecuteHolder() will
not allow new
+ // executions to be added for this session anymore. Because both
+ // SessionHolder.addExecuteHolder() and
+ // SparkConnectExecutionManager.removeAllExecutionsForSession() are
executed under
+ // executionsLock, this guarantees that removeAllExecutionsForSession
triggered here will
+ // remove all executions and no new executions will be added in the
meanwhile.
+
SparkConnectService.executionManager.removeAllExecutionsForSession(this.key)
+
+ // Clean up ML cache (only if ML models were created)
+ mlCache.close()
+ } {
+ // Close the catalogs loaded by this session, then clean up all
artifacts, even if an
+ // earlier step failed. Catalogs are closed first because they may have
been loaded by the
+ // session classloader, which is closed with the artifacts. Otherwise
resources held by the
+ // catalogs (e.g. thread pools) keep the session classloader alive.
Cloned sessions load
+ // their own catalogs, but plans copied into them, e.g. temporary views,
can still refer to
+ // the catalogs closed here.
+ // Note: interrupted executions may still be running; they fail if they
try to load a
+ // catalog after this point. There can also be concurrent AddArtifact
calls still adding
+ // something.
+ session.sessionState.close()
Review Comment:
**Nit (P3):** Running this release in the `finally` block guarantees that
catalogs and artifacts are released when an earlier step such as
`removeAllPipelineExecutions()` throws. But every new test tears down
successfully, so moving this call back into the body, the shape of the previous
revision, would keep all tests green. Add a test that makes one earlier step
fail, for example a pipeline execution whose stop throws, and then asserts that
the catalog is closed, `getAddedJars` is empty, and the original exception
still propagates.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59952#discussion_r4429037137).
**Verification:**
- **Regression:** Closing a Spark Connect session whose earlier teardown
step throws still closes a loaded closeable catalog and releases the session's
artifacts, and the close still reports the original failure.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala:
##########
@@ -526,6 +551,34 @@ private[sql] class DefaultCatalogManager(
}
v1SessionCatalog.setCurrentDatabase(conf.defaultDatabase)
}
+
+ // Unregister every catalog loaded by this manager, and close the ones that
are `AutoCloseable`.
+ // `defaultSessionCatalog` is not closed since it is owned by the caller,
not by this manager.
+ // Closing is terminal: loading a catalog afterwards fails instead of
creating an instance
+ // that would never be closed.
+ //
+ // SPARK-56939: catalog implementations are closed outside of the manager
lock so that a
+ // catalog calling back into the manager while closing cannot deadlock.
+ override private[sql] def close(): Unit = {
+ val loaded = synchronized {
+ closed = true
+ val entries = catalogs.toSeq
+ catalogs.clear()
+ entries
+ }
+ loaded.foreach {
+ case (name, closeable: AutoCloseable) =>
+ try {
+ closeable.close()
Review Comment:
**Non-blocking (P2):** This also closes instances that stored temp-view
plans can still use from other sessions. `prepareTemporaryViewPlan` in
`views.scala` converts a `DataSourceV2Relation` to a by-name `V2TableReference`
only at the top level, because it uses `transform`, which skips subquery
expressions. It also leaves time-travel and bounded changelog relations as they
are. A Dataset-based global temp view therefore keeps session A's `ice`
instance, for example `spark.sql("SELECT * FROM ice.db.t WHERE id IN (SELECT id
FROM ice.db.u)").createOrReplaceGlobalTempView("v")` or a view over `VERSION AS
OF`. After A is released, session B's query on `global_temp.v` goes through
that closed instance: `V2TableRefreshUtil.refresh` reloads a versioned table
through it, and otherwise the scan uses a `Table` it loaded. For a catalog that
fails after `close()`, as the new Javadoc asks, B's query now fails where it
worked before this change. Temp views copied into a cloned session break the
same way. Catalogs that are not `AutoCloseable`, views created from SQL text
and top-level ordinary reads are unaffected.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59952#pullrequestreview-5719371234).
**Recommended change:** Make temp-view storage convert every v2 relation in
a Dataset-based view into a by-name reference, including relations inside
subquery expressions, time-travel relations (carrying their version or
timestamp), and bounded changelog relations (carrying their range). Make the
reading session resolve those references inside subquery plans as it already
does at the top level. Then update the CatalogPlugin Javadoc and the
SessionState and SessionHolder comments, which currently say plans copied into
clones can still use closed instances. Add temp view and global temp view tests
for these relation shapes across a session release.
**Why this works:** A stored view plan then holds only catalog names,
identifiers and captured table metadata. Each reading session resolves them
through its own catalog manager, loading the same time-travel version or
changelog range, and validates them as top-level references already are. No
session-scoped instance outlives its session in a shared plan, so closing a
released session's catalogs cannot break another session's query.
**Scope:** Complete the existing by-name conversion of v2 relations in
stored temp-view plans and its resolution, covering subquery, time-travel and
bounded changelog relations.
**Risks:** Relations inside subqueries of Dataset-based views are now
re-resolved and validated per read like top-level ones, so such a view now
fails when its subquery table is dropped or incompatibly changed, or when the
reading session does not configure that catalog, exactly as a top-level
relation already does. Time-travel and bounded changelog relations in stored
views now reload their table through the reading session's catalog on each
read, which adds one table load per relation per query on that path.
**Constraints:** A re-resolved time-travel or bounded changelog reference
must load exactly the captured version, timestamp or range, so the view's
results do not change. Resolution of stored references must cover subquery
plans in the same analyzer path that resolves top-level temp-view references.
Views created from SQL text and views over v1 or builtin session catalog tables
must behave as before.
**Success:** No stored Dataset-based temp or global temp view plan holds a
session-scoped catalog instance. A global temp view, or a temp view copied into
a clone, keeps working in other live sessions after the session that created it
is released, for subquery, time-travel and bounded changelog reads of a
closeable catalog. Reading such a view returns the same results as before the
change.
**Verification:**
- **Regression:** A global temp view created from a Dataset whose subquery
reads a table of a closeable catalog keeps returning the same rows from another
session after the creating Spark Connect session is released.
- **Regression:** A global temp view over a time-travel read and one over a
bounded changelog read of a closeable catalog keep returning the same rows from
another session after the creating session is released, and a temp view of each
shape copied into a cloned session keeps working in the clone after the source
is released.
- **Compatibility:** Temp views over top-level ordinary v2 reads, views
created from SQL text, and views over session catalog tables return the same
results and errors as before, and a stored subquery relation whose table was
dropped fails the same way a stored top-level relation does.
--
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]
