private void fillDirtyData(JdbcExec exec, String tableIdentifier) {
        String[] identifiers = tableIdentifier.split("\\.");
        String database;
        String table;
        String schema = null;
        if (identifiers.length == 2) {
            database = identifiers[0];
            table = identifiers[1];
        } else {
            database = identifiers[0];
            schema = identifiers[1];
            table = identifiers[2];
        }
        TableMetricStatementExecutor executor = null;
        try {
            Field subExecutor;
            if (exec instanceof TableMetricStatementExecutor) {
                executor = (TableMetricStatementExecutor) exec;
            } else if (exec instanceof TableBufferReducedStatementExecutor) {
                subExecutor = TableBufferReducedStatementExecutor.class.getDeclaredField("upsertExecutor");
                subExecutor.setAccessible(true);
                executor = (TableMetricStatementExecutor) subExecutor.get(exec);
            } else if (exec instanceof TableBufferedStatementExecutor) {
                subExecutor = TableBufferedStatementExecutor.class.getDeclaredField("statementExecutor");
                subExecutor.setAccessible(true);
                executor = (TableMetricStatementExecutor) subExecutor.get(exec);
            }
        } catch (Exception e) {
            LOG.error("parse executor failed", e);
        }

        try {
            DirtyOptions dirtyOptions = dirtySinkHelper.getDirtyOptions();
            String dirtyLabel = DirtySinkHelper.regexReplace(dirtyOptions.getLabels(), DirtyType.BATCH_LOAD_ERROR, null,
                    database, table, schema);
            String dirtyLogTag =
                    DirtySinkHelper.regexReplace(dirtyOptions.getLogTag(), DirtyType.BATCH_LOAD_ERROR, null,
                            database, table, schema);
            String dirtyIdentifier =
                    DirtySinkHelper.regexReplace(dirtyOptions.getIdentifier(), DirtyType.BATCH_LOAD_ERROR,
                            null, database, table, schema);

            if (executor != null) {
                executor.setDirtyMetaData(dirtyLabel, dirtyLogTag, dirtyIdentifier);
            } else {
                LOG.error("executor is null, can not set metaData");
            }
        } catch (Exception e) {
            LOG.error("filling dirty metadata failed", e);
        }
    }
Extract and write record to recordsMap(buffer)

@param row The record to write.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/jdbc/src/main/java/org/apache/inlong/sort/jdbc/internal/JdbcMultiBatchingOutputFormat.java