code: def _map_reduce(self, map, reduce, out, session, read_pref, **kwargs):
          
          cmd = SON([("mapReduce", self.__name),
                     ("map", map),
                     ("reduce", reduce),
                     ("out", out)])
          collation = validate_collation_or_none(kwargs.pop('collation', None))
          cmd.update(kwargs)
  
          inline = 'inline' in out
  
          if inline:
              user_fields = {'results': 1}
          else:
              user_fields = None
  
          read_pref = ((session and session._txn_read_preference())
                       or read_pref)
  
          with self.__database.client._socket_for_reads(read_pref, session) as (
                  sock_info, slave_ok):
              if (sock_info.max_wire_version >= 4 and
                      ('readConcern' not in cmd) and
                      inline):
                  read_concern = self.read_concern
              else:
                  read_concern = None
              if 'writeConcern' not in cmd and not inline:
                  write_concern = self._write_concern_for(session)
              else:
                  write_concern = None
  
              return self._command(
                  sock_info, cmd, slave_ok, read_pref,
                  read_concern=read_concern,
                  write_concern=write_concern,
                  collation=collation, session=session,
                  user_fields=user_fields)
docs: Internal mapReduce helper.