[FLINK-24862][hive]Fix hive user-defined function cannot be used normally in hive dialect (apache#17761)

Author: xiangqiao123

flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectITCase.java

@@ -58,6 +58,7 @@
 import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
 import org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat;
 import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFCount;
 import org.apache.hadoop.hive.ql.udf.generic.GenericUDFAbs;
 import org.apache.hadoop.hive.serde.serdeConstants;
 import org.apache.hadoop.hive.serde2.columnar.LazyBinaryColumnarSerDe;
@@ -635,6 +636,35 @@ public void testTemporaryFunction() throws Exception {
         tableEnv.executeSql("drop temporary function if exists foo");
     }
 
+    @Test
+    public void testTemporaryFunctionUDAF() throws Exception {
+        // create temp function
+        tableEnv.executeSql(
+                String.format(
+                        "create temporary function temp_count as '%s'",
+                        GenericUDAFCount.class.getName()));
+        String[] functions = tableEnv.listUserDefinedFunctions();
+        assertArrayEquals(new String[] {"temp_count"}, functions);
+        // call the function
+        tableEnv.executeSql("create table src(x int)");
+        tableEnv.executeSql("insert into src values (1),(-1)").await();
+        assertEquals(
+                "[+I[2]]",
+                queryResult(tableEnv.sqlQuery("select temp_count(x) from src")).toString());
+        // switch DB and the temp function can still be used
+        tableEnv.executeSql("create database db1");
+        tableEnv.useDatabase("db1");
+        assertEquals(
+                "[+I[2]]",
+                queryResult(tableEnv.sqlQuery("select temp_count(x) from `default`.src"))
+                        .toString());
+        // drop the function
+        tableEnv.executeSql("drop temporary function temp_count");
+        functions = tableEnv.listUserDefinedFunctions();
+        assertEquals(0, functions.length);
+        tableEnv.executeSql("drop temporary function if exists foo");
+    }
+
     @Test
     public void testCatalog() {
         List<Row> catalogs =

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java

@@ -631,16 +631,14 @@ private void validateAndPrepareFunction(CatalogFunction function)
         // In this situation the UDF have not been validated and cleaned, so we need to validate it
         // and clean its closure here.
         // If the input is instance of `ScalarFunctionDefinition`, `TableFunctionDefinition` and so
-        // on,
-        // it means it uses the old type inference. We assume that they have been validated before
-        // being
-        // wrapped.
-        if (function instanceof InlineCatalogFunction
-                && ((InlineCatalogFunction) function).getDefinition()
-                        instanceof UserDefinedFunction) {
-
+        // on, it means it uses the old type inference. We assume that they have been validated
+        // before being wrapped.
+        if (function instanceof InlineCatalogFunction) {
             FunctionDefinition definition = ((InlineCatalogFunction) function).getDefinition();
-            UserDefinedFunctionHelper.prepareInstance(config, (UserDefinedFunction) definition);
+            if (definition instanceof UserDefinedFunction) {
+                UserDefinedFunctionHelper.prepareInstance(config, (UserDefinedFunction) definition);
+            }
+            // Skip validation if it's not a UserDefinedFunction.
         } else if (function.getFunctionLanguage() == FunctionLanguage.JAVA) {
             ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
             UserDefinedFunctionHelper.validateClass(