@Override
    public Optional<String> getUpsertStatement(
            String tableName, String[] fieldNames, String[] uniqueKeyFields) {
        StringBuilder sb = new StringBuilder();
        sb.append("MERGE INTO ")
                .append(tableName)
                .append(" T1 USING (")
                .append(buildQueryStatement(fieldNames))
                .append(") T2 ON (")
                .append(buildConditions(uniqueKeyFields))
                .append(") ");
        String updateSql = buildUpdateConnection(fieldNames, uniqueKeyFields);
        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());
    }
SqlServer merge data.
https://docs.microsoft.com/en-us/sql/t-sql/statements/merge-transact-sql?view=sql-server-ver15
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/dialect/SqlServerDialect.java