[KYUUBI #7377] KSHC create database respects catalog-level warehouse

### Why are the changes needed?

When creating a database using KSHC, the database location does not respect
`spark.sql.catalog.<catalog>.hive.metastore.warehouse.dir`. Instead, it always falls back to `spark.sql.warehouse.dir`.

### How was this patch tested?

UT

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #7378 from maomaodev/kyuubi-7377.

Closes #7377

7eb12dd [lifumao] inline vars
62e1be0 [lifumao] KSHC create database respects catalog-level warehouse
5d12669 [lifumao] KSHC create database respects catalog-level warehouse

Authored-by: lifumao <lifumao@tencent.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: maomaodev

extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTableCatalog.scala

@@ -477,7 +477,7 @@ class HiveTableCatalog(sparkSession: SparkSession)
       namespace match {
         case Array(db) if !catalog.databaseExists(db) =>
           catalog.createDatabase(
-            toCatalogDatabase(db, metadata, defaultLocation = Some(catalog.getDefaultDBPath(db))),
+            toCatalogDatabase(db, metadata, defaultLocation = Some(getCatalogDefaultDBPath(db))),
             ignoreIfExists = false)
 
         case Array(_) =>
@@ -488,6 +488,25 @@ class HiveTableCatalog(sparkSession: SparkSession)
       }
     }
 
+  /**
+   * Returns the default database path with catalog-level warehouse configuration precedence.
+   *
+   * This method resolves the database path using the following priority order:
+   *   1. Catalog-level `spark.sql.catalog.<catalog>.hive.metastore.warehouse.dir`
+   *   2. Global-level `spark.sql.warehouse.dir` (Underlying)
+   *
+   * @param db database name
+   * @return qualified URI path for the database
+   */
+  private def getCatalogDefaultDBPath(db: String): URI = {
+    Option(catalogOptions.get("hive.metastore.warehouse.dir")).filter(_.nonEmpty) match {
+      case Some(dir) =>
+        CatalogUtils.makeQualifiedDBObjectPath(catalog.getDefaultDBPath(db), dir, hadoopConf)
+      case None =>
+        catalog.getDefaultDBPath(db)
+    }
+  }
+
   override def alterNamespace(namespace: Array[String], changes: NamespaceChange*): Unit =
     withSparkSQLConf(LEGACY_NON_IDENTIFIER_OUTPUT_CATALOG_NAME -> "true") {
       namespace match {

extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/HiveCatalogSuite.scala

@@ -29,7 +29,7 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
-import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+import org.apache.spark.sql.connector.catalog.{Identifier, SupportsNamespaces, TableCatalog}
 import org.apache.spark.sql.connector.expressions.Transform
 import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
 import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan
@@ -54,12 +54,13 @@ class HiveCatalogSuite extends KyuubiHiveTest {
 
   var catalog: HiveTableCatalog = _
 
-  private def newCatalog(): HiveTableCatalog = {
+  private def newCatalog(extraOptions: Map[String, String] = Map.empty): HiveTableCatalog = {
     val catalog = new HiveTableCatalog
     val catalogName = "hive"
     val properties = Maps.newHashMap[String, String]()
     properties.put("javax.jdo.option.ConnectionURL", "jdbc:derby:memory:memorydb;create=true")
     properties.put("javax.jdo.option.ConnectionDriverName", "org.apache.derby.jdbc.EmbeddedDriver")
+    extraOptions.foreach { case (k, v) => properties.put(k, v) }
     catalog.initialize(catalogName, new CaseInsensitiveStringMap(properties))
     catalog
   }
@@ -71,8 +72,15 @@ class HiveCatalogSuite extends KyuubiHiveTest {
   }
 
   def makeQualifiedPathWithWarehouse(path: String): URI = {
-    val p = new Path(catalog.conf.warehousePath, path)
-    val fs = p.getFileSystem(catalog.hadoopConfiguration())
+    makeQualifiedPathWithWarehouse(path, catalog.conf.warehousePath, catalog)
+  }
+
+  private def makeQualifiedPathWithWarehouse(
+      path: String,
+      warehousePath: String,
+      targetCatalog: HiveTableCatalog): URI = {
+    val p = new Path(warehousePath, path)
+    val fs = p.getFileSystem(targetCatalog.hadoopConfiguration())
     fs.makeQualified(p).toUri
   }
 
@@ -357,6 +365,45 @@ class HiveCatalogSuite extends KyuubiHiveTest {
     catalog.dropNamespace(testNs, cascade = false)
   }
 
+  test("createNamespace location: use global-level warehouse dir") {
+    val ns = Array("ns_default_path")
+    try {
+      catalog.createNamespace(ns, emptyProps)
+      val location = catalog.loadNamespaceMetadata(ns)
+        .asScala(SupportsNamespaces.PROP_LOCATION)
+
+      val expectedUri = makeQualifiedPathWithWarehouse(s"${ns.head}.db")
+      assert(new URI(location) === expectedUri)
+    } finally {
+      catalog.dropNamespace(ns, cascade = true)
+    }
+  }
+
+  test("createNamespace location: use catalog-level warehouse dir") {
+    withTempDir { tmpDir =>
+      val customWarehouseDir = tmpDir.getCanonicalPath
+      val customCatalog = newCatalog(Map("hive.metastore.warehouse.dir" -> customWarehouseDir))
+      val ns = Array("ns_custom_path")
+      try {
+        customCatalog.createNamespace(ns, emptyProps)
+        val location = customCatalog.loadNamespaceMetadata(ns)
+          .asScala(SupportsNamespaces.PROP_LOCATION)
+
+        val expectedUri =
+          makeQualifiedPathWithWarehouse(s"${ns.head}.db", customWarehouseDir, customCatalog)
+        assert(new URI(location) === expectedUri)
+
+        val defaultUri = makeQualifiedPathWithWarehouse(
+          s"${ns.head}.db",
+          customCatalog.conf.warehousePath,
+          customCatalog)
+        assert(new URI(location) !== defaultUri)
+      } finally {
+        customCatalog.dropNamespace(ns, cascade = true)
+      }
+    }
+  }
+
   test("Support Parquet/Orc provider is splitable") {
     val parquet_table = Identifier.of(testNs, "parquet_table")
     val parProps: util.Map[String, String] = new util.HashMap[String, String]()