private SingleOutputStreamOperator<WriteResult> appendWriter(DataStream<RowData> input, RowType flinkRowType,
                List<Integer> equalityFieldIds) {
            // Fallback to use upsert mode parsed from table properties if don't specify in job level.
            // Only if not appendMode, upsert can be valid.
            boolean upsertMode = flinkWriteConf.upsertMode() && !appendMode;

            LOG.info("The iceberg sink is using {} mode.", upsertMode ? "upsert" : "append");
            // Validate the equality fields and partition fields if we enable the upsert mode.

            if (upsertMode) {
                Preconditions.checkState(!flinkWriteConf.overwriteMode(),
                        "OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream.");
                Preconditions.checkState(!equalityFieldIds.isEmpty(),
                        "Equality field columns shouldn't be empty when configuring to use UPSERT data stream.");
                if (!table.spec().isUnpartitioned()) {
                    for (PartitionField partitionField : table.spec().fields()) {
                        Preconditions.checkState(equalityFieldIds.contains(partitionField.sourceId()),
                                "In UPSERT mode, partition field '%s' should be included in equality fields: '%s'",
                                partitionField, equalityFieldColumns);
                    }
                }
            }

            // Add rate limit if necessary
            DataStream<RowData> inputWithRateLimit = appendWithRateLimit(input);
            DataStream<RowData> inputWithMiniBatch = appendWithMiniBatchGroup(
                    inputWithRateLimit, flinkRowType, new HashSet<>(equalityFieldIds));

            IcebergProcessOperator<RowData, WriteResult> streamWriter = createStreamWriter(
                    table, flinkRowType, equalityFieldIds, flinkWriteConf, appendMode, inlongMetric,
                    auditHostAndPorts, dirtyOptions, dirtySink, tableSchema,
                    switchAppendUpsertEnable, tableOptions.get(WRITE_MINI_BATCH_ENABLE), auditKeys);

            int parallelism = writeParallelism == null ? input.getParallelism() : writeParallelism;
            SingleOutputStreamOperator<WriteResult> writerStream = inputWithMiniBatch
                    .transform(operatorName(ICEBERG_STREAM_WRITER_NAME),
                            TypeInformation.of(WriteResult.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