[SPARK-14123][SPARK-14384][SQL] Handle CreateFunction/DropFunction

## What changes were proposed in this pull request?
This PR implements CreateFunction and DropFunction commands. Besides implementing these two commands, we also change how to manage functions. Here are the main changes.
* `FunctionRegistry` will be a container to store all functions builders and it will not actively load any functions. Because of this change, we do not need to maintain a separate registry for HiveContext. So, `HiveFunctionRegistry` is deleted.
* SessionCatalog takes care the job of loading a function if this function is not in the `FunctionRegistry` but its metadata is stored in the external catalog. For this case, SessionCatalog will (1) load the metadata from the external catalog, (2) load all needed resources (i.e. jars and files), (3) create a function builder based on the function definition, (4) register the function builder in the `FunctionRegistry`.
* A `UnresolvedGenerator` is created. So, the parser will not need to call `FunctionRegistry` directly during parsing, which is not a good time to create a Hive UDTF. In the analysis phase, we will resolve `UnresolvedGenerator`.

This PR is based on viirya's https://github.com/apache/spark/pull/12036/

## How was this patch tested?
Existing tests and new tests.

## TODOs
[x] Self-review
[x] Cleanup
[x] More tests for create/drop functions (we need to more tests for permanent functions).
[ ] File JIRAs for all TODOs
[x] Standardize the error message when a function does not exist.

Author: Yin Huai <yhuai@databricks.com>
Author: Liang-Chi Hsieh <simonh@tw.ibm.com>

Closes #12117 from yhuai/function.

1 files changed, 51 insertions, 43 deletions

diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2Suites.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2Suites.scala
index 2c7358e59a..a1268b8e94 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2Suites.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveThriftServer2Suites.scala
@@ -491,46 +491,50 @@ class HiveThriftBinaryServerSuite extends HiveThriftJdbcTest {
 
   test("SPARK-11595 ADD JAR with input path having URL scheme") {
     withJdbcStatement { statement =>
-      val jarPath = "../hive/src/test/resources/TestUDTF.jar"
-      val jarURL = s"file://${System.getProperty("user.dir")}/$jarPath"
+      try {
+        val jarPath = "../hive/src/test/resources/TestUDTF.jar"
+        val jarURL = s"file://${System.getProperty("user.dir")}/$jarPath"
 
-      Seq(
-        s"ADD JAR $jarURL",
-        s"""CREATE TEMPORARY FUNCTION udtf_count2
-           |AS 'org.apache.spark.sql.hive.execution.GenericUDTFCount2'
-         """.stripMargin
-      ).foreach(statement.execute)
+        Seq(
+          s"ADD JAR $jarURL",
+          s"""CREATE TEMPORARY FUNCTION udtf_count2
+             |AS 'org.apache.spark.sql.hive.execution.GenericUDTFCount2'
+           """.stripMargin
+        ).foreach(statement.execute)
 
-      val rs1 = statement.executeQuery("DESCRIBE FUNCTION udtf_count2")
+        val rs1 = statement.executeQuery("DESCRIBE FUNCTION udtf_count2")
 
-      assert(rs1.next())
-      assert(rs1.getString(1) === "Function: udtf_count2")
+        assert(rs1.next())
+        assert(rs1.getString(1) === "Function: udtf_count2")
 
-      assert(rs1.next())
-      assertResult("Class: org.apache.spark.sql.hive.execution.GenericUDTFCount2") {
-        rs1.getString(1)
-      }
+        assert(rs1.next())
+        assertResult("Class: org.apache.spark.sql.hive.execution.GenericUDTFCount2") {
+          rs1.getString(1)
+        }
 
-      assert(rs1.next())
-      assert(rs1.getString(1) === "Usage: To be added.")
+        assert(rs1.next())
+        assert(rs1.getString(1) === "Usage: To be added.")
 
-      val dataPath = "../hive/src/test/resources/data/files/kv1.txt"
+        val dataPath = "../hive/src/test/resources/data/files/kv1.txt"
 
-      Seq(
-        s"CREATE TABLE test_udtf(key INT, value STRING)",
-        s"LOAD DATA LOCAL INPATH '$dataPath' OVERWRITE INTO TABLE test_udtf"
-      ).foreach(statement.execute)
+        Seq(
+          s"CREATE TABLE test_udtf(key INT, value STRING)",
+          s"LOAD DATA LOCAL INPATH '$dataPath' OVERWRITE INTO TABLE test_udtf"
+        ).foreach(statement.execute)
 
-      val rs2 = statement.executeQuery(
-        "SELECT key, cc FROM test_udtf LATERAL VIEW udtf_count2(value) dd AS cc")
+        val rs2 = statement.executeQuery(
+          "SELECT key, cc FROM test_udtf LATERAL VIEW udtf_count2(value) dd AS cc")
 
-      assert(rs2.next())
-      assert(rs2.getInt(1) === 97)
-      assert(rs2.getInt(2) === 500)
+        assert(rs2.next())
+        assert(rs2.getInt(1) === 97)
+        assert(rs2.getInt(2) === 500)
 
-      assert(rs2.next())
-      assert(rs2.getInt(1) === 97)
-      assert(rs2.getInt(2) === 500)
+        assert(rs2.next())
+        assert(rs2.getInt(1) === 97)
+        assert(rs2.getInt(2) === 500)
+      } finally {
+        statement.executeQuery("DROP TEMPORARY FUNCTION udtf_count2")
+      }
     }
   }
 
@@ -565,24 +569,28 @@ class SingleSessionSuite extends HiveThriftJdbcTest {
       },
 
       { statement =>
-        val rs1 = statement.executeQuery("SET foo")
+        try {
+          val rs1 = statement.executeQuery("SET foo")
 
-        assert(rs1.next())
-        assert(rs1.getString(1) === "foo")
-        assert(rs1.getString(2) === "bar")
+          assert(rs1.next())
+          assert(rs1.getString(1) === "foo")
+          assert(rs1.getString(2) === "bar")
 
-        val rs2 = statement.executeQuery("DESCRIBE FUNCTION udtf_count2")
+          val rs2 = statement.executeQuery("DESCRIBE FUNCTION udtf_count2")
 
-        assert(rs2.next())
-        assert(rs2.getString(1) === "Function: udtf_count2")
+          assert(rs2.next())
+          assert(rs2.getString(1) === "Function: udtf_count2")
 
-        assert(rs2.next())
-        assertResult("Class: org.apache.spark.sql.hive.execution.GenericUDTFCount2") {
-          rs2.getString(1)
-        }
+          assert(rs2.next())
+          assertResult("Class: org.apache.spark.sql.hive.execution.GenericUDTFCount2") {
+            rs2.getString(1)
+          }
 
-        assert(rs2.next())
-        assert(rs2.getString(1) === "Usage: To be added.")
+          assert(rs2.next())
+          assert(rs2.getString(1) === "Usage: To be added.")
+        } finally {
+          statement.executeQuery("DROP TEMPORARY FUNCTION udtf_count2")
+        }
       }
     )
   }