This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new a80620a49f3c [SPARK-55982][SQL] Fix function cache coherence on drop
namespace in V2 catalogs
a80620a49f3c is described below
commit a80620a49f3c0eb6c960a299e4651e283e6585f8
Author: Shrirang Mhalgi <[email protected]>
AuthorDate: Sun Jun 21 21:21:02 2026 -0700
[SPARK-55982][SQL] Fix function cache coherence on drop namespace in V2
catalogs
### What changes were proposed in this pull request?
Override `dropNamespace` in `InMemoryCatalog` to remove functions and
procedures belonging to the namespace before calling `super.dropNamespace`.
This ensures functions and procedures don't remain resolvable after their
schema is dropped.
### Why are the changes needed?
V2 catalog functions and procedures remain resolvable after `DROP NAMESPACE
... CASCADE` because the catalog's `dropNamespace` doesn't clean up its
function / procedure storage. The session catalog path already handles this
(`SessionCatalog.dropDatabase` calls
`functionRegistry.dropFunctionsInDatabase`), but V2 catalogs were missing this
cleanup.
### Why this fix is at the catalog level, not the framework level:
V2 functions are resolved fresh on every access via
`catalog.loadFunction(ident)` - there is no session-level function cache for V2
catalogs (unlike the session catalog which has `functionRegistry`). The same
applies to procedures via `catalog.loadProcedure(ident)`. Therefore, cache
coherence depends entirely on the catalog implementation correctly removing
functions and procedures during `dropNamespace(cascade=true)`. This PR fixes
the reference implementation (`InMemoryCatalog`) to [...]
### Does this PR introduce _any_ user-facing change?
No. The PR fixes the `InMemoryCatalog` (test catalog) to correctly
implement the `dropNamespace(cascade=true)` contract by removing functions and
procedures before dropping the namespace. Production V2 catalogs (Iceberg,
Unity) are responsible for their own cleanup. The fix demonstrates and enforces
the expected behavior.
### How was this patch tested?
Added regression test in `DataSourceV2FunctionSuite` (functions) and
`ProcedureSuite` (procedures).
### Was this patch authored or co-authored using generative AI tooling?
Yes. Co-authored using Claude Opus 4.6.
Closes #56179 from shrirangmhalgi/SPARK-55982-drop-namespace-functions.
Authored-by: Shrirang Mhalgi <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/connector/catalog/InMemoryCatalog.scala | 13 +++++++++++++
.../sql/connector/DataSourceV2FunctionSuite.scala | 22 ++++++++++++++++++++++
.../spark/sql/connector/ProcedureSuite.scala | 13 +++++++++++++
3 files changed, 48 insertions(+)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalog.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalog.scala
index 4f0588498ec4..655a3a717deb 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalog.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryCatalog.scala
@@ -27,6 +27,19 @@ import
org.apache.spark.sql.connector.catalog.functions.UnboundFunction
import org.apache.spark.sql.connector.catalog.procedures.UnboundProcedure
class InMemoryCatalog extends InMemoryTableCatalog with FunctionCatalog with
ProcedureCatalog {
+ override def dropNamespace(namespace: Array[String], cascade: Boolean):
Boolean = {
+ if (cascade) {
+ // SPARK-55982: Remove functions and procedures in this namespace before
dropping.
+ // Only needed for cascade=true because without cascade,
super.dropNamespace
+ // will fail if the namespace still contains tables or child namespaces.
+ listFunctions(namespace).foreach(ident => functions.remove(ident))
+ procedures.keySet.asScala
+ .filter(_.namespace.sameElements(namespace))
+ .foreach(ident => procedures.remove(ident))
+ }
+ super.dropNamespace(namespace, cascade)
+ }
+
protected val functions: util.Map[Identifier, UnboundFunction] =
new ConcurrentHashMap[Identifier, UnboundFunction]()
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala
index 6243d3b65f6c..8b4f10070399 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2FunctionSuite.scala
@@ -726,6 +726,28 @@ class DataSourceV2FunctionSuite extends
DatasourceV2SQLBase {
checkAnswer(sql("SELECT testcat.ns.simple_strlen('abc')"), Row(3) :: Nil)
checkAnswer(sql("SELECT testcat.ns.simple_strlen('hello world')"), Row(11)
:: Nil)
}
+
+ test("SPARK-55982: function should not resolve after namespace is dropped") {
+ withNamespace("testcat.dropns") {
+ val cat = catalog("testcat").asInstanceOf[InMemoryCatalog]
+ cat.createNamespace(Array("dropns"), new java.util.HashMap[String,
String]())
+ addFunction(Identifier.of(Array("dropns"), "strlen"), SimpleStrLen)
+ checkAnswer(sql("SELECT testcat.dropns.strlen('abc')"), Row(3) :: Nil)
+ sql("DROP NAMESPACE testcat.dropns CASCADE")
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql("SELECT testcat.dropns.strlen('abc')").collect()
+ },
+ condition = "UNRESOLVED_ROUTINE",
+ parameters = Map(
+ "routineName" -> "`testcat`.`dropns`.`strlen`",
+ "searchPath" -> "[`system`.`builtin`, `system`.`session`,
`spark_catalog`.`default`]"),
+ context = ExpectedContext(
+ fragment = "testcat.dropns.strlen('abc')",
+ start = 7,
+ stop = 34))
+ }
+ }
}
case object SimpleStrLen extends SimpleFunction with ScalarFunction[Int] {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
index f6b0dae9b362..cf45f4f416dc 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/ProcedureSuite.scala
@@ -598,6 +598,19 @@ class ProcedureSuite extends SharedSparkSession with
BeforeAndAfter {
Row("Parameters: ()") :: Nil)
}
+ test("SPARK-55982: DROP NAMESPACE CASCADE removes procedures") {
+ sql("CREATE NAMESPACE cat.dropns")
+ catalog.createProcedure(Identifier.of(Array("dropns"), "sum"), UnboundSum)
+ // Procedure resolves before drop.
+ checkAnswer(sql("CALL cat.dropns.sum(1, 2)"), Row(3) :: Nil)
+ sql("DROP NAMESPACE cat.dropns CASCADE")
+ // After cascade drop, the procedure must no longer resolve.
+ checkError(
+ exception = intercept[AnalysisException](sql("CALL cat.dropns.sum(1,
2)")),
+ condition = "FAILED_TO_LOAD_ROUTINE",
+ parameters = Map("routineName" -> "`cat`.`dropns`.`sum`"))
+ }
+
object UnboundBindFailProcedure extends UnboundProcedure {
override def name: String = "bind_fail"
override def description: String = "bind fail procedure"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
