Merge pull request #6 from DTStack/v1.4.1

V1.4.1

Author: yangsishu

.gitignore

@@ -1,11 +1,11 @@
 target
 .idea/
 /.idea/*
-/target
-target
+target/
 .class
 .project
 .classpath
 *.eclipse.*
 *.iml
 plugins/
+lib/

README.md

@@ -38,14 +38,18 @@
 
 ```
 mvn clean package -Dmaven.test.skip
+打包结束后，项目根目录下会产生plugins目录，plugins目录下存放编译好的数据同步插件包,在lib目下存放job提交的包
 ```
 
-打包结束后，项目根目录下会产生plugins目录，plugins目录下存放编译好的数据同步插件包
+### 1.4 启动
 
+#### 1.4.1 启动命令
 
-### 1.4 启动
+```
+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}
+```
 
-#### 1.4.1 命令行参数选项
+#### 1.4.2 命令行参数选项
 
 * **model**
 	* 描述：执行模式，也就是flink集群的工作模式

bin/submit.sh

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+
+#
+# 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.
+#
+
+set -e
+
+export SQL_HOME="$(cd "`dirname "$0"`"/..; pwd)"
+
+# Find the java binary
+if [ -n "${JAVA_HOME}" ]; then
+  JAVA_RUN="${JAVA_HOME}/bin/java"
+else
+  if [ `command -v java` ]; then
+    JAVA_RUN="java"
+  else
+    echo "JAVA_HOME is not set" >&2
+    exit 1
+  fi
+fi
+
+JAR_DIR=$SQL_HOME/lib/*
+CLASS_NAME=com.dtstack.flink.sql.launcher.LauncherMain
+
+echo "sql submit ..."
+nohup $JAVA_RUN -cp $JAR_DIR $CLASS_NAME $@ &

core/pom.xml

@@ -2,7 +2,7 @@
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <parent>
         <artifactId>flink.sql</artifactId>
-        <groupId>com.dtstack.flinkx</groupId>
+        <groupId>com.dtstack.flink</groupId>
         <version>1.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>

core/src/main/java/com/dtstack/flink/sql/enums/ECacheType.java

@@ -27,7 +27,7 @@
  * @author xuchao
  */
 public enum ECacheType {
-    NONE, LRU;
+    NONE, LRU, ALL;
 
     public static boolean isValid(String type){
         for(ECacheType tmpType : ECacheType.values()){

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

@@ -20,8 +20,10 @@
 
 package com.dtstack.flink.sql.parser;
 
+import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 
+import java.util.List;
 import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -70,9 +72,9 @@ private Map parseProp(String propsStr){
         String[] strs = propsStr.trim().split("'\\s*,");
         Map<String, Object> propMap = Maps.newHashMap();
         for(int i=0; i<strs.length; i++){
-            String[] ss = strs[i].split("=");
-            String key = ss[0].trim();
-            String value = ss[1].trim().replaceAll("'", "").trim();
+            List<String> ss = DtStringUtil.splitIgnoreQuota(strs[i], '=');
+            String key = ss.get(0).trim();
+            String value = ss.get(1).trim().replaceAll("'", "").trim();
             propMap.put(key, value);
         }
 

core/src/main/java/com/dtstack/flink/sql/side/AllReqRow.java

@@ -0,0 +1,69 @@
+/*
+ * 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.
+ */
+
+
+
+package com.dtstack.flink.sql.side;
+
+import com.dtstack.flink.sql.threadFactory.DTThreadFactory;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.types.Row;
+
+import java.sql.SQLException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Reason:
+ * Date: 2018/9/18
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public abstract class AllReqRow extends RichFlatMapFunction<Row, Row>{
+
+    protected SideInfo sideInfo;
+
+    private ScheduledExecutorService es;
+
+    public AllReqRow(SideInfo sideInfo){
+        this.sideInfo = sideInfo;
+
+    }
+
+    protected abstract Row fillData(Row input, Object sideInput);
+
+    protected abstract void initCache() throws SQLException;
+
+    protected abstract void reloadCache();
+
+    @Override
+    public void open(Configuration parameters) throws Exception {
+        super.open(parameters);
+        initCache();
+        System.out.println("----- all cacheRef init end-----");
+
+        //start reload cache thread
+        SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
+        es = Executors.newSingleThreadScheduledExecutor(new DTThreadFactory("cache-all-reload"));
+        es.scheduleAtFixedRate(() -> reloadCache(), sideTableInfo.getCacheTimeout(), sideTableInfo.getCacheTimeout(), TimeUnit.MILLISECONDS);
+    }
+
+}

core/src/main/java/com/dtstack/flink/sql/side/AsyncReqRow.java

@@ -25,23 +25,12 @@
 import com.dtstack.flink.sql.side.cache.CacheObj;
 import com.dtstack.flink.sql.side.cache.LRUSideCache;
 import org.apache.calcite.sql.JoinType;
-import org.apache.calcite.sql.SqlBasicCall;
-import org.apache.calcite.sql.SqlIdentifier;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlNode;
-import org.apache.flink.api.java.typeutils.RowTypeInfo;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
-import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.async.ResultFuture;
 import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
 import org.apache.flink.types.Row;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
-import java.util.List;
-import java.util.Map;
 
 /**
  * All interfaces inherit naming rules: type + "AsyncReqRow" such as == "MysqlAsyncReqRow
@@ -53,50 +42,24 @@
 
 public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> {
 
-    private static final Logger LOG = LoggerFactory.getLogger(AsyncReqRow.class);
-
     private static final long serialVersionUID = 2098635244857937717L;
 
-    protected RowTypeInfo rowTypeInfo;
-
-    protected List<FieldInfo> outFieldInfoList;
-
-    protected List<String> equalFieldList = Lists.newArrayList();
-
-    protected List<Integer> equalValIndex = Lists.newArrayList();
-
-    protected String sqlCondition = "";
-
-    protected String sideSelectFields = "";
-
-    protected JoinType joinType;
+    protected SideInfo sideInfo;
 
-    //key:Returns the value of the position, returns the index values ​​in the input data
-    protected Map<Integer, Integer> inFieldIndex = Maps.newHashMap();
-
-    protected Map<Integer, Integer> sideFieldIndex = Maps.newHashMap();
-
-    protected SideTableInfo sideTableInfo;
-
-    protected AbsSideCache sideCache;
-
-    public AsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList,
-                       SideTableInfo sideTableInfo){
-        this.rowTypeInfo = rowTypeInfo;
-        this.outFieldInfoList = outFieldInfoList;
-        this.joinType = joinInfo.getJoinType();
-        this.sideTableInfo = sideTableInfo;
-        parseSelectFields(joinInfo);
-        buildEqualInfo(joinInfo, sideTableInfo);
+    public AsyncReqRow(SideInfo sideInfo){
+        this.sideInfo = sideInfo;
     }
 
     private void initCache(){
+        SideTableInfo sideTableInfo = sideInfo.getSideTableInfo();
         if(sideTableInfo.getCacheType() == null || ECacheType.NONE.name().equalsIgnoreCase(sideTableInfo.getCacheType())){
             return;
         }
 
+        AbsSideCache sideCache;
         if(ECacheType.LRU.name().equalsIgnoreCase(sideTableInfo.getCacheType())){
             sideCache = new LRUSideCache(sideTableInfo);
+            sideInfo.setSideCache(sideCache);
         }else{
             throw new RuntimeException("not support side cache with type:" + sideTableInfo.getCacheType());
         }
@@ -105,101 +68,22 @@ private void initCache(){
     }
 
     protected CacheObj getFromCache(String key){
-        return sideCache.getFromCache(key);
+        return sideInfo.getSideCache().getFromCache(key);
     }
 
     protected void putCache(String key, CacheObj value){
-        sideCache.putCache(key, value);
+        sideInfo.getSideCache().putCache(key, value);
     }
 
     protected boolean openCache(){
-        return sideCache != null;
+        return sideInfo.getSideCache() != null;
     }
 
-    public void parseSelectFields(JoinInfo joinInfo){
-        String sideTableName = joinInfo.getSideTableName();
-        String nonSideTableName = joinInfo.getNonSideTable();
-        List<String> fields = Lists.newArrayList();
-
-        int sideIndex = 0;
-        for( int i=0; i<outFieldInfoList.size(); i++){
-            FieldInfo fieldInfo = outFieldInfoList.get(i);
-            if(fieldInfo.getTable().equalsIgnoreCase(sideTableName)){
-                fields.add(fieldInfo.getFieldName());
-                sideFieldIndex.put(i, sideIndex);
-                sideIndex++;
-            }else if(fieldInfo.getTable().equalsIgnoreCase(nonSideTableName)){
-                int nonSideIndex = rowTypeInfo.getFieldIndex(fieldInfo.getFieldName());
-                inFieldIndex.put(i, nonSideIndex);
-            }else{
-                throw new RuntimeException("unknown table " + fieldInfo.getTable());
-            }
-        }
-
-        if(fields.size() == 0){
-            throw new RuntimeException("select non field from table " +  sideTableName);
-        }
-
-        sideSelectFields = String.join(",", fields);
-    }
-
-    public abstract void buildEqualInfo(JoinInfo joinInfo, SideTableInfo sideTableInfo);
-
-    public void dealOneEqualCon(SqlNode sqlNode, String sideTableName){
-        if(sqlNode.getKind() != SqlKind.EQUALS){
-            throw new RuntimeException("not equal operator.");
-        }
-
-        SqlIdentifier left = (SqlIdentifier)((SqlBasicCall)sqlNode).getOperands()[0];
-        SqlIdentifier right = (SqlIdentifier)((SqlBasicCall)sqlNode).getOperands()[1];
-
-        String leftTableName = left.getComponent(0).getSimple();
-        String leftField = left.getComponent(1).getSimple();
-
-        String rightTableName = right.getComponent(0).getSimple();
-        String rightField = right.getComponent(1).getSimple();
-
-        if(leftTableName.equalsIgnoreCase(sideTableName)){
-            equalFieldList.add(leftField);
-            int equalFieldIndex = -1;
-            for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
-                String fieldName = rowTypeInfo.getFieldNames()[i];
-                if(fieldName.equalsIgnoreCase(rightField)){
-                    equalFieldIndex = i;
-                }
-            }
-            if(equalFieldIndex == -1){
-                throw new RuntimeException("can't find equal field " + rightField);
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        }else if(rightTableName.equalsIgnoreCase(sideTableName)){
-
-            equalFieldList.add(rightField);
-            int equalFieldIndex = -1;
-            for(int i=0; i<rowTypeInfo.getFieldNames().length; i++){
-                String fieldName = rowTypeInfo.getFieldNames()[i];
-                if(fieldName.equalsIgnoreCase(leftField)){
-                    equalFieldIndex = i;
-                }
-            }
-            if(equalFieldIndex == -1){
-                throw new RuntimeException("can't find equal field " + rightField);
-            }
-
-            equalValIndex.add(equalFieldIndex);
-
-        }else{
-            throw new RuntimeException("resolve equalFieldList error:" + sqlNode.toString());
-        }
-
-    }
 
     protected abstract Row fillData(Row input, Object sideInput);
 
     protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
-        if(joinType == JoinType.LEFT){
+        if(sideInfo.getJoinType() == JoinType.LEFT){
             //Reserved left table data
             Row row = fillData(input, null);
             resultFuture.complete(Collections.singleton(row));