virtual bool run(const string &db, BSONObj &cmdObj, int options, string &errmsg,
                         BSONObjBuilder &result, bool fromRepl) {

            string ns = parseNs(db, cmdObj);

            intrusive_ptr<ExpressionContext> pCtx =
                new ExpressionContext(InterruptStatusMongod::status, NamespaceString(ns));

            /* try to parse the command; if this fails, then we didn't run */
            intrusive_ptr<Pipeline> pPipeline = Pipeline::parseCommand(errmsg, cmdObj, pCtx);
            if (!pPipeline.get())
                return false;

            if (pPipeline->getSplitMongodPipeline()) {
                // This is only used in testing
                return executeSplitPipeline(result, errmsg, ns, db, pPipeline, pCtx);
            }

#if _DEBUG
            // This is outside of the if block to keep the object alive until the pipeline is finished.
            BSONObj parsed;
            if (!pPipeline->isExplain() && !pCtx->inShard) {
                // Make sure all operations round-trip through Pipeline::toBson()
                // correctly by reparsing every command on DEBUG builds. This is
                // important because sharded aggregations rely on this ability.
                // Skipping when inShard because this has already been through the
                // transformation (and this unsets pCtx->inShard).
                parsed = pPipeline->serialize().toBson();
                pPipeline = Pipeline::parseCommand(errmsg, parsed, pCtx);
                verify(pPipeline);
            }
#endif

            // This does the mongod-specific stuff like creating a cursor
            PipelineD::prepareCursorSource(pPipeline, nsToDatabase(ns), pCtx);
            pPipeline->stitch();

            if (isCursorCommand(cmdObj)) {
                CursorId id;
                {
                    // Set up cursor
                    Client::ReadContext ctx(ns);
                    shared_ptr<Cursor> cursor(new PipelineCursor(pPipeline));
                    // cc will be owned by cursor manager
                    ClientCursor* cc = new ClientCursor(0, cursor, ns, cmdObj.getOwned());
                    id = cc->cursorid();
                }

                handleCursorCommand(id, cmdObj, result);
            }
            else {
                pPipeline->run(result);
            }

            if (DocumentSourceOut* out = dynamic_cast<DocumentSourceOut*>(pPipeline->output())) {
                result.append("outputNs", out->getOutputNs());
            }

            return true;
        }

 virtual void addRequiredPrivileges(const std::string& dbname,
                                    const BSONObj& cmdObj,
                                    std::vector<Privilege>* out) {
     ActionSet actions;
     actions.addAction(ActionType::find);
     out->push_back(Privilege(parseNs(dbname, cmdObj), actions));
 }

 Status checkAuthForCommand(ClientBasic* client,
                            const std::string& dbname,
                            const BSONObj& cmdObj) override {
     NamespaceString nss(parseNs(dbname, cmdObj));
     auto hasTerm = cmdObj.hasField(kTermField);
     return AuthorizationSession::get(client)->checkAuthForFind(nss, hasTerm);
 }

 void WriteCmd::addRequiredPrivileges(const std::string& dbname,
                                      const BSONObj& cmdObj,
                                      std::vector<Privilege>* out) {
     ActionSet actions;
     actions.addAction(_action);
     out->push_back(Privilege(parseNs(dbname, cmdObj), actions));
 }

    Status FindCmd::explain(OperationContext* txn,
                             const std::string& dbname,
                             const BSONObj& cmdObj,
                             ExplainCommon::Verbosity verbosity,
                             BSONObjBuilder* out) const {
        const string fullns = parseNs(dbname, cmdObj);

        // Parse the command BSON to a LiteParsedQuery.
        LiteParsedQuery* rawLpq;
        bool isExplain = true;
        Status lpqStatus = LiteParsedQuery::make(fullns, cmdObj, isExplain, &rawLpq);
        if (!lpqStatus.isOK()) {
            return lpqStatus;
        }
        auto_ptr<LiteParsedQuery> lpq(rawLpq);

        const NamespaceString nss(fullns);

        // Finish the parsing step by using the LiteParsedQuery to create a CanonicalQuery.
        // This requires a lock on the collection in case we're parsing $where: where-specific
        // parsing code assumes we have a lock and creates execution machinery that requires it.
        CanonicalQuery* rawCq;
        WhereCallbackReal whereCallback(txn, nss.db());
        Status canonStatus = CanonicalQuery::canonicalize(lpq.release(), &rawCq, whereCallback);
        if (!canonStatus.isOK()) {
            return canonStatus;
        }
        auto_ptr<CanonicalQuery> cq(rawCq);

        AutoGetCollectionForRead ctx(txn, nss);
        // The collection may be NULL. If so, getExecutor() should handle it by returning
        // an execution tree with an EOFStage.
        Collection* collection = ctx.getCollection();

        // We have a parsed query. Time to get the execution plan for it.
        PlanExecutor* rawExec;
        Status execStatus = Status::OK();
        if (cq->getParsed().getOptions().oplogReplay) {
            execStatus = getOplogStartHack(txn, collection, cq.release(), &rawExec);
        }
        else {
            size_t options = QueryPlannerParams::DEFAULT;
            if (shardingState.needCollectionMetadata(cq->getParsed().ns())) {
                options |= QueryPlannerParams::INCLUDE_SHARD_FILTER;
            }

            execStatus = getExecutor(txn, collection, cq.release(), &rawExec, options);
        }

        if (!execStatus.isOK()) {
            return execStatus;
        }

        scoped_ptr<PlanExecutor> exec(rawExec);
        exec->setYieldPolicy(PlanExecutor::YIELD_AUTO);

        // Got the execution tree. Explain it.
        return Explain::explainStages(exec.get(), verbosity, out);
    }

        bool run(OperationContext* txn, const string& dbname, BSONObj& jsobj, int, string& errmsg, BSONObjBuilder& result, bool fromRepl ) {

            if ( !globalScriptEngine ) {
                errmsg = "server-side JavaScript execution is disabled";
                return false;
            }
            
            /* db.$cmd.findOne( { group : <p> } ) */
            const BSONObj& p = jsobj.firstElement().embeddedObjectUserCheck();

            BSONObj q;
            if ( p["cond"].type() == Object )
                q = p["cond"].embeddedObject();
            else if ( p["condition"].type() == Object )
                q = p["condition"].embeddedObject();
            else
                q = getQuery( p );

            BSONObj key;
            string keyf;
            if ( p["key"].type() == Object ) {
                key = p["key"].embeddedObjectUserCheck();
                if ( ! p["$keyf"].eoo() ) {
                    errmsg = "can't have key and $keyf";
                    return false;
                }
            }
            else if ( p["$keyf"].type() ) {
                keyf = p["$keyf"]._asCode();
            }
            else {
                // no key specified, will use entire object as key
            }

            BSONElement reduce = p["$reduce"];
            if ( reduce.eoo() ) {
                errmsg = "$reduce has to be set";
                return false;
            }

            BSONElement initial = p["initial"];
            if ( initial.type() != Object ) {
                errmsg = "initial has to be an object";
                return false;
            }


            string finalize;
            if (p["finalize"].type())
                finalize = p["finalize"]._asCode();

            const string ns = parseNs(dbname, jsobj);
            Client::ReadContext ctx(txn, ns);

            return group( txn, ctx.ctx().db() , ns , q ,
                          key , keyf , reduce._asCode() , reduce.type() != CodeWScope ? 0 : reduce.codeWScopeScopeDataUnsafe() ,
                          initial.embeddedObject() , finalize ,
                          errmsg , result );
        }

 ResourcePattern Command::parseResourcePattern(const std::string& dbname,
                                               const BSONObj& cmdObj) const {
     std::string ns = parseNs(dbname, cmdObj);
     if (ns.find('.') == std::string::npos) {
         return ResourcePattern::forDatabaseName(ns);
     }
     return ResourcePattern::forExactNamespace(NamespaceString(ns));
 }

        /**
         * Parses a count command object, 'cmdObj'.
         *
         * On success, fills in the out-parameter 'request' and returns an OK status.
         *
         * Returns a failure status if 'cmdObj' is not well formed.
         */
        Status parseRequest(const std::string& dbname,
                            const BSONObj& cmdObj,
                            CountRequest* request) const {

            long long skip = 0;
            if (cmdObj["skip"].isNumber()) {
                skip = cmdObj["skip"].numberLong();
                if (skip < 0) {
                    return Status(ErrorCodes::BadValue, "skip value is negative in count query");
                }
            }
            else if (cmdObj["skip"].ok()) {
                return Status(ErrorCodes::BadValue, "skip value is not a valid number");
            }

            long long limit = 0;
            if (cmdObj["limit"].isNumber()) {
                limit = cmdObj["limit"].numberLong();
            }
            else if (cmdObj["limit"].ok()) {
                return Status(ErrorCodes::BadValue, "limit value is not a valid number");
            }

            // For counts, limit and -limit mean the same thing.
            if (limit < 0) {
                limit = -limit;
            }

            // We don't validate that "query" is a nested object due to SERVER-15456.
            BSONObj query = cmdObj.getObjectField("query");

            BSONObj hintObj;
            if (Object == cmdObj["hint"].type()) {
                hintObj = cmdObj["hint"].Obj();
            }
            else if (String == cmdObj["hint"].type()) {
                const std::string hint = cmdObj.getStringField("hint");
                hintObj = BSON("$hint" << hint);
            }

            std::string ns = parseNs(dbname, cmdObj);

            if (!nsIsFull(ns)) {
                return Status(ErrorCodes::BadValue, "collection name missing");
            }

            // Parsed correctly. Fill out 'request' with the results.
            request->ns = ns;
            request->query = query;
            request->hint = hintObj;
            request->limit = limit;
            request->skip = skip;

            // By default, count requests are regular count not explain of count.
            request->explain = false;

            return Status::OK();
        }

        Status checkAuthForCommand( ClientBasic* client,
                                    const std::string& dbname,
                                    const BSONObj& cmdObj ) {

            return auth::checkAuthForWriteCommand( client->getAuthorizationSession(),
                                                   _writeType,
                                                   NamespaceString( parseNs( dbname, cmdObj ) ),
                                                   cmdObj );
        }

        virtual bool run(const string &db, BSONObj &cmdObj, int options, string &errmsg,
                         BSONObjBuilder &result, bool fromRepl) {

            string ns = parseNs(db, cmdObj);

            intrusive_ptr<ExpressionContext> pCtx =
                new ExpressionContext(InterruptStatusMongod::status, NamespaceString(ns));
            pCtx->tempDir = storageGlobalParams.dbpath + "/_tmp";

            /* try to parse the command; if this fails, then we didn't run */
            intrusive_ptr<Pipeline> pPipeline = Pipeline::parseCommand(errmsg, cmdObj, pCtx);
            if (!pPipeline.get())
                return false;

#if _DEBUG
            // This is outside of the if block to keep the object alive until the pipeline is finished.
            BSONObj parsed;
            if (!pPipeline->isExplain() && !pCtx->inShard) {
                // Make sure all operations round-trip through Pipeline::toBson()
                // correctly by reparsing every command on DEBUG builds. This is
                // important because sharded aggregations rely on this ability.
                // Skipping when inShard because this has already been through the
                // transformation (and this unsets pCtx->inShard).
                parsed = pPipeline->serialize().toBson();
                pPipeline = Pipeline::parseCommand(errmsg, parsed, pCtx);
                verify(pPipeline);
            }
#endif

            // This does the mongod-specific stuff like creating a cursor
            PipelineD::prepareCursorSource(pPipeline, pCtx);
            pPipeline->stitch();

            if (pPipeline->isExplain()) {
                result << "stages" << Value(pPipeline->writeExplainOps());
                return true; // don't do any actual execution
            }

            if (isCursorCommand(cmdObj)) {
                CursorId id;
                {
                    // Set up cursor
                    Client::ReadContext ctx(ns);
                    ClientCursor* cc = new ClientCursor(new PipelineRunner(pPipeline));
                    cc->isAggCursor = true; // enable special locking and ns deletion behavior
                    id = cc->cursorid();
                }

                handleCursorCommand(id, cmdObj, result);
            }
            else {
                pPipeline->run(result);
            }

            return true;
        }

 Status checkAuthForCommand(Client* client,
                            const std::string& dbname,
                            const BSONObj& cmdObj) const override {
     if (!AuthorizationSession::get(client)->isAuthorizedForActionsOnResource(
             ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
             ActionType::splitVector)) {
         return Status(ErrorCodes::Unauthorized, "Unauthorized");
     }
     return Status::OK();
 }

 Status checkAuthForCommand(ClientBasic* client,
                            const std::string& dbname,
                            const BSONObj& cmdObj) {
     std::string ns = parseNs(dbname, cmdObj);
     if (!client->getAuthorizationSession()->isAuthorizedForActionsOnNamespace(
             NamespaceString(ns), ActionType::find)) {
         return Status(ErrorCodes::Unauthorized, "unauthorized");
     }
     return Status::OK();
 }

 virtual Status checkAuthForCommand(ClientBasic* client,
                                    const std::string& dbname,
                                    const BSONObj& cmdObj) {
     if (!client->getAuthorizationSession()->isAuthorizedForActionsOnResource(
             ResourcePattern::forExactNamespace(NamespaceString(parseNs(dbname, cmdObj))),
             ActionType::getShardVersion)) {
         return Status(ErrorCodes::Unauthorized, "Unauthorized");
     }
     return Status::OK();
 }

    Status CmdCount::parseRequest(const std::string& dbname,
                                  const BSONObj& cmdObj,
                                  CountRequest* request) const {
        const string ns = parseNs(dbname, cmdObj);

        long long skip = 0;
        if (cmdObj["skip"].isNumber()) {
            skip = cmdObj["skip"].numberLong();
            if (skip < 0) {
                return Status(ErrorCodes::BadValue, "skip value is negative in count query");
            }
        }
        else if (cmdObj["skip"].ok()) {
            return Status(ErrorCodes::BadValue, "skip value is not a valid number");
        }

        long long limit = 0;
        if (cmdObj["limit"].isNumber()) {
            limit = cmdObj["limit"].numberLong();
        }
        else if (cmdObj["limit"].ok()) {
            return Status(ErrorCodes::BadValue, "limit value is not a valid number");
        }

        // For counts, limit and -limit mean the same thing.
        if (limit < 0) {
            limit = -limit;
        }

        BSONObj query;
        if (!cmdObj["query"].eoo()) {
            if (Object != cmdObj["query"].type()) {
                return Status(ErrorCodes::BadValue, "query field for count must be an object");
            }
            query = cmdObj.getObjectField("query");
        }

        BSONObj hintObj;
        if (Object == cmdObj["hint"].type()) {
            hintObj = cmdObj["hint"].Obj();
        }
        else if (String == cmdObj["hint"].type()) {
            const std::string hint = cmdObj.getStringField("hint");
            hintObj = BSON("$hint" << hint);
        }

        // Parsed correctly. Fill out 'request' with the results.
        request->ns = ns;
        request->query = query;
        request->hint = hintObj;
        request->limit = limit;
        request->skip = skip;

        return Status::OK();
    }

    Status GroupCommand::parseRequest(const string& dbname,
                                      const BSONObj& cmdObj,
                                      GroupRequest* request) const {
        request->ns = parseNs(dbname, cmdObj);

        const BSONObj& p = cmdObj.firstElement().embeddedObjectUserCheck();

        if (p["cond"].type() == Object) {
            request->query = p["cond"].embeddedObject().getOwned();
        }
        else if (p["condition"].type() == Object) {
            request->query = p["condition"].embeddedObject().getOwned();
        }
        else if (p["query"].type() == Object) {
            request->query = p["query"].embeddedObject().getOwned();
        }
        else if (p["q"].type() == Object) {
            request->query = p["q"].embeddedObject().getOwned();
        }

        if (p["key"].type() == Object) {
            request->keyPattern = p["key"].embeddedObjectUserCheck().getOwned();
            if (!p["$keyf"].eoo()) {
                return Status(ErrorCodes::BadValue, "can't have key and $keyf");
            }
        }
        else if (!p["$keyf"].eoo()) {
            request->keyFunctionCode = p["$keyf"]._asCode();
        }
        else {
            // No key specified.  Use the entire object as the key.
        }

        BSONElement reduce = p["$reduce"];
        if (reduce.eoo()) {
            return Status(ErrorCodes::BadValue, "$reduce has to be set");
        }
        request->reduceCode = reduce._asCode();

        if (reduce.type() == CodeWScope) {
            request->reduceScope = reduce.codeWScopeScopeDataUnsafe();
        }

        if (p["initial"].type() != Object) {
            return Status(ErrorCodes::BadValue, "initial has to be an object");
        }
        request->initial = p["initial"].embeddedObject().getOwned();

        if (!p["finalize"].eoo()) {
            request->finalize = p["finalize"]._asCode();
        }

        return Status::OK();
    }

    bool CmdExplain::run(OperationContext* txn,
                         const string& dbname,
                         BSONObj& cmdObj, int options,
                         string& errmsg,
                         BSONObjBuilder& result,
                         bool fromRepl) {
        // Get the verbosity.
        Explain::Verbosity verbosity = Explain::QUERY_PLANNER;
        if (!cmdObj["verbosity"].eoo()) {
            const char* verbStr = cmdObj["verbosity"].valuestrsafe();
            if (mongoutils::str::equals(verbStr, "executionStats")) {
                verbosity = Explain::EXEC_STATS;
            }
            else if (mongoutils::str::equals(verbStr, "allPlansExecution")) {
                verbosity = Explain::EXEC_ALL_PLANS;
            }
            else if (!mongoutils::str::equals(verbStr, "queryPlanner")) {
               errmsg = "verbosity string must be one of "
                        "{'queryPlanner', 'executionStats', 'allPlansExecution'}";
               return false;
            }
        }

        if (Object != cmdObj.firstElement().type()) {
            errmsg = "explain command requires a nested object";
            return false;
        }

        // This is the nested command which we are explaining.
        BSONObj explainObj = cmdObj.firstElement().Obj();

        const string ns = parseNs(dbname, explainObj);

        Command* commToExplain = Command::findCommand(explainObj.firstElementFieldName());
        if (NULL == commToExplain) {
            mongoutils::str::stream ss;
            ss << "unknown command: " << explainObj.firstElementFieldName();
            Status explainStatus(ErrorCodes::CommandNotFound, ss);
            return appendCommandStatus(result, explainStatus);
        }

        // Actually call the nested command's explain(...) method.
        Status explainStatus = commToExplain->explain(txn, dbname, explainObj, verbosity, &result);
        if (!explainStatus.isOK()) {
            return appendCommandStatus(result, explainStatus);
        }

        return true;
    }

    Status WriteCmd::checkAuthForCommand( ClientBasic* client,
                                          const std::string& dbname,
                                          const BSONObj& cmdObj ) {

        Status status( auth::checkAuthForWriteCommand( client->getAuthorizationSession(),
                _writeType,
                NamespaceString( parseNs( dbname, cmdObj ) ),
                cmdObj ));

        if ( !status.isOK() ) {
            setLastError( status.code(), status.reason().c_str() );
        }

        return status;
    }

        virtual Status checkAuthForCommand(ClientBasic* client,
                                           const std::string& dbname,
                                           const BSONObj& cmdObj) {
            std::string ns = parseNs(dbname, cmdObj);

            ActionSet actions;
            actions.addAction(ActionType::insert);
            actions.addAction(ActionType::createIndex); // SERVER-11418

            if (!client->getAuthorizationSession()->isAuthorizedForActionsOnResource(
                    ResourcePattern::forExactNamespace(NamespaceString(ns)), actions)) {
                return Status(ErrorCodes::Unauthorized, "Unauthorized");
            }
            return Status::OK();
        }

    Status explain(OperationContext* txn,
                   const std::string& dbname,
                   const BSONObj& cmdObj,
                   ExplainCommon::Verbosity verbosity,
                   const rpc::ServerSelectionMetadata&,
                   BSONObjBuilder* out) const override {
        const std::string fullns = parseNs(dbname, cmdObj);
        const NamespaceString nss(fullns);
        if (!nss.isValid()) {
            return {ErrorCodes::InvalidNamespace,
                    str::stream() << "Invalid collection name: " << nss.ns()};
        }

        // Parse the command BSON to a LiteParsedQuery.
        const bool isExplain = true;
        auto lpqStatus = LiteParsedQuery::makeFromFindCommand(nss, cmdObj, isExplain);
        if (!lpqStatus.isOK()) {
            return lpqStatus.getStatus();
        }

        // Finish the parsing step by using the LiteParsedQuery to create a CanonicalQuery.

        ExtensionsCallbackReal extensionsCallback(txn, &nss);
        auto statusWithCQ =
            CanonicalQuery::canonicalize(lpqStatus.getValue().release(), extensionsCallback);
        if (!statusWithCQ.isOK()) {
            return statusWithCQ.getStatus();
        }
        std::unique_ptr<CanonicalQuery> cq = std::move(statusWithCQ.getValue());

        AutoGetCollectionForRead ctx(txn, nss);
        // The collection may be NULL. If so, getExecutor() should handle it by returning
        // an execution tree with an EOFStage.
        Collection* collection = ctx.getCollection();

        // We have a parsed query. Time to get the execution plan for it.
        auto statusWithPlanExecutor =
            getExecutorFind(txn, collection, nss, std::move(cq), PlanExecutor::YIELD_AUTO);
        if (!statusWithPlanExecutor.isOK()) {
            return statusWithPlanExecutor.getStatus();
        }
        std::unique_ptr<PlanExecutor> exec = std::move(statusWithPlanExecutor.getValue());

        // Got the execution tree. Explain it.
        Explain::explainStages(exec.get(), verbosity, out);
        return Status::OK();
    }

        virtual bool run(OperationContext* txn,
                         const string& dbname,
                         BSONObj& cmdObj,
                         int,
                         string& errmsg,
                         BSONObjBuilder& result,
                         bool fromRepl) {

            string fromhost = cmdObj.getStringField("from");
            if ( fromhost.empty() ) {
                errmsg = "missing 'from' parameter";
                return false;
            }

            {
                HostAndPort h(fromhost);
                if (repl::isSelf(h)) {
                    errmsg = "can't cloneCollection from self";
                    return false;
                }
            }

            string collection = parseNs(dbname, cmdObj);
            if ( collection.empty() ) {
                errmsg = "bad 'cloneCollection' value";
                return false;
            }
            BSONObj query = cmdObj.getObjectField("query");
            if ( query.isEmpty() )
                query = BSONObj();

            BSONElement copyIndexesSpec = cmdObj.getField("copyindexes");
            bool copyIndexes = copyIndexesSpec.isBoolean() ? copyIndexesSpec.boolean() : true;

            log() << "cloneCollection.  db:" << dbname << " collection:" << collection << " from: " << fromhost
                  << " query: " << query << " " << ( copyIndexes ? "" : ", not copying indexes" ) << endl;

            Cloner cloner;
            auto_ptr<DBClientConnection> myconn;
            myconn.reset( new DBClientConnection() );
            if ( ! myconn->connect( fromhost , errmsg ) )
                return false;

            cloner.setConnection( myconn.release() );

            return cloner.copyCollection(txn, collection, query, errmsg, true, false, copyIndexes);
        }