code: def _run_aggregation_cmd(self, session, explicit_session):
          
          read_preference = self._target._read_preference_for(session)
          client = self._database.client
  
          def _cmd(session, server, sock_info, slave_ok):
              pipeline = self._full_pipeline()
              cmd = SON([("aggregate", self._aggregation_target),
                         ("pipeline", pipeline),
                         ("cursor", {})])
  
              result = sock_info.command(
                  self._database.name,
                  cmd,
                  slave_ok,
                  read_preference,
                  self._target.codec_options,
                  parse_write_concern_error=True,
                  read_concern=self._target.read_concern,
                  collation=self._collation,
                  session=session,
                  client=self._database.client)
  
              cursor = result["cursor"]
  
              if (self._start_at_operation_time is None and
                  self._resume_token is None and
                  cursor.get("_id") is None and
                  sock_info.max_wire_version >= 7):
                  self._start_at_operation_time = result["operationTime"]
  
              ns = cursor["ns"]
              _, collname = ns.split(".", 1)
              aggregation_collection = self._database.get_collection(
                  collname, codec_options=self._target.codec_options,
                  read_preference=read_preference,
                  write_concern=self._target.write_concern,
                  read_concern=self._target.read_concern
              )
  
              return CommandCursor(
                  aggregation_collection, cursor, sock_info.address,
                  batch_size=self._batch_size or 0,
                  max_await_time_ms=self._max_await_time_ms,
                  session=session, explicit_session=explicit_session)
  
          return client._retryable_read(_cmd, read_preference, session)
docs: Run the full aggregation pipeline for this ChangeStream and return
          the corresponding CommandCursor.