SERVER-25444 Enforce max size of view pipeline

4 files changed, 192 insertions, 48 deletions

diff --git a/src/mongo/db/views/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
index a8be7f54290..68b142658fa 100644
--- a/src/mongo/db/views/view_catalog.cpp
+++ b/src/mongo/db/views/view_catalog.cpp
@@ -37,6 +37,7 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/base/string_data.h"
+#include "mongo/bson/util/builder.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/pipeline/aggregation_request.h"
@@ -48,6 +49,7 @@
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/views/resolved_view.h"
 #include "mongo/db/views/view.h"
+#include "mongo/db/views/view_graph.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -169,15 +171,20 @@ Status ViewCatalog::_upsertIntoGraph(OperationContext* txn, const ViewDefinition
         std::vector<NamespaceString> refs = pipelineStatus.getValue()->getInvolvedCollections();
         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(txn, viewDef, refs);
             if (!collationStatus.isOK()) {
                 return collationStatus;
             }
-            return _viewGraph.insertAndValidate(viewDef.name(), refs);
+            return _viewGraph.insertAndValidate(viewDef.name(), refs, pipelineSize);
         } else {
-            _viewGraph.insertWithoutValidating(viewDef.name(), refs);
+            _viewGraph.insertWithoutValidating(viewDef.name(), refs, pipelineSize);
             return Status::OK();
         }
     };
@@ -355,8 +362,19 @@ StatusWith<ResolvedView> ViewCatalog::resolveView(OperationContext* txn,
 
     for (int i = 0; i < ViewGraph::kMaxViewDepth; i++) {
         auto view = _lookup_inlock(txn, resolvedNss->ns());
-        if (!view)
+        if (!view) {
+            // Return error status if pipeline is too large.
+            int pipelineSize = 0;
+            for (auto obj : resolvedPipeline) {
+                pipelineSize += obj.objsize();
+            }
+            if (pipelineSize > ViewGraph::kMaxViewPipelineSizeBytes) {
+                return {ErrorCodes::ViewPipelineMaxSizeExceeded,
+                        str::stream() << "View pipeline exceeds maximum size; maximum size is "
+                                      << ViewGraph::kMaxViewPipelineSizeBytes};
+            }
             return StatusWith<ResolvedView>({*resolvedNss, resolvedPipeline});
+        }
 
         resolvedNss = &(view->viewOn());
 
diff --git a/src/mongo/db/views/view_catalog_test.cpp b/src/mongo/db/views/view_catalog_test.cpp
index 77f2da66ad6..b780beb83be 100644
--- a/src/mongo/db/views/view_catalog_test.cpp
+++ b/src/mongo/db/views/view_catalog_test.cpp
@@ -210,6 +210,83 @@ TEST_F(ViewCatalogFixture, CreateViewCycles) {
     }
 }
 
+TEST_F(ViewCatalogFixture, CreateViewWithPipelineExactMaxSize) {
+    BSONArrayBuilder builder;
+    int objsize = BSON("$match" << BSON("x"
+                                        << "foobar"))
+                      .objsize();
+
+    int pipelineSize = 0;
+
+    for (; pipelineSize < ViewGraph::kMaxViewPipelineSizeBytes; pipelineSize += objsize) {
+        builder << BSON("$match" << BSON("x"
+                                         << "foobar"));
+    }
+
+    ASSERT_EQ(pipelineSize, ViewGraph::kMaxViewPipelineSizeBytes);
+
+    const NamespaceString viewName("db.view");
+    const NamespaceString viewOn("db.coll");
+    const BSONObj collation;
+
+    auto pipeline = builder.arr();
+
+    ASSERT_OK(viewCatalog.createView(opCtx.get(), viewName, viewOn, pipeline, collation));
+}
+
+TEST_F(ViewCatalogFixture, CreateViewWithPipelineExceedingMaxSize) {
+    BSONArrayBuilder builder;
+
+    int objsize = BSON("$match" << BSON("x"
+                                        << "foo"))
+                      .objsize();
+
+    int pipelineSize = 0;
+
+    for (; pipelineSize < ViewGraph::kMaxViewPipelineSizeBytes + 1; pipelineSize += objsize) {
+        builder << BSON("$match" << BSON("x"
+                                         << "foo"));
+    }
+
+    const NamespaceString viewName("db.view");
+    const NamespaceString viewOn("db.coll");
+    const BSONObj collation;
+
+    auto pipeline = builder.arr();
+
+    ASSERT_NOT_OK(viewCatalog.createView(opCtx.get(), viewName, viewOn, pipeline, collation));
+}
+
+TEST_F(ViewCatalogFixture, CreateViewWithCumulativePipelineExceedingMaxSize) {
+    BSONArrayBuilder builder1;
+    BSONArrayBuilder builder2;
+
+    int objsize = BSON("$match" << BSON("x"
+                                        << "foo"))
+                      .objsize();
+
+    int pipelineSize = 0;
+
+    for (; pipelineSize < ViewGraph::kMaxViewPipelineSizeBytes + 1; pipelineSize += objsize * 2) {
+        builder1 << BSON("$match" << BSON("x"
+                                          << "foo"));
+        builder2 << BSON("$match" << BSON("x"
+                                          << "foo"));
+    }
+
+    auto pipeline1 = builder1.arr();
+    auto pipeline2 = builder2.arr();
+
+    const NamespaceString view1("db.view1");
+    const NamespaceString view2("db.view2");
+    const NamespaceString viewOn("db.coll");
+    const BSONObj collation1;
+    const BSONObj collation2;
+
+    ASSERT_OK(viewCatalog.createView(opCtx.get(), view1, viewOn, pipeline1, collation1));
+    ASSERT_NOT_OK(viewCatalog.createView(opCtx.get(), view2, view1, pipeline2, collation2));
+}
+
 TEST_F(ViewCatalogFixture, DropMissingView) {
     NamespaceString viewName("db.view");
     ASSERT_NOT_OK(viewCatalog.dropView(opCtx.get(), viewName));
diff --git a/src/mongo/db/views/view_graph.cpp b/src/mongo/db/views/view_graph.cpp
index 1f8d4b667e2..05037db7b03 100644
--- a/src/mongo/db/views/view_graph.cpp
+++ b/src/mongo/db/views/view_graph.cpp
@@ -34,6 +34,10 @@
 
 namespace mongo {
 
+// Leave room for view name and type in documents returned from listCollections, or an actual query
+// on a sharded system.
+const int ViewGraph::kMaxViewPipelineSizeBytes = 16 * 1000 * 1000;
+
 const int ViewGraph::kMaxViewDepth = 20;
 uint64_t ViewGraph::_idCounter = 0;
 
@@ -43,8 +47,9 @@ void ViewGraph::clear() {
 }
 
 Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
-                                    const std::vector<NamespaceString>& refs) {
-    insertWithoutValidating(viewNss, refs);
+                                    const std::vector<NamespaceString>& refs,
+                                    const int pipelineSize) {
+    insertWithoutValidating(viewNss, 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)
@@ -58,13 +63,22 @@ Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
                                     << 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);
+    };
+
     // Check for cycles and get the height of the children.
-    HeightMap heightMap;
+    StatsMap statsMap;
     std::vector<uint64_t> cycleVertices;
     ErrorCodes::Error childRes =
-        _getChildrenHeightAndCheckCycle(nodeId, nodeId, 0, &heightMap, &cycleVertices);
+        _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;
@@ -79,17 +93,22 @@ Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
     }
 
     // Subtract one since the child height includes the non-view leaf node(s).
-    int childrenHeight = heightMap[nodeId].height - 1;
+    int childrenHeight = statsMap[nodeId].height - 1;
+
+    int childrenSize = statsMap[nodeId].cumulativeSize;
 
-    // Get the height of the parents to obtain the diameter through this node.
-    heightMap.clear();
-    ErrorCodes::Error parentRes = _getParentsHeight(nodeId, 0, &heightMap);
+    // 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();
     }
 
     // Check the combined heights of the children and parents.
-    int parentsHeight = heightMap[nodeId].height;
+    int parentsHeight = statsMap[nodeId].height;
     // Subtract one since the parent and children heights include the current node.
     int diameter = parentsHeight + childrenHeight - 1;
 
@@ -97,16 +116,29 @@ Status ViewGraph::insertAndValidate(const NamespaceString& viewNss,
         return viewDepthLimitExceeded();
     }
 
+    // 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 viewPipelineMaxSizeExceeded();
+    }
+
     return Status::OK();
 }
 
 void ViewGraph::insertWithoutValidating(const NamespaceString& viewNss,
-                                        const std::vector<NamespaceString>& refs) {
+                                        const std::vector<NamespaceString>& refs,
+                                        const int pipelineSize) {
     uint64_t nodeId = _getNodeId(viewNss);
     // 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());
 
     for (const NamespaceString& childNss : refs) {
@@ -150,11 +182,12 @@ void ViewGraph::remove(const NamespaceString& viewNss) {
     }
 }
 
-ErrorCodes::Error ViewGraph::_getParentsHeight(uint64_t currentId,
-                                               int currentDepth,
-                                               HeightMap* heightMap) {
+ErrorCodes::Error ViewGraph::_getParentsStats(uint64_t currentId,
+                                              int currentDepth,
+                                              StatsMap* statsMap) {
     const Node& currentNode = _graph[currentId];
     int maxHeightOfParents = 0;
+    int maxSizeOfParents = 0;
 
     // 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.
@@ -163,26 +196,32 @@ ErrorCodes::Error ViewGraph::_getParentsHeight(uint64_t currentId,
     }
 
     for (uint64_t parentId : currentNode.parents) {
-        if (!(*heightMap)[parentId].checked) {
-            auto res = _getParentsHeight(parentId, currentDepth + 1, heightMap);
+        if (!(*statsMap)[parentId].checked) {
+            auto res = _getParentsStats(parentId, currentDepth + 1, statsMap);
             if (res != ErrorCodes::OK) {
                 return res;
             }
         }
-        maxHeightOfParents = std::max(maxHeightOfParents, (*heightMap)[parentId].height);
+        maxHeightOfParents = std::max(maxHeightOfParents, (*statsMap)[parentId].height);
+        maxSizeOfParents = std::max(maxSizeOfParents, (*statsMap)[parentId].cumulativeSize);
     }
 
-    (*heightMap)[currentId].checked = true;
-    (*heightMap)[currentId].height = maxHeightOfParents + 1;
+    (*statsMap)[currentId].checked = true;
+    (*statsMap)[currentId].height = maxHeightOfParents + 1;
+    (*statsMap)[currentId].cumulativeSize += maxSizeOfParents + currentNode.size;
+
+    if ((*statsMap)[currentId].cumulativeSize > kMaxViewPipelineSizeBytes) {
+        return ErrorCodes::ViewPipelineMaxSizeExceeded;
+    }
 
     return ErrorCodes::OK;
 }
 
-ErrorCodes::Error ViewGraph::_getChildrenHeightAndCheckCycle(uint64_t startingId,
-                                                             uint64_t currentId,
-                                                             int currentDepth,
-                                                             HeightMap* heightMap,
-                                                             std::vector<uint64_t>* cycleIds) {
+ErrorCodes::Error ViewGraph::_getChildrenStatsAndCheckCycle(uint64_t startingId,
+                                                            uint64_t currentId,
+                                                            int currentDepth,
+                                                            StatsMap* statsMap,
+                                                            std::vector<uint64_t>* cycleIds) {
     // Check children of current node.
     const Node& currentNode = _graph[currentId];
     if (currentDepth > 0 && currentId == startingId) {
@@ -196,10 +235,11 @@ ErrorCodes::Error ViewGraph::_getChildrenHeightAndCheckCycle(uint64_t startingId
     }
 
     int maxHeightOfChildren = 0;
+    int maxSizeOfChildren = 0;
     for (uint64_t childId : currentNode.children) {
-        if (!(*heightMap)[childId].checked) {
-            auto res = _getChildrenHeightAndCheckCycle(
-                startingId, childId, currentDepth + 1, heightMap, cycleIds);
+        if (!(*statsMap)[childId].checked) {
+            auto res = _getChildrenStatsAndCheckCycle(
+                startingId, childId, currentDepth + 1, statsMap, cycleIds);
             if (res == ErrorCodes::GraphContainsCycle) {
                 cycleIds->push_back(currentId);
                 return res;
@@ -207,11 +247,13 @@ ErrorCodes::Error ViewGraph::_getChildrenHeightAndCheckCycle(uint64_t startingId
                 return res;
             }
         }
-        maxHeightOfChildren = std::max(maxHeightOfChildren, (*heightMap)[childId].height);
+        maxHeightOfChildren = std::max(maxHeightOfChildren, (*statsMap)[childId].height);
+        maxSizeOfChildren = std::max(maxSizeOfChildren, (*statsMap)[childId].cumulativeSize);
     }
 
-    (*heightMap)[currentId].checked = true;
-    (*heightMap)[currentId].height = maxHeightOfChildren + 1;
+    (*statsMap)[currentId].checked = true;
+    (*statsMap)[currentId].height = maxHeightOfChildren + 1;
+    (*statsMap)[currentId].cumulativeSize += maxSizeOfChildren + currentNode.size;
     return ErrorCodes::OK;
 }
 
diff --git a/src/mongo/db/views/view_graph.h b/src/mongo/db/views/view_graph.h
index 65b192280ce..7ff3029188a 100644
--- a/src/mongo/db/views/view_graph.h
+++ b/src/mongo/db/views/view_graph.h
@@ -50,23 +50,27 @@ class ViewGraph {
 
 public:
     static const int kMaxViewDepth;
+    static const int kMaxViewPipelineSizeBytes;
 
     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 or max diameter. If an error is detected, it will not insert.
+     * a cycle, max diameter, or max pipeline size. If an error is detected, it will not insert.
      */
     Status insertAndValidate(const NamespaceString& viewNss,
-                             const std::vector<NamespaceString>& refs);
+                             const std::vector<NamespaceString>& refs,
+                             const 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 or max diameter.
+     * Does the same as insertAndValidate except does not check for cycles, max diameter, or max
+     * pipeline size.
      */
     void insertWithoutValidating(const NamespaceString& viewNss,
-                                 const std::vector<NamespaceString>& refs);
+                                 const std::vector<NamespaceString>& refs,
+                                 const int pipelineSize);
 
     /**
      * Called when a view is removed from the catalog. If the view does not exist in the graph it is
@@ -90,31 +94,34 @@ private:
         stdx::unordered_set<uint64_t> parents;
         stdx::unordered_set<uint64_t> children;
         std::string ns;
+        int size = 0;
     };
 
     // Bookkeeping for graph traversals.
-    struct NodeHeight {
+    struct NodeStats {
         bool checked = false;
         int height = 0;
+        int cumulativeSize = 0;
     };
 
-    using HeightMap = stdx::unordered_map<uint64_t, NodeHeight>;
+    using StatsMap = stdx::unordered_map<uint64_t, NodeStats>;
 
     /**
-     * Recursively traverses parents of this node and computes their heights. Returns an error
-     * if the maximum depth is exceeded.
+     * 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.
      */
-    ErrorCodes::Error _getParentsHeight(uint64_t currentId, int currentDepth, HeightMap* heightMap);
+    ErrorCodes::Error _getParentsStats(uint64_t currentId, int currentDepth, StatsMap* statsMap);
 
     /**
-     * Recursively traverses children of the starting node and computes their heights. Returns an
-     * error if the maximum depth is exceeded or a cycle is detected through the starting node.
+     * 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.
      */
-    ErrorCodes::Error _getChildrenHeightAndCheckCycle(uint64_t startingId,
-                                                      uint64_t currentId,
-                                                      int currentDepth,
-                                                      HeightMap* heightMap,
-                                                      std::vector<uint64_t>* cycleIds);
+    ErrorCodes::Error _getChildrenStatsAndCheckCycle(uint64_t startingId,
+                                                     uint64_t currentId,
+                                                     int currentDepth,
+                                                     StatsMap* statsMap,
+                                                     std::vector<uint64_t>* cycleIds);
 
     /**
      * Gets the id for this namespace, and creates an id if it doesn't exist.