@SuppressWarnings({"unchecked"})
    private void addRow(RowKind rowKind, JsonNode rootNode, JsonNode physicalNode, JsonNode updateBeforeNode,
            Map<String, String> physicalData, Map<String, String> updateBeforeData) throws IOException {
        String database = jsonDynamicSchemaFormat.parse(rootNode, databasePattern);
        String table = jsonDynamicSchemaFormat.parse(rootNode, tablePattern);
        String tableIdentifier = StringUtils.join(database, ".", table);
        if (dirtySinkHelper.getDirtySink() != null) {
            try {
                fillDirtySink(rootNode, physicalData, updateBeforeData, database, table);
            } catch (Exception e) {
                LOG.warn("fill dirty sink parameters failed");
            }
        }
        switch (rowKind) {
            case INSERT:
            case UPDATE_AFTER:
                handleColumnsChange(tableIdentifier, rootNode, physicalNode);
                batchBytes += physicalData.toString().getBytes(StandardCharsets.UTF_8).length;
                size++;
                batchMap.computeIfAbsent(tableIdentifier, k -> new ArrayList<>())
                        .add(physicalData);
                if (enableBatchDelete()) {
                    physicalData.put(DORIS_DELETE_SIGN, DORIS_DELETE_FALSE);
                }
                break;
            case DELETE:
                handleColumnsChange(tableIdentifier, rootNode, physicalNode);
                batchBytes += physicalData.toString().getBytes(StandardCharsets.UTF_8).length;
                size++;
                // add doris delete sign
                if (enableBatchDelete()) {
                    physicalData.put(DORIS_DELETE_SIGN, DORIS_DELETE_TRUE);
                }
                batchMap.computeIfAbsent(tableIdentifier, k -> new ArrayList<>())
                        .add(physicalData);
                break;
            case UPDATE_BEFORE:
                if (updateBeforeData != null) {
                    handleColumnsChange(tableIdentifier, rootNode, updateBeforeNode);
                    batchBytes += updateBeforeData.toString()
                            .getBytes(StandardCharsets.UTF_8).length;
                    size++;
                    // add doris delete sign
                    if (enableBatchDelete()) {
                        updateBeforeData.put(DORIS_DELETE_SIGN, DORIS_DELETE_TRUE);
                    }
                    batchMap.computeIfAbsent(tableIdentifier, k -> new ArrayList<>())
                            .add(updateBeforeData);
                }
                break;
            default:
        }
    }
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