SERVER-27761 track collation in ViewGraphr3.4.4-rc0

Keeps track of the collation of views in the ViewGraph, to ensure that all views in the same
connected component share the same collation. This prohibits users from dropping a view and
recreating it with a different collation if other views depend on it.

(cherry picked from commit 476c7733f5dc01011e1ebd9d7818aad8d63a1781)

 Conflicts:
	jstests/core/views/views_collation.js
	src/mongo/db/views/SConscript
	src/mongo/db/views/view_graph.cpp

6 files changed, 491 insertions, 107 deletions

diff --git a/jstests/views/views_collation.js b/jstests/views/views_collation.js
index 04abf106292..915d06d31c4 100644
--- a/jstests/views/views_collation.js
+++ b/jstests/views/views_collation.js
@@ -266,6 +266,69 @@
             viewsDB.runCommand(
                 {collMod: "esView", viewOn: "simpleCollection", pipeline: [graphLookupFilView]}),
             ErrorCodes.OptionNotSupportedOnView);
+
+        // Views cannot be dropped and recreated with a different collation if other views depend on
+        // that view.
+        assert.commandWorked(viewsDB.runCommand(
+            {create: "filView2", viewOn: "filView", collation: {locale: "fil"}}));
+        assert.commandWorked(viewsDB.runCommand({drop: "filView"}));
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({create: "filView", viewOn: "simpleCollection"}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandFailedWithCode(
+            viewsDB.runCommand(
+                {create: "filView", viewOn: "simpleCollection", collation: {locale: "en"}}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandWorked(
+            viewsDB.createView("filView", "ukCollection", [], {collation: {locale: "fil"}}));
+
+        // Views cannot be dropped and recreated with a different collation if other views depend on
+        // that view via $lookup or $graphLookup.
+        assert.commandWorked(viewsDB.runCommand(
+            {collMod: "filView2", viewOn: "simpleCollection", pipeline: [lookupFilView]}));
+        assert.commandWorked(viewsDB.runCommand({drop: "filView"}));
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({create: "filView", viewOn: "simpleCollection"}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandFailedWithCode(
+            viewsDB.runCommand(
+                {create: "filView", viewOn: "simpleCollection", collation: {locale: "en"}}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandWorked(viewsDB.runCommand(
+            {create: "filView", viewOn: "ukCollection", pipeline: [], collation: {locale: "fil"}}));
+
+        assert.commandWorked(viewsDB.runCommand(
+            {collMod: "filView2", viewOn: "simpleCollection", pipeline: [graphLookupFilView]}));
+        assert.commandWorked(viewsDB.runCommand({drop: "filView"}));
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({create: "filView", viewOn: "simpleCollection"}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandFailedWithCode(
+            viewsDB.runCommand(
+                {create: "filView", viewOn: "simpleCollection", collation: {locale: "en"}}),
+            ErrorCodes.OptionNotSupportedOnView);
+
+        // If two views "A" and "C" have different collations and depend on the namespace "B", then
+        // "B"
+        // cannot be created as a view.
+        assert.commandWorked(
+            viewsDB.runCommand({create: "A", viewOn: "B", collation: {locale: "hsb"}}));
+        assert.commandWorked(
+            viewsDB.runCommand({create: "B", viewOn: "other", collation: {locale: "hsb"}}));
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({create: "C", viewOn: "B", collation: {locale: "wae"}}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandWorked(viewsDB.runCommand({drop: "B"}));
+        assert.commandWorked(
+            viewsDB.runCommand({create: "C", viewOn: "B", collation: {locale: "wae"}}));
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({create: "B", viewOn: "other", collation: {locale: "hsb"}}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({create: "B", viewOn: "other", collation: {locale: "wae"}}),
+            ErrorCodes.OptionNotSupportedOnView);
+        assert.commandFailedWithCode(viewsDB.runCommand({create: "B", viewOn: "other"}),
+                                     ErrorCodes.OptionNotSupportedOnView);
     }
 
     // Run the test on a standalone.
diff --git a/src/mongo/db/views/SConscript b/src/mongo/db/views/SConscript
index 9a70543f300..e016301729f 100644
--- a/src/mongo/db/views/SConscript
+++ b/src/mongo/db/views/SConscript
@@ -37,6 +37,7 @@ env.CppUnitTest(
     target='view_catalog_test',
     source=[
         'view_catalog_test.cpp',
+        'view_graph_test.cpp',
     ],
     LIBDEPS=[
         'views',
diff --git a/src/mongo/db/views/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
index d8e0e842824..e0f3c0beb5e 100644
--- a/src/mongo/db/views/view_catalog.cpp
+++ b/src/mongo/db/views/view_catalog.cpp
@@ -198,9 +198,9 @@ Status ViewCatalog::_upsertIntoGraph(OperationContext* txn, const ViewDefinition
             if (!collationStatus.isOK()) {
                 return collationStatus;
             }
-            return _viewGraph.insertAndValidate(viewDef.name(), refs, pipelineSize);
+            return _viewGraph.insertAndValidate(viewDef, refs, pipelineSize);
         } else {
-            _viewGraph.insertWithoutValidating(viewDef.name(), refs, pipelineSize);
+            _viewGraph.insertWithoutValidating(viewDef, refs, pipelineSize);
             return Status::OK();
         }
     };
diff --git a/src/mongo/db/views/view_graph.cpp b/src/mongo/db/views/view_graph.cpp
index 05037db7b03..fc40fc4720a 100644
--- a/src/mongo/db/views/view_graph.cpp
+++ b/src/mongo/db/views/view_graph.cpp
@@ -30,7 +30,9 @@
 
 #include "mongo/db/views/view_graph.h"
 
+#include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/db/views/view.h"
+#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 
@@ -46,50 +48,33 @@ void ViewGraph::clear() {
     _namespaceIds.clear();
 }
 
-Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
+size_t ViewGraph::size() const {
+    return _graph.size();
+}
+
+Status ViewGraph::insertAndValidate(const ViewDefinition& view,
                                     const std::vector<NamespaceString>& refs,
-                                    const int pipelineSize) {
-    insertWithoutValidating(viewNss, refs, pipelineSize);
+                                    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);
 
-    auto viewDepthLimitExceeded = [this, &viewNss]() -> Status {
-        this->remove(viewNss);
-        return Status(ErrorCodes::ViewDepthLimitExceeded,
-                      str::stream() << "View depth limit exceeded; maximum depth is "
-                                    << kMaxViewDepth);
-    };
-
-    auto viewPipelineMaxSizeExceeded = [this, &viewNss]() -> Status {
-        this->remove(viewNss);
-        return Status(ErrorCodes::ViewPipelineMaxSizeExceeded,
-                      str::stream() << "View pipeline exceeds maximum size; maximum size is "
-                                    << ViewGraph::kMaxViewPipelineSizeBytes);
-    };
+    // If the graph fails validation for any reason, the insert is automatically rolled back on
+    // exiting this method.
+    auto rollBackInsert = [&]() -> void { remove(viewNss); };
+    auto guard = MakeGuard(rollBackInsert);
 
     // Check for cycles and get the height of the children.
     StatsMap statsMap;
     std::vector<uint64_t> cycleVertices;
-    ErrorCodes::Error childRes =
-        _getChildrenStatsAndCheckCycle(nodeId, nodeId, 0, &statsMap, &cycleVertices);
-    if (childRes == ErrorCodes::ViewDepthLimitExceeded) {
-        return viewDepthLimitExceeded();
-    } else if (childRes == ErrorCodes::ViewPipelineMaxSizeExceeded) {
-        return viewPipelineMaxSizeExceeded();
-    } else if (childRes == ErrorCodes::GraphContainsCycle) {
-        // Make the error message with the namespaces of the cycle and remove the node.
-        str::stream ss;
-        ss << "View cycle detected: ";
-        for (auto cycleIter = cycleVertices.rbegin(); cycleIter != cycleVertices.rend();
-             cycleIter++) {
-            ss << _graph[*cycleIter].ns << " => ";
-        }
-        ss << viewNss.ns();
-        remove(viewNss);
-        return Status(ErrorCodes::GraphContainsCycle, ss);
+    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).
@@ -100,11 +85,9 @@ Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
     // 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();
-    ErrorCodes::Error parentRes = _getParentsStats(nodeId, 0, &statsMap);
-    if (parentRes == ErrorCodes::ViewDepthLimitExceeded) {
-        return viewDepthLimitExceeded();
-    } else if (parentRes == ErrorCodes::ViewPipelineMaxSizeExceeded) {
-        return viewPipelineMaxSizeExceeded();
+    auto parentRes = _validateParents(nodeId, 0, &statsMap);
+    if (!parentRes.isOK()) {
+        return parentRes;
     }
 
     // Check the combined heights of the children and parents.
@@ -113,7 +96,8 @@ Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
     int diameter = parentsHeight + childrenHeight - 1;
 
     if (diameter > kMaxViewDepth) {
-        return viewDepthLimitExceeded();
+        return {ErrorCodes::ViewDepthLimitExceeded,
+                str::stream() << "View depth limit exceeded; maximum depth is " << kMaxViewDepth};
     }
 
     // Check the combined sizes of the children and parents.
@@ -124,22 +108,30 @@ Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
     int pipelineTotalSize = parentsSize + childrenSize - currentNode.size;
 
     if (pipelineTotalSize > kMaxViewPipelineSizeBytes) {
-        return viewPipelineMaxSizeExceeded();
+        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();
 }
 
-void ViewGraph::insertWithoutValidating(const NamespaceString& viewNss,
+void ViewGraph::insertWithoutValidating(const ViewDefinition& view,
                                         const std::vector<NamespaceString>& refs,
-                                        const int pipelineSize) {
-    uint64_t nodeId = _getNodeId(viewNss);
+                                        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]);
-    node->size = pipelineSize;
     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);
@@ -172,8 +164,10 @@ void ViewGraph::remove(const NamespaceString& viewNss) {
         }
     }
 
-    // Remove all child pointers since this view no longer references anything.
+    // This node no longer represents a view, so its children must be cleared and its collator
+    // unset.
     node->children.clear();
+    node->collator = boost::none;
 
     // Only remove node if there are no remaining references to this node.
     if (node->parents.size() == 0) {
@@ -182,9 +176,7 @@ void ViewGraph::remove(const NamespaceString& viewNss) {
     }
 }
 
-ErrorCodes::Error ViewGraph::_getParentsStats(uint64_t currentId,
-                                              int currentDepth,
-                                              StatsMap* statsMap) {
+Status ViewGraph::_validateParents(uint64_t currentId, int currentDepth, StatsMap* statsMap) {
     const Node& currentNode = _graph[currentId];
     int maxHeightOfParents = 0;
     int maxSizeOfParents = 0;
@@ -192,13 +184,25 @@ ErrorCodes::Error ViewGraph::_getParentsStats(uint64_t currentId,
     // Return early if we've already exceeded the maximum depth. This will also be triggered if
     // we're traversing a cycle introduced through unvalidated inserts.
     if (currentDepth > kMaxViewDepth) {
-        return ErrorCodes::ViewDepthLimitExceeded;
+        return {ErrorCodes::ViewDepthLimitExceeded,
+                str::stream() << "View depth limit exceeded; maximum depth is "
+                              << ViewGraph::kMaxViewDepth};
     }
 
     for (uint64_t parentId : currentNode.parents) {
+        const auto& parentNode = _graph[parentId];
+        if (parentNode.isView() &&
+            !CollatorInterface::collatorsMatch(currentNode.collator.get(),
+                                               parentNode.collator.get())) {
+            return {ErrorCodes::OptionNotSupportedOnView,
+                    str::stream() << "View " << currentNode.ns
+                                  << " has a collation that does not match the collation of view "
+                                  << parentNode.ns};
+        }
+
         if (!(*statsMap)[parentId].checked) {
-            auto res = _getParentsStats(parentId, currentDepth + 1, statsMap);
-            if (res != ErrorCodes::OK) {
+            auto res = _validateParents(parentId, currentDepth + 1, statsMap);
+            if (!res.isOK()) {
                 return res;
             }
         }
@@ -210,51 +214,77 @@ ErrorCodes::Error ViewGraph::_getParentsStats(uint64_t currentId,
     (*statsMap)[currentId].height = maxHeightOfParents + 1;
     (*statsMap)[currentId].cumulativeSize += maxSizeOfParents + currentNode.size;
 
-    if ((*statsMap)[currentId].cumulativeSize > kMaxViewPipelineSizeBytes) {
-        return ErrorCodes::ViewPipelineMaxSizeExceeded;
+    const auto size = (*statsMap)[currentId].cumulativeSize;
+    if (size > kMaxViewPipelineSizeBytes) {
+        return {ErrorCodes::ViewPipelineMaxSizeExceeded,
+                str::stream() << "View pipeline is too large and exceeds the maximum size of "
+                              << ViewGraph::kMaxViewPipelineSizeBytes
+                              << " bytes"};
     }
 
-    return ErrorCodes::OK;
+    return Status::OK();
 }
 
-ErrorCodes::Error ViewGraph::_getChildrenStatsAndCheckCycle(uint64_t startingId,
-                                                            uint64_t currentId,
-                                                            int currentDepth,
-                                                            StatsMap* statsMap,
-                                                            std::vector<uint64_t>* cycleIds) {
-    // Check children of current node.
+Status ViewGraph::_validateChildren(uint64_t startingId,
+                                    uint64_t currentId,
+                                    int currentDepth,
+                                    StatsMap* statsMap,
+                                    std::vector<uint64_t>* traversalIds) {
     const Node& currentNode = _graph[currentId];
+    traversalIds->push_back(currentId);
+
+    // If we've encountered the id of the starting node, we've found a cycle in the graph.
     if (currentDepth > 0 && currentId == startingId) {
-        return ErrorCodes::GraphContainsCycle;
+        auto iterator = traversalIds->rbegin();
+        StringBuilder errmsg;
+
+        errmsg << "View cycle detected: ";
+        errmsg << _graph[*iterator].ns;
+        for (; iterator != traversalIds->rend(); ++iterator) {
+            errmsg << " => " << _graph[*iterator].ns;
+        }
+        return {ErrorCodes::GraphContainsCycle, errmsg.str()};
     }
 
     // Return early if we've already exceeded the maximum depth. This will also be triggered if
     // we're traversing a cycle introduced through unvalidated inserts.
     if (currentDepth > kMaxViewDepth) {
-        return ErrorCodes::ViewDepthLimitExceeded;
+        return {ErrorCodes::ViewDepthLimitExceeded,
+                str::stream() << "View depth limit exceeded; maximum depth is "
+                              << ViewGraph::kMaxViewDepth};
     }
 
     int maxHeightOfChildren = 0;
     int maxSizeOfChildren = 0;
     for (uint64_t childId : currentNode.children) {
-        if (!(*statsMap)[childId].checked) {
-            auto res = _getChildrenStatsAndCheckCycle(
-                startingId, childId, currentDepth + 1, statsMap, cycleIds);
-            if (res == ErrorCodes::GraphContainsCycle) {
-                cycleIds->push_back(currentId);
-                return res;
-            } else if (res != ErrorCodes::OK) {
-                return res;
-            }
+        if ((*statsMap)[childId].checked) {
+            continue;
         }
+
+        const auto& childNode = _graph[childId];
+        if (childNode.isView() &&
+            !CollatorInterface::collatorsMatch(currentNode.collator.get(),
+                                               childNode.collator.get())) {
+            return {ErrorCodes::OptionNotSupportedOnView,
+                    str::stream() << "View " << currentNode.ns
+                                  << " has a collation that does not match the collation of view "
+                                  << childNode.ns};
+        }
+
+        auto res = _validateChildren(startingId, childId, currentDepth + 1, statsMap, traversalIds);
+        if (!res.isOK()) {
+            return res;
+        }
+
         maxHeightOfChildren = std::max(maxHeightOfChildren, (*statsMap)[childId].height);
         maxSizeOfChildren = std::max(maxSizeOfChildren, (*statsMap)[childId].cumulativeSize);
     }
 
+    traversalIds->pop_back();
     (*statsMap)[currentId].checked = true;
     (*statsMap)[currentId].height = maxHeightOfChildren + 1;
     (*statsMap)[currentId].cumulativeSize += maxSizeOfChildren + currentNode.size;
-    return ErrorCodes::OK;
+    return Status::OK();
 }
 
 uint64_t ViewGraph::_getNodeId(const NamespaceString& nss) {
diff --git a/src/mongo/db/views/view_graph.h b/src/mongo/db/views/view_graph.h
index 7ff3029188a..8f0e3c8c140 100644
--- a/src/mongo/db/views/view_graph.h
+++ b/src/mongo/db/views/view_graph.h
@@ -27,10 +27,12 @@
 */
 #pragma once
 
+#include <boost/optional.hpp>
 #include <vector>
 
 #include "mongo/base/status.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/stdx/unordered_map.h"
 #include "mongo/stdx/unordered_set.h"
 #include "mongo/util/string_map.h"
@@ -39,9 +41,10 @@ namespace mongo {
 class ViewDefinition;
 
 /**
- * Validates that the graph of view dependencies is acyclic and within the allowed depth.
- * Each node is represented by an integer id, and stores integer ids for its parents and children
- * in the graph.
+ * Represents the dependencies of views on other namespaces and validates that this graph is acyclic
+ * and smaller than the maximum diameter. It also checks that the views in each connected component
+ * have compatible collations, and that all possible view pipelines stay within the maximum size in
+ * bytes.
  *
  * This is owned and managed by the ViewCatalog.
  */
@@ -55,22 +58,30 @@ public:
     ViewGraph() = default;
 
     /**
-     * Called when a view is added to the catalog. 'refs' are a list of namespaces that the view
-     * represented by 'viewNss' references in its viewOn or pipeline. Checks if this view introduces
-     * a cycle, max diameter, or max pipeline size. If an error is detected, it will not insert.
+     * Inserts a new node for 'view' in the graph, which must not already be present. 'refs'
+     * contains the list of namespaces referred to by the view in its "viewOn" or "pipeline".
+     * 'pipelineSize' is the size of the view's pipeline, in bytes. If inserting the view would
+     * violate one of the graph's validity properties, the insertion is reverted and a non-OK status
+     * is returned.
+     *
+     * This method is intended for validating a view that is created or modified by a user
+     * operation.
      */
-    Status insertAndValidate(const NamespaceString& viewNss,
+    Status insertAndValidate(const ViewDefinition& view,
                              const std::vector<NamespaceString>& refs,
-                             const int pipelineSize);
+                             int pipelineSize);
 
     /**
-     * Called when view definitions are being reloaded from the catalog (e.g. on restart of mongod).
-     * Does the same as insertAndValidate except does not check for cycles, max diameter, or max
-     * pipeline size.
+     * Like insertAndValidate(), inserts a new node for 'view' in the graph, which must not already
+     * be present. However, no validation is performed. The insertion is not rolled back even if it
+     * puts the graph into an invalid state.
+     *
+     * This method is intended for quickly repopulating the graph with view definitions that are
+     * assumed to be already valid.
      */
-    void insertWithoutValidating(const NamespaceString& viewNss,
+    void insertWithoutValidating(const ViewDefinition& view,
                                  const std::vector<NamespaceString>& refs,
-                                 const int pipelineSize);
+                                 int pipelineSize);
 
     /**
      * Called when a view is removed from the catalog. If the view does not exist in the graph it is
@@ -83,17 +94,43 @@ public:
      */
     void clear();
 
+    /**
+     * Returns the number of namespaces tracked by the view graph. Only exposed for testing.
+     */
+    size_t size() const;
+
 private:
-    // A graph node represents a namespace. The parent-child relation is defined as a parent
-    // references the child either through viewOn or in $lookup/$graphLookup/$facet in its pipeline.
-    // E.g. the command {create: "a", viewOn: "b", pipeline: [{$lookup: {from: "c"}}]}
-    // means the node for "a" is a parent of nodes for "b" and "c" since it references them.
+    // A graph node represents a namespace. We say that a node A is a parent of B if A is a view and
+    // it references B via its "viewOn" or $lookup/$graphLookup/$facet in its "pipeline".
+    //
+    // This node represents a view namespace if and only if 'children' is nonempty and 'collator' is
+    // set.
     struct Node {
-        // Note, a view may refer to the same child more than once, but we only need to know the
-        // set of children and parents, since we do not need to traverse duplicates.
-        stdx::unordered_set<uint64_t> parents;
-        stdx::unordered_set<uint64_t> children;
+        /**
+         * Returns true if this node represents a view.
+         */
+        bool isView() const {
+            invariant(children.empty() == !static_cast<bool>(collator));
+            return !children.empty();
+        }
+
+        // The fully-qualified namespace that this node represents.
         std::string ns;
+
+        // Represents the namespaces depended on by this view. A view may refer to the same child
+        // more than once, but we store the children as a set because each namespace need not be
+        // traversed more than once.
+        stdx::unordered_set<uint64_t> children;
+
+        // Represents the views that depend on this namespace.
+        stdx::unordered_set<uint64_t> parents;
+
+        // When set, this is an unowned pointer to the view's collation, or nullptr if the view has
+        // the binary collation. When not set, this namespace is not a view and we don't care about
+        // its collator.
+        boost::optional<const CollatorInterface*> collator;
+
+        // The size of this view's "pipeline", in bytes.
         int size = 0;
     };
 
@@ -107,21 +144,28 @@ private:
     using StatsMap = stdx::unordered_map<uint64_t, NodeStats>;
 
     /**
-     * Recursively traverses parents of this node and computes their heights and sizes. Returns an
-     * error if the maximum depth is exceeded or the pipeline size exceeds the max.
+     * Recursively validates the parents of 'currentId', filling out statistics about the node
+     * represented by that id in 'statsMap'. The recursion level is tracked in 'currentDepth' to
+     * limit recursive calls.
+     *
+     * A non-OK status is returned if 'currentId' or its parents violate any of the graph's
+     * validity properties.
      */
-    ErrorCodes::Error _getParentsStats(uint64_t currentId, int currentDepth, StatsMap* statsMap);
+    Status _validateParents(uint64_t currentId, int currentDepth, StatsMap* statsMap);
 
     /**
-     * Recursively traverses children of the starting node and computes their heights and sizes.
-     * Returns an error if the maximum depth is exceeded, a cycle is detected through the starting
-     * node, or the pipeline size exceeds the max.
+     * Recursively validates the children of 'currentId', filling out statistics about the node
+     * represented by that id in 'statsMap'. The recursion level is tracked in 'currentDepth' to
+     * limit recursive calls. Both 'startingId' and 'traversalIds' are used to detect cycles.
+     *
+     * A non-OK status is returned if 'currentId' or its children violate any of the graph's
+     * validity properties.
      */
-    ErrorCodes::Error _getChildrenStatsAndCheckCycle(uint64_t startingId,
-                                                     uint64_t currentId,
-                                                     int currentDepth,
-                                                     StatsMap* statsMap,
-                                                     std::vector<uint64_t>* cycleIds);
+    Status _validateChildren(uint64_t startingId,
+                             uint64_t currentId,
+                             int currentDepth,
+                             StatsMap* statsMap,
+                             std::vector<uint64_t>* traversalIds);
 
     /**
      * Gets the id for this namespace, and creates an id if it doesn't exist.
@@ -132,7 +176,7 @@ private:
     // i.e. existing views, collections, and non-existing namespaces.
     StringMap<uint64_t> _namespaceIds;
 
-    // Maps node ids to nodes. There is a 1-1 correspondance with _namespaceIds, hence the lifetime
+    // Maps node ids to nodes. There is a 1-1 correspondence with _namespaceIds, hence the lifetime
     // of a node is the same as the lifetime as its corresponding node id.
     stdx::unordered_map<uint64_t, Node> _graph;
     static uint64_t _idCounter;
diff --git a/src/mongo/db/views/view_graph_test.cpp b/src/mongo/db/views/view_graph_test.cpp
new file mode 100644
index 00000000000..15638204c5c
--- /dev/null
+++ b/src/mongo/db/views/view_graph_test.cpp
@@ -0,0 +1,246 @@
+/**
+ * Copyright (C) 2017 MongoDB Inc.
+ *
+ * This program is free software: you can redistribute it and/or  modify
+ * it under the terms of the GNU Affero General Public License, version 3,
+ * as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * As a special exception, the copyright holders give permission to link the
+ * code of portions of this program with the OpenSSL library under certain
+ * conditions as described in each individual source file and distribute
+ * linked combinations including the program with the OpenSSL library. You
+ * must comply with the GNU Affero General Public License in all respects
+ * for all of the code used other than as permitted herein. If you modify
+ * file(s) with this exception, you may extend this exception to your
+ * version of the file(s), but you are not obligated to do so. If you do not
+ * wish to do so, delete this exception statement from your version. If you
+ * delete this exception statement from all source files in the program,
+ * then also delete it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/bson/bsonmisc.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/query/collation/collator_factory_interface.h"
+#include "mongo/db/query/query_test_service_context.h"
+#include "mongo/db/views/view.h"
+#include "mongo/db/views/view_graph.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+constexpr auto kEmptyPipelineSize = 0;
+constexpr auto kTestDb = "test"_sd;
+constexpr auto kFooName = "foo"_sd;
+constexpr auto kBarName = "bar"_sd;
+constexpr auto kQuxName = "qux"_sd;
+const auto kFooNamespace = NamespaceString(kTestDb, kFooName);
+const auto kBarNamespace = NamespaceString(kTestDb, kBarName);
+const auto kQuxNamespace = NamespaceString(kTestDb, kQuxName);
+const auto kEmptyPipeline = BSONArray();
+const auto kBinaryCollation = BSONObj();
+const auto kFilipinoCollation = BSON("locale"
+                                     << "fil");
+
+class ViewGraphFixture : public unittest::Test {
+public:
+    ViewGraphFixture()
+        : _queryServiceContext(), _opCtx(_queryServiceContext.makeOperationContext()) {}
+
+    const OperationContext* opCtx() const {
+        return _opCtx.get();
+    }
+
+    ViewGraph* viewGraph() {
+        return &_viewGraph;
+    }
+
+    ViewDefinition makeViewDefinition(StringData db,
+                                      StringData view,
+                                      StringData viewOn,
+                                      BSONArray pipeline,
+                                      BSONObj collatorSpec) const {
+        auto collator = std::unique_ptr<CollatorInterface>(nullptr);
+        if (!collatorSpec.isEmpty()) {
+            auto factoryCollator = CollatorFactoryInterface::get(_opCtx->getServiceContext())
+                                       ->makeFromBSON(collatorSpec);
+            ASSERT_OK(factoryCollator.getStatus());
+            collator = std::move(factoryCollator.getValue());
+        }
+
+        return {db, view, viewOn, pipeline, std::move(collator)};
+    }
+
+private:
+    QueryTestServiceContext _queryServiceContext;
+    ServiceContext::UniqueOperationContext _opCtx;
+    ViewGraph _viewGraph;
+};
+
+TEST_F(ViewGraphFixture, CanInsertViewsWithMatchingBinaryCollations) {
+    const auto fooView =
+        makeViewDefinition(kTestDb, kFooName, kBarName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(fooView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barView =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(barView, {kQuxNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 3UL);
+}
+
+TEST_F(ViewGraphFixture, CanInsertViewsWithMatchingNonTrivialCollations) {
+    const auto fooView =
+        makeViewDefinition(kTestDb, kFooName, kBarName, kEmptyPipeline, kFilipinoCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(fooView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barView =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kFilipinoCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(barView, {kQuxNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 3UL);
+}
+
+TEST_F(ViewGraphFixture, CannotInsertViewsWithNonMatchingCollations) {
+    const auto fooView =
+        makeViewDefinition(kTestDb, kFooName, kBarName, kEmptyPipeline, kFilipinoCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(fooView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barView =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_EQ(viewGraph()->insertAndValidate(barView, {kQuxNamespace}, kEmptyPipelineSize),
+              ErrorCodes::OptionNotSupportedOnView);
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+}
+
+TEST_F(ViewGraphFixture, CannotRecreateViewWithDifferentCollationIfDependedOnByOtherViews) {
+    const auto fooView =
+        makeViewDefinition(kTestDb, kFooName, kBarName, kEmptyPipeline, kFilipinoCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(fooView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barView =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kFilipinoCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(barView, {kQuxNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 3UL);
+
+    viewGraph()->remove(kBarNamespace);
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barViewBinaryCollation =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_EQ(
+        viewGraph()->insertAndValidate(barViewBinaryCollation, {kQuxNamespace}, kEmptyPipelineSize),
+        ErrorCodes::OptionNotSupportedOnView);
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+}
+
+// Tests that an insertion that would create a view cycle is rejected, but will later be accepted if
+// the cycle is broken by removing another existing view.
+TEST_F(ViewGraphFixture, CanCreateViewThatReferencesDroppedView) {
+    const auto fooView =
+        makeViewDefinition(kTestDb, kFooName, kBarName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(fooView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barView =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(barView, {kQuxNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 3UL);
+
+    const auto quxView =
+        makeViewDefinition(kTestDb, kQuxName, kBarName, kEmptyPipeline, kBinaryCollation);
+
+    // Inserting qux should fail, as it cycles with bar.
+    ASSERT_EQ(viewGraph()->insertAndValidate(quxView, {kBarNamespace}, kEmptyPipelineSize),
+              ErrorCodes::GraphContainsCycle);
+    ASSERT_EQ(viewGraph()->size(), 3UL);
+
+    viewGraph()->remove(kBarNamespace);
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    // With bar removed, we expect qux to be inserted successfully.
+    ASSERT_OK(viewGraph()->insertAndValidate(quxView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 3UL);
+}
+
+// Tests that an insertion that would create mismatching collators is rejected, but will later be
+// accepted if the existing view with the conflicting collator is removed.
+TEST_F(ViewGraphFixture, CanCreateViewWithDifferentCollationThanDroppedView) {
+    const auto fooView =
+        makeViewDefinition(kTestDb, kFooName, kBarName, kEmptyPipeline, kFilipinoCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(fooView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barView =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kBinaryCollation);
+
+    // Inserting bar should fail, as foo depends on bar and has a different collation.
+    ASSERT_EQ(viewGraph()->insertAndValidate(barView, {kQuxNamespace}, kEmptyPipelineSize),
+              ErrorCodes::OptionNotSupportedOnView);
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    viewGraph()->remove(kFooNamespace);
+    ASSERT_EQ(viewGraph()->size(), 0UL);
+
+    // Now bar should be inserted successfully, as there are no existing views in the graph that
+    // depend on it.
+    ASSERT_OK(viewGraph()->insertAndValidate(barView, {kFooNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+}
+
+// Tests that a node in the graph is properly converted from a "view" node to a "non-view" node when
+// a view with that namespace is removed.
+TEST_F(ViewGraphFixture, DroppingViewPreservesNodeInGraphIfDependedOnByOtherViews) {
+    const auto fooView =
+        makeViewDefinition(kTestDb, kFooName, kBarName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(fooView, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 2UL);
+
+    const auto barView =
+        makeViewDefinition(kTestDb, kBarName, kQuxName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(barView, {kQuxNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 3UL);
+
+    // Inserts baz into the graph so that qux has another namespace that depends on it. This way,
+    // the node for qux won't be destroyed when baz is removed.
+    const auto bazView =
+        makeViewDefinition(kTestDb, "baz"_sd, kQuxName, kEmptyPipeline, kBinaryCollation);
+    ASSERT_OK(viewGraph()->insertAndValidate(bazView, {kQuxNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 4UL);
+
+    // Inserting a view that depends on bar but has a different collation should fail.
+    const auto viewWithDifferentCollation = makeViewDefinition(
+        kTestDb, "badCollation"_sd, kBarName, kEmptyPipeline, kFilipinoCollation);
+    ASSERT_EQ(viewGraph()->insertAndValidate(
+                  viewWithDifferentCollation, {kBarNamespace}, kEmptyPipelineSize),
+              ErrorCodes::OptionNotSupportedOnView);
+    ASSERT_EQ(viewGraph()->size(), 4UL);
+
+    // Removes bar from the graph. The graph's size should remain at 4, since bar is still depended
+    // on by qux.
+    viewGraph()->remove(kBarNamespace);
+    ASSERT_EQ(viewGraph()->size(), 4UL);
+
+    // Inserting the viewWithDifferentCollation from above should now succeed, since bar is no
+    // longer a view.
+    ASSERT_OK(viewGraph()->insertAndValidate(
+        viewWithDifferentCollation, {kBarNamespace}, kEmptyPipelineSize));
+    ASSERT_EQ(viewGraph()->size(), 5UL);
+}
+}  // namespace
+}  // namespace mongo