virtual bool run(OperationContext* opCtx,
                     const string& dbname,
                     const BSONObj& cmdObj,
                     BSONObjBuilder& result) {
        const bool isExplain = false;
        Lock::DBLock dbLock(opCtx, dbname, MODE_IS);
        auto nss = parseNsOrUUID(opCtx, dbname, cmdObj);
        auto request = CountRequest::parseFromBSON(nss, cmdObj, isExplain);
        if (!request.isOK()) {
            return appendCommandStatus(result, request.getStatus());
        }

        AutoGetCollectionOrViewForReadCommand ctx(
            opCtx, request.getValue().getNs(), std::move(dbLock));
        Collection* collection = ctx.getCollection();

        if (ctx.getView()) {
            ctx.releaseLocksForView();

            auto viewAggregation = request.getValue().asAggregationCommand();
            if (!viewAggregation.isOK()) {
                return appendCommandStatus(result, viewAggregation.getStatus());
            }

            BSONObj aggResult = Command::runCommandDirectly(
                opCtx, OpMsgRequest::fromDBAndBody(dbname, std::move(viewAggregation.getValue())));

            if (ResolvedView::isResolvedViewErrorResponse(aggResult)) {
                result.appendElements(aggResult);
                return false;
            }

            ViewResponseFormatter formatter(aggResult);
            Status formatStatus = formatter.appendAsCountResponse(&result);
            if (!formatStatus.isOK()) {
                return appendCommandStatus(result, formatStatus);
            }
            return true;
        }

        // Prevent chunks from being cleaned up during yields - this allows us to only check the
        // version on initial entry into count.
        auto rangePreserver =
            CollectionShardingState::get(opCtx, request.getValue().getNs())->getMetadata();

        auto statusWithPlanExecutor = getExecutorCount(opCtx,
                                                       collection,
                                                       request.getValue(),
                                                       false,  // !explain
                                                       PlanExecutor::YIELD_AUTO);
        if (!statusWithPlanExecutor.isOK()) {
            return appendCommandStatus(result, statusWithPlanExecutor.getStatus());
        }

        auto exec = std::move(statusWithPlanExecutor.getValue());

        // Store the plan summary string in CurOp.
        auto curOp = CurOp::get(opCtx);
        {
            stdx::lock_guard<Client> lk(*opCtx->getClient());
            curOp->setPlanSummary_inlock(Explain::getPlanSummary(exec.get()));
        }

        Status execPlanStatus = exec->executePlan();
        if (!execPlanStatus.isOK()) {
            return appendCommandStatus(result, execPlanStatus);
        }

        PlanSummaryStats summaryStats;
        Explain::getSummaryStats(*exec, &summaryStats);
        if (collection) {
            collection->infoCache()->notifyOfQuery(opCtx, summaryStats.indexesUsed);
        }
        curOp->debug().setPlanSummaryMetrics(summaryStats);

        if (curOp->shouldDBProfile()) {
            BSONObjBuilder execStatsBob;
            Explain::getWinningPlanStats(exec.get(), &execStatsBob);
            curOp->debug().execStats = execStatsBob.obj();
        }

        // Plan is done executing. We just need to pull the count out of the root stage.
        invariant(STAGE_COUNT == exec->getRootStage()->stageType());
        CountStage* countStage = static_cast<CountStage*>(exec->getRootStage());
        const CountStats* countStats =
            static_cast<const CountStats*>(countStage->getSpecificStats());

        result.appendNumber("n", countStats->nCounted);
        return true;
    }