private DataStream<RowData> appendWithMiniBatchGroup(DataStream<RowData> input,
                RowType flinkType,
                Set<Integer> equalityFieldIds) {
            if (!tableOptions.get(WRITE_MINI_BATCH_ENABLE)) {
                return input;
            }

            FieldGetter[] fieldGetters =
                    IntStream.range(0, flinkType.getFieldCount())
                            .mapToObj(i -> RowData.createFieldGetter(flinkType.getTypeAt(i), i))
                            .toArray(RowData.FieldGetter[]::new);
            Schema writeSchema = TypeUtil.reassignIds(
                    FlinkSchemaUtil.convert(FlinkSchemaUtil.toSchema(flinkType)), table.schema());
            Schema deleteSchema = TypeUtil.select(table.schema(), equalityFieldIds);
            PartitionKey partitionKey = new PartitionKey(table.spec(), table.schema());

            int parallelism = writeParallelism == null ? input.getParallelism() : writeParallelism;
            boolean preAggregation = tableOptions.get(WRITE_MINI_BATCH_PRE_AGG_ENABLE) && !equalityFieldIds.isEmpty();
            BufferType bufferType = tableOptions.get(WRITE_MINI_BATCH_BUFFER_TYPE);
            PartitionGroupBuffer buffer = preAggregation
                    ? PartitionGroupBuffer.preAggInstance(
                            fieldGetters, deleteSchema, writeSchema, partitionKey, bufferType)
                    : PartitionGroupBuffer.nonPreAggInstance(writeSchema, partitionKey, bufferType);
            SingleOutputStreamOperator<RowData> inputWithMiniBatchGroup = input
                    .transform("mini_batch_group",
                            input.getType(),
                            new IcebergMiniBatchGroupOperator(buffer))
                    .setParallelism(parallelism);
            if (uidPrefix != null) {
                inputWithMiniBatchGroup.uid(uidPrefix + "mini_batch_group");
            }
            return inputWithMiniBatchGroup;
        }
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