Status ViewCatalog::dropView(OperationContext* opCtx, const NamespaceString& viewName) {
    stdx::lock_guard<stdx::mutex> lk(_mutex);
    _requireValidCatalog(lk, opCtx);

    // Save a copy of the view definition in case we need to roll back.
    auto viewPtr = _lookup(lk, opCtx, viewName.ns());
    if (!viewPtr) {
        return {ErrorCodes::NamespaceNotFound,
                str::stream() << "cannot drop missing view: " << viewName.ns()};
    }

    ViewDefinition savedDefinition = *viewPtr;

    invariant(_valid.load());
    _durable->remove(opCtx, viewName);
    _viewGraph.remove(savedDefinition.name());
    _viewMap.erase(viewName.ns());
    opCtx->recoveryUnit()->onRollback([this, viewName, savedDefinition]() {
        this->_viewGraphNeedsRefresh = true;
        this->_viewMap[viewName.ns()] = std::make_shared<ViewDefinition>(savedDefinition);
    });

    // We may get invalidated, but we're exclusively locked, so the change must be ours.
    opCtx->recoveryUnit()->onCommit(
        [this](boost::optional<Timestamp>) { this->_valid.store(true); });
    return Status::OK();
}

Status ViewCatalog::_upsertIntoGraph(OperationContext* opCtx, const ViewDefinition& viewDef) {

    // Performs the insert into the graph.
    auto doInsert = [this, &opCtx](const ViewDefinition& viewDef, bool needsValidation) -> Status {
        // Validate that the pipeline is eligible to serve as a view definition. If it is, this
        // will also return the set of involved namespaces.
        auto pipelineStatus = _validatePipeline_inlock(opCtx, viewDef);
        if (!pipelineStatus.isOK()) {
            if (needsValidation) {
                uassertStatusOKWithContext(pipelineStatus.getStatus(),
                                           str::stream() << "Invalid pipeline for view "
                                                         << viewDef.name().ns());
            }
            return pipelineStatus.getStatus();
        }

        auto involvedNamespaces = pipelineStatus.getValue();
        std::vector<NamespaceString> refs(involvedNamespaces.begin(), involvedNamespaces.end());
        refs.push_back(viewDef.viewOn());

        int pipelineSize = 0;
        for (auto obj : viewDef.pipeline()) {
            pipelineSize += obj.objsize();
        }

        if (needsValidation) {
            // Check the collation of all the dependent namespaces before updating the graph.
            auto collationStatus = _validateCollation_inlock(opCtx, viewDef, refs);
            if (!collationStatus.isOK()) {
                return collationStatus;
            }
            return _viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
        } else {
            _viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
            return Status::OK();
        }
    };

    if (_viewGraphNeedsRefresh) {
        _viewGraph.clear();
        for (auto&& iter : _viewMap) {
            auto status = doInsert(*(iter.second.get()), false);
            // If we cannot fully refresh the graph, we will keep '_viewGraphNeedsRefresh' true.
            if (!status.isOK()) {
                return status;
            }
        }
        // Only if the inserts completed without error will we no longer need a refresh.
        opCtx->recoveryUnit()->onRollback([this]() { this->_viewGraphNeedsRefresh = true; });
        _viewGraphNeedsRefresh = false;
    }

    // Remove the view definition first in case this is an update. If it is not in the graph, it
    // is simply a no-op.
    _viewGraph.remove(viewDef.name());

    return doInsert(viewDef, true);
}

StatusWith<stdx::unordered_set<NamespaceString>> ViewCatalog::_validatePipeline_inlock(
    OperationContext* opCtx, const ViewDefinition& viewDef) const {
    AggregationRequest request(viewDef.viewOn(), viewDef.pipeline());
    const LiteParsedPipeline liteParsedPipeline(request);
    const auto involvedNamespaces = liteParsedPipeline.getInvolvedNamespaces();

    // Verify that this is a legitimate pipeline specification by making sure it parses
    // correctly. In order to parse a pipeline we need to resolve any namespaces involved to a
    // collection and a pipeline, but in this case we don't need this map to be accurate since
    // we will not be evaluating the pipeline.
    StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
    for (auto&& nss : involvedNamespaces) {
        resolvedNamespaces[nss.coll()] = {nss, {}};
    }
    boost::intrusive_ptr<ExpressionContext> expCtx =
        new ExpressionContext(opCtx,
                              request,
                              CollatorInterface::cloneCollator(viewDef.defaultCollator()),
                              // We can use a stub MongoProcessInterface because we are only parsing
                              // the Pipeline for validation here. We won't do anything with the
                              // pipeline that will require a real implementation.
                              std::make_shared<StubMongoProcessInterface>(),
                              std::move(resolvedNamespaces));

    // Save this to a variable to avoid reading the atomic variable multiple times.
    auto currentFCV = serverGlobalParams.featureCompatibility.getVersion();

    // If the feature compatibility version is not 4.0, and we are validating features as master,
    // ban the use of new agg features introduced in 4.0 to prevent them from being persisted in the
    // catalog.
    if (serverGlobalParams.validateFeaturesAsMaster.load() &&
        currentFCV != ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo40) {
        expCtx->maxFeatureCompatibilityVersion = currentFCV;
    }
    auto pipelineStatus = Pipeline::parse(viewDef.pipeline(), std::move(expCtx));
    if (!pipelineStatus.isOK()) {
        return pipelineStatus.getStatus();
    }

    // Validate that the view pipeline does not contain any ineligible stages.
    auto sources = pipelineStatus.getValue()->getSources();
    if (!sources.empty() && sources.front()->constraints().isChangeStreamStage()) {
        return {ErrorCodes::OptionNotSupportedOnView,
                "$changeStream cannot be used in a view definition"};
    }

    return std::move(involvedNamespaces);
}

Status ViewCatalog::_validateCollation_inlock(OperationContext* opCtx,
                                              const ViewDefinition& view,
                                              const std::vector<NamespaceString>& refs) {
    for (auto&& potentialViewNss : refs) {
        auto otherView = _lookup_inlock(opCtx, potentialViewNss.ns());
        if (otherView &&
            !CollatorInterface::collatorsMatch(view.defaultCollator(),
                                               otherView->defaultCollator())) {
            return {ErrorCodes::OptionNotSupportedOnView,
                    str::stream() << "View " << view.name().toString()
                                  << " has conflicting collation with view "
                                  << otherView->name().toString()};
        }
    }
    return Status::OK();
}

void ViewGraph::insertWithoutValidating(const ViewDefinition& view,
                                        const std::vector<NamespaceString>& refs,
                                        int pipelineSize) {
    uint64_t nodeId = _getNodeId(view.name());
    // Note, the parent pointers of this node are set when the parents are inserted.
    // This sets the children pointers of the node for this view, as well as the parent
    // pointers for its children.
    Node* node = &(_graph[nodeId]);
    invariant(node->children.empty());
    invariant(!static_cast<bool>(node->collator));

    node->size = pipelineSize;
    node->collator = view.defaultCollator();

    for (const NamespaceString& childNss : refs) {
        uint64_t childId = _getNodeId(childNss);
        node->children.insert(childId);
        _graph[childId].parents.insert(nodeId);
    }
}

StatusWith<ResolvedViewDefinition> ViewCatalog::resolveView(OperationContext* txn,
                                                            const NamespaceString& nss) {
    const NamespaceString* resolvedNss = &nss;
    std::vector<BSONObj> resolvedPipeline;

    for (std::uint32_t i = 0; i < ViewCatalog::kMaxViewDepth; i++) {
        ViewDefinition* view = lookup(resolvedNss->ns());
        if (!view)
            return StatusWith<ResolvedViewDefinition>({*resolvedNss, resolvedPipeline});

        resolvedNss = &(view->viewOn());

        // Prepend the underlying view's pipeline to the current working pipeline.
        const std::vector<BSONObj>& toPrepend = view->pipeline();
        resolvedPipeline.insert(resolvedPipeline.begin(), toPrepend.begin(), toPrepend.end());
    }

    return {ErrorCodes::ViewDepthLimitExceeded,
            str::stream() << "View depth too deep or view cycle detected; maximum depth is "
                          << kMaxViewDepth};
}

StatusWith<stdx::unordered_set<NamespaceString>> ViewCatalog::_validatePipeline_inlock(
    OperationContext* opCtx, const ViewDefinition& viewDef) const {
    AggregationRequest request(viewDef.viewOn(), viewDef.pipeline());
    const LiteParsedPipeline liteParsedPipeline(request);
    const auto involvedNamespaces = liteParsedPipeline.getInvolvedNamespaces();

    // Verify that this is a legitimate pipeline specification by making sure it parses
    // correctly. In order to parse a pipeline we need to resolve any namespaces involved to a
    // collection and a pipeline, but in this case we don't need this map to be accurate since
    // we will not be evaluating the pipeline.
    StringMap<ExpressionContext::ResolvedNamespace> resolvedNamespaces;
    for (auto&& nss : involvedNamespaces) {
        resolvedNamespaces[nss.coll()] = {nss, {}};
    }
    boost::intrusive_ptr<ExpressionContext> expCtx =
        new ExpressionContext(opCtx,
                              request,
                              CollatorInterface::cloneCollator(viewDef.defaultCollator()),
                              // We can use a stub MongoProcessInterface because we are only parsing
                              // the Pipeline for validation here. We won't do anything with the
                              // pipeline that will require a real implementation.
                              std::make_shared<StubMongoProcessInterface>(),
                              std::move(resolvedNamespaces));
    auto pipelineStatus = Pipeline::parse(viewDef.pipeline(), std::move(expCtx));
    if (!pipelineStatus.isOK()) {
        return pipelineStatus.getStatus();
    }

    // Validate that the view pipeline does not contain any ineligible stages.
    auto sources = pipelineStatus.getValue()->getSources();
    if (!sources.empty() && sources.front()->constraints().isChangeStreamStage()) {
        return {ErrorCodes::OptionNotSupportedOnView,
                "$changeStream cannot be used in a view definition"};
    }

    return std::move(involvedNamespaces);
}

Status ViewCatalog::_upsertIntoGraph(OperationContext* txn, const ViewDefinition& viewDef) {

    // Performs the insert into the graph.
    auto doInsert = [this, &txn](const ViewDefinition& viewDef, bool needsValidation) -> Status {
        // Parse the pipeline for this view to get the namespaces it references.
        AggregationRequest request(viewDef.viewOn(), viewDef.pipeline());
        boost::intrusive_ptr<ExpressionContext> expCtx = new ExpressionContext(txn, request);
        auto pipelineStatus = Pipeline::parse(viewDef.pipeline(), expCtx);
        if (!pipelineStatus.isOK()) {
            uassert(40255,
                    str::stream() << "Invalid pipeline for existing view " << viewDef.name().ns()
                                  << "; "
                                  << pipelineStatus.getStatus().reason(),
                    !needsValidation);
            return pipelineStatus.getStatus();
        }

        std::vector<NamespaceString> refs = pipelineStatus.getValue()->getInvolvedCollections();
        refs.push_back(viewDef.viewOn());

        if (needsValidation) {
            return _viewGraph.insertAndValidate(viewDef.name(), refs);
        } else {
            _viewGraph.insertWithoutValidating(viewDef.name(), refs);
            return Status::OK();
        }
    };

    if (_viewGraphNeedsRefresh) {
        _viewGraph.clear();
        for (auto&& iter : _viewMap) {
            auto status = doInsert(*(iter.second.get()), false);
            // If we cannot fully refresh the graph, we will keep '_viewGraphNeedsRefresh' true.
            if (!status.isOK()) {
                return status;
            }
        }
        // Only if the inserts completed without error will we no longer need a refresh.
        _viewGraphNeedsRefresh = false;
    }

    // Remove the view definition first in case this is an update. If it is not in the graph, it
    // is simply a no-op.
    _viewGraph.remove(viewDef.name());

    return doInsert(viewDef, true);
}

Status ViewGraph::insertAndValidate(const ViewDefinition& view,
                                    const std::vector<NamespaceString>& refs,
                                    int pipelineSize) {
    insertWithoutValidating(view, refs, pipelineSize);

    // Perform validation on this newly inserted view. Note, if the graph was put in an invalid
    // state through unvalidated inserts (e.g. if the user manually edits system.views)
    // this may not necessarily be detected. We only check for errors introduced by this view.
    const auto& viewNss = view.name();
    uint64_t nodeId = _getNodeId(viewNss);

    // If the graph fails validation for any reason, the insert is automatically rolled back on
    // exiting this method.
    auto rollBackInsert = [&]() -> auto {
        remove(viewNss);
    };
    auto guard = MakeGuard(rollBackInsert);

    // Check for cycles and get the height of the children.
    StatsMap statsMap;
    std::vector<uint64_t> cycleVertices;
    cycleVertices.reserve(kMaxViewDepth);
    auto childRes = _validateChildren(nodeId, nodeId, 0, &statsMap, &cycleVertices);
    if (!childRes.isOK()) {
        return childRes;
    }

    // Subtract one since the child height includes the non-view leaf node(s).
    int childrenHeight = statsMap[nodeId].height - 1;

    int childrenSize = statsMap[nodeId].cumulativeSize;

    // Get the height of the parents to obtain the diameter through this node, as well as the size
    // of the pipeline to check if if the combined pipeline exceeds the max size.
    statsMap.clear();
    auto parentRes = _validateParents(nodeId, 0, &statsMap);
    if (!parentRes.isOK()) {
        return parentRes;
    }

    // Check the combined heights of the children and parents.
    int parentsHeight = statsMap[nodeId].height;
    // Subtract one since the parent and children heights include the current node.
    int diameter = parentsHeight + childrenHeight - 1;

    if (diameter > kMaxViewDepth) {
        return {ErrorCodes::ViewDepthLimitExceeded,
                str::stream() << "View depth limit exceeded; maximum depth is " << kMaxViewDepth};
    }

    // Check the combined sizes of the children and parents.
    int parentsSize = statsMap[nodeId].cumulativeSize;
    // Subtract the current node's size since the parent and children sizes include the current
    // node.
    const Node& currentNode = _graph[nodeId];
    int pipelineTotalSize = parentsSize + childrenSize - currentNode.size;

    if (pipelineTotalSize > kMaxViewPipelineSizeBytes) {
        return {ErrorCodes::ViewPipelineMaxSizeExceeded,
                str::stream() << "Operation would result in a resolved view pipeline that exceeds "
                                 "the maximum size of "
                              << kMaxViewPipelineSizeBytes
                              << " bytes"};
    }

    guard.Dismiss();
    return Status::OK();
}