@Override
    public String getInsertIntoStatement(String tableName, String[] fieldNames) {
        String columns =
                Arrays.stream(fieldNames)
                        .map(this::quoteIdentifier)
                        .collect(Collectors.joining(", "));
        String placeholders =
                Arrays.stream(fieldNames).map(f -> ":" + f).collect(Collectors.joining(", "));
        Pair<String, String> databaseAndTableName = getDatabaseAndTableName(tableName);
        return "INSERT INTO "
                + quoteIdentifier(databaseAndTableName.getLeft())
                + POINT
                + quoteIdentifier(databaseAndTableName.getRight())
                + "("
                + columns
                + ")"
                + " VALUES ("
                + placeholders
                + ")";
    }
Get delete one row statement by condition fields
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/dialect/jdbc/JdbcDialect.java