@Override
    public PreparedStatement setQueryPrimaryKeySql(Connection conn, String tableIdentifier) throws SQLException {
        PreparedStatement st = conn.prepareStatement(QUERY_PRIMARY_KEY_SQL);
        st.setString(1, JdbcMultiBatchingComm.getDatabaseNameFromIdentifier(tableIdentifier));
        st.setString(2, JdbcMultiBatchingComm.getTableNameFromIdentifier(tableIdentifier));
        return st;
    }
Mysql upsert query use DUPLICATE KEY UPDATE.

<p>NOTE: It requires Mysql's primary key to be consistent with pkFields.

<p>We don't use REPLACE INTO, if there are other fields, we can keep their previous values.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/dialect/MySQLDialect.java