public List<String> getPkNamesFromDb(String tableIdentifier,
            JdbcOptions jdbcOptions) {
        PreparedStatement st = null;
        try {
            JdbcOptions jdbcExecOptions = JdbcMultiBatchingComm.getExecJdbcOptions(jdbcOptions, tableIdentifier);
            SimpleJdbcConnectionProvider tableConnectionProvider = new SimpleJdbcConnectionProvider(jdbcExecOptions);
            Connection conn = tableConnectionProvider.getOrEstablishConnection();
            st = setQueryPrimaryKeySql(conn, tableIdentifier);
            ResultSet rs = st.executeQuery();
            if (rs.next()) {
                String pkColumn = rs.getString(PRIMARY_KEY_COLUMN);
                LOG.info("TableIdentifier:{} get pkColumn:{}", tableIdentifier, pkColumn);
                checkAndClose(st);
                return Arrays.asList(pkColumn.split(","));
            } else {
                LOG.info("TableIdentifier:{} get pkColumn: null", tableIdentifier);
                checkAndClose(st);
                return null;
            }
        } catch (Exception e) {
            LOG.error("TableIdentifier:{} getAndSetPkNamesFromDb get err:", tableIdentifier, e);
            checkAndClose(st);
        }
        return null;
    }
get getPkNames from query db.tb

@return a list of PkNames.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/table/AbstractJdbcDialect.java