@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 ")
            .append(tableName)
            .append(" T1 USING (")
            .append(buildDualQueryStatement(fieldNames))
            .append(") T2 ON (")
            .append(buildConnectionConditions(uniqueKeyFields))
            .append(") ");

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

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

        sb.append(" WHEN NOT MATCHED THEN " + "INSERT (")
            .append(
                Arrays
                    .stream(fieldNames)
                    .map(this::quoteIdentifier)
                    .collect(Collectors.joining(",")))
            .append(") VALUES (")
            .append(
                Arrays
                    .stream(fieldNames)
                    .map(col -> "T2." + quoteIdentifier(col))
                    .collect(Collectors.joining(",")))
            .append(");");
        return Optional.of(sb.toString());
    }
Date: 2020/1/15
Company: www.dtstack.com

@author maqi
sqlserver/sqlserver-sink/src/main/java/com/dtstack/flink/sql/sink/sqlserver/SqlserverDialect.java