private JdbcExec enhanceExecutor(JdbcExec exec) throws NoSuchFieldException, IllegalAccessException {
        if (dirtySinkHelper.getDirtySink() == null) {
            return null;
        }
        // enhance the actual executor to tablemetricstatementexecutor
        Field subExecutor;
        if (exec instanceof TableBufferReducedStatementExecutor) {
            subExecutor = TableBufferReducedStatementExecutor.class.getDeclaredField("upsertExecutor");
        } else if (exec instanceof TableBufferedStatementExecutor) {
            subExecutor = TableBufferedStatementExecutor.class.getDeclaredField("statementExecutor");
        } else {
            throw new RuntimeException("table enhance failed, can't enhance " + exec.getClass());
        }
        subExecutor.setAccessible(true);
        TableSimpleStatementExecutor executor = (TableSimpleStatementExecutor) subExecutor.get(exec);
        // get the stamentfactory and rowconverter in order to initialize tablemetricstatementExecutor
        Field statementFactory = TableSimpleStatementExecutor.class.getDeclaredField("stmtFactory");
        Field rowConverter = TableSimpleStatementExecutor.class.getDeclaredField("converter");
        statementFactory.setAccessible(true);
        rowConverter.setAccessible(true);
        final StatementFactory stmtFactory = (StatementFactory) statementFactory.get(executor);
        final JdbcRowConverter converter = (JdbcRowConverter) rowConverter.get(executor);
        TableMetricStatementExecutor newExecutor =
                new TableMetricStatementExecutor(stmtFactory, converter, dirtySinkHelper, sinkMetricData);
        newExecutor.setMultipleSink(true);
        // replace the original TableBufferedStatementExecutor with metric executor
        if (exec instanceof TableBufferedStatementExecutor) {
            subExecutor = TableBufferedStatementExecutor.class.getDeclaredField("valueTransform");
            subExecutor.setAccessible(true);
            Function<RowData, RowData> valueTransform = (Function<RowData, RowData>) subExecutor.get(exec);
            newExecutor.setValueTransform(valueTransform);
            return (JdbcExec) newExecutor;
        }
        // replace the sub-executor that generates flinkSQL for executors such as
        // TableBufferReducedExecutor or InsertOrUpdateExecutor
        subExecutor.set(exec, newExecutor);
        return null;
    }
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