private JdbcExec getOrCreateStatementExecutor(
            String tableIdentifier) throws IOException {
        if (StringUtils.isBlank(tableIdentifier)) {
            return null;
        }
        JdbcExec jdbcExec = jdbcExecMap.get(tableIdentifier);
        if (null != jdbcExec) {
            return jdbcExec;
        }
        if (!pkNameMap.containsKey(tableIdentifier)) {
            getAndSetPkNamesFromDb(tableIdentifier);
        }
        RowType rowType = rowTypeMap.get(tableIdentifier);
        LogicalType[] logicalTypes = rowType.getFields().stream()
                .map(RowType.RowField::getType)
                .toArray(LogicalType[]::new);
        String[] filedNames = rowType.getFields().stream()
                .map(RowType.RowField::getName)
                .toArray(String[]::new);
        TypeInformation<RowData> rowDataTypeInfo = InternalTypeInfo.of(rowType);
        List<String> pkNameList = null;
        if (null != pkNameMap.get(tableIdentifier)) {
            pkNameList = pkNameMap.get(tableIdentifier);
        }
        StatementExecutorFactory<JdbcExec> statementExecutorFactory;
        if (CollectionUtils.isNotEmpty(pkNameList) && !appendMode) {
            // upsert query
            JdbcDmlOptions createDmlOptions = JdbcDmlOptions.builder()
                    .withTableName(JdbcMultiBatchingComm.getTableNameFromIdentifier(tableIdentifier))
                    .withDialect(jdbcOptions.getDialect())
                    .withFieldNames(filedNames)
                    .withKeyFields(pkNameList.toArray(new String[pkNameList.size()]))
                    .build();
            statementExecutorFactory = ctx -> (JdbcExec) JdbcMultiBatchingComm.createBufferReduceExecutor(
                    createDmlOptions, ctx, rowDataTypeInfo, logicalTypes);

        } else {
            // append only query
            final String sql = dmlOptions
                    .getDialect()
                    .getInsertIntoStatement(JdbcMultiBatchingComm.getTableNameFromIdentifier(tableIdentifier),
                            filedNames);
            statementExecutorFactory = ctx -> (JdbcExec) JdbcMultiBatchingComm.createSimpleBufferedExecutor(
                    ctx,
                    dmlOptions.getDialect(),
                    filedNames,
                    logicalTypes,
                    sql,
                    rowDataTypeInfo);
        }

        jdbcExec = statementExecutorFactory.apply(getRuntimeContext());
        try {
            JdbcOptions jdbcExecOptions = JdbcMultiBatchingComm.getExecJdbcOptions(jdbcOptions, tableIdentifier);
            SimpleJdbcConnectionProvider tableConnectionProvider = new SimpleJdbcConnectionProvider(jdbcExecOptions);
            try {
                tableConnectionProvider.getOrEstablishConnection();
            } catch (Exception e) {
                LOG.error("unable to open JDBC writer, tableIdentifier:{} err:", tableIdentifier, e);
                return null;
            }
            connectionExecProviderMap.put(tableIdentifier, tableConnectionProvider);

            if (!stopWritingWhenTableException && dirtySinkHelper.getDirtySink() != null) {
                try {
                    JdbcExec newExecutor = enhanceExecutor(jdbcExec);
                    if (newExecutor != null) {
                        jdbcExec = newExecutor;
                    }
                } catch (Exception e) {
                    LOG.warn("enhance executor failed for class :" +
                            jdbcExec.getClass(), e);
                }
            }

            jdbcExec.prepareStatements(tableConnectionProvider.getConnection());
        } catch (Exception e) {
            return null;
        }
        jdbcExecMap.put(tableIdentifier, jdbcExec);
        return jdbcExec;
    }
get or create  StatementExecutor for one table.

@param tableIdentifier The table identifier for which to get statementExecutor.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java