private JdbcExec enhanceExecutor(JdbcExec exec) throws NoSuchFieldException, IllegalAccessException {
        if (dirtySink == null) {
            return null;
        }
        final DirtySinkHelper dirtySinkHelper = new DirtySinkHelper<>(dirtyOptions, dirtySink);
        // enhance the actual executor to tablemetricstatementexecutor
        Field executorType;
        if (exec instanceof TableBufferReducedStatementExecutor) {
            executorType = TableBufferReducedStatementExecutor.class.getDeclaredField("upsertExecutor");
        } else if (exec instanceof TableBufferedStatementExecutor) {
            executorType = TableBufferedStatementExecutor.class.getDeclaredField("statementExecutor");
        } else {
            throw new RuntimeException("table enhance failed, can't enhance " + exec.getClass());
        }
        executorType.setAccessible(true);
        TableSimpleStatementExecutor executor = (TableSimpleStatementExecutor) executorType.get(exec);
        // get the factory and rowconverter 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);
        // for TableBufferedStatementExecutor, replace the executor
        if (exec instanceof TableBufferedStatementExecutor) {
            Field transform = TableBufferedStatementExecutor.class.getDeclaredField("valueTransform");
            transform.setAccessible(true);
            Function<RowData, RowData> valueTransform = (Function<RowData, RowData>) transform.get(exec);
            newExecutor.setValueTransform(valueTransform);
            return (JdbcExec) newExecutor;
        }
        // replace the sub-executor that generates flinkSQL for executors such as
        // TableBufferReducedExecutor or InsertOrUpdateExecutor
        executorType.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/JdbcBatchingOutputFormat.java