rdb sink add update mode

Author: todd5167

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

@@ -22,7 +22,6 @@
 
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
-import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 
 import java.util.List;
 import java.util.Map;

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/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;
@@ -626,7 +627,7 @@ private boolean checkJoinCondition(SqlNode conditionNode, String sideTableAlias,
 
     private List<String> convertPrimaryAlias(SideTableInfo sideTableInfo) {
         List<String> res = Lists.newArrayList();
-        sideTableInfo.getPrimaryKeys().forEach(field -> {
+        Arrays.stream(sideTableInfo.getPrimaryKeys()).forEach(field -> {
             res.add(sideTableInfo.getPhysicalFields().getOrDefault(field, field));
         });
         return res;

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

@@ -24,22 +24,21 @@
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.MeterView;
+import org.apache.flink.types.Row;
 
 /**
  * Created by sishu.yss on 2018/11/28.
  */
-public abstract  class MetricOutputFormat extends  RichOutputFormat<Tuple2>{
+public abstract class MetricOutputFormat extends RichOutputFormat<Tuple2<Boolean, Row>> {
 
-     protected  transient Counter outRecords;
+    public transient Counter outRecords;
+    public transient Counter outDirtyRecords;
+    public transient Meter outRecordsRate;
 
-     protected  transient Counter outDirtyRecords;
-
-     protected transient Meter outRecordsRate;
-
-     public void initMetric() {
+    public void initMetric() {
         outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
         outDirtyRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_DIRTY_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/AbsTableParser.java

@@ -115,8 +115,7 @@ public void parseFieldsInfo(String fieldsInfo, TableInfo tableInfo){
     public void dealPrimaryKey(Matcher matcher, TableInfo tableInfo){
         String primaryFields = matcher.group(1).trim();
         String[] splitArry = primaryFields.split(",");
-        List<String> primaryKes = Lists.newArrayList(splitArry);
-        tableInfo.setPrimaryKeys(primaryKes);
+        tableInfo.setPrimaryKeys(splitArry);
     }
 
     /**

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

@@ -59,7 +59,7 @@ public abstract class TableInfo implements Serializable {
 
     private final List<FieldExtraInfo> fieldExtraInfoList = Lists.newArrayList();
 
-    private List<String> primaryKeys;
+    private String[] primaryKeys;
 
     private Integer parallelism = -1;
 
@@ -85,11 +85,11 @@ public Class<?>[] getFieldClasses() {
         return fieldClasses;
     }
 
-    public List<String> getPrimaryKeys() {
+    public String[] getPrimaryKeys() {
         return primaryKeys;
     }
 
-    public void setPrimaryKeys(List<String> primaryKeys) {
+    public void setPrimaryKeys(String[] primaryKeys) {
         this.primaryKeys = primaryKeys;
     }
 

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

@@ -1,7 +1,6 @@
 package com.dtstack.flink.sql.sink.db;
 
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
-import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 
 import java.util.List;
 import java.util.Map;

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaOutputFormat.java

@@ -20,7 +20,6 @@
 package com.dtstack.flink.sql.sink.impala;
 
 import com.dtstack.flink.sql.sink.impala.table.ImpalaTableInfo;
-import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 import com.dtstack.flink.sql.util.JDBCUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;

impala/impala-sink/src/main/java/com/dtstack/flink/sql/sink/impala/ImpalaSink.java

@@ -21,7 +21,6 @@
 import com.dtstack.flink.sql.sink.IStreamSinkGener;
 import com.dtstack.flink.sql.sink.impala.table.ImpalaTableInfo;
 import com.dtstack.flink.sql.sink.rdb.RdbSink;
-import com.dtstack.flink.sql.sink.rdb.format.RetractJDBCOutputFormat;
 import com.dtstack.flink.sql.table.TargetTableInfo;
 import org.apache.commons.lang.StringUtils;
 import org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction;

mysql/mysql-sink/src/main/java/com/dtstack/flink/sql/sink/mysql/MySQLDialect.java

@@ -0,0 +1,81 @@
+/*
+ * 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.mysql;
+
+import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Date: 2019/12/31
+ * Company: www.dtstack.com
+ * @author maqi
+ */
+public class MySQLDialect implements JDBCDialect {
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public boolean canHandle(String url) {
+        return url.startsWith("jdbc:mysql:");
+    }
+
+    @Override
+    public Optional<String> defaultDriverName() {
+        return Optional.of("com.mysql.jdbc.Driver");
+    }
+
+    @Override
+    public String quoteIdentifier(String identifier) {
+        return "`" + identifier + "`";
+    }
+
+    /**
+     *  根据ALLReplace参数，选择使用replace语句还是ON DUPLICATE KEY UPDATE 语句
+     * @param tableName
+     * @param fieldNames
+     * @param uniqueKeyFields
+     * @param allReplace
+     * @return
+     */
+    @Override
+    public Optional<String> getUpsertStatement(String tableName, String[] fieldNames, String[] uniqueKeyFields, boolean allReplace) {
+        return allReplace ? buildReplaceIntoStatement(tableName, fieldNames) : buildDuplicateUpsertStatement(tableName, fieldNames);
+    }
+
+    public Optional<String> buildDuplicateUpsertStatement(String tableName, String[] fieldNames) {
+        String updateClause = Arrays.stream(fieldNames).map(f -> quoteIdentifier(f) + "=IFNULL(VALUES(" + quoteIdentifier(f) + ")," + quoteIdentifier(f) + ")")
+                .collect(Collectors.joining(", "));
+        return Optional.of(getInsertIntoStatement(tableName, fieldNames) +
+                " ON DUPLICATE KEY UPDATE " + updateClause
+        );
+    }
+
+    public Optional<String> buildReplaceIntoStatement(String tableName, String[] fieldNames) {
+        String columns = Arrays.stream(fieldNames)
+                .map(this::quoteIdentifier)
+                .collect(Collectors.joining(", "));
+        String placeholders = Arrays.stream(fieldNames)
+                .map(f -> "?")
+                .collect(Collectors.joining(", "));
+        return Optional.of("REPLACE INTO " + quoteIdentifier(tableName) +
+                "(" + columns + ")" + " VALUES (" + placeholders + ")");
+    }
+}