private void fillDirtySink(JsonNode rootNode, Map<String, String> physicalData,
            Map<String, String> updateBeforeData, String database, String table)
            throws IOException {
        DirtyOptions dirtyOptions = dirtySinkHelper.getDirtyOptions();
        String dirtyLabel = null;
        String dirtyLogTag = null;
        String dirtyIdentifier = null;
        if (dirtyOptions.ignoreDirty()) {
            dirtyLabel = jsonDynamicSchemaFormat.parse(rootNode,
                    DirtySinkHelper.regexReplace(dirtyOptions.getLabels(), DirtyType.BATCH_LOAD_ERROR, null));
            dirtyLogTag = jsonDynamicSchemaFormat.parse(rootNode,
                    DirtySinkHelper.regexReplace(dirtyOptions.getLogTag(), DirtyType.BATCH_LOAD_ERROR, null));
            dirtyIdentifier = jsonDynamicSchemaFormat.parse(rootNode,
                    DirtySinkHelper.regexReplace(dirtyOptions.getIdentifier(), DirtyType.BATCH_LOAD_ERROR, null));
        }
        physicalData.put(DIRTY_LOG_TAG, dirtyLogTag);
        physicalData.put(DIRTY_IDENTIFIER, dirtyIdentifier);
        physicalData.put(DIRTY_LABEL, dirtyLabel);
        physicalData.put(DATABASE, database);
        physicalData.put(TABLE, table);
        if (updateBeforeData != null) {
            updateBeforeData.put(DIRTY_LOG_TAG, dirtyLogTag);
            updateBeforeData.put(DIRTY_IDENTIFIER, dirtyIdentifier);
            updateBeforeData.put(DIRTY_LABEL, dirtyLabel);
            updateBeforeData.put(DATABASE, database);
            updateBeforeData.put(TABLE, table);
        }
    }
A builder used to set parameters to the output format's configuration in a fluent way.

@return builder
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/table/DorisDynamicSchemaOutputFormat.java