private void handleColumnsChange(String tableIdentifier, JsonNode rootNode, JsonNode physicalData) {
        String columns = parseColumns(rootNode, physicalData);
        String oldColumns = columnsMap.get(tableIdentifier);
        if (columns == null && oldColumns != null || (columns != null && !columns.equals(oldColumns))) {
            flushSingleTable(tableIdentifier, batchMap.get(tableIdentifier));
            if (!errorTables.contains(tableIdentifier)) {
                columnsMap.put(tableIdentifier, columns);
            } else {
                batchMap.remove(tableIdentifier);
                columnsMap.remove(tableIdentifier);
                errorTables.remove(tableIdentifier);
            }
        }
    }
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