You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ya...@apache.org on 2024/08/06 08:50:39 UTC
(spark) branch master updated: [SPARK-49099][SQL] CatalogManager.setCurrentNamespace should respect custom session catalog
This is an automated email from the ASF dual-hosted git repository.
yao 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 97fe0e896c6a [SPARK-49099][SQL] CatalogManager.setCurrentNamespace should respect custom session catalog
97fe0e896c6a is described below
commit 97fe0e896c6aa15eee02c0f15fa2bfd8d21624ee
Author: Rui Wang <ru...@databricks.com>
AuthorDate: Tue Aug 6 16:50:26 2024 +0800
[SPARK-49099][SQL] CatalogManager.setCurrentNamespace should respect custom session catalog
### What changes were proposed in this pull request?
Refactor CatalogManager.setCurrentNamespace so it unifies the handling of `SupportsNamespaces` and SessionCatalog, thus it can respect custom session catalog.
### Why are the changes needed?
This is a bug.
CatalogManager.setCurrentNamespace should respect custom session catalog. Without this PR, a custom session catalog won't work as Spark uses `spark_catalog` to see if the catalog is session log and then directly call V1SessionCatalog.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UT
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #47592 from amaliujia/refactor_setCurrentNamespace.
Lead-authored-by: Rui Wang <ru...@databricks.com>
Co-authored-by: Wenchen Fan <we...@databricks.com>
Co-authored-by: Kent Yao <ya...@apache.org>
Signed-off-by: Kent Yao <ya...@apache.org>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 18 ++++++++++++++++--
.../spark/sql/catalyst/catalog/SessionCatalog.scala | 6 +++++-
.../spark/sql/connector/catalog/CatalogManager.scala | 20 +++++++++++++-------
.../sql/catalyst/analysis/LookupFunctionsSuite.scala | 2 +-
.../spark/sql/connector/DataSourceV2SQLSuite.scala | 10 +++++++++-
.../org/apache/spark/sql/test/SQLTestUtils.scala | 5 +++--
.../org/apache/spark/sql/hive/test/TestHive.scala | 1 +
7 files changed, 48 insertions(+), 14 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 1b194da5ab0a..539f05e64ca1 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -23,6 +23,7 @@ import java.util.concurrent.atomic.AtomicBoolean
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
import scala.util.{Failure, Random, Success, Try}
import org.apache.spark.{SparkException, SparkUnsupportedOperationException}
@@ -80,7 +81,7 @@ object SimpleAnalyzer extends Analyzer(
override def resolver: Resolver = caseSensitiveResolution
}
-object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog {
+object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog with SupportsNamespaces {
private def fail() = throw SparkUnsupportedOperationException()
override def listTables(namespace: Array[String]): Array[Identifier] = fail()
override def loadTable(ident: Identifier): Table = {
@@ -94,10 +95,23 @@ object FakeV2SessionCatalog extends TableCatalog with FunctionCatalog {
override def alterTable(ident: Identifier, changes: TableChange*): Table = fail()
override def dropTable(ident: Identifier): Boolean = fail()
override def renameTable(oldIdent: Identifier, newIdent: Identifier): Unit = fail()
- override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = fail()
+ override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {}
override def name(): String = CatalogManager.SESSION_CATALOG_NAME
override def listFunctions(namespace: Array[String]): Array[Identifier] = fail()
override def loadFunction(ident: Identifier): UnboundFunction = fail()
+ override def listNamespaces(): Array[Array[String]] = fail()
+ override def listNamespaces(namespace: Array[String]): Array[Array[String]] = fail()
+ override def loadNamespaceMetadata(namespace: Array[String]): util.Map[String, String] = {
+ if (namespace.length == 1) {
+ mutable.HashMap[String, String]().asJava
+ } else {
+ throw new NoSuchNamespaceException(namespace)
+ }
+ }
+ override def createNamespace(
+ namespace: Array[String], metadata: util.Map[String, String]): Unit = fail()
+ override def alterNamespace(namespace: Array[String], changes: NamespaceChange*): Unit = fail()
+ override def dropNamespace(namespace: Array[String], cascade: Boolean): Boolean = fail()
}
/**
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
index 701c68684c34..7c3a81879c26 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala
@@ -334,12 +334,16 @@ class SessionCatalog(
def getCurrentDatabase: String = synchronized { currentDb }
def setCurrentDatabase(db: String): Unit = {
+ setCurrentDatabaseWithNameCheck(db, requireDbExists)
+ }
+
+ def setCurrentDatabaseWithNameCheck(db: String, nameCheck: String => Unit): Unit = {
val dbName = format(db)
if (dbName == globalTempDatabase) {
throw QueryCompilationErrors.cannotUsePreservedDatabaseAsCurrentDatabaseError(
globalTempDatabase)
}
- requireDbExists(dbName)
+ nameCheck(dbName)
synchronized { currentDb = dbName }
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala
index 16c387a82373..db94659b1033 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogManager.scala
@@ -106,19 +106,25 @@ class CatalogManager(
}
}
- def setCurrentNamespace(namespace: Array[String]): Unit = synchronized {
+ private def assertNamespaceExist(namespace: Array[String]): Unit = {
currentCatalog match {
- case _ if isSessionCatalog(currentCatalog) && namespace.length == 1 =>
- v1SessionCatalog.setCurrentDatabase(namespace.head)
- case _ if isSessionCatalog(currentCatalog) =>
- throw QueryCompilationErrors.noSuchNamespaceError(namespace)
case catalog: SupportsNamespaces if !catalog.namespaceExists(namespace) =>
- throw QueryCompilationErrors.noSuchNamespaceError(namespace)
+ throw QueryCompilationErrors.noSuchNamespaceError(catalog.name() +: namespace)
case _ =>
- _currentNamespace = Some(namespace)
}
}
+ def setCurrentNamespace(namespace: Array[String]): Unit = synchronized {
+ if (isSessionCatalog(currentCatalog) && namespace.length == 1) {
+ v1SessionCatalog.setCurrentDatabaseWithNameCheck(
+ namespace.head,
+ _ => assertNamespaceExist(namespace))
+ } else {
+ assertNamespaceExist(namespace)
+ }
+ _currentNamespace = Some(namespace)
+ }
+
private var _currentCatalogName: Option[String] = None
def currentCatalog: CatalogPlugin = synchronized {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/LookupFunctionsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/LookupFunctionsSuite.scala
index ae32365e69bb..1fd81349ac72 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/LookupFunctionsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/LookupFunctionsSuite.scala
@@ -47,7 +47,7 @@ class LookupFunctionsSuite extends PlanTest {
ignoreIfExists = false)
val catalog = new SessionCatalog(externalCatalog, new SimpleFunctionRegistry)
val catalogManager = new CatalogManager(new CustomV2SessionCatalog(catalog), catalog)
- catalog.setCurrentDatabase("db1")
+ catalogManager.setCurrentNamespace(Array("db1"))
try {
val analyzer = new Analyzer(catalogManager)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
index fe45905497e8..c032de4bc260 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala
@@ -1443,7 +1443,7 @@ class DataSourceV2SQLSuiteV1Filter
}
checkError(exception,
errorClass = "SCHEMA_NOT_FOUND",
- parameters = Map("schemaName" -> "`ns1`.`ns2`"))
+ parameters = Map("schemaName" -> "`testcat`.`ns1`.`ns2`"))
}
test("SPARK-31100: Use: v2 catalog that does not implement SupportsNameSpaces is used " +
@@ -3514,6 +3514,14 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ test("SPARK-49099: Switch current schema with custom spark_catalog") {
+ withSQLConf(V2_SESSION_CATALOG_IMPLEMENTATION.key -> classOf[InMemoryCatalog].getName) {
+ sql("CREATE DATABASE test_db")
+ sql("USE test_db")
+ }
+ }
+
+
test("SPARK-36680: Supports Dynamic Table Options for Spark SQL") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala
index 8799ef4b0868..b607b98ea078 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala
@@ -423,8 +423,9 @@ private[sql] trait SQLTestUtilsBase
* `f` returns.
*/
protected def activateDatabase(db: String)(f: => Unit): Unit = {
- spark.sessionState.catalog.setCurrentDatabase(db)
- Utils.tryWithSafeFinally(f)(spark.sessionState.catalog.setCurrentDatabase("default"))
+ spark.sessionState.catalogManager.setCurrentNamespace(Array(db))
+ Utils.tryWithSafeFinally(f)(
+ spark.sessionState.catalogManager.setCurrentNamespace(Array("default")))
}
/**
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
index 3f8de93b330b..ec5648064ebc 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
@@ -545,6 +545,7 @@ private[hive] class TestHiveSparkSession(
sharedState.cacheManager.clearCache()
sharedState.loadedTables.clear()
sessionState.catalog.reset()
+ sessionState.catalogManager.reset()
metadataHive.reset()
// HDFS root scratch dir requires the write all (733) permission. For each connecting user,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org