[SPARK-33692][SQL] View should use captured catalog and namespace to lookup function

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

@@ -1488,12 +1488,28 @@ class SessionCatalog(
     // We probably shouldn't use a single FunctionRegistry to register all three kinds of functions
     // (built-in, temp, and external).
     if (name.database.isEmpty && functionRegistry.functionExists(name)) {
-      // This function has been already loaded into the function registry.
-      return functionRegistry.lookupFunction(name, children)
+      val isResolvingView = AnalysisContext.get.catalogAndNamespace.nonEmpty
+      // We lookup function without database in two cases:
+      // 1. the function is not a temporary function
+      // 2. the function is a temporary function but we are not resolving view
+      if (!isTemporaryFunction(name) || !isResolvingView) {
+        // This function has been already loaded into the function registry.
+        return functionRegistry.lookupFunction(name, children)
+      }
+    }
+
+    // Get the database from AnalysisContext if it's defined, otherwise, use current database
+    val currentDatabase = AnalysisContext.get.catalogAndNamespace match {
+      case Seq() => getCurrentDatabase
+      case Seq(_, db) => db
+      case Seq(catalog, namespace @ _*) =>
+        throw new AnalysisException(
+          s"Unsupported catalog ${catalog} and " +
+            s"namespace '${namespace.mkString("[", ".", "]")}' for function '$name'")
     }
 
     // If the name itself is not qualified, add the current database to it.
-    val database = formatDatabaseName(name.database.getOrElse(getCurrentDatabase))
+    val database = formatDatabaseName(name.database.getOrElse(currentDatabase))
     val qualifiedName = name.copy(database = Some(database))
 
     if (functionRegistry.functionExists(qualifiedName)) {

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

@@ -209,4 +209,32 @@ class UDFSuite
       assert(e.getMessage.contains("Can not get an evaluator of the empty UDAF"))
     }
   }
+
+  test("SPARK-33692: permanent view should use captured catalog and namespace for function") {
+    val upperFuncClass =
+      classOf[org.apache.hadoop.hive.ql.udf.generic.GenericUDFUpper].getCanonicalName
+    val lowerFuncClass =
+      classOf[org.apache.hadoop.hive.ql.udf.generic.GenericUDFLower].getCanonicalName
+    val functionName = "test_udf"
+    withTempDatabase { dbName =>
+      withUserDefinedFunction(
+        s"default.$functionName" -> false,
+        s"$dbName.$functionName" -> false,
+        functionName -> true) {
+        withView("default.v1") {
+          sql("USE DEFAULT")
+          sql(s"CREATE FUNCTION $functionName AS '$upperFuncClass'")
+          // create a view using a function in 'default' database
+          sql(s"CREATE VIEW v1 AS SELECT $functionName('TeSt_STr')")
+          sql(s"USE $dbName")
+          // create function in another database with the same function name
+          sql(s"CREATE FUNCTION $functionName AS '$lowerFuncClass'")
+          // create temporary function with the same function name
+          sql(s"CREATE TEMPORARY FUNCTION $functionName AS '$lowerFuncClass'")
+          // view v1 should still using function defined in `default` database
+          checkAnswer(sql("SELECT * FROM default.v1"), Seq(Row("TEST_STR")))
+        }
+      }
+    }
+  }
 }

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveSQLViewSuite.scala

@@ -22,6 +22,7 @@ import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
 import org.apache.spark.sql.execution.SQLViewSuite
 import org.apache.spark.sql.hive.test.TestHiveSingleton
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{NullType, StructType}
 
 /**
@@ -72,9 +73,13 @@ class HiveSQLViewSuite extends SQLViewSuite with TestHiveSingleton {
           withTable("tab1") {
             (1 to 10).map(i => s"$i").toDF("id").write.saveAsTable("tab1")
 
-            // temporary view
-            sql(s"CREATE TEMPORARY VIEW tempView1 AS SELECT $tempFunctionName(id) from tab1")
-            checkAnswer(sql("select count(*) FROM tempView1"), Row(10))
+            // TODO: temporary function support for temporary view with sql text stored will
+            // be fixed in another PR
+            withSQLConf(SQLConf.STORE_ANALYZED_PLAN_FOR_VIEW.key -> "true") {
+              // temporary view
+              sql(s"CREATE TEMPORARY VIEW tempView1 AS SELECT $tempFunctionName(id) from tab1")
+              checkAnswer(sql("select count(*) FROM tempView1"), Row(10))
+            }
 
             // permanent view
             val e = intercept[AnalysisException] {