@Override
    public Optional<String> getUpsertStatement(
            String tableName, String[] fieldNames, String[] uniqueKeyFields) {
        String uniqueColumns =
                Arrays.stream(uniqueKeyFields)
                        .map(this::quoteIdentifier)
                        .collect(Collectors.joining(", "));
        List<String> uniqueKeyFieldList = Arrays.asList(uniqueKeyFields);
        String updateClause =
                Arrays.stream(fieldNames)
                        .filter(f -> !uniqueKeyFieldList.contains(f))
                        .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
                        .collect(Collectors.joining(", "));
        String str = getInsertIntoStatement(tableName, fieldNames);
        if (StringUtils.isNotEmpty(updateClause)) {
            str = str + " ON CONFLICT ("
                    + uniqueColumns
                    + ")"
                    + " DO UPDATE SET "
                    + updateClause;
        }
        return Optional.of(str);
    }
Postgres upsert query. It use ON CONFLICT ... DO UPDATE SET.. to replace into Postgres.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/dialect/TDSQLPostgresDialect.java