@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)
                .map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
                .collect(Collectors.joining(", "));

        return Optional.of(getInsertIntoStatement(schema, tableName, fieldNames, null) +
                " ON CONFLICT (" + uniqueColumns + ")" +
                " DO UPDATE SET " + updateClause
        );
    }
Date: 2020/09/10
Company: www.dtstack.com

@author tiezhu
kingbase/kingbase-sink/src/main/java/com/dtstack/flink/sql/sink/kingbase/KingbaseDialect.java