[SPARK-31407][SQL][TEST] TestHiveQueryExecution should respect databa…

…se when creating table

### What changes were proposed in this pull request?

In `TestHiveQueryExecution`, if we detect a database in the referenced table, we should create the table under that database.

### Why are the changes needed?

This fix the test `Fix hive/SQLQuerySuite.derived from Hive query file: drop_database_removes_partition_dirs.q` which currently only pass when we run it with the whole test suit but fail when run it separately.

### Does this PR introduce any user-facing change?

No.

### How was this patch tested?

Run the test separately and together with the whole test suite.

Closes #28177 from Ngone51/fix_derived.

Authored-by: yi.wu <yi.wu@databricks.com>
Signed-off-by: HyukjinKwon <gurwls223@apache.org>

sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala

@@ -40,6 +40,7 @@ import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
 import org.apache.spark.sql.catalyst.catalog.ExternalCatalogWithListener
 import org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, OneRowRelation}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
 import org.apache.spark.sql.execution.{QueryExecution, SQLExecution}
 import org.apache.spark.sql.execution.command.CacheTableCommand
 import org.apache.spark.sql.hive._
@@ -588,20 +589,29 @@ private[hive] class TestHiveQueryExecution(
 
   override lazy val analyzed: LogicalPlan = {
     val describedTables = logical match {
-      case CacheTableCommand(tbl, _, _, _) => tbl.table :: Nil
+      case CacheTableCommand(tbl, _, _, _) => tbl :: Nil
       case _ => Nil
     }
 
     // Make sure any test tables referenced are loaded.
     val referencedTables =
       describedTables ++
-        logical.collect { case UnresolvedRelation(ident) => ident.last }
+        logical.collect { case UnresolvedRelation(ident) => ident.asTableIdentifier }
     val resolver = sparkSession.sessionState.conf.resolver
-    val referencedTestTables = sparkSession.testTables.keys.filter { testTable =>
-      referencedTables.exists(resolver(_, testTable))
+    val referencedTestTables = referencedTables.flatMap { tbl =>
+      val testTableOpt = sparkSession.testTables.keys.find(resolver(_, tbl.table))
+      testTableOpt.map(testTable => tbl.copy(table = testTable))
+    }
+    logDebug(s"Query references test tables: ${referencedTestTables.map(_.table).mkString(", ")}")
+    referencedTestTables.foreach { tbl =>
+      val curDB = sparkSession.catalog.currentDatabase
+      try {
+        tbl.database.foreach(db => sparkSession.catalog.setCurrentDatabase(db))
+        sparkSession.loadTestTable(tbl.table)
+      } finally {
+        tbl.database.foreach(_ => sparkSession.catalog.setCurrentDatabase(curDB))
+      }
     }
-    logDebug(s"Query references test tables: ${referencedTestTables.mkString(", ")}")
-    referencedTestTables.foreach(sparkSession.loadTestTable)
     // Proceed with analysis.
     sparkSession.sessionState.analyzer.executeAndCheck(logical, tracker)
   }