[SPARK-31999][SQL] Add REFRESH FUNCTION command

docs/_data/menu-sql.yaml

@@ -208,6 +208,8 @@
                   url: sql-ref-syntax-aux-cache-clear-cache.html
                 - text: REFRESH TABLE
                   url: sql-ref-syntax-aux-refresh-table.html
+                - text: REFRESH FUNCTION
+                  url: sql-ref-syntax-aux-refresh-function.html
                 - text: REFRESH
                   url: sql-ref-syntax-aux-cache-refresh.html
             - text: DESCRIBE

docs/sql-ref-syntax-aux-cache-cache-table.md

@@ -80,3 +80,4 @@ CACHE TABLE testCache OPTIONS ('storageLevel' 'DISK_ONLY') SELECT * FROM testDat
 * [UNCACHE TABLE](sql-ref-syntax-aux-cache-uncache-table.html)
 * [REFRESH TABLE](sql-ref-syntax-aux-refresh-table.html)
 * [REFRESH](sql-ref-syntax-aux-cache-refresh.html)
+* [REFRESH FUNCTION](sql-ref-syntax-aux-refresh-function.html)

docs/sql-ref-syntax-aux-cache-clear-cache.md

@@ -41,3 +41,4 @@ CLEAR CACHE;
 * [UNCACHE TABLE](sql-ref-syntax-aux-cache-uncache-table.html)
 * [REFRESH TABLE](sql-ref-syntax-aux-refresh-table.html)
 * [REFRESH](sql-ref-syntax-aux-cache-refresh.html)
+* [REFRESH FUNCTION](sql-ref-syntax-aux-refresh-function.html)

docs/sql-ref-syntax-aux-cache-refresh.md

@@ -54,3 +54,4 @@ REFRESH "hdfs://path/to/table";
 * [CLEAR CACHE](sql-ref-syntax-aux-cache-clear-cache.html)
 * [UNCACHE TABLE](sql-ref-syntax-aux-cache-uncache-table.html)
 * [REFRESH TABLE](sql-ref-syntax-aux-refresh-table.html)
+* [REFRESH FUNCTION](sql-ref-syntax-aux-refresh-function.html)

docs/sql-ref-syntax-aux-cache-uncache-table.md

@@ -50,3 +50,4 @@ UNCACHE TABLE t1;
 * [CLEAR CACHE](sql-ref-syntax-aux-cache-clear-cache.html)
 * [REFRESH TABLE](sql-ref-syntax-aux-refresh-table.html)
 * [REFRESH](sql-ref-syntax-aux-cache-refresh.html)
+* [REFRESH FUNCTION](sql-ref-syntax-aux-refresh-function.html)

docs/sql-ref-syntax-aux-cache.md

@@ -23,4 +23,5 @@ license: |
  * [UNCACHE TABLE statement](sql-ref-syntax-aux-cache-uncache-table.html)
  * [CLEAR CACHE statement](sql-ref-syntax-aux-cache-clear-cache.html)
  * [REFRESH TABLE statement](sql-ref-syntax-aux-refresh-table.html)
- * [REFRESH statement](sql-ref-syntax-aux-cache-refresh.html)
+ * [REFRESH statement](sql-ref-syntax-aux-cache-refresh.html)
+ * [REFRESH FUNCTION statement](sql-ref-syntax-aux-refresh-function.html)

docs/sql-ref-syntax-aux-refresh-function.md

@@ -0,0 +1,60 @@
+---
+layout: global
+title: REFRESH FUNCTION
+displayTitle: REFRESH FUNCTION
+license: |
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+ 
+     http://www.apache.org/licenses/LICENSE-2.0
+ 
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+---
+
+### Description
+
+`REFRESH FUNCTION` statement invalidates the cached function entry, which include class name
+and resource location of the given function. The invalidated cache is populated right away.
+Note that, refresh function only works for permanent function. Refresh native function or temporary function will cause exception.
+
+### Syntax
+
+```sql
+REFRESH FUNCTION function_identifier
+```
+
+### Parameters
+
+* **function_identifier**
+
+    Specifies a function name, which is either a qualified or unqualified name. If no database identifier is provided, use the current database.
+
+    **Syntax:** `[ database_name. ] function_name`
+
+### Examples
+
+```sql
+-- The cached entries of the function will be refreshed  
+-- The function is resolved from the current database as the function name is unqualified.
+REFRESH FUNCTION func1;
+
+-- The cached entries of the function will be refreshed
+-- The function is resolved from tempDB database as the function name is qualified.
+REFRESH FUNCTION tempDB.func1;   
+```
+
+### Related Statements
+
+* [CACHE TABLE](sql-ref-syntax-aux-cache-cache-table.html)
+* [CLEAR CACHE](sql-ref-syntax-aux-cache-clear-cache.html)
+* [UNCACHE TABLE](sql-ref-syntax-aux-cache-uncache-table.html)
+* [REFRESH TABLE](sql-ref-syntax-aux-refresh-table.html)
+* [REFRESH](sql-ref-syntax-aux-cache-refresh.html)

docs/sql-ref-syntax-aux-refresh-table.md

@@ -57,3 +57,4 @@ REFRESH TABLE tempDB.view1;
 * [CLEAR CACHE](sql-ref-syntax-aux-cache-clear-cache.html)
 * [UNCACHE TABLE](sql-ref-syntax-aux-cache-uncache-table.html)
 * [REFRESH](sql-ref-syntax-aux-cache-refresh.html)
+* [REFRESH FUNCTION](sql-ref-syntax-aux-refresh-function.html)

docs/sql-ref-syntax.md

@@ -83,6 +83,7 @@ Spark SQL is Apache Spark's module for working with structured data. The SQL Syn
  * [LIST JAR](sql-ref-syntax-aux-resource-mgmt-list-jar.html)
  * [REFRESH](sql-ref-syntax-aux-cache-refresh.html)
  * [REFRESH TABLE](sql-ref-syntax-aux-refresh-table.html)
+ * [REFRESH FUNCTION](sql-ref-syntax-aux-refresh-function.html)
  * [RESET](sql-ref-syntax-aux-conf-mgmt-reset.html)
  * [SET](sql-ref-syntax-aux-conf-mgmt-set.html)
  * [SHOW COLUMNS](sql-ref-syntax-aux-show-columns.html)

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -229,6 +229,7 @@ statement
         comment=(STRING | NULL)                                        #commentNamespace
     | COMMENT ON TABLE multipartIdentifier IS comment=(STRING | NULL)  #commentTable
     | REFRESH TABLE multipartIdentifier                                #refreshTable
+    | REFRESH FUNCTION multipartIdentifier                             #refreshFunction
     | REFRESH (STRING | .*?)                                           #refreshResource
     | CACHE LAZY? TABLE multipartIdentifier
         (OPTIONS options=tablePropertyList)? (AS? query)?              #cacheTable

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

@@ -1341,6 +1341,16 @@ class SessionCatalog(
     functionRegistry.registerFunction(func, info, builder)
   }
 
+  /**
+   * Unregister a temporary or permanent function from a session-specific [[FunctionRegistry]]
+   */
+  def unregisterFunction(name: FunctionIdentifier, ignoreIfNotExists: Boolean): Unit = {
+    if (!functionRegistry.dropFunction(name) && !ignoreIfNotExists) {
+      throw new NoSuchFunctionException(
+        formatDatabaseName(name.database.getOrElse(currentDb)), name.funcName)
+    }
+  }
+
   /**
    * Drop a temporary function.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -3650,6 +3650,11 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
       ctx.REPLACE != null)
   }
 
+  override def visitRefreshFunction(ctx: RefreshFunctionContext): LogicalPlan = withOrigin(ctx) {
+    val functionIdentifier = visitMultipartIdentifier(ctx.multipartIdentifier)
+    RefreshFunction(functionIdentifier)
+  }
+
   override def visitCommentNamespace(ctx: CommentNamespaceContext): LogicalPlan = withOrigin(ctx) {
     val comment = ctx.comment.getType match {
       case SqlBaseParser.NULL => ""

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala

@@ -516,3 +516,8 @@ case class CommentOnNamespace(child: LogicalPlan, comment: String) extends Comma
 case class CommentOnTable(child: LogicalPlan, comment: String) extends Command {
   override def children: Seq[LogicalPlan] = child :: Nil
 }
+
+/**
+ * The logical plan of the REFRESH FUNCTION command that works for v2 catalogs.
+ */
+case class RefreshFunction(func: Seq[String]) extends Command

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala

@@ -2113,6 +2113,12 @@ class DDLParserSuite extends AnalysisTest {
       "Operation not allowed: CREATE FUNCTION with resource type 'other'")
   }
 
+  test("REFRESH FUNCTION") {
+    parseCompare("REFRESH FUNCTION c", RefreshFunction(Seq("c")))
+    parseCompare("REFRESH FUNCTION b.c", RefreshFunction(Seq("b", "c")))
+    parseCompare("REFRESH FUNCTION a.b.c", RefreshFunction(Seq("a", "b", "c")))
+  }
+
   private case class TableSpec(
       name: Seq[String],
       schema: Option[StructType],

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -611,6 +611,12 @@ class ResolveSessionCatalog(
         CreateFunctionCommand(database, function, className, resources, isTemp, ignoreIfExists,
           replace)
       }
+
+    case RefreshFunction(func) =>
+      val FunctionIdentifier(function, database) =
+        parseSessionCatalogFunctionIdentifier(func, "REFRESH FUNCTION")
+      // Fallback to v1 command
+      RefreshFunctionCommand(database, function)
   }
 
   // TODO: move function related v2 statements to the new framework.

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

@@ -236,6 +236,58 @@ case class ShowFunctionsCommand(
   }
 }
 
+
+/**
+ * A command for users to refresh the persistent function.
+ * The syntax of using this command in SQL is:
+ * {{{
+ *    REFRESH FUNCTION functionName
+ * }}}
+ */
+case class RefreshFunctionCommand(
+    databaseName: Option[String],
+    functionName: String)
+  extends RunnableCommand {
+
+  override def run(sparkSession: SparkSession): Seq[Row] = {
+    val catalog = sparkSession.sessionState.catalog
+    if (FunctionRegistry.builtin.functionExists(FunctionIdentifier(functionName))) {
+      throw new AnalysisException(s"Cannot refresh native function $functionName")
+    } else if (catalog.isTemporaryFunction(FunctionIdentifier(functionName, databaseName))) {
+      throw new AnalysisException(s"Cannot refresh temp function $functionName")
+    } else {
+      // we only refresh the permanent function.
+      // there are 4 cases:
+      // 1. registry exists externalCatalog exists
+      // 2. registry exists externalCatalog not exists
+      // 3. registry not exists externalCatalog exists
+      // 4. registry not exists externalCatalog not exists
+      val identifier = FunctionIdentifier(
+        functionName, Some(databaseName.getOrElse(catalog.getCurrentDatabase)))
+      val isRegisteredFunction = catalog.isRegisteredFunction(identifier)
+      val isPersistentFunction = catalog.isPersistentFunction(identifier)
+      if (isRegisteredFunction && isPersistentFunction) {
+        // re-register function
+        catalog.unregisterFunction(identifier, true)
+        val func = catalog.getFunctionMetadata(identifier)
+        catalog.registerFunction(func, true)
+      } else if (isRegisteredFunction && !isPersistentFunction) {
+        // unregister function and throw NoSuchFunctionException
+        catalog.unregisterFunction(identifier, true)
+        throw new NoSuchFunctionException(identifier.database.get, functionName)
+      } else if (!isRegisteredFunction && isPersistentFunction) {
+        // register function
+        val func = catalog.getFunctionMetadata(identifier)
+        catalog.registerFunction(func, true)
+      } else {
+        throw new NoSuchFunctionException(identifier.database.get, functionName)
+      }
+    }
+
+    Seq.empty[Row]
+  }
+}
+
 object FunctionsCommand {
   // operators that do not have corresponding functions.
   // They should be handled `DescribeFunctionCommand`, `ShowFunctionsCommand`

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

@@ -28,8 +28,8 @@ import org.apache.spark.{SparkException, SparkFiles}
 import org.apache.spark.internal.config
 import org.apache.spark.internal.config.RDD_PARALLEL_LISTING_THRESHOLD
 import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
-import org.apache.spark.sql.catalyst.{QualifiedTableName, TableIdentifier}
-import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, NoSuchDatabaseException, NoSuchPartitionException, NoSuchTableException, TempTableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.{FunctionIdentifier, QualifiedTableName, TableIdentifier}
+import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, NoSuchDatabaseException, NoSuchFunctionException, NoSuchPartitionException, NoSuchTableException, TempTableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.connector.catalog.SupportsNamespaces.PROP_OWNER
@@ -3030,6 +3030,49 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
       }
     }
   }
+
+  test("REFRESH FUNCTION") {
+    val msg = intercept[AnalysisException] {
+      sql("REFRESH FUNCTION md5")
+    }.getMessage
+    assert(msg.contains("Cannot refresh native function"))
+
+    withUserDefinedFunction("func1" -> true) {
+      sql("CREATE TEMPORARY FUNCTION func1 AS 'test.org.apache.spark.sql.MyDoubleAvg'")
+      val msg = intercept[AnalysisException] {
+        sql("REFRESH FUNCTION func1")
+      }.getMessage
+      assert(msg.contains("Cannot refresh temp function"))
+    }
+
+    withUserDefinedFunction("func1" -> false) {
+      intercept[NoSuchFunctionException] {
+        sql("REFRESH FUNCTION func1")
+      }
+
+      val func = FunctionIdentifier("func1", Some("default"))
+      sql("CREATE FUNCTION func1 AS 'test.org.apache.spark.sql.MyDoubleAvg'")
+      assert(spark.sessionState.catalog.isRegisteredFunction(func) == false)
+      sql("REFRESH FUNCTION func1")
+      assert(spark.sessionState.catalog.isRegisteredFunction(func) == true)
+
+      spark.sessionState.catalog.externalCatalog.dropFunction("default", "func1")
+      assert(spark.sessionState.catalog.isRegisteredFunction(func) == true)
+      intercept[NoSuchFunctionException] {
+        sql("REFRESH FUNCTION func1")
+      }
+      assert(spark.sessionState.catalog.isRegisteredFunction(func) == false)
+
+      val function = CatalogFunction(func, "test.non.exists.udf", Seq.empty)
+      spark.sessionState.catalog.createFunction(function, false)
+      assert(spark.sessionState.catalog.isRegisteredFunction(func) == false)
+      val err = intercept[AnalysisException] {
+        sql("REFRESH FUNCTION func1")
+      }.getMessage
+      assert(err.contains("Can not load class"))
+      assert(spark.sessionState.catalog.isRegisteredFunction(func) == false)
+    }
+  }
 }
 
 object FakeLocalFsFileSystem {