Merge branch 'v1.5.0_dev' of https://github.com/XuQianJin-Stars/flinkStreamSQL into v1.5.0_dev

Author: XuQianJin-Stars

README.md

@@ -5,6 +5,8 @@
 >  >  * 自定义create function 语法
 >  >  * 实现了流与维表的join
 >  >  * 支持原生FLinkSQL所有的语法
+>  >  * 扩展了输入和输出的性能指标到promethus
+
 
 # 已支持
   * 源表：kafka 0.9，1.x版本
@@ -17,6 +19,7 @@
   * 增加kafka结果表功能
   * 增加SQL支持CEP
   * 维表快照
+  * sql优化（谓词下移等）
 
 ## 1 快速起步
 ### 1.1 运行模式

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

@@ -0,0 +1,33 @@
+/*
+ * 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.enums;
+
+/**
+ * Database type
+ *
+ * Company: www.dtstack.com
+ * @author jiangbo
+ */
+public enum EDatabaseType {
+
+    MYSQL,
+    SQLSERVER,
+    ORACLE,
+
+}

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

@@ -20,6 +20,7 @@
 
 package com.dtstack.flink.sql.parser;
 
+import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.calcite.sql.*;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
@@ -77,7 +78,8 @@ public void parseSql(String sql, SqlTree sqlTree) {
             parseNode(sqlNode, sqlParseResult);
 
             sqlParseResult.setTableName(tableName);
-            sqlParseResult.setExecSql(selectSql.toUpperCase());
+            String transformSelectSql = DtStringUtil.replaceIgnoreQuota(sqlNode.toString(), "`", "");
+            sqlParseResult.setExecSql(transformSelectSql);
             sqlTree.addTmpSql(sqlParseResult);
             sqlTree.addTmplTableInfo(tableName, sqlParseResult);
         } else {

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

@@ -37,7 +37,7 @@
  * @author xuchao
  */
 
-public abstract class AllReqRow extends RichFlatMapFunction<Row, Row>{
+public abstract class AllReqRow extends RichFlatMapFunction<Row, Row> implements ISideReqRow {
 
     protected SideInfo sideInfo;
 
@@ -48,8 +48,6 @@ public AllReqRow(SideInfo sideInfo){
 
     }
 
-    protected abstract Row fillData(Row input, Object sideInput);
-
     protected abstract void initCache() throws SQLException;
 
     protected abstract void reloadCache();

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

@@ -40,7 +40,7 @@
  * @author xuchao
  */
 
-public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> {
+public abstract class AsyncReqRow extends RichAsyncFunction<Row, Row> implements ISideReqRow {
 
     private static final long serialVersionUID = 2098635244857937717L;
 
@@ -79,9 +79,6 @@ protected boolean openCache(){
         return sideInfo.getSideCache() != null;
     }
 
-
-    protected abstract Row fillData(Row input, Object sideInput);
-
     protected void dealMissKey(Row input, ResultFuture<Row> resultFuture){
         if(sideInfo.getJoinType() == JoinType.LEFT){
             //Reserved left table data

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

@@ -0,0 +1,33 @@
+/*
+ * 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 org.apache.flink.types.Row;
+
+/**
+ *
+ * Date: 2018/12/4
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+public interface ISideReqRow {
+
+    Row fillData(Row input, Object sideInput);
+
+}

core/src/main/java/com/dtstack/flink/sql/sink/MetricOutputFormat.java

@@ -0,0 +1,42 @@
+/*
+ * 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.sink;
+
+import com.dtstack.flink.sql.metric.MetricConstant;
+import org.apache.flink.api.java.tuple.Tuple2;
+
+import org.apache.flink.api.common.io.RichOutputFormat;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Meter;
+import org.apache.flink.metrics.MeterView;
+
+/**
+ * Created by sishu.yss on 2018/11/28.
+ */
+public abstract  class MetricOutputFormat extends  RichOutputFormat<Tuple2>{
+
+     protected  transient Counter outRecords;
+
+     protected transient Meter outRecordsRate;
+
+     public void initMetric() {
+        outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
+        outRecordsRate = getRuntimeContext().getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(outRecords, 20));
+     }
+
+}

core/src/main/java/com/dtstack/flink/sql/table/AbsSideTableParser.java

@@ -25,6 +25,8 @@
 import com.dtstack.flink.sql.util.MathUtil;
 
 import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * Reason:
@@ -35,6 +37,19 @@
 
 public abstract class AbsSideTableParser extends AbsTableParser {
 
+    private final static String SIDE_SIGN_KEY = "sideSignKey";
+
+    private final static Pattern SIDE_TABLE_SIGN = Pattern.compile("(?i)^PERIOD\\s+FOR\\s+SYSTEM_TIME$");
+
+    static {
+        keyPatternMap.put(SIDE_SIGN_KEY, SIDE_TABLE_SIGN);
+        keyHandlerMap.put(SIDE_SIGN_KEY, AbsSideTableParser::dealSideSign);
+    }
+
+    private static void dealSideSign(Matcher matcher, TableInfo tableInfo){
+        //FIXME SIDE_TABLE_SIGN current just used as a sign for side table; and do nothing
+    }
+
     //Analytical create table attributes ==> Get information cache
     protected void parseCacheProp(SideTableInfo sideTableInfo, Map<String, Object> props){
         if(props.containsKey(SideTableInfo.CACHE_KEY.toLowerCase())){

core/src/main/java/com/dtstack/flink/sql/table/TableInfo.java

@@ -22,6 +22,7 @@
 
 import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 
+import java.io.Serializable;
 import java.util.List;
 
 /**
@@ -31,7 +32,7 @@
  * @author xuchao
  */
 
-public abstract class TableInfo {
+public abstract class TableInfo implements Serializable {
 
     public static final String PARALLELISM_KEY = "parallelism";
 
@@ -121,6 +122,29 @@ public void addFieldType(String fieldType){
         fieldTypeList.add(fieldType);
     }
 
+    public void setFields(String[] fields) {
+        this.fields = fields;
+    }
+
+    public void setFieldTypes(String[] fieldTypes) {
+        this.fieldTypes = fieldTypes;
+    }
+
+    public void setFieldClasses(Class<?>[] fieldClasses) {
+        this.fieldClasses = fieldClasses;
+    }
+
+    public List<String> getFieldList() {
+        return fieldList;
+    }
+
+    public List<String> getFieldTypeList() {
+        return fieldTypeList;
+    }
+
+    public List<Class> getFieldClassList() {
+        return fieldClassList;
+    }
 
     public void finish(){
         this.fields = fieldList.toArray(new String[fieldList.size()]);

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

@@ -21,8 +21,10 @@
 package com.dtstack.flink.sql.util;
 
 import com.dtstack.flink.sql.enums.ColumnType;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 
 import java.math.BigDecimal;
 import java.util.ArrayList;
@@ -42,6 +44,9 @@ public class DtStringUtil {
 
     private static final Pattern NO_VERSION_PATTERN = Pattern.compile("([a-zA-Z]+).*");
 
+    private static ObjectMapper objectMapper = new ObjectMapper();
+
+
     /**
      * Split the specified string delimiter --- ignored quotes delimiter
      * @param str
@@ -207,4 +212,17 @@ public static String addJdbcParam(String dbUrl, Map<String, String> addParams, b
 
         return preStr + "?" + sb.toString();
     }
+
+    public  static boolean isJosn(String str){
+        boolean flag = false;
+        if(StringUtils.isNotBlank(str)){
+            try {
+                objectMapper.readValue(str,Map.class);
+                flag = true;
+            } catch (Throwable e) {
+                flag=false;
+            }
+        }
+        return flag;
+    }
 }