public void addSingle(T row) {
        if (row instanceof RowData) {
            try {
                RowData rowData = (RowData) row;
                Map<String, String> valueMap = new HashMap<>();
                StringJoiner value = new StringJoiner(this.fieldDelimiter);
                for (int i = 0; i < rowData.getArity() && i < fieldGetters.length; ++i) {
                    Object field = fieldGetters[i].getFieldOrNull(rowData);
                    if (jsonFormat) {
                        String data = field != null ? field.toString() : null;
                        valueMap.put(this.fieldNames[i], data);
                        batchBytes += this.fieldNames[i].getBytes(StandardCharsets.UTF_8).length;
                        if (data != null) {
                            batchBytes += data.getBytes(StandardCharsets.UTF_8).length;
                        }
                    } else {
                        String data = field != null ? field.toString() : NULL_VALUE;
                        value.add(data);
                        batchBytes += data.getBytes(StandardCharsets.UTF_8).length;
                    }
                }
                // add doris delete sign
                if (enableBatchDelete()) {
                    if (jsonFormat) {
                        valueMap.put(DORIS_DELETE_SIGN, DorisParseUtils.parseDeleteSign(rowData.getRowKind()));
                    } else {
                        value.add(DorisParseUtils.parseDeleteSign(rowData.getRowKind()));
                    }
                }
                Object data = jsonFormat ? valueMap : value.toString();
                List mapData = batchMap.getOrDefault(tableIdentifier, new ArrayList<String>());
                mapData.add(data);
                batchMap.putIfAbsent(tableIdentifier, mapData);
            } catch (Exception e) {
                LOG.error(String.format("serialize error, raw data: %s", row), e);
                if (SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE == schemaUpdatePolicy) {
                    handleDirtyData(row, DirtyType.SERIALIZE_ERROR, e);
                }
            }
        } else if (row instanceof String) {
            batchBytes += ((String) row).getBytes(StandardCharsets.UTF_8).length;
            List mapData = batchMap.getOrDefault(tableIdentifier, new ArrayList<String>());
            mapData.add(row);
            batchMap.putIfAbsent(tableIdentifier, mapData);
        } else {
            LOG.error(String.format("The type of element should be 'RowData' or 'String' 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' or 'String' 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