[SPARK-48851][SQL] Change the value of SCHEMA_NOT_FOUND from namespace to catalog.namespace

[panbingkun]

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 #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.

[cloud-fan]

Suggested change

	def this(db: Seq[String]) = {
	def this(namespace: Seq[String]) = {

[cloud-fan]

shall we use NoSuchNamespaceException?

[panbingkun]

Okay

[cloud-fan]

why do we need this? It can only be SESSION_CATALOG_NAME

[panbingkun]

• Give a simple example, eg:

package org.apache.spark.sql.connector

import java.util

import scala.collection.mutable
import scala.jdk.CollectionConverters._

import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.catalog.{CatalogDatabase, SessionCatalog}
import org.apache.spark.sql.connector.catalog.{CatalogPlugin, NamespaceChange, SupportsNamespaces}
import org.apache.spark.sql.util.CaseInsensitiveStringMap

class BaiduSessionCatalog extends SupportsNamespaces with SQLConfHelper with CatalogPlugin {

  var sparkSessionCatalog: SessionCatalog = _

  override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {}

  override def name(): String = "baidu"

  override def listNamespaces(): Array[Array[String]] = {
    Array.empty
  }

  override def listNamespaces(namespace: Array[String]): Array[Array[String]] = {
    Array.empty
  }

  override def loadNamespaceMetadata(namespace: Array[String]): util.Map[String, String] = {
    new util.HashMap[String, String]()
  }

  override def createNamespace(namespace: Array[String], metadata: util.Map[String, String]): Unit = {
    println("baidu catalog createNamespace ...")
  }

  override def alterNamespace(namespace: Array[String], changes: NamespaceChange*): Unit = {
    sparkSessionCatalog.catalogName = Some(name())
    namespace match {
      case Array(db) =>
        val metadata = sparkSessionCatalog.getDatabaseMetadata(db).toMetadata
      case _ =>
    }
  }

  override def dropNamespace(namespace: Array[String], cascade: Boolean): Boolean = {
    true
  }

  private implicit class CatalogDatabaseHelper(catalogDatabase: CatalogDatabase) {
    def toMetadata: util.Map[String, String] = {
      val metadata = mutable.HashMap[String, String]()

      catalogDatabase.properties.foreach {
        case (key, value) => metadata.put(key, value)
      }
      metadata.put(SupportsNamespaces.PROP_LOCATION, catalogDatabase.locationUri.toString)
      metadata.put(SupportsNamespaces.PROP_COMMENT, catalogDatabase.description)

      metadata.asJava
    }
  }
}

• Test code as following:

package org.apache.spark.sql.connector

import org.scalatest.BeforeAndAfter

import org.apache.spark.sql.QueryTest
import org.apache.spark.sql.connector.catalog.SupportsNamespaces
import org.apache.spark.sql.test.SharedSparkSession

class BaiduSessionCatalogSuite
  extends QueryTest
  with SharedSparkSession
  with BeforeAndAfter {

  protected val catalogClassName: String = classOf[BaiduSessionCatalog].getName
  private lazy val baiduCatalog = spark.sessionState.catalogManager.catalog("baidu")

  before {
    spark.conf.set("spark.sql.catalog.baidu", catalogClassName)
    baiduCatalog.asInstanceOf[BaiduSessionCatalog].sparkSessionCatalog = spark.sessionState.catalog
  }

  test("alter namespace") {
    baiduCatalog.asInstanceOf[SupportsNamespaces].alterNamespace(Array("ns"), null)
    println("alter namespace successfully.")
  }
}

• If we use hardcode SESSION_CATALOG_NAME , the error message above will always be.

[SCHEMA_NOT_FOUND] The schema `spark_catalog`.`ns` cannot be found. Verify the spelling and correctness of the schema and catalog.
If you did not qualify the name with a catalog, verify the current_schema() output, or qualify the name with the correct catalog.
To tolerate the error on drop use DROP SCHEMA IF EXISTS. SQLSTATE: 42704

• Otherwise, it will display
  

• Do we need to consider this scenario? @cloud-fan

[panbingkun]

I hardcoded SESSION_CATALOG_NAME first and ignored the above scenario.
I will add it when needed.

[cloud-fan]

It's very hacky to introduce custom catalogs with the SessionCatalog class... SessionCatalog should always point to the HMS.

[dongjoon-hyun]

+1, LGTM too (Pending CIs).

[cloud-fan]

thanks, merging to master!

[panbingkun]

thanks, merging to master!

Thank you for your review.
I will fix similar scenarios (when table not exist), by adding catalog name for the table to make it more consistent and reasonable.

spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

Lines 257 to 267 in 4c99c4d

	private def requireTableExists(name: TableIdentifier): Unit = {
	if (!tableExists(name)) {
	throw new NoSuchTableException(db = name.database.get, table = name.table)
	}
	}
	private def requireTableNotExists(name: TableIdentifier): Unit = {
	if (tableExists(name)) {
	throw new TableAlreadyExistsException(db = name.database.get, table = name.table)
	}
	}

[zhengruifeng]

This change is not only a user-facing change, it actually breaks the exception handling in external catalogs like:

try {

} catch {
  case e: NoSuchDatabaseException => // ignore or handle

}

I am going to restore this in #47433