Merge branch '1.10_release_4.1.x' into 1.10_release_4.2.x

Author: gituser

localTest/pom.xml

@@ -129,6 +129,32 @@
             <version>1.0-SNAPSHOT</version>
         </dependency>
 
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.postgresql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.side.all.postgresql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.sink.postgresql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.dtstack.flink</groupId>
+            <artifactId>sql.side.async.postgresql</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+
         <dependency>
             <groupId>com.dtstack.flink</groupId>
             <artifactId>sql.hbase</artifactId>

postgresql/postgresql-sink/src/main/java/com/dtstack/flink/sql/sink/postgresql/PostgresqlDialect.java

@@ -20,6 +20,7 @@
 
 
 import com.dtstack.flink.sql.sink.rdb.dialect.JDBCDialect;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.Arrays;
 import java.util.Optional;
@@ -48,9 +49,18 @@ public Optional<String> getUpsertStatement(String schema, String tableName, Stri
                 .collect(Collectors.joining(", "));
 
         String updateClause = Arrays.stream(fieldNames)
+                .filter(item -> !Arrays.asList(uniqueKeyFields).contains(item))
                 .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
                 .collect(Collectors.joining(", "));
 
+        if (StringUtils.isBlank(updateClause)) {
+            return Optional.of(
+                    getInsertIntoStatement(schema, tableName, fieldNames, null)
+                            + " ON CONFLICT ("
+                            + uniqueColumns
+                            + ") DO NOTHING");
+        }
+
         return Optional.of(getInsertIntoStatement(schema, tableName, fieldNames, null) +
                 " ON CONFLICT (" + uniqueColumns + ")" +
                 " DO UPDATE SET " + updateClause