Merge branch 'v1.8.0_dev_feature_udaf' into 'v1.8.0_dev'

flinksql增加udaf函数功能

flinksql增加udaf函数功能

See merge request !70

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/parser/CreateFuncParser.java

@@ -32,7 +32,7 @@
 
 public class CreateFuncParser implements IParser {
 
-    private static final String funcPatternStr = "(?i)\\s*create\\s+(scala|table)\\s+function\\s+(\\S+)\\s+WITH\\s+(\\S+)";
+    private static final String funcPatternStr = "(?i)\\s*create\\s+(scala|table|aggregate)\\s+function\\s+(\\S+)\\s+WITH\\s+(\\S+)";
 
     private static final Pattern funcPattern = Pattern.compile(funcPatternStr);
 

core/src/main/java/com/dtstack/flink/sql/util/FlinkUtil.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.util;
 
@@ -34,6 +34,8 @@
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.functions.ScalarFunction;
 import org.apache.flink.table.functions.TableFunction;
+import org.apache.flink.table.functions.AggregateFunction;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -156,10 +158,8 @@ public static void setStreamTimeCharacteristic(StreamExecutionEnvironment env, P
     }
 
 
-
     /**
-     * FIXME 暂时不支持 UDF 实现类--有参构造方法
-     * TABLE|SCALA
+     * TABLE|SCALA|AGGREGATE
      * 注册UDF到table env
      */
     public static void registerUDF(String type, String classPath, String funcName, TableEnvironment tableEnv,
@@ -168,10 +168,11 @@ public static void registerUDF(String type, String classPath, String funcName, T
             registerScalaUDF(classPath, funcName, tableEnv, classLoader);
         }else if("TABLE".equalsIgnoreCase(type)){
             registerTableUDF(classPath, funcName, tableEnv, classLoader);
+        }else if("AGGREGATE".equalsIgnoreCase(type)){
+            registerAggregateUDF(classPath, funcName, tableEnv, classLoader);
         }else{
-            throw new RuntimeException("not support of UDF which is not in (TABLE, SCALA)");
+            throw new RuntimeException("not support of UDF which is not in (TABLE, SCALA, AGGREGATE)");
         }
-
     }
 
     /**
@@ -195,7 +196,7 @@ public static void registerScalaUDF(String classPath, String funcName, TableEnvi
 
     /**
      * 注册自定义TABLEFFUNC方法到env上
-     * TODO 对User-Defined Aggregate Functions的支持
+     *
      * @param classPath
      * @param funcName
      * @param tableEnv
@@ -206,11 +207,11 @@ public static void registerTableUDF(String classPath, String funcName, TableEnvi
             TableFunction udfFunc = Class.forName(classPath, false, classLoader)
                     .asSubclass(TableFunction.class).newInstance();
 
-            if(tableEnv instanceof StreamTableEnvironment){
-                ((StreamTableEnvironment)tableEnv).registerFunction(funcName, udfFunc);
-            }else if(tableEnv instanceof BatchTableEnvironment){
-                ((BatchTableEnvironment)tableEnv).registerFunction(funcName, udfFunc);
-            }else{
+            if (tableEnv instanceof StreamTableEnvironment) {
+                ((StreamTableEnvironment) tableEnv).registerFunction(funcName, udfFunc);
+            } else if (tableEnv instanceof BatchTableEnvironment) {
+                ((BatchTableEnvironment) tableEnv).registerFunction(funcName, udfFunc);
+            } else {
                 throw new RuntimeException("no support tableEnvironment class for " + tableEnv.getClass().getName());
             }
 
@@ -221,6 +222,33 @@ public static void registerTableUDF(String classPath, String funcName, TableEnvi
         }
     }
 
+    /**
+     * 注册自定义Aggregate FUNC方法到env上
+     *
+     * @param classPath
+     * @param funcName
+     * @param tableEnv
+     */
+    public static void registerAggregateUDF(String classPath, String funcName, TableEnvironment tableEnv,
+                                            ClassLoader classLoader) {
+        try {
+            AggregateFunction udfFunc = Class.forName(classPath, false, classLoader)
+                    .asSubclass(AggregateFunction.class).newInstance();
+
+            if (tableEnv instanceof StreamTableEnvironment) {
+                ((StreamTableEnvironment) tableEnv).registerFunction(funcName,  udfFunc);
+            } else if (tableEnv instanceof BatchTableEnvironment) {
+                ((BatchTableEnvironment) tableEnv).registerFunction(funcName,  udfFunc);
+            } else {
+                throw new RuntimeException("no support tableEnvironment class for " + tableEnv.getClass().getName());
+            }
+
+            logger.info("register Aggregate function:{} success.", funcName);
+        } catch (Exception e) {
+            logger.error("", e);
+            throw new RuntimeException("register Aggregate UDF exception:", e);
+        }
+    }
 
     /**
      *