@Override
    public Optional<String> getUpsertStatement(String schema, String tableName, String[] fieldNames, String[] uniqueKeyFields, boolean allReplace) {
        tableName = DtStringUtil.getTableFullPath(schema, tableName);
        StringBuilder sb = new StringBuilder();
        sb.append("MERGE INTO " + tableName + " T1 USING "
                + "(" + buildValuesStatement(fieldNames) + ") T2 ("
                + buildFiledNameStatement(fieldNames) +
                ") ON ("
                + buildConnectionConditions(uniqueKeyFields) + ") ");

        String updateSql = buildUpdateConnection(fieldNames, uniqueKeyFields, allReplace);

        if (StringUtils.isNotEmpty(updateSql)) {
            sb.append(" WHEN MATCHED THEN UPDATE SET ");
            sb.append(updateSql);
        }

        sb.append(" WHEN NOT MATCHED THEN "
                + "INSERT (" + Arrays.stream(fieldNames).map(this::quoteIdentifier).collect(Collectors.joining(",")) + ") VALUES ("
                + Arrays.stream(fieldNames).map(col -> "T2." + quoteIdentifier(col)).collect(Collectors.joining(",")) + ")");
        return Optional.of(sb.toString());
    }
Date: 2020/1/19
Company: www.dtstack.com
@author maqi
db2/db2-sink/src/main/java/com/dtstack/flink/sql/sink/db/DbDialect.java