public void getAndSetPkNamesFromDb(String tableIdentifier) {
        try {
            AbstractJdbcDialect jdbcDialect = (AbstractJdbcDialect) jdbcOptions.getDialect();
            List<String> pkNames = jdbcDialect.getPkNamesFromDb(tableIdentifier, jdbcOptions);
            pkNameMap.put(tableIdentifier, pkNames);
        } catch (Exception e) {
            LOG.error("TableIdentifier:{} getAndSetPkNamesFromDb get err:", tableIdentifier, e);
        }
    }
Use reflection to initialize TableMetricStatementExecutor, and replace the original executor
 or upsertExecutor to calculate metrics.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java