private SingleOutputStreamOperator<MultipleWriteResult> appendMultipleWriter(DataStream<RowData> input) {
            // equality field will be initialized at runtime
            // upsert mode will be initialized at runtime

            int parallelism = writeParallelism == null ? input.getParallelism() : writeParallelism;
            DynamicSchemaHandleOperator routeOperator = new DynamicSchemaHandleOperator(
                    catalogLoader, multipleSinkOption, dirtyOptions, dirtySink, inlongMetric, auditHostAndPorts,
                    enableSchemaChange, schemaChangePolicies, autoCreateTableWhenSnapshot, auditKeys);
            SingleOutputStreamOperator<RecordWithSchema> routeStream = input
                    .transform(operatorName(ICEBERG_WHOLE_DATABASE_MIGRATION_NAME),
                            TypeInformation.of(RecordWithSchema.class),
                            routeOperator)
                    .setParallelism(parallelism);
            RowType tableSchemaRowType = (RowType) tableSchema.toRowDataType().getLogicalType();
            IcebergProcessOperator streamWriter =
                    new IcebergProcessOperator(new IcebergMultipleStreamWriter(
                            appendMode, catalogLoader, inlongMetric, auditHostAndPorts,
                            multipleSinkOption, dirtyOptions, dirtySink, tableSchemaRowType,
                            getMetaFieldIndex(tableSchema),
                            switchAppendUpsertEnable, auditKeys));
            SingleOutputStreamOperator<MultipleWriteResult> writerStream = routeStream
                    .transform(operatorName(ICEBERG_MULTIPLE_STREAM_WRITER_NAME),
                            TypeInformation.of(IcebergProcessOperator.class),
                            streamWriter)
                    .setParallelism(parallelism);
            if (uidPrefix != null) {
                writerStream = writerStream.uid(uidPrefix + "-writer");
            }
            return writerStream;
        }
This operator is used to buffer a mini batch data group by field before writing to save memory resources.

This operator must have the same degree of parallelism as the writer operator to ensure that the data of
the same batch of equality fields in a checkpoint period will only be processed once by a certain writer
subtask (rather than processed multiple times).

@param input
@return
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java