Merge remote-tracking branch 'origin/v1.8.0_dev' into feat_1.8_asyncE…

…xception_mergeDev # Conflicts: # cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java # core/src/main/java/com/dtstack/flink/sql/side/AbstractSideTableInfo.java # core/src/main/java/com/dtstack/flink/sql/side/BaseAsyncReqRow.java # core/src/main/java/com/dtstack/flink/sql/table/AbstractSideTableParser.java # rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java # redis5/redis5-side/redis-async-side/src/main/java/com/dtstack/flink/sql/side/redis/RedisAsyncReqRow.java
DTStack · Apr 8, 2020 · f05e92b · f05e92b
2 parents caa8824 + fe59047
commit f05e92b
Show file tree

Hide file tree

Showing 119 changed files with 3,027 additions and 1,827 deletions.
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
@@ -0,0 +1,10 @@
+build:
+  stage: test
+  script:
+    - mvn clean org.jacoco:jacoco-maven-plugin:0.7.8:prepare-agent package -Dmaven.test.failure.ignore=true -q
+    - mvn sonar:sonar -Dsonar.projectKey="dt-insight-engine/flinkStreamSQL"  -Dsonar.login=11974c5e9a29625efa09fdc3c3fdc031efb1aab1 -Dsonar.host.url=http://172.16.100.198:9000 -Dsonar.jdbc.url=jdbc:postgresql://172.16.100.198:5432/sonar -Dsonar.java.binaries=target/sonar
+    - sh ci/sonar_notify.sh
+  only:
+    - v1.8.0_dev
+  tags:
+    - dt-insight-engine
diff --git a/README.md b/README.md
@@ -7,34 +7,11 @@
 >  >  * 支持原生FLinkSQL所有的语法
 >  >  * 扩展了输入和输出的性能指标到promethus
 
- ## 新特性:
-  * 1.kafka源表支持not null语法,支持字符串类型的时间转换。
-  * 2.rdb维表与DB建立连接时，周期进行连接，防止连接断开。rdbsink写入时，对连接进行检查。
-  * 3.异步维表支持非等值连接，比如：<>,<,>。
-  * 4.增加kafka数组解析
-  * 5.增加kafka1.0以上版本的支持
-  * 6.增加postgresql、kudu、clickhouse维表、结果表的支持
-  * 7.支持插件的依赖方式,参考pluginLoadMode参数
-  * 8.支持cep处理
-  * 9.支持udaf
-  * 10.支持谓词下移
-  * 11.支持状态的ttl
-
- ## BUG修复：
-  * 1.修复不能解析sql中orderby,union语法。
-  * 2.修复yarnPer模式提交失败的异常。
-  * 3.一些bug的修复
-
 # 已支持
   * 源表：kafka 0.9、0.10、0.11、1.x版本
   * 维表：mysql, SQlServer,oracle, hbase, mongo, redis, cassandra, serversocket, kudu, postgresql, clickhouse, impala, db2, sqlserver
   * 结果表：mysql, SQlServer, oracle, hbase, elasticsearch5.x, mongo, redis, cassandra, console, kudu, postgresql, clickhouse, impala, db2, sqlserver
 
-# 后续开发计划
-  * 维表快照
-  * kafka avro格式
-  * topN
-
 ## 1 快速起步
 ### 1.1 运行模式
 
@@ -205,6 +182,7 @@ sh submit.sh -sql D:\sideSql.txt  -name xctest -remoteSqlPluginPath /opt/dtstack
 * [impala 结果表插件](docs/impalaSink.md)
 * [db2 结果表插件](docs/db2Sink.md)
 * [sqlserver 结果表插件](docs/sqlserverSink.md)
+* [kafka 结果表插件](docs/kafkaSink.md)
 
 ### 2.3 维表插件
 * [hbase 维表插件](docs/hbaseSide.md)

diff --git a/...andra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java b/...andra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java
@@ -267,7 +267,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
                         String connInfo = "address:" + tableInfo.getAddress() + ";userName:" + tableInfo.getUserName()
                                 + ",pwd:" + tableInfo.getPassword();
                         LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
-                        Thread.sleep(5 * 1000);
+                        Thread.sleep(LOAD_DATA_ERROR_SLEEP_TIME);
                     } catch (InterruptedException e1) {
                         LOG.error("", e1);
                     }

diff --git a/...a-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java b/...a-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java
@@ -169,6 +169,7 @@ public void handleAsyncInvoke(Map<String, Object> inputParams, CRow input, Resul
         connCassandraDB(cassandraSideTableInfo);
 
         String sqlCondition = sideInfo.getSqlCondition() + " " + buildWhereCondition(inputParams) + "  ALLOW FILTERING ";
+        LOG.info("sqlCondition:{}" + sqlCondition);
 
         ListenableFuture<ResultSet> resultSet = Futures.transformAsync(session,
                 new AsyncFunction<Session, ResultSet>() {

diff --git a/...async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncSideInfo.java b/...async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncSideInfo.java
@@ -30,6 +30,8 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 
@@ -42,6 +44,8 @@
 public class CassandraAsyncSideInfo extends BaseSideInfo {
 
     private static final long serialVersionUID = -4403313049809013362L;
+    private static final Logger LOG = LoggerFactory.getLogger(CassandraAsyncSideInfo.class.getSimpleName());
+
 
     public CassandraAsyncSideInfo(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, AbstractSideTableInfo sideTableInfo) {
         super(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo);
@@ -63,9 +67,9 @@ public void buildEqualInfo(JoinInfo joinInfo, AbstractSideTableInfo sideTableInf
         }
 
         sqlCondition = "select ${selectField} from ${tableName}";
-
         sqlCondition = sqlCondition.replace("${tableName}", cassandraSideTableInfo.getDatabase()+"."+cassandraSideTableInfo.getTableName()).replace("${selectField}", sideSelectFields);
-        System.out.println("---------side_exe_sql-----\n" + sqlCondition);
+
+        LOG.info("---------side_exe_sql-----\n{}" + sqlCondition);
     }
 
 

diff --git a/ci/sonar_notify.sh b/ci/sonar_notify.sh
@@ -0,0 +1,14 @@
+#!/bin/bash
+#参考钉钉文档 https://open-doc.dingtalk.com/microapp/serverapi2/qf2nxq
+ sonarreport=$(curl -s http://172.16.100.198:8082/?projectname=dt-insight-engine/flinkStreamSQL)
+ curl -s "https://oapi.dingtalk.com/robot/send?access_token=71555061297a53d3ac922a6f4d94285d8e23bccdca0c00b4dc6df0a2d49da724" \
+   -H "Content-Type: application/json" \
+   -d "{
+     \"msgtype\": \"markdown\",
+     \"markdown\": {
+         \"title\":\"sonar代码质量\",
+         \"text\": \"## sonar代码质量报告: \n
+> [sonar地址](http://172.16.100.198:9000/dashboard?id=dt-insight-engine/flinkStreamSQL) \n
+> ${sonarreport} \n\"
+     }
+ }"
diff --git a/...async-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAsyncReqRow.java b/...async-side/src/main/java/com/dtstack/flink/sql/side/clickhouse/ClickhouseAsyncReqRow.java
@@ -48,14 +48,18 @@ public void open(Configuration parameters) throws Exception {
         RdbSideTableInfo rdbSideTableInfo = (RdbSideTableInfo) sideInfo.getSideTableInfo();
         clickhouseClientConfig.put("url", rdbSideTableInfo.getUrl())
                 .put("driver_class", CLICKHOUSE_DRIVER)
-                .put("max_pool_size", DEFAULT_MAX_DB_CONN_POOL_SIZE)
+                .put("max_pool_size", rdbSideTableInfo.getAsyncPoolSize())
                 .put("user", rdbSideTableInfo.getUserName())
                 .put("password", rdbSideTableInfo.getPassword())
-                .put("provider_class", DT_PROVIDER_CLASS);
+                .put("provider_class", DT_PROVIDER_CLASS)
+                .put("preferred_test_query", PREFERRED_TEST_QUERY_SQL)
+                .put("idle_connection_test_period", DEFAULT_IDLE_CONNECTION_TEST_PEROID)
+                .put("test_connection_on_checkin", DEFAULT_TEST_CONNECTION_ON_CHECKIN);
+
         System.setProperty("vertx.disableFileCPResolving", "true");
         VertxOptions vo = new VertxOptions();
-        vo.setEventLoopPoolSize(DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE);
-        vo.setWorkerPoolSize(DEFAULT_VERTX_WORKER_POOL_SIZE);
+        vo.setEventLoopPoolSize(rdbSideTableInfo.getAsyncPoolSize());
+        vo.setWorkerPoolSize(rdbSideTableInfo.getAsyncPoolSize());
         vo.setFileResolverCachingEnabled(false);
         Vertx vertx = Vertx.vertx(vo);
         setRdbSqlClient(JDBCClient.createNonShared(vertx, clickhouseClientConfig));

diff --git a/...le/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java b/...le/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java
@@ -56,7 +56,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
 
     @Override
     public void writeRecord(Tuple2 tuple2) throws IOException {
-        System.out.println("received oriainal data:" + tuple2);
+        LOG.info("received oriainal data:{}" + tuple2);
         Tuple2<Boolean, Row> tupleTrans = tuple2;
         Boolean retract = tupleTrans.getField(0);
         if (!retract) {

diff --git a/...e/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/table/TablePrintUtil.java b/...e/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/table/TablePrintUtil.java
@@ -95,7 +95,7 @@ public static <T> TablePrintUtil build(List<T> data) {
                     try {
                         value = obj.getClass().getMethod(colList.get(j).getMethodName).invoke(data.get(i)).toString();
                     } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
-                        e.printStackTrace();
+                        LOG.error("", e);
                     }
                     item[j] = value == null ? "null" : value;
                 }

diff --git a/core/src/main/java/com/dtstack/flink/sql/environment/MyLocalStreamEnvironment.java b/core/src/main/java/com/dtstack/flink/sql/environment/MyLocalStreamEnvironment.java
@@ -113,14 +113,13 @@ public JobExecutionResult execute(String jobName) throws Exception {
             LOG.info("Running job on local embedded Flink mini cluster");
         }
 
-        MiniCluster exec = new MiniCluster(configBuilder.build());
-        try {
+        try (MiniCluster exec = new MiniCluster(configBuilder.build());) {
             exec.start();
-            return exec.executeJobBlocking(jobGraph);
-        }
-        finally {
+            JobExecutionResult jobExecutionResult = exec.executeJobBlocking(jobGraph);
             transformations.clear();
-            exec.closeAsync();
+            return jobExecutionResult;
+        } catch (Exception e) {
+            throw new RuntimeException(e);
         }
     }
 }
diff --git a/core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java b/core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java
@@ -95,11 +95,8 @@ public class ExecuteProcessHelper {
 
     public static ParamsInfo parseParams(String[] args) throws Exception {
         LOG.info("------------program params-------------------------");
-        System.out.println("------------program params-------------------------");
         Arrays.stream(args).forEach(arg -> LOG.info("{}", arg));
-        Arrays.stream(args).forEach(System.out::println);
         LOG.info("-------------------------------------------");
-        System.out.println("----------------------------------------");
 
         OptionParser optionParser = new OptionParser(args);
         Options options = optionParser.getOptions();
@@ -206,7 +203,6 @@ private static void sqlTranslation(String localSqlPluginPath,
             if (LOG.isInfoEnabled()) {
                 LOG.info("exe-sql:\n" + result.getExecSql());
             }
-
             boolean isSide = false;
             for (String tableName : result.getTargetTableList()) {
                 if (sqlTree.getTmpTableMap().containsKey(tableName)) {
@@ -228,12 +224,10 @@ private static void sqlTranslation(String localSqlPluginPath,
                         //sql-dimensional table contains the dimension table of execution
                         sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig, null);
                     } else {
-                        System.out.println("----------exec sql without dimension join-----------");
-                        System.out.println("----------real sql exec is--------------------------");
-                        System.out.println(result.getExecSql());
+                        LOG.info("----------exec sql without dimension join-----------");
+                        LOG.info("----------real sql exec is--------------------------\n{}", result.getExecSql());
                         FlinkSQLExec.sqlUpdate(tableEnv, result.getExecSql(), queryConfig);
                         if (LOG.isInfoEnabled()) {
-                            System.out.println();
                             LOG.info("exec sql: " + result.getExecSql());
                         }
                     }
@@ -355,6 +349,7 @@ public static StreamExecutionEnvironment getStreamExeEnv(Properties confProperti
         return env;
     }
 
+
     public static void setLogLevel(ParamsInfo paramsInfo){
         String logLevel = paramsInfo.getConfProp().getProperty(ConfigConstrant.LOG_LEVEL_KEY);
         if(org.apache.commons.lang3.StringUtils.isBlank(logLevel)){

diff --git a/core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java b/core/src/main/java/com/dtstack/flink/sql/format/SerializationMetricWrapper.java
@@ -24,7 +24,7 @@
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.MeterView;
-import org.apache.flink.types.Row;
+import org.apache.flink.table.runtime.types.CRow;
 
 
 /**
@@ -34,11 +34,11 @@
  * author: toutian
  * create: 2019/12/24
  */
-public class SerializationMetricWrapper implements SerializationSchema<Row> {
+public class SerializationMetricWrapper implements SerializationSchema<CRow> {
 
     private static final long serialVersionUID = 1L;
 
-    private SerializationSchema<Row> serializationSchema;
+    private SerializationSchema<CRow> serializationSchema;
 
     private transient RuntimeContext runtimeContext;
 
@@ -47,7 +47,7 @@ public class SerializationMetricWrapper implements SerializationSchema<Row> {
     protected transient Meter dtNumRecordsOutRate;
 
 
-    public SerializationMetricWrapper(SerializationSchema<Row> serializationSchema) {
+    public SerializationMetricWrapper(SerializationSchema<CRow> serializationSchema) {
         this.serializationSchema = serializationSchema;
     }
 
@@ -57,7 +57,7 @@ public void initMetric() {
     }
 
     @Override
-    public byte[] serialize(Row element) {
+    public byte[] serialize(CRow element) {
         beforeSerialize();
         byte[] row = serializationSchema.serialize(element);
         afterSerialize();
@@ -79,7 +79,7 @@ public void setRuntimeContext(RuntimeContext runtimeContext) {
         this.runtimeContext = runtimeContext;
     }
 
-    public SerializationSchema<Row> getSerializationSchema() {
+    public SerializationSchema<CRow> getSerializationSchema() {
         return serializationSchema;
     }
 

diff --git a/core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java b/core/src/main/java/com/dtstack/flink/sql/format/dtnest/DtNestRowDeserializationSchema.java
@@ -28,8 +28,9 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.*;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.NullNode;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
 import org.apache.flink.types.Row;
 
 import java.io.IOException;

diff --git a/core/src/main/java/com/dtstack/flink/sql/option/OptionParser.java b/core/src/main/java/com/dtstack/flink/sql/option/OptionParser.java
@@ -29,9 +29,10 @@
 import java.util.List;
 import java.util.Map;
 import java.io.File;
-import java.io.FileInputStream;
 import java.net.URLEncoder;
+
 import org.apache.commons.codec.Charsets;
+import org.apache.flink.util.FileUtils;
 
 
 /**
@@ -92,29 +93,21 @@ public Options getOptions(){
     }
 
     public List<String> getProgramExeArgList() throws Exception {
-        Map<String,Object> mapConf = PluginUtil.objectToMap(properties);
+        Map<String, Object> mapConf = PluginUtil.objectToMap(properties);
         List<String> args = Lists.newArrayList();
-        for(Map.Entry<String, Object> one : mapConf.entrySet()){
+        for (Map.Entry<String, Object> one : mapConf.entrySet()) {
             String key = one.getKey();
             Object value = one.getValue();
-            if(value == null){
+            if (value == null) {
                 continue;
-            }else if(OPTION_SQL.equalsIgnoreCase(key)){
+            } else if (OPTION_SQL.equalsIgnoreCase(key)) {
                 File file = new File(value.toString());
-                FileInputStream in = new FileInputStream(file);
-                byte[] filecontent = new byte[(int) file.length()];
-                in.read(filecontent);
-                String content = new String(filecontent, Charsets.UTF_8.name());
+                String content = FileUtils.readFile(file, "UTF-8");
                 value = URLEncoder.encode(content, Charsets.UTF_8.name());
             }
             args.add("-" + key);
             args.add(value.toString());
         }
         return args;
     }
-
-    public static void main(String[] args) throws Exception {
-        OptionParser optionParser = new OptionParser(args);
-        System.out.println(optionParser.getOptions());
-    }
 }
diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/CreateTmpTableParser.java
@@ -22,7 +22,11 @@
 
 import com.dtstack.flink.sql.util.DtStringUtil;
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import com.google.common.collect.Lists;

diff --git a/core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java b/core/src/main/java/com/dtstack/flink/sql/parser/InsertSqlParser.java
@@ -21,7 +21,14 @@
 package com.dtstack.flink.sql.parser;
 
 import org.apache.calcite.config.Lex;
-import org.apache.calcite.sql.*;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlInsert;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlMatchRecognize;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlSelect;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
 import org.apache.commons.lang3.StringUtils;