public static AbstractUpsertWriter create(
            JDBCDialect dialect,
            String schema,
            String tableName,
            String[] fieldNames,
            int[] fieldTypes,
            String[] keyFields,
            String[] partitionFields,
            boolean objectReuse,
            boolean allReplace,
            AbstractDtRichOutputFormat metricOutputFormat,
            long errorLimit) {

        checkNotNull(keyFields);

        List<String> nameList = Arrays.asList(fieldNames);
        int[] pkFields = Arrays.stream(keyFields).mapToInt(nameList::indexOf).toArray();
        int[] pkTypes = fieldTypes == null ? null :
                Arrays.stream(pkFields).map(f -> fieldTypes[f]).toArray();

        String deleteSql = dialect.getDeleteStatement(schema, tableName, keyFields);
        // LOG.info("deleteSQL is :{}", deleteSql);

        Optional<String> upsertSql = dialect.getUpsertStatement(schema, tableName, fieldNames, keyFields, allReplace);
        LOG.info("execute UpsertStatement: {}", upsertSql.orElse("use UsingInsertUpdateStatement"));

        return upsertSql.map((Function<String, AbstractUpsertWriter>) sql ->
                new UpsertWriterUsingUpsertStatement(
                        fieldTypes, pkFields, pkTypes, objectReuse, deleteSql, sql, metricOutputFormat, errorLimit))
                .orElseGet(() ->
                        new UpsertWriterUsingInsertUpdateStatement(
                                fieldTypes, pkFields, pkTypes, objectReuse, deleteSql,
                                dialect.getRowExistsStatement(tableName, keyFields),
                                dialect.getInsertIntoStatement(schema, tableName, fieldNames, partitionFields),
                                dialect.getUpdateStatement(tableName, fieldNames, keyFields),
                                metricOutputFormat, errorLimit));
    }
Upsert writer to deal with upsert, delete message.dd
@author maqi
rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/writer/AbstractUpsertWriter.java