private void doMerge(SourceRecord record) throws SQLException{
        String sinkTableName = mappingSourceSinkTables.get(getSourceTableName(record).toLowerCase());
        String[] pks = getSourcePrimaryKeys(record);
        List<String> columns = getColumns(record);

        if (batchPS == null) {
            StringBuilder sqlCmd = new StringBuilder();
            sqlCmd.append("MERGE INTO " + sinkTableName + " trg USING (SELECT ");

            for (int i = 0; i <= columns.size() - 1; i++) {
                if (i > 0) sqlCmd.append(",");
                sqlCmd.append("? as ").append(columns.get(i));
            }
            sqlCmd.append(" FROM DUAL) src ON (");

            for (int i = 0; i <= pks.length - 1; i++) {
                if (i > 0) sqlCmd.append(" AND ");
                sqlCmd.append("src.").append(pks[i]).append(" = trg.").append(pks[i]);
            }

            sqlCmd.append(" ) WHEN MATCHED THEN UPDATE SET ");
            // Set all columns for UPDATE SET statement
            for (int i = 0; i <= columns.size() - 1; i++) {
                boolean contains = Arrays.asList(pks).contains(columns.get(i));
                boolean containsUppercase = Arrays.asList(pks).contains(columns.get(i).toUpperCase());
                boolean containsQuoted = Arrays.asList(pks).contains("\""+columns.get(i).toUpperCase()+"\"");
                if (!contains && !containsUppercase && !containsQuoted)
                    sqlCmd.append(" trg.").append(columns.get(i)).append(" = src.").append(columns.get(i)).append(" ,");
            }
            // Delete the last comma
            sqlCmd.setLength(sqlCmd.length() - 1);

            sqlCmd.append(" WHEN NOT MATCHED THEN INSERT ( ");
            sqlCmd.append(String.join(",",columns));
            // Todas las columnas
            sqlCmd.append(" ) VALUES ( ");
            for (int i = 0; i <= columns.size() - 1; i++) {
                if (i > 0) sqlCmd.append(",");
                sqlCmd.append(" src.").append(columns.get(i));
            }
            sqlCmd.append(" ) ");

            LOG.info("Merging record with " + sqlCmd);
            batchPS = this.getConnection().prepareStatement(String.valueOf(sqlCmd));
        }

        // Binding values to the statement
        Struct value = ((Struct) record.value()).getStruct("after");
        // SET values
        bindValuesToBatchPS(value);
        batchPS.addBatch();
    }
Constructs the SqlManager.

@param opts   the ReplicaDB ToolOptions describing the user's requested action.
@param dsType
src/main/java/org/replicadb/manager/cdc/OracleManagerCDC.java