static RowType toFlinkRowType(Schema schema, TableSchema requestedSchema) {
        if (requestedSchema != null) {
            // Convert the flink schema to iceberg schema firstly, then reassign ids to match the existing iceberg
            // schema.
            List<NestedField> filteredFields = FlinkSchemaUtil.convert(requestedSchema)
                    .columns()
                    .stream()
                    .filter(nestedField -> !META_INCREMENTAL.equals(nestedField.name()))
                    .collect(Collectors.toList());
            Schema writeSchema = TypeUtil.reassignIds(new Schema(filteredFields), schema);
            TypeUtil.validateWriteSchema(schema, writeSchema, true, true);

            // We use this flink schema to read values from RowData. The flink's TINYINT and SMALLINT will be promoted
            // to iceberg INTEGER, that means if we use iceberg's table schema to read TINYINT (backend by 1 'byte'),
            // we will read 4 bytes rather than 1 byte, it will mess up the byte array in BinaryRowData. So here we must
            // use flink schema.
            return (RowType) filterOutMetaField(requestedSchema).getLogicalType();
        } else {
            return FlinkSchemaUtil.convert(schema);
        }
    }
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