Merge branch 'master' of github.com:DTStack/flinkStreamSQL into v1.5.0_dev

Author: yanxi0227

README.md

@@ -4,6 +4,7 @@
 >  >  * 自定义create view 语法
 >  >  * 自定义create function 语法
 >  >  * 实现了流与维表的join
+>  >  * 支持原生FLinkSQL所有的语法
  
 # 已支持
   * 源表：kafka 0.9，1.x版本
@@ -14,6 +15,8 @@
   * 增加oracle维表，结果表功能
   * 增加SQlServer维表，结果表功能
   * 增加kafka结果表功能
+  * 增加SQL支持CEP
+  * 维表快照
 
 ## 1 快速起步
 ### 1.1 运行模式
@@ -26,7 +29,7 @@
 ### 1.2 执行环境
 
 * Java: JDK8及以上
-* Flink集群: 1.4（单机模式不需要安装Flink集群）
+* Flink集群: 1.4,1.5（单机模式不需要安装Flink集群）
 * 操作系统：理论上不限
 
 ### 1.3 打包
@@ -44,12 +47,12 @@ mvn clean package -Dmaven.test.skip
 #### 1.4.1 启动命令
 
 ```
-sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack/150_flinkplugin/sqlplugin   -localSqlPluginPath D:\gitspace\flinkStreamSQL\plugins   -mode yarn -flinkconf D:\flink_home\kudu150etc  -yarnconf D:\hadoop\etc\hadoopkudu -confProp {\"time.characteristic\":\"EventTime\",\"sql.checkpoint.interval\":10000}
+sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack/150_flinkplugin/sqlplugin   -localSqlPluginPath D:\gitspace\flinkStreamSQL\plugins   -addjar \["udf.jar\"\] -mode yarn -flinkconf D:\flink_home\kudu150etc  -yarnconf D:\hadoop\etc\hadoopkudu -confProp \{\"time.characteristic\":\"EventTime\",\"sql.checkpoint.interval\":10000\}
 ```
 
 #### 1.4.2 命令行参数选项
 
-* **model**
+* **mode**
 	* 描述：执行模式，也就是flink集群的工作模式
 		* local: 本地模式
 		* standalone: 提交到独立部署模式的flink集群
@@ -80,6 +83,7 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 
 * **addjar**
     * 描述：扩展jar路径,当前主要是UDF定义的jar；
+    * 格式：json
     * 必选：否
     * 默认值：无
 
@@ -180,12 +184,16 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 ## 4 样例
 
 ```
+
+CREATE (scala|table) FUNCTION CHARACTER_LENGTH WITH com.dtstack.Kun
+
+
 CREATE TABLE MyTable(
     name varchar,
     channel varchar,
     pv int,
     xctime bigint,
-    CHARACTER_LENGTH(channel) AS timeLeng
+    CHARACTER_LENGTH(channel) AS timeLeng //自定义的函数
  )WITH(
     type ='kafka09',
     bootstrapServers ='172.16.8.198:9092',
@@ -223,7 +231,7 @@ CREATE TABLE sideTable(
     cf:name varchar as name,
     cf:info varchar as info,
     PRIMARY KEY(name),
-    PERIOD FOR SYSTEM_TIME
+    PERIOD FOR SYSTEM_TIME //维表标识
  )WITH(
     type ='hbase',
     zookeeperQuorum ='rdos1:2181',

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -91,8 +91,6 @@ public class Main {
 
     private static final Logger LOG = LoggerFactory.getLogger(Main.class);
 
-    private static final String LOCAL_MODE = "local";
-
     private static final int failureRate = 3;
 
     private static final int failureInterval = 6; //min
@@ -141,7 +139,7 @@ public static void main(String[] args) throws Exception {
         Thread.currentThread().setContextClassLoader(dtClassLoader);
 
         URLClassLoader parentClassloader;
-        if(!LOCAL_MODE.equals(deployMode)){
+        if(!ClusterMode.local.name().equals(deployMode)){
             parentClassloader = (URLClassLoader) threadClassLoader.getParent();
         }else{
             parentClassloader = dtClassLoader;
@@ -313,7 +311,7 @@ private static void registerTable(SqlTree sqlTree, StreamExecutionEnvironment en
     }
 
     private static StreamExecutionEnvironment getStreamExeEnv(Properties confProperties, String deployMode) throws IOException {
-        StreamExecutionEnvironment env = !LOCAL_MODE.equals(deployMode) ?
+        StreamExecutionEnvironment env = !ClusterMode.local.name().equals(deployMode) ?
                 StreamExecutionEnvironment.getExecutionEnvironment() :
                 new MyLocalStreamEnvironment();
 

mysql/mysql-side/mysql-all-side/src/main/java/com/dtstack/flink/sql/side/mysql/MysqlAllReqRow.java

@@ -71,7 +71,6 @@ public MysqlAllReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo
         super(new MysqlAllSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
     }
 
-
     @Override
     protected Row fillData(Row input, Object sideInput) {
         Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;

redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java

@@ -144,14 +144,14 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
         List<String> value = async.keys(key + ":*").get();
         String[] values = value.toArray(new String[value.size()]);
         RedisFuture<List<KeyValue<String, String>>> future =  async.mget(values);
-        while (future.isDone()){
-            try {
-                List<KeyValue<String, String>> kvList = future.get();
-                if (kvList.size() != 0){
-                    for (int i=0; i<kvList.size(); i++){
-                        String[] splitKeys = kvList.get(i).getKey().split(":");
+        future.thenAccept(new Consumer<List<KeyValue<String, String>>>() {
+            @Override
+            public void accept(List<KeyValue<String, String>> keyValues) {
+                if (keyValues.size() != 0){
+                    for (int i=0; i<keyValues.size(); i++){
+                        String[] splitKeys = keyValues.get(i).getKey().split(":");
                         keyValue.put(splitKeys[1], splitKeys[2]);
-                        keyValue.put(splitKeys[3], kvList.get(i).getValue());
+                        keyValue.put(splitKeys[3], keyValues.get(i).getValue());
                     }
                     Row row = fillData(input, keyValue);
                     resultFuture.complete(Collections.singleton(row));
@@ -164,13 +164,8 @@ public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Except
                         putCache(key, CacheMissVal.getMissKeyObj());
                     }
                 }
-            } catch (InterruptedException e1) {
-                e1.printStackTrace();
-            } catch (ExecutionException e1) {
-                e1.printStackTrace();
             }
-        }
-
+        });
     }
 
     private String buildCacheKey(List<String> keyData) {