[SPARK-32677][SQL] Load function resource before create

python/pyspark/sql/tests/test_catalog.py

@@ -128,8 +128,9 @@ def test_list_functions(self):
 
             with self.function("func1", "some_db.func2"):
                 spark.catalog.registerFunction("temp_func", lambda x: str(x))
-                spark.sql("CREATE FUNCTION func1 AS 'org.apache.spark.data.bricks'")
-                spark.sql("CREATE FUNCTION some_db.func2 AS 'org.apache.spark.data.bricks'")
+                spark.sql("CREATE FUNCTION func1 AS 'test.org.apache.spark.sql.MyDoubleAvg'")
+                spark.sql(
+                    "CREATE FUNCTION some_db.func2 AS 'test.org.apache.spark.sql.MyDoubleAvg'")
                 newFunctions = dict((f.name, f) for f in spark.catalog.listFunctions())
                 newFunctionsSomeDb = \
                     dict((f.name, f) for f in spark.catalog.listFunctions("some_db"))

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

@@ -1359,16 +1359,23 @@ class SessionCatalog(
     val info = new ExpressionInfo(funcDefinition.className, func.database.orNull, func.funcName)
     val builder =
       functionBuilder.getOrElse {
-        val className = funcDefinition.className
-        if (!Utils.classIsLoadable(className)) {
-          throw new AnalysisException(s"Can not load class '$className' when registering " +
-            s"the function '$func', please make sure it is on the classpath")
-        }
-        makeFunctionBuilder(func.unquotedString, className)
+        requireFunctionClassExists(funcDefinition)
+        makeFunctionBuilder(func.unquotedString, funcDefinition.className)
       }
     functionRegistry.registerFunction(func, info, builder)
   }
 
+  /**
+   * Make sure function class is on the classpath.
+   */
+  def requireFunctionClassExists(funcDefinition: CatalogFunction): Unit = {
+    val className = funcDefinition.className
+    if (!Utils.classIsLoadable(className)) {
+      throw new AnalysisException(s"Can not load class '$className' when registering " +
+        s"the function '${funcDefinition.identifier}', please make sure it is on the classpath")
+    }
+  }
+
   /**
    * Unregister a temporary or permanent function from a session-specific [[FunctionRegistry]]
    * Return true if function exists.

sql/core/src/main/scala/org/apache/spark/sql/execution/command/functions.scala

@@ -76,14 +76,15 @@ case class CreateFunctionCommand(
   override def run(sparkSession: SparkSession): Seq[Row] = {
     val catalog = sparkSession.sessionState.catalog
     val func = CatalogFunction(FunctionIdentifier(functionName, databaseName), className, resources)
+    catalog.loadFunctionResources(resources)
+    // We fail fast if function class is not exists.
+    catalog.requireFunctionClassExists(func)
     if (isTemp) {
-      // We first load resources and then put the builder in the function registry.
-      catalog.loadFunctionResources(resources)
       catalog.registerFunction(func, overrideIfExists = replace)
     } else {
       // Handles `CREATE OR REPLACE FUNCTION AS ... USING ...`
       if (replace && catalog.functionExists(func.identifier)) {
-        // alter the function in the metastore
+        // Alter the function in the metastore
         catalog.alterFunction(func)
       } else {
         // For a permanent, we will store the metadata into underlying external catalog.

sql/core/src/test/resources/sql-tests/results/udaf.sql.out

@@ -42,7 +42,8 @@ CREATE FUNCTION udaf1 AS 'test.non.existent.udaf'
 -- !query schema
 struct<>
 -- !query output
-
+org.apache.spark.sql.AnalysisException
+Can not load class 'test.non.existent.udaf' when registering the function 'udaf1', please make sure it is on the classpath;
 
 
 -- !query
@@ -51,7 +52,7 @@ SELECT default.udaf1(int_col1) as udaf1 from t1
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Can not load class 'test.non.existent.udaf' when registering the function 'default.udaf1', please make sure it is on the classpath; line 1 pos 7
+Undefined function: 'udaf1'. This function is neither a registered temporary function nor a permanent function registered in the database 'default'.; line 1 pos 7
 
 
 -- !query
@@ -67,4 +68,6 @@ DROP FUNCTION udaf1
 -- !query schema
 struct<>
 -- !query output
+org.apache.spark.sql.catalyst.analysis.NoSuchPermanentFunctionException
+Function 'default.udaf1' not found in database 'default';
 

sql/core/src/test/resources/sql-tests/results/udf/udf-udaf.sql.out

@@ -42,7 +42,8 @@ CREATE FUNCTION udaf1 AS 'test.non.existent.udaf'
 -- !query schema
 struct<>
 -- !query output
-
+org.apache.spark.sql.AnalysisException
+Can not load class 'test.non.existent.udaf' when registering the function 'udaf1', please make sure it is on the classpath;
 
 
 -- !query
@@ -51,7 +52,7 @@ SELECT default.udaf1(udf(int_col1)) as udaf1, udf(default.udaf1(udf(int_col1)))
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Can not load class 'test.non.existent.udaf' when registering the function 'default.udaf1', please make sure it is on the classpath; line 1 pos 94
+Undefined function: 'udaf1'. This function is neither a registered temporary function nor a permanent function registered in the database 'default'.; line 1 pos 7
 
 
 -- !query
@@ -67,4 +68,6 @@ DROP FUNCTION udaf1
 -- !query schema
 struct<>
 -- !query output
+org.apache.spark.sql.catalyst.analysis.NoSuchPermanentFunctionException
+Function 'default.udaf1' not found in database 'default';
 

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -3176,6 +3176,21 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
       }
     }
   }
+
+  test("SPARK-32677: Load function resource before create") {
+    withUserDefinedFunction("func1" -> false) {
+      val func = FunctionIdentifier("func1", Some("default"))
+      val msg = intercept[AnalysisException] {
+        sql("CREATE FUNCTION func1 AS 'test.non.exists.udf'")
+      }.getMessage
+      assert(msg.contains("please make sure it is on the classpath"))
+      assert(!spark.sessionState.catalog.functionExists(func))
+      assert(!spark.sessionState.catalog.isRegisteredFunction(func))
+
+      sql("CREATE FUNCTION func1 AS 'test.org.apache.spark.sql.MyDoubleAvg'")
+      assert(spark.sessionState.catalog.functionExists(func))
+    }
+  }
 }
 
 object FakeLocalFsFileSystem {

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

@@ -156,7 +156,9 @@ class HiveUDFDynamicLoadSuite extends QueryTest with SQLTestUtils with TestHiveS
 
           sql(s"CREATE FUNCTION ${udfInfo.funcName} AS '${udfInfo.className}' USING JAR '$jarUrl'")
 
-          assert(Thread.currentThread().getContextClassLoader eq sparkClassLoader)
+          assert(Thread.currentThread().getContextClassLoader ne sparkClassLoader)
+          assert(Thread.currentThread().getContextClassLoader eq
+            spark.sqlContext.sharedState.jarClassLoader)
 
           // JAR will be loaded at first usage, and it will change the current thread's
           // context classloader to jar classloader in sharedState.