[hotfix][Connector/JDBC] Remove unnecessary unique fields of PostgreSQL upsert statement

flink-connector-jdbc-postgres/src/main/java/org/apache/flink/connector/jdbc/postgres/database/dialect/PostgresDialect.java

@@ -25,6 +25,7 @@
 
 import java.util.Arrays;
 import java.util.EnumSet;
+import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -73,17 +74,22 @@ public Optional<String> getUpsertStatement(
                 Arrays.stream(uniqueKeyFields)
                         .map(this::quoteIdentifier)
                         .collect(Collectors.joining(", "));
+        final Set<String> uniqueKeyFieldsSet = new HashSet<>(Arrays.asList(uniqueKeyFields));
         String updateClause =
                 Arrays.stream(fieldNames)
+                        .filter(f -> !uniqueKeyFieldsSet.contains(f))
                         .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
                         .collect(Collectors.joining(", "));
+        String conflictAction =
+                updateClause.isEmpty()
+                        ? " DO NOTHING"
+                        : String.format(" DO UPDATE SET %s", updateClause);
         return Optional.of(
                 getInsertIntoStatement(tableName, fieldNames)
                         + " ON CONFLICT ("
                         + uniqueColumns
                         + ")"
-                        + " DO UPDATE SET "
-                        + updateClause);
+                        + conflictAction);
     }
 
     @Override

flink-connector-jdbc-postgres/src/test/java/org/apache/flink/connector/jdbc/postgres/database/dialect/PostgresDialectTest.java

@@ -21,9 +21,13 @@
 import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialectTest;
 import org.apache.flink.connector.jdbc.postgres.PostgresTestBase;
 
+import org.junit.jupiter.api.Test;
+
 import java.util.Arrays;
 import java.util.List;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /** The PostgresSql params for {@link JdbcDialectTest}. */
 class PostgresDialectTest extends JdbcDialectTest implements PostgresTestBase {
 
@@ -58,4 +62,24 @@ protected List<TestItem> testData() {
                         "The precision of field 'f0' is out of the TIMESTAMP precision range [1, 6] supported by PostgreSQL dialect."),
                 createTestItem("TIMESTAMP_LTZ(3)", "Unsupported type:TIMESTAMP_LTZ(3)"));
     }
+
+    @Test
+    void testUpsertStatement() {
+        PostgresDialect dialect = new PostgresDialect();
+        final String tableName = "tbl";
+        final String[] fieldNames = {
+            "id", "name", "email", "ts", "field1", "field_2", "__field_3__"
+        };
+        final String[] doUpdatekeyFields = {"id", "__field_3__"};
+        final String[] doNothingkeyFields = {
+            "id", "name", "email", "ts", "field1", "field_2", "__field_3__"
+        };
+
+        assertThat(dialect.getUpsertStatement(tableName, fieldNames, doUpdatekeyFields).get())
+                .isEqualTo(
+                        "INSERT INTO tbl(id, name, email, ts, field1, field_2, __field_3__) VALUES (:id, :name, :email, :ts, :field1, :field_2, :__field_3__) ON CONFLICT (id, __field_3__) DO UPDATE SET name=EXCLUDED.name, email=EXCLUDED.email, ts=EXCLUDED.ts, field1=EXCLUDED.field1, field_2=EXCLUDED.field_2");
+        assertThat(dialect.getUpsertStatement(tableName, fieldNames, doNothingkeyFields).get())
+                .isEqualTo(
+                        "INSERT INTO tbl(id, name, email, ts, field1, field_2, __field_3__) VALUES (:id, :name, :email, :ts, :field1, :field_2, :__field_3__) ON CONFLICT (id, name, email, ts, field1, field_2, __field_3__) DO NOTHING");
+    }
 }