Merge branch 'hotfix_1.8_3.10.x_27373' into 1.8_test_3.10.x

Author: gituser

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

@@ -20,13 +20,11 @@
 
 import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
 import com.google.common.collect.Lists;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 
-import java.util.Arrays;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -41,8 +39,11 @@ public class ImpalaDialect implements JDBCDialect {
 
     private TypeInformation[] fieldTypes;
 
-    public ImpalaDialect(TypeInformation[] fieldTypes){
+    private List<String> primaryKeys;
+
+    public ImpalaDialect(TypeInformation[] fieldTypes, List<String> primaryKeys){
         this.fieldTypes = fieldTypes;
+        this.primaryKeys = primaryKeys;
     }
 
     @Override
@@ -62,7 +63,17 @@ public String quoteIdentifier(String identifier) {
 
     @Override
     public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
-        throw new RuntimeException("impala does not support update sql, please remove primary key or use append mode");
+        //跳过primary key字段
+        String setClause = Arrays.stream(fieldNames)
+                .filter(f -> CollectionUtils.isNotEmpty(primaryKeys) ? !primaryKeys.contains(f) : true)
+                .map(f -> quoteIdentifier(f) + "=?")
+                .collect(Collectors.joining(", "));
+        String conditionClause = Arrays.stream(conditionFields)
+                .map(f -> quoteIdentifier(f) + "=?")
+                .collect(Collectors.joining(" AND "));
+        return "UPDATE " + quoteIdentifier(tableName) +
+                " SET " + setClause +
+                " WHERE " + conditionClause;
     }
 
     @Override

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

@@ -48,7 +48,7 @@ public ImpalaSink() {
     public JDBCUpsertOutputFormat getOutputFormat() {
         JDBCOptions jdbcOptions = JDBCOptions.builder()
                 .setDbUrl(getImpalaJdbcUrl())
-                .setDialect(new ImpalaDialect(getFieldTypes()))
+                .setDialect(new ImpalaDialect(getFieldTypes(), primaryKeys))
                 .setUsername(userName)
                 .setPassword(password)
                 .setTableName(tableName)

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/JDBCTypeConvertUtils.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.sink.rdb;
 
+import com.google.common.collect.Lists;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -51,6 +52,10 @@ public class JDBCTypeConvertUtils {
 	 * @see PreparedStatement
 	 */
 	public static void setRecordToStatement(PreparedStatement upload, int[] typesArray, Row row) throws SQLException {
+		setRecordToStatement(upload, typesArray, row, null);
+	}
+
+	public static void setRecordToStatement(PreparedStatement upload, int[] typesArray, Row row, int[] pkFieldIndices) throws SQLException {
 		if (typesArray != null && typesArray.length > 0 && typesArray.length != row.getArity()) {
 			LOG.warn("Column SQL types array doesn't match arity of passed Row! Check the passed array...");
 		}
@@ -62,10 +67,34 @@ public static void setRecordToStatement(PreparedStatement upload, int[] typesArr
 			}
 		} else {
 			// types provided
+			int placeIndex = 0;
 			for (int i = 0; i < row.getArity(); i++) {
-				setField(upload, typesArray[i], row.getField(i), i);
+				if(isPrimaryKeyField(pkFieldIndices, i)){
+					continue;
+				}
+				setField(upload, typesArray[i], row.getField(i), placeIndex ++);
+			}
+
+			//填充whereClause中的主键占位符
+			if (pkFieldIndices != null && pkFieldIndices.length > 0) {
+				for (int j = 0; j < pkFieldIndices.length; j++) {
+					int pkIndex = pkFieldIndices[j];
+					setField(upload, typesArray[pkIndex], row.getField(pkIndex), placeIndex + j);
+				}
+			}
+		}
+	}
+
+	private static boolean isPrimaryKeyField(int[] pkFieldIndices, int fieldIndex) {
+		if (pkFieldIndices == null || pkFieldIndices.length <= 0) {
+			return false;
+		}
+		for (int index : pkFieldIndices) {
+			if (index == fieldIndex) {
+				return true;
 			}
 		}
+		return false;
 	}
 
 	public static void setField(PreparedStatement upload, int type, Object field, int index) throws SQLException {

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/AbstractUpsertWriter.java

@@ -268,6 +268,7 @@ private static final class UpsertWriterUsingInsertUpdateStatement extends Abstra
         private final String existSql;
         private final String insertSql;
         private final String updateSql;
+        private final int[] pkFields;
 
         private transient PreparedStatement existStatement;
         private transient PreparedStatement insertStatement;
@@ -287,6 +288,7 @@ private UpsertWriterUsingInsertUpdateStatement(
             this.existSql = existSql;
             this.insertSql = insertSql;
             this.updateSql = updateSql;
+            this.pkFields = pkFields;
         }
 
         @Override
@@ -310,7 +312,7 @@ void processOneRowInBatch(Row pk, Row row) throws SQLException {
             resultSet.close();
             if (exist) {
                 // do update
-                setRecordToStatement(updateStatement, fieldTypes, row);
+                setRecordToStatement(updateStatement, fieldTypes, row, pkFields);
                 updateStatement.addBatch();
             } else {
                 // do insert