Merge branch 'v1.8.0_dev' of ssh://git.dtstack.cn:10022/dt-insight-engine/flinkStreamSQL into feat_elasticsearch6-side

Author: dengdejiushini

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -69,7 +69,7 @@
  * @see Tuple
  * @see DriverManager
  */
-public class CassandraOutputFormat extends DtRichOutputFormat {
+public class CassandraOutputFormat extends DtRichOutputFormat<Tuple2> {
     private static final long serialVersionUID = -7994311331389155692L;
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraOutputFormat.class);

clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseDialect.java

@@ -0,0 +1,46 @@
+/*
+ * 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.clickhouse;
+
+import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+
+import java.util.Optional;
+
+/**
+ * Date: 2020/1/15
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class ClickhouseDialect implements JDBCDialect {
+
+    @Override
+    public boolean canHandle(String url) {
+        return url.startsWith("jdbc:clickhouse:");
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("ru.yandex.clickhouse.ClickHouseDriver");
+    }
+
+    @Override
+    public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
+        throw new RuntimeException("Clickhouse does not support update sql, please remove primary key or use append mode");
+    }
+}

clickhouse/clickhouse-sink/src/main/java/com/dtstack/flink/sql/sink/clickhouse/ClickhouseSink.java

@@ -21,58 +21,39 @@
 
 
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
+import com.dtstack.flink.sql.sink.rdb.JDBCOptions;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
-import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
+import com.dtstack.flink.sql.sink.rdb.format.JDBCUpsertOutputFormat;
 
 import java.util.List;
 import java.util.Map;
 
 
 public class ClickhouseSink extends RdbSink implements IStreamSinkGener<RdbSink> {
-
-    private static final String CLICKHOUSE_DRIVER = "ru.yandex.clickhouse.ClickHouseDriver";
-
     public ClickhouseSink() {
+        super(new ClickhouseDialect());
     }
 
     @Override
-    public RetractJDBCOutputFormat getOutputFormat() {
-        return new RetractJDBCOutputFormat();
-    }
-
-    @Override
-    public void buildSql(String scheam, String tableName, List<String> fields) {
-        buildInsertSql(tableName, fields);
-    }
-
-    @Override
-    public String buildUpdateSql(String schema, String tableName, List<String> fieldNames, Map<String, List<String>> realIndexes, List<String> fullField) {
-        return null;
-    }
-
-    private void buildInsertSql(String tableName, List<String> fields) {
-        String sqlTmp = "insert into " +  tableName + " (${fields}) values (${placeholder})";
-        String fieldsStr = "";
-        String placeholder = "";
-
-        for (String fieldName : fields) {
-            fieldsStr += ",`" + fieldName + "`";
-            placeholder += ",?";
-        }
-
-        fieldsStr = fieldsStr.replaceFirst(",", "");
-        placeholder = placeholder.replaceFirst(",", "");
-
-        sqlTmp = sqlTmp.replace("${fields}", fieldsStr).replace("${placeholder}", placeholder);
-        this.sql = sqlTmp;
-        System.out.println("---insert sql----");
-        System.out.println(sql);
-    }
-
-
-    @Override
-    public String getDriverName() {
-        return CLICKHOUSE_DRIVER;
+    public JDBCUpsertOutputFormat getOutputFormat() {
+        JDBCOptions jdbcOptions = JDBCOptions.builder()
+                .setDBUrl(dbURL)
+                .setDialect(jdbcDialect)
+                .setUsername(userName)
+                .setPassword(password)
+                .setTableName(tableName)
+                .build();
+
+        return JDBCUpsertOutputFormat.builder()
+                .setOptions(jdbcOptions)
+                .setFieldNames(fieldNames)
+                .setFlushMaxSize(batchNum)
+                .setFlushIntervalMills(batchWaitInterval)
+                .setFieldTypes(sqlTypes)
+                .setKeyFields(primaryKeys)
+                .setAllReplace(allReplace)
+                .setUpdateMode(updateMode)
+                .build();
     }
 
 

console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java

@@ -37,7 +37,7 @@
  *
  * @author xuqianjin
  */
-public class ConsoleOutputFormat extends DtRichOutputFormat {
+public class ConsoleOutputFormat extends DtRichOutputFormat<Tuple2> {
 
     private static final Logger LOG = LoggerFactory.getLogger(ConsoleOutputFormat.class);
 

core/pom.xml

@@ -116,6 +116,12 @@
             <version>${flink.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+        </dependency>
+
     </dependencies>
 
     <build>

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

@@ -181,8 +181,12 @@ private static void sqlTranslation(String localSqlPluginPath, StreamTableEnviron
                         //sql-dimensional table contains the dimension table of execution
                         sideSqlExec.exec(result.getExecSql(), sideTableMap, tableEnv, registerTableCache, queryConfig);
                     }else{
+                        System.out.println("----------exec sql without dimension join-----------" );
+                        System.out.println("----------real sql exec is--------------------------");
+                        System.out.println(result.getExecSql());
                         FlinkSQLExec.sqlUpdate(tableEnv, result.getExecSql(), queryConfig);
                         if(LOG.isInfoEnabled()){
+                            System.out.println();
                             LOG.info("exec sql: " + result.getExecSql());
                         }
                     }

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

@@ -0,0 +1,44 @@
+/*
+ * 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;
+
+/**
+ * restract stream数据处理模式
+ * 
+ * Reason:
+ * Date: 2019/1/2
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public enum EUpdateMode {
+    // 不回撤数据，只下发增量数据
+    APPEND(0),
+    // 先删除回撤数据，然后更新
+    UPSERT(1);
+
+    private int type;
+
+    EUpdateMode(int type) {
+        this.type = type;
+    }
+
+    public int getType() {
+        return this.type;
+    }
+}

core/src/main/java/com/dtstack/flink/sql/outputformat/DtRichOutputFormat.java

@@ -18,8 +18,6 @@
 package com.dtstack.flink.sql.outputformat;
 
 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;
@@ -29,16 +27,13 @@
  * extend RichOutputFormat with metric 'dtNumRecordsOut', 'dtNumDirtyRecordsOut', 'dtNumRecordsOutRate'
  * Created by sishu.yss on 2018/11/28.
  */
-public abstract class DtRichOutputFormat extends RichOutputFormat<Tuple2>{
-
-    protected transient Counter outRecords;
+public abstract class DtRichOutputFormat<T> extends RichOutputFormat<T>{
 
-    protected transient Counter outDirtyRecords;
-
-    protected transient Meter outRecordsRate;
+    public transient Counter outRecords;
+    public transient Counter outDirtyRecords;
+    public transient Meter outRecordsRate;
 
     protected static int ROW_PRINT_FREQUENCY = 1000;
-
     protected static int DIRTY_PRINT_FREQUENCY = 1000;
 
     public void initMetric() {

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

@@ -21,9 +21,10 @@
 package com.dtstack.flink.sql.side;
 
 import com.google.common.collect.HashBasedTable;
+import org.apache.commons.lang3.StringUtils;
 
 /**
- * Reason:
+ * 用于记录转换之后的表和原来表直接字段的关联关系
  * Date: 2018/8/30
  * Company: www.dtstack.com
  * @author xuchao
@@ -37,6 +38,8 @@ public class FieldReplaceInfo {
 
     private String targetTableAlias = null;
 
+    private FieldReplaceInfo preNode = null;
+
     public void setMappingTable(HashBasedTable<String, String, String> mappingTable) {
         this.mappingTable = mappingTable;
     }
@@ -57,7 +60,39 @@ public String getTargetTableAlias() {
         return targetTableAlias;
     }
 
+    public FieldReplaceInfo getPreNode() {
+        return preNode;
+    }
+
+    public void setPreNode(FieldReplaceInfo preNode) {
+        this.preNode = preNode;
+    }
+
     public void setTargetTableAlias(String targetTableAlias) {
         this.targetTableAlias = targetTableAlias;
     }
+
+    /**
+     * 根据原始的tableName + fieldName 获取转换之后的fieldName
+     * @param tableName
+     * @param fieldName
+     * @return
+     */
+    public String getTargetFieldName(String tableName, String fieldName){
+        String targetFieldName = mappingTable.get(tableName, fieldName);
+        if(StringUtils.isNotBlank(targetFieldName)){
+            return targetFieldName;
+        }
+
+        if(preNode == null){
+            return null;
+        }
+
+        String preNodeTargetFieldName = preNode.getTargetFieldName(tableName, fieldName);
+        if(StringUtils.isBlank(preNodeTargetFieldName)){
+            return null;
+        }
+
+        return mappingTable.get(preNode.getTargetTableAlias(), preNodeTargetFieldName);
+    }
 }

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

@@ -41,8 +41,8 @@ public class JoinInfo implements Serializable {
 
     //左表是否是维表
     private boolean leftIsSideTable;
-    //左表是 转换后的中间表
-    private boolean leftIsMidTable;
+
+    private boolean leftIsTmpTable = false;
 
     //右表是否是维表
     private boolean rightIsSideTable;
@@ -66,8 +66,6 @@ public class JoinInfo implements Serializable {
     private SqlNode selectNode;
 
     private JoinType joinType;
-    // 左边是中间转换表，做表映射关系，给替换属性名称使用
-    private Map<String, String> leftTabMapping;
 
     public String getSideTableName(){
         if(leftIsSideTable){
@@ -92,21 +90,6 @@ public String getNewTableName(){
         return leftStr + "_" + rightTableName;
     }
 
-    public boolean isLeftIsMidTable() {
-        return leftIsMidTable;
-    }
-
-    public void setLeftIsMidTable(boolean leftIsMidTable) {
-        this.leftIsMidTable = leftIsMidTable;
-    }
-
-    public Map<String, String> getLeftTabMapping() {
-        return leftTabMapping;
-    }
-
-    public void setLeftTabMapping(Map<String, String> leftTabMapping) {
-        this.leftTabMapping = leftTabMapping;
-    }
 
     public String getNewTableAlias(){
         return leftTableAlias + "_" + rightTableAlias;
@@ -211,4 +194,12 @@ public JoinType getJoinType() {
     public void setJoinType(JoinType joinType) {
         this.joinType = joinType;
     }
+
+    public boolean isLeftIsTmpTable() {
+        return leftIsTmpTable;
+    }
+
+    public void setLeftIsTmpTable(boolean leftIsTmpTable) {
+        this.leftIsTmpTable = leftIsTmpTable;
+    }
 }