Merge branch 'feat_update_sql' into feat_upsertMode_mergedDev

# Conflicts:
#	cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java
#	console/console-sink/src/main/java/com/dtstack/flink/sql/sink/console/ConsoleOutputFormat.java
#	core/src/main/java/com/dtstack/flink/sql/outputformat/DtRichOutputFormat.java
#	hbase/hbase-sink/src/main/java/com/dtstack/flink/sql/sink/hbase/HbaseOutputFormat.java
#	kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java
#	mongo/mongo-sink/src/main/java/com/dtstack/flink/sql/sink/mongo/MongoOutputFormat.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java
#	redis5/redis5-sink/src/main/java/com/dtstack/flink/sql/sink/redis/RedisOutputFormat.java

Author: todd5167

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,8 +21,9 @@
 
 
 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;
@@ -33,46 +34,25 @@ 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();
     }
 
 

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>{
+public abstract class DtRichOutputFormat<T> extends RichOutputFormat<T>{
 
     protected transient Counter outRecords;
-
     protected transient Counter outDirtyRecords;
-
     protected 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/SideSqlExec.java

@@ -60,6 +60,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import java.sql.Timestamp;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;

db2/db2-sink/src/main/java/com/dtstack/flink/sql/sink/db/DbDialect.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.db;
+
+import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+
+import java.util.Optional;
+
+/**
+ * Date: 2020/1/19
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class DbDialect implements JDBCDialect {
+    @Override
+    public boolean canHandle(String url) {
+        return url.startsWith("jdbc:db2:");
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("com.ibm.db2.jcc.DB2Driver");
+    }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return identifier;
+    }
+
+}

db2/db2-sink/src/main/java/com/dtstack/flink/sql/sink/db/DbSink.java

@@ -1,7 +1,8 @@
 package com.dtstack.flink.sql.sink.db;
 
+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;
@@ -11,42 +12,23 @@ public class DbSink extends RdbSink {
     private static final String DB2_DRIVER = "com.ibm.db2.jcc.DB2Driver";
 
     public DbSink() {
+        super(new DbDialect());
     }
-
-    @Override
-    public void buildSql(String schema, String tableName, List<String> fields) {
-        buildInsertSql(tableName, fields);
-    }
-
-    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;
-    }
-
-    @Override
-    public String buildUpdateSql(String schema, String tableName, List<String> fieldNames, Map<String, List<String>> realIndexes, List<String> fullField) {
-        return null;
-    }
-
-    @Override
-    public String getDriverName() {
-        return DB2_DRIVER;
-    }
-
     @Override
-    public RetractJDBCOutputFormat getOutputFormat() {
-        return new RetractJDBCOutputFormat();
+    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();
     }
 }

docs/postgresqlSink.md

@@ -16,7 +16,7 @@ CREATE TABLE tableName(
 ```
 
 ## 2.支持版本
- postgresql-8.2+
+ postgresql-9.5+
 
 ## 3.表结构定义
 
@@ -36,8 +36,6 @@ CREATE TABLE tableName(
 | password | postgresql连接密码|是||
 | tableName | postgresqll表名称|是||
 | parallelism | 并行度设置|否|1|
-| isUpsert | 使用upsert模式插入数据(版本9.5之后才支持upsert) |否|false
-| keyField | 设置更新主键字段名（isupsert为true时为必填项）|否|
 
 ## 5.样例：
 ```