private void handleDirtyData(JsonNode data, byte[] originData, String database,
            String table, DirtyType dirtyType, Throwable e) {
        if (exceptionPolicy == SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE) {
            String label = parseValue(data, dirtySinkHelper.getDirtyOptions().getLabels());
            String logTag = parseValue(data, dirtySinkHelper.getDirtyOptions().getLogTag());
            String identifier = parseValue(data, dirtySinkHelper.getDirtyOptions().getIdentifier());
            dirtySinkHelper.invoke(new String(originData), dirtyType, label, logTag, identifier, e);
        }
        if (metricData != null) {
            metricData.outputDirtyMetricsWithEstimate(database, table, 1, originData.length);
        }
    }
Process schema change for Doris

@param data The origin data
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/schema/SchemaChangeHelper.java