private void addBatch(T row) throws IOException {
        readInNum.incrementAndGet();
        if (!multipleSink) {
            addSingle(row);
            return;
        }
        if (row instanceof RowData) {
            RowData rowData = (RowData) row;
            JsonNode rootNode;
            try {
                rootNode = jsonDynamicSchemaFormat.deserialize(rowData.getBinary(0));
            } catch (Exception e) {
                LOG.error(String.format("deserialize error, raw data: %s", new String(rowData.getBinary(0))), e);
                if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdatePolicy) {
                    handleDirtyData(new String(rowData.getBinary(0)), DirtyType.DESERIALIZE_ERROR, e);
                }
                return;
            }
            boolean isDDL = jsonDynamicSchemaFormat.extractDDLFlag(rootNode);
            if (isDDL) {
                ddlNum.incrementAndGet();
                helper.process(rowData.getBinary(0), rootNode);
                return;
            }
            String tableIdentifier;
            List<RowKind> rowKinds;
            JsonNode physicalData;
            JsonNode updateBeforeNode;
            List<Map<String, String>> physicalDataList;
            List<Map<String, String>> updateBeforeList = null;
            try {
                tableIdentifier = StringUtils.join(
                        jsonDynamicSchemaFormat.parse(rootNode, databasePattern), ".",
                        jsonDynamicSchemaFormat.parse(rootNode, tablePattern));
                if (checkFlushException(tableIdentifier)) {
                    return;
                }
                rowKinds = Preconditions.checkNotNull(jsonDynamicSchemaFormat
                        .opType2RowKind(jsonDynamicSchemaFormat.getOpType(rootNode)));
                physicalData = Preconditions.checkNotNull(jsonDynamicSchemaFormat.getPhysicalData(rootNode));
                Preconditions.checkArgument(!physicalData.isEmpty());
                physicalDataList = Preconditions.checkNotNull(jsonDynamicSchemaFormat.jsonNode2Map(physicalData));
                updateBeforeNode = jsonDynamicSchemaFormat.getUpdateBefore(rootNode);
                if (updateBeforeNode != null) {
                    updateBeforeList = jsonDynamicSchemaFormat.jsonNode2Map(updateBeforeNode);
                }
            } catch (Exception e) {
                LOG.error(String.format("json parse error, raw data: %s", new String(rowData.getBinary(0))), e);
                if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdatePolicy) {
                    handleDirtyData(new String(rowData.getBinary(0)), DirtyType.JSON_PROCESS_ERROR, e);
                }
                return;
            }
            for (int i = 0; i < physicalDataList.size(); i++) {
                for (RowKind rowKind : rowKinds) {
                    if (updateBeforeList != null && updateBeforeList.size() > i) {
                        addRow(rowKind, rootNode, physicalData, updateBeforeNode,
                                physicalDataList.get(i), updateBeforeList.get(i));
                    } else {
                        addRow(rowKind, rootNode, physicalData, updateBeforeNode,
                                physicalDataList.get(i), null);
                    }
                }
            }
        } else {
            LOG.error(String.format("The type of element should be 'RowData' only, raw data: %s", row));
            if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdatePolicy) {
                handleDirtyData(row, DirtyType.UNSUPPORTED_DATA_TYPE,
                        new RuntimeException("The type of element should be 'RowData' only."));
            }
        }
    }
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