@Override
    public Optional<String> getUpsertStatement(String schema, String tableName, String[] fieldNames, String[] uniqueKeyFields, boolean allReplace) {
        String uniqueColumns = Arrays.stream(uniqueKeyFields)
                .map(this::quoteIdentifier)
                .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
        );

    }
Date: 2020/1/3
Company: www.dtstack.com
@author maqi
postgresql/postgresql-sink/src/main/java/com/dtstack/flink/sql/sink/postgresql/PostgresqlDialect.java