[SPARK-48851][SQL] Change the value of SCHEMA_NOT_FOUND from `names…

…pace` to `catalog.namespace` ### What changes were proposed in this pull request? The pr aims to change the value of `SCHEMA_NOT_FOUND` from `namespace` to `catalog.namespace`. ### Why are the changes needed? As discussing apache#47038 (comment), we should provide more friendly and clear prompt error message. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Update existed UT & Pass GA. ### Was this patch authored or co-authored using generative AI tooling? No. Closes apache#47276 from panbingkun/db_with_catalog. Authored-by: panbingkun <panbingkun@baidu.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
panbingkun · Jul 11, 2024 · b4e3c2a · b4e3c2a
1 parent 225b6cb
commit b4e3c2a
Show file tree

Hide file tree

Showing 30 changed files with 96 additions and 94 deletions.
diff --git a/...ct/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala b/...ct/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala
@@ -31,7 +31,7 @@ import org.apache.spark.{QueryContext, QueryContextType, SparkArithmeticExceptio
 import org.apache.spark.connect.proto.{FetchErrorDetailsRequest, FetchErrorDetailsResponse, SparkConnectServiceGrpc, UserContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.{NamespaceAlreadyExistsException, NoSuchDatabaseException, NoSuchTableException, TableAlreadyExistsException, TempTableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.analysis.{NamespaceAlreadyExistsException, NoSuchDatabaseException, NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException, TempTableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.trees.Origin
 import org.apache.spark.sql.streaming.StreamingQueryException
@@ -219,6 +219,8 @@ private[client] object GrpcExceptionConverter {
         params.errorClass.orNull,
         params.messageParameters,
         params.cause)),
+    errorConstructor(params =>
+      new NoSuchNamespaceException(params.errorClass.orNull, params.messageParameters)),
     errorConstructor(params =>
       new NoSuchTableException(params.errorClass.orNull, params.messageParameters, params.cause)),
     errorConstructor[NumberFormatException](params =>

diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala
@@ -33,7 +33,7 @@ import org.scalatest.PrivateMethodTester
 
 import org.apache.spark.{SparkArithmeticException, SparkException, SparkUpgradeException}
 import org.apache.spark.SparkBuildInfo.{spark_version => SPARK_VERSION}
-import org.apache.spark.sql.catalyst.analysis.{NamespaceAlreadyExistsException, NoSuchDatabaseException, TableAlreadyExistsException, TempTableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.analysis.{NamespaceAlreadyExistsException, NoSuchNamespaceException, TableAlreadyExistsException, TempTableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.StringEncoder
 import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
 import org.apache.spark.sql.catalyst.parser.ParseException
@@ -165,8 +165,8 @@ class ClientE2ETestSuite
     }
   }
 
-  test("throw NoSuchDatabaseException") {
-    val ex = intercept[NoSuchDatabaseException] {
+  test("throw NoSuchNamespaceException") {
+    val ex = intercept[NoSuchNamespaceException] {
       spark.sql("use database123")
     }
     assert(ex.getErrorClass != null)

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
@@ -250,7 +250,7 @@ class SessionCatalog(
 
   private def requireDbExists(db: String): Unit = {
     if (!databaseExists(db)) {
-      throw new NoSuchDatabaseException(db)
+      throw new NoSuchNamespaceException(Seq(CatalogManager.SESSION_CATALOG_NAME, db))
     }
   }
 
@@ -291,7 +291,8 @@ class SessionCatalog(
   def dropDatabase(db: String, ignoreIfNotExists: Boolean, cascade: Boolean): Unit = {
     val dbName = format(db)
     if (dbName == DEFAULT_DATABASE) {
-      throw QueryCompilationErrors.cannotDropDefaultDatabaseError(dbName)
+      throw QueryCompilationErrors.cannotDropDefaultDatabaseError(
+        Seq(CatalogManager.SESSION_CATALOG_NAME, dbName))
     }
     if (!ignoreIfNotExists) {
       requireDbExists(dbName)
@@ -527,7 +528,7 @@ class SessionCatalog(
    * We replace char/varchar with "annotated" string type in the table schema, as the query
    * engine doesn't support char/varchar yet.
    */
-  @throws[NoSuchDatabaseException]
+  @throws[NoSuchNamespaceException]
   @throws[NoSuchTableException]
   def getTableMetadata(name: TableIdentifier): CatalogTable = {
     val t = getTableRawMetadata(name)
@@ -538,7 +539,7 @@ class SessionCatalog(
    * Retrieve the metadata of an existing permanent table/view. If no database is specified,
    * assume the table/view is in the current database.
    */
-  @throws[NoSuchDatabaseException]
+  @throws[NoSuchNamespaceException]
   @throws[NoSuchTableException]
   def getTableRawMetadata(name: TableIdentifier): CatalogTable = {
     val qualifiedIdent = qualifyIdentifier(name)
@@ -556,7 +557,7 @@ class SessionCatalog(
    * For example, if none of the requested tables could be retrieved, an empty list is returned.
    * There is no guarantee of ordering of the returned tables.
    */
-  @throws[NoSuchDatabaseException]
+  @throws[NoSuchNamespaceException]
   def getTablesByName(names: Seq[TableIdentifier]): Seq[CatalogTable] = {
     if (names.nonEmpty) {
       val qualifiedIdents = names.map(qualifyIdentifier)
@@ -1056,7 +1057,6 @@ class SessionCatalog(
         getTempViewOrPermanentTableMetadata(ident).tableType == CatalogTableType.VIEW
       } catch {
         case _: NoSuchTableException => false
-        case _: NoSuchDatabaseException => false
         case _: NoSuchNamespaceException => false
       }
     }

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -1072,10 +1072,10 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
       messageParameters = Map("database" -> database))
   }
 
-  def cannotDropDefaultDatabaseError(database: String): Throwable = {
+  def cannotDropDefaultDatabaseError(nameParts: Seq[String]): Throwable = {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE.DROP_DATABASE",
-      messageParameters = Map("database" -> toSQLId(database)))
+      messageParameters = Map("database" -> toSQLId(nameParts)))
   }
 
   def cannotUsePreservedDatabaseAsCurrentDatabaseError(database: String): Throwable = {

diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalogSuite.scala
@@ -229,7 +229,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("get database should throw exception when the database does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.getDatabaseMetadata("db_that_does_not_exist")
       }
     }
@@ -283,7 +283,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("drop database when the database does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.dropDatabase("db_that_does_not_exist", ignoreIfNotExists = false, cascade = false)
       }
       catalog.dropDatabase("db_that_does_not_exist", ignoreIfNotExists = true, cascade = false)
@@ -295,7 +295,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
       catalog.setCurrentDatabase("db1")
       assert(catalog.getCurrentDatabase == "db1")
       catalog.dropDatabase("db1", ignoreIfNotExists = false, cascade = true)
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.createTable(newTable("tbl1", "db1"), ignoreIfExists = false)
       }
       catalog.setCurrentDatabase("default")
@@ -321,7 +321,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("alter database should throw exception when the database does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.alterDatabase(newDb("unknown_db"))
       }
     }
@@ -332,7 +332,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
       assert(catalog.getCurrentDatabase == "default")
       catalog.setCurrentDatabase("db2")
       assert(catalog.getCurrentDatabase == "db2")
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.setCurrentDatabase("deebo")
       }
       catalog.createDatabase(newDb("deebo"), ignoreIfExists = false)
@@ -370,10 +370,10 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
   test("create table when database does not exist") {
     withBasicCatalog { catalog =>
       // Creating table in non-existent database should always fail
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.createTable(newTable("tbl1", "does_not_exist"), ignoreIfExists = false)
       }
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.createTable(newTable("tbl1", "does_not_exist"), ignoreIfExists = true)
       }
       // Table already exists
@@ -419,11 +419,11 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
   test("drop table when database/table does not exist") {
     withBasicCatalog { catalog =>
       // Should always throw exception when the database does not exist
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.dropTable(TableIdentifier("tbl1", Some("unknown_db")), ignoreIfNotExists = false,
           purge = false)
       }
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.dropTable(TableIdentifier("tbl1", Some("unknown_db")), ignoreIfNotExists = true,
           purge = false)
       }
@@ -494,7 +494,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("rename table when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.renameTable(TableIdentifier("tbl1", Some("unknown_db")), TableIdentifier("tbl2"))
       }
       intercept[NoSuchTableException] {
@@ -543,7 +543,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("alter table when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.alterTable(newTable("tbl1", "unknown_db"))
       }
       intercept[NoSuchTableException] {
@@ -608,7 +608,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("get table when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.getTableMetadata(TableIdentifier("tbl1", Some("unknown_db")))
       }
       intercept[NoSuchTableException] {
@@ -856,7 +856,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
           TableIdentifier("tbl4"),
           TableIdentifier("tbl1", Some("db2")),
           TableIdentifier("tbl2", Some("db2"))))
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.listTables("unknown_db")
       }
     }
@@ -876,7 +876,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
           TableIdentifier("tbl2", Some("db2"))))
       assert(catalog.listTables("db2", "*1").toSet ==
         Set(TableIdentifier("tbl1"), TableIdentifier("tbl1", Some("db2"))))
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.listTables("unknown_db", "*")
       }
     }
@@ -970,7 +970,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("create partitions when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.createPartitions(
           TableIdentifier("tbl1", Some("unknown_db")), Seq(), ignoreIfExists = false)
       }
@@ -1077,7 +1077,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("drop partitions when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.dropPartitions(
           TableIdentifier("tbl1", Some("unknown_db")),
           Seq(),
@@ -1177,7 +1177,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("get partition when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.getPartition(TableIdentifier("tbl1", Some("unknown_db")), part1.spec)
       }
       intercept[NoSuchTableException] {
@@ -1258,7 +1258,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("rename partitions when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.renamePartitions(
           TableIdentifier("tbl1", Some("unknown_db")), Seq(part1.spec), Seq(part2.spec))
       }
@@ -1349,7 +1349,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("alter partitions when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.alterPartitions(TableIdentifier("tbl1", Some("unknown_db")), Seq(part1))
       }
       intercept[NoSuchTableException] {
@@ -1497,7 +1497,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("list partitions when database/table does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.listPartitions(TableIdentifier("tbl1", Some("unknown_db")))
       }
       intercept[NoSuchTableException] {
@@ -1544,7 +1544,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("create function when database does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.createFunction(
           newFunc("func5", Some("does_not_exist")), ignoreIfExists = false)
       }
@@ -1687,7 +1687,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("drop function when database/function does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.dropFunction(
           FunctionIdentifier("something", Some("unknown_db")), ignoreIfNotExists = false)
       }
@@ -1746,7 +1746,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("get function when database/function does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.getFunctionMetadata(FunctionIdentifier("func1", Some("unknown_db")))
       }
       intercept[NoSuchFunctionException] {
@@ -1799,7 +1799,7 @@ abstract class SessionCatalogSuite extends AnalysisTest with Eventually {
 
   test("list functions when database does not exist") {
     withBasicCatalog { catalog =>
-      intercept[NoSuchDatabaseException] {
+      intercept[NoSuchNamespaceException] {
         catalog.listFunctions("unknown_db", "func*")
       }
     }

diff --git a/...catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala b/...catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
@@ -210,7 +210,7 @@ class InMemoryTableCatalog extends BasicInMemoryTableCatalog with SupportsNamesp
       case _ if namespaceExists(namespace) =>
         util.Collections.emptyMap[String, String]
       case _ =>
-        throw new NoSuchNamespaceException(namespace)
+        throw new NoSuchNamespaceException(name() +: namespace)
     }
   }
 
@@ -256,7 +256,7 @@ class InMemoryTableCatalog extends BasicInMemoryTableCatalog with SupportsNamesp
     if (namespace.isEmpty || namespaceExists(namespace)) {
       super.listTables(namespace)
     } else {
-      throw new NoSuchNamespaceException(namespace)
+      throw new NoSuchNamespaceException(name() +: namespace)
     }
   }
 }

diff --git a/...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DropNamespaceExec.scala b/...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DropNamespaceExec.scala
@@ -45,7 +45,7 @@ case class DropNamespaceExec(
           throw QueryCompilationErrors.cannotDropNonemptyNamespaceError(namespace)
       }
     } else if (!ifExists) {
-      throw QueryCompilationErrors.noSuchNamespaceError(ns)
+      throw QueryCompilationErrors.noSuchNamespaceError(catalog.name() +: ns)
     }
 
     Seq.empty