SERVER-40361 Don't store debug info once plan cache size grows large

Introduces a new setParameter,
'internalQueryCacheMaxSizeBytesBeforeStripDebugInfo'. When
the cumulative size of a mongod's plan caches exceeds this
threshold, additional plan cache entries are stored without
any debug info. This should help to prevent problems where
the plan caches collectively consume too much memory.

The default setting of the parameter is 0.5 GB, but it can
be configured by the operator at startup or at runtime.

(cherry picked from commit eeb4b8aaffbcbb236b2d02e35dad919b4fa0aa80)
(cherry picked from commit 65ad41f1df99bbdfabeb8235351d9c21f9eea142)
(cherry picked from commit 52b11e90efa467dbe6b55977e5d2239aba3f6ec4)
(cherry picked from commit e31f945ddc59c270ba61c44ca792f4d7058c1703)

13 files changed, 750 insertions, 277 deletions

diff --git a/jstests/noPassthrough/plan_cache_memory_debug_info.js b/jstests/noPassthrough/plan_cache_memory_debug_info.js
new file mode 100644
index 00000000000..31dab5aeba9
--- /dev/null
+++ b/jstests/noPassthrough/plan_cache_memory_debug_info.js
@@ -0,0 +1,270 @@
+/**
+ * Tests that detailed debug information is excluded from new plan cache entries once the estimated
+ * cumulative size of the system's plan caches exceeds a pre-configured threshold.
+ */
+(function() {
+    "use strict";
+
+    /**
+     * Creates two indexes for the given collection. In order for plans to be cached, there need to
+     * be at least two possible indexed plans.
+     */
+    function createIndexesForColl(coll) {
+        assert.commandWorked(coll.createIndex({a: 1}));
+        assert.commandWorked(coll.createIndex({b: 1}));
+    }
+
+    function totalPlanCacheSize() {
+        const serverStatus = assert.commandWorked(db.serverStatus());
+        return serverStatus.metrics.query.planCacheTotalSizeEstimateBytes;
+    }
+
+    function planCacheContents(coll) {
+        const result =
+            assert.commandWorked(db.runCommand({planCacheListQueryShapes: coll.getName()}));
+        assert(result.hasOwnProperty("shapes"), tojson(result));
+        return result.shapes;
+    }
+
+    /**
+     * Retrieve the cache entry associated with the query shape defined by the given 'filter'
+     * (assuming the query has no projection, sort, or collation) using the 'planCacheListPlans'
+     * command.  Asserts that the plan cache entry exists, and returns it.
+     */
+    function getPlanCacheEntryForFilter(coll, filter) {
+        const cmdResult = assert.commandWorked(
+            db.runCommand({planCacheListPlans: coll.getName(), query: filter}));
+        // Ensure that an entry actually exists in the cache for this query shape.
+        assert.gt(cmdResult.plans.length, 0, tojson(cmdResult));
+        return cmdResult;
+    }
+
+    function assertExistenceOfRequiredCacheEntryFields(entry) {
+        assert(entry.hasOwnProperty("works"), tojson(entry));
+        assert(entry.hasOwnProperty("timeOfCreation"), tojson(entry));
+        assert(entry.hasOwnProperty("estimatedSizeBytes"), tojson(entry));
+
+        assert(entry.hasOwnProperty("plans"), tojson(entry));
+        for (let plan of entry.plans) {
+            assert(plan.hasOwnProperty("details"), tojson(plan));
+            assert(plan.details.hasOwnProperty("solution"), tojson(plan));
+            assert(plan.hasOwnProperty("filterSet"), tojson(plan));
+        }
+    }
+
+    function assertCacheEntryHasDebugInfo(entry) {
+        assertExistenceOfRequiredCacheEntryFields(entry);
+
+        // Check that fields deriving from debug info are present.
+        for (let i = 0; i < entry.plans.length; ++i) {
+            const plan = entry.plans[i];
+            assert(plan.hasOwnProperty("reason"), tojson(plan));
+            assert(plan.reason.hasOwnProperty("score"), tojson(plan));
+            assert(plan.reason.hasOwnProperty("stats"), tojson(plan));
+        }
+        assert(entry.plans[0].hasOwnProperty("feedback"), tojson(entry));
+        assert(entry.plans[0].feedback.hasOwnProperty("nfeedback"), tojson(entry));
+        assert(entry.plans[0].feedback.hasOwnProperty("scores"), tojson(entry));
+    }
+
+    function assertCacheEntryIsMissingDebugInfo(entry) {
+        assertExistenceOfRequiredCacheEntryFields(entry);
+
+        // Verify that fields deriving from debug info  are missing for the legacy format.
+        for (let i = 0; i < entry.plans.length; ++i) {
+            const plan = entry.plans[i];
+            assert(!plan.hasOwnProperty("reason"), tojson(plan));
+            assert(!plan.hasOwnProperty("feedback"), tojson(plan));
+        }
+
+        // We expect cache entries to be reasonably small when their debug info is stripped.
+        // Although there are no strict guarantees on the size of the entry, we can expect that the
+        // size estimate should always remain under 4kb.
+        assert.lt(entry.estimatedSizeBytes, 4 * 1024, tojson(entry));
+    }
+
+    /**
+     * Given a match expression 'filter' describing a query shape, obtains the associated plan cache
+     * information using 'planCacheListPlans'. Verifies that the plan cache entry exists and
+     * contains the expected debug info.
+     */
+    function assertQueryShapeHasDebugInfoInCache(coll, filter) {
+        const cacheEntry = getPlanCacheEntryForFilter(coll, filter);
+        assertCacheEntryHasDebugInfo(cacheEntry);
+    }
+
+    /**
+     * Given a match expression 'filter' describing a query shape, obtains the associated plan cache
+     * information using 'planCacheListPlans'. Verifies that the plan cache entry exists but has had
+     * its debug info stripped.
+     */
+    function assertQueryShapeIsMissingDebugInfoInCache(coll, filter) {
+        const cacheEntry = getPlanCacheEntryForFilter(coll, filter);
+        assertCacheEntryIsMissingDebugInfo(cacheEntry);
+    }
+
+    /**
+     * Given a previous total plan cache size, 'oldCacheSize', and a newer observation of the plan
+     * cache size, 'newCacheSize', asserts that this growth is consistent with the size reported by
+     * 'cacheEntry'.
+     */
+    function assertChangeInCacheSizeIsDueToEntry(cacheEntry, oldCacheSize, newCacheSize) {
+        const cacheSizeGrowth = newCacheSize - oldCacheSize;
+        // Instead of asserting that the cache size growth is exactly equal to the cache entry's
+        // size, we assert that the difference between them is sufficiently small. This wiggle room
+        // is necessary since the reported cache size is an estimate and may not be precise.
+        assert.lt(Math.abs(cacheEntry.estimatedSizeBytes - cacheSizeGrowth), 50);
+    }
+
+    const conn = MongoRunner.runMongod({});
+    assert.neq(conn, null, "mongod failed to start");
+    const db = conn.getDB("test");
+    const coll = db.plan_cache_memory_debug_info;
+    coll.drop();
+    createIndexesForColl(coll);
+
+    const smallQuery = {
+        a: 1,
+        b: 1,
+    };
+
+    // Create a plan cache entry, and verify that the estimated plan cache size has increased.
+    let oldPlanCacheSize = totalPlanCacheSize();
+    assert.eq(0, coll.find(smallQuery).itcount());
+    let newPlanCacheSize = totalPlanCacheSize();
+    assert.gt(newPlanCacheSize, oldPlanCacheSize);
+
+    // Verify that the cache now has a single entry whose estimated size explains the increase in
+    // the total plan cache size reported by serverStatus(). The cache entry should contain all
+    // expected debug info.
+    let cacheContents = planCacheContents(coll);
+    assert.eq(cacheContents.length, 1, cacheContents);
+    const cacheEntry = getPlanCacheEntryForFilter(coll, cacheContents[0].query);
+    assertCacheEntryHasDebugInfo(cacheEntry);
+    assertQueryShapeHasDebugInfoInCache(coll, smallQuery);
+    assertChangeInCacheSizeIsDueToEntry(cacheEntry, oldPlanCacheSize, newPlanCacheSize);
+
+    // Configure the server so that new plan cache entries should not preserve debug info.
+    const setParamRes = assert.commandWorked(
+        db.adminCommand({setParameter: 1, internalQueryCacheMaxSizeBytesBeforeStripDebugInfo: 0}));
+    const stripDebugInfoThresholdDefault = setParamRes.was;
+
+    // Generate a query which includes a 10,000 element $in predicate.
+    const kNumInElements = 10 * 1000;
+    const largeQuery = {
+        a: 1,
+        b: 1,
+        c: {$in: Array.from({length: kNumInElements}, (_, i) => i)},
+    };
+
+    // Create a new cache entry using the query with the large $in predicate. Verify that the
+    // estimated total plan cache size has increased again, and check that there are now two entries
+    // in the cache.
+    oldPlanCacheSize = totalPlanCacheSize();
+    assert.eq(0, coll.find(largeQuery).itcount());
+    newPlanCacheSize = totalPlanCacheSize();
+    assert.gt(newPlanCacheSize, oldPlanCacheSize);
+    cacheContents = planCacheContents(coll);
+    assert.eq(cacheContents.length, 2, cacheContents);
+
+    // The cache entry associated with 'smallQuery' should retain its debug info, whereas the cache
+    // entry associated with 'largeQuery' should have had its debug info stripped.
+    assertQueryShapeHasDebugInfoInCache(coll, smallQuery);
+    assertQueryShapeIsMissingDebugInfoInCache(coll, largeQuery);
+
+    // The second cache entry should be smaller than the first, despite the query being much larger.
+    const smallQueryCacheEntry = getPlanCacheEntryForFilter(coll, smallQuery);
+    let largeQueryCacheEntry = getPlanCacheEntryForFilter(coll, largeQuery);
+    assert.lt(largeQueryCacheEntry.estimatedSizeBytes,
+              smallQueryCacheEntry.estimatedSizeBytes,
+              cacheContents);
+
+    // The new cache entry's size should account for the latest observed increase in total plan
+    // cache size.
+    assertChangeInCacheSizeIsDueToEntry(largeQueryCacheEntry, oldPlanCacheSize, newPlanCacheSize);
+
+    // Verify that a new cache entry in a different collection also has its debug info stripped.
+    // This demonstrates that the size threshold applies on a server-wide basis as opposed to on a
+    // per-collection basis.
+    const secondColl = db.plan_cache_memory_debug_info_other;
+    secondColl.drop();
+    createIndexesForColl(secondColl);
+
+    // Introduce a new cache entry in the second collection's cache and verify that the cumulative
+    // plan cache size has increased.
+    oldPlanCacheSize = totalPlanCacheSize();
+    assert.eq(0, secondColl.find(smallQuery).itcount());
+    newPlanCacheSize = totalPlanCacheSize();
+    assert.gt(newPlanCacheSize, oldPlanCacheSize);
+
+    // Ensure that the second collection's cache now has one entry, and that entry's debug info is
+    // stripped.
+    cacheContents = planCacheContents(secondColl);
+    assert.eq(cacheContents.length, 1, cacheContents);
+    assertQueryShapeIsMissingDebugInfoInCache(secondColl, smallQuery);
+
+    // Meanwhile, the contents of the original collection's plan cache should remain unchanged.
+    cacheContents = planCacheContents(coll);
+    assert.eq(cacheContents.length, 2, cacheContents);
+    assertQueryShapeHasDebugInfoInCache(coll, smallQuery);
+    assertQueryShapeIsMissingDebugInfoInCache(coll, largeQuery);
+
+    // Ensure that 'planCacheListQueryShapes' works when debug info has been stripped. For a cache
+    // entry which is missing debug info, we expect 'planCacheListQueryShapes' to display an empty
+    // object.
+    const listQueryShapesResult =
+        assert.commandWorked(db.runCommand({planCacheListQueryShapes: secondColl.getName()}));
+    assert(listQueryShapesResult.hasOwnProperty("shapes"), tojson(listQueryShapesResult));
+    assert.eq(listQueryShapesResult.shapes.length, 1, listQueryShapesResult);
+    const listedShape = listQueryShapesResult.shapes[0];
+    assert.eq(listedShape, {}, listQueryShapesResult);
+
+    // Restore the threshold for stripping debug info to its default. Verify that if we add a third
+    // cache entry to the original collection 'coll', the plan cache size increases once again, and
+    // the new cache entry stores debug info.
+    assert.commandWorked(db.adminCommand({
+        setParameter: 1,
+        internalQueryCacheMaxSizeBytesBeforeStripDebugInfo: stripDebugInfoThresholdDefault,
+    }));
+    const smallQuery2 = {
+        a: 1,
+        b: 1,
+        c: 1,
+    };
+    oldPlanCacheSize = totalPlanCacheSize();
+    assert.eq(0, coll.find(smallQuery2).itcount());
+    newPlanCacheSize = totalPlanCacheSize();
+    assert.gt(newPlanCacheSize, oldPlanCacheSize);
+
+    // Verify that there are now three cache entries.
+    cacheContents = planCacheContents(coll);
+    assert.eq(cacheContents.length, 3, cacheContents);
+
+    // Make sure that the cache entries have or are missing debug info as expected.
+    assertQueryShapeHasDebugInfoInCache(coll, smallQuery);
+    assertQueryShapeHasDebugInfoInCache(coll, smallQuery2);
+    assertQueryShapeIsMissingDebugInfoInCache(coll, largeQuery);
+    assertQueryShapeIsMissingDebugInfoInCache(secondColl, smallQuery);
+
+    // Clear the cache entry for 'largeQuery' and regenerate it. The cache should grow larger, since
+    // the regenerated cache entry should now contain debug info. Also, check that the size of the
+    // new cache entry is estimated to be at least 10kb, since the query itself is known to be at
+    // least 10kb.
+    oldPlanCacheSize = totalPlanCacheSize();
+    assert.commandWorked(coll.runCommand("planCacheClear", {query: largeQuery}));
+    cacheContents = planCacheContents(coll);
+    assert.eq(cacheContents.length, 2, cacheContents);
+
+    assert.eq(0, coll.find(largeQuery).itcount());
+    cacheContents = planCacheContents(coll);
+    assert.eq(cacheContents.length, 3, cacheContents);
+
+    newPlanCacheSize = totalPlanCacheSize();
+    assert.gt(newPlanCacheSize, oldPlanCacheSize);
+
+    assertQueryShapeHasDebugInfoInCache(coll, largeQuery);
+    largeQueryCacheEntry = getPlanCacheEntryForFilter(coll, largeQuery);
+    assert.gt(largeQueryCacheEntry.estimatedSizeBytes, 10 * 1024, largeQueryCacheEntry);
+
+    MongoRunner.stopMongod(conn);
+}());
diff --git a/jstests/noPassthrough/query_knobs_validation.js b/jstests/noPassthrough/query_knobs_validation.js
new file mode 100644
index 00000000000..abe7869d666
--- /dev/null
+++ b/jstests/noPassthrough/query_knobs_validation.js
@@ -0,0 +1,96 @@
+/**
+ * Tests to validate the input values accepted by internal query server parameters. The test
+ * verfies that the system responds with the expected error code for input values that fall outside
+ * each parameter's valid bounds, and correctly applies input values which fall within that
+ * parameter's valid bounds.
+ */
+
+(function() {
+    "use strict";
+
+    const conn = MongoRunner.runMongod();
+    const testDB = conn.getDB("admin");
+    const expectedParamDefaults = {
+        internalQueryPlanEvaluationWorks: 10000,
+        internalQueryPlanEvaluationCollFraction: 0.3,
+        internalQueryPlanEvaluationMaxResults: 101,
+        internalQueryCacheSize: 5000,
+        internalQueryCacheEvictionRatio: 10.0,
+        internalQueryCacheMaxSizeBytesBeforeStripDebugInfo: 512 * 1024 * 1024,
+        internalQueryPlannerMaxIndexedSolutions: 64,
+        internalQueryEnumerationMaxOrSolutions: 10,
+        internalQueryEnumerationMaxIntersectPerAnd: 3,
+        internalQueryForceIntersectionPlans: false,
+        internalQueryPlannerEnableIndexIntersection: true,
+        internalQueryPlannerEnableHashIntersection: false,
+        internalQueryPlanOrChildrenIndependently: true,
+        internalQueryMaxScansToExplode: 200,
+        internalQueryExecMaxBlockingSortBytes: 32 * 1024 * 1024,
+        internalQueryExecYieldIterations: 128,
+        internalQueryExecYieldPeriodMS: 10,
+        internalQueryFacetBufferSizeBytes: 100 * 1024 * 1024,
+        internalDocumentSourceCursorBatchSizeBytes: 4 * 1024 * 1024,
+        internalDocumentSourceLookupCacheSizeBytes: 100 * 1024 * 1024,
+        internalLookupStageIntermediateDocumentMaxSizeBytes: 100 * 1024 * 1024,
+        internalQueryMaxPushBytes: 100 * 1024 * 1024,
+        internalQueryMaxAddToSetBytes: 100 * 1024 * 1024,
+        // Should be half the value of 'internalQueryExecYieldIterations' parameter.
+        internalInsertMaxBatchSize: 64,
+        internalQueryPlannerGenerateCoveredWholeIndexScans: false,
+        internalQueryIgnoreUnknownJSONSchemaKeywords: false,
+        internalQueryProhibitBlockingMergeOnMongoS: false,
+    };
+
+    function assertDefaultParameterValues() {
+        // For each parameter in 'expectedParamDefaults' verify that the value returned by
+        // 'getParameter' is same as the expected value.
+        for (let paramName in expectedParamDefaults) {
+            const expectedParamValue = expectedParamDefaults[paramName];
+            const getParamRes =
+                assert.commandWorked(testDB.adminCommand({getParameter: 1, [paramName]: 1}));
+            assert.eq(getParamRes[paramName], expectedParamValue);
+        }
+    }
+
+    function assertSetParameterSucceeds(paramName, value) {
+        assert.commandWorked(testDB.adminCommand({setParameter: 1, [paramName]: value}));
+        // Verify that the set parameter actually worked by doing a get and verifying the value.
+        const getParamRes =
+            assert.commandWorked(testDB.adminCommand({getParameter: 1, [paramName]: 1}));
+        assert.eq(getParamRes[paramName], value);
+    }
+
+    function assertSetParameterFails(paramName, value) {
+        assert.commandFailed(testDB.adminCommand({setParameter: 1, [paramName]: value}));
+    }
+
+    // Verify that the default values are set as expected when the server starts up.
+    assertDefaultParameterValues();
+
+    assertSetParameterSucceeds("internalQueryCacheMaxSizeBytesBeforeStripDebugInfo", 1);
+    assertSetParameterSucceeds("internalQueryCacheMaxSizeBytesBeforeStripDebugInfo", 0);
+    assertSetParameterFails("internalQueryCacheMaxSizeBytesBeforeStripDebugInfo", -1);
+
+    assertSetParameterSucceeds("internalQueryFacetMaxOutputDocSizeBytes", 1);
+    assertSetParameterFails("internalQueryFacetMaxOutputDocSizeBytes", 0);
+    assertSetParameterFails("internalQueryFacetMaxOutputDocSizeBytes", -1);
+
+    assertSetParameterSucceeds("internalQueryMaxPushBytes", 10);
+    assertSetParameterFails("internalQueryMaxPushBytes", 0);
+    assertSetParameterFails("internalQueryMaxPushBytes", -1);
+
+    assertSetParameterSucceeds("internalQueryMaxAddToSetBytes", 10);
+    assertSetParameterFails("internalQueryMaxAddToSetBytes", 0);
+    assertSetParameterFails("internalQueryMaxAddToSetBytes", -1);
+
+    // Internal BSON max object size is slightly larger than the max user object size, to
+    // accommodate command metadata.
+    const bsonUserSizeLimit = assert.commandWorked(testDB.hello()).maxBsonObjectSize;
+    const bsonObjMaxInternalSize = bsonUserSizeLimit + 16 * 1024;
+
+    assertSetParameterFails("internalLookupStageIntermediateDocumentMaxSizeBytes", 1);
+    assertSetParameterSucceeds("internalLookupStageIntermediateDocumentMaxSizeBytes",
+                               bsonObjMaxInternalSize);
+
+    MongoRunner.stopMongod(conn);
+})();
diff --git a/src/mongo/db/commands/index_filter_commands_test.cpp b/src/mongo/db/commands/index_filter_commands_test.cpp
index 6c0500c8a80..f991b96ba89 100644
--- a/src/mongo/db/commands/index_filter_commands_test.cpp
+++ b/src/mongo/db/commands/index_filter_commands_test.cpp
@@ -176,15 +176,16 @@ bool planCacheContains(const PlanCache& planCache,
     bool found = false;
     for (vector<PlanCacheEntry*>::const_iterator i = entries.begin(); i != entries.end(); i++) {
         PlanCacheEntry* entry = *i;
+        const auto& createdFromQuery = entry->debugInfo->createdFromQuery;
 
         // Canonicalizing query shape in cache entry to get cache key.
         // Alternatively, we could add key to PlanCacheEntry but that would be used in one place
         // only.
         auto qr = stdx::make_unique<QueryRequest>(nss);
-        qr->setFilter(entry->query);
-        qr->setSort(entry->sort);
-        qr->setProj(entry->projection);
-        qr->setCollation(entry->collation);
+        qr->setFilter(createdFromQuery.filter);
+        qr->setSort(createdFromQuery.sort);
+        qr->setProj(createdFromQuery.projection);
+        qr->setCollation(createdFromQuery.collation);
         auto statusWithCurrentQuery = CanonicalQuery::canonicalize(opCtx.get(), std::move(qr));
         ASSERT_OK(statusWithCurrentQuery.getStatus());
         unique_ptr<CanonicalQuery> currentQuery = std::move(statusWithCurrentQuery.getValue());
diff --git a/src/mongo/db/commands/plan_cache_commands.cpp b/src/mongo/db/commands/plan_cache_commands.cpp
index 0d9cba677a1..e0d6d0c6928 100644
--- a/src/mongo/db/commands/plan_cache_commands.cpp
+++ b/src/mongo/db/commands/plan_cache_commands.cpp
@@ -104,8 +104,8 @@ namespace mongo {
 
 using std::string;
 using std::stringstream;
-using std::vector;
 using std::unique_ptr;
+using std::vector;
 
 PlanCacheCommand::PlanCacheCommand(const string& name,
                                    const string& helpText,
@@ -259,11 +259,18 @@ Status PlanCacheListQueryShapes::list(const PlanCache& planCache, BSONObjBuilder
         invariant(entry);
 
         BSONObjBuilder shapeBuilder(arrayBuilder.subobjStart());
-        shapeBuilder.append("query", entry->query);
-        shapeBuilder.append("sort", entry->sort);
-        shapeBuilder.append("projection", entry->projection);
-        if (!entry->collation.isEmpty()) {
-            shapeBuilder.append("collation", entry->collation);
+
+        // Most details demonstrating the shape of the query must be omitted if the plan cache entry
+        // has had its debug info stripped. Debug info may be omitted from the plan cache entry in
+        // order to reduce the plan cache's memory footprint.
+        if (entry->debugInfo) {
+            const auto& createdFromQuery = entry->debugInfo->createdFromQuery;
+            shapeBuilder.append("query", createdFromQuery.filter);
+            shapeBuilder.append("sort", createdFromQuery.sort);
+            shapeBuilder.append("projection", createdFromQuery.projection);
+            if (!createdFromQuery.collation.isEmpty()) {
+                shapeBuilder.append("collation", createdFromQuery.collation);
+            }
         }
         shapeBuilder.doneFast();
 
@@ -403,8 +410,11 @@ Status PlanCacheListPlans::list(OperationContext* opCtx,
 
     BSONArrayBuilder plansBuilder(bob->subarrayStart("plans"));
     size_t numPlans = entry->plannerData.size();
-    invariant(numPlans == entry->decision->stats.size());
-    invariant(numPlans == entry->decision->scores.size());
+    if (entry->debugInfo) {
+        invariant(numPlans == entry->debugInfo->decision->stats.size());
+        invariant(numPlans == entry->debugInfo->decision->scores.size());
+    }
+
     for (size_t i = 0; i < numPlans; ++i) {
         BSONObjBuilder planBob(plansBuilder.subobjStart());
 
@@ -414,30 +424,34 @@ Status PlanCacheListPlans::list(OperationContext* opCtx,
         detailsBob.append("solution", entry->plannerData[i]->toString());
         detailsBob.doneFast();
 
-        // reason is comprised of score and initial stats provided by
-        // multi plan runner.
-        BSONObjBuilder reasonBob(planBob.subobjStart("reason"));
-        reasonBob.append("score", entry->decision->scores[i]);
-        BSONObjBuilder statsBob(reasonBob.subobjStart("stats"));
-        PlanStageStats* stats = entry->decision->stats[i].get();
-        if (stats) {
-            Explain::statsToBSON(*stats, &statsBob);
-        }
-        statsBob.doneFast();
-        reasonBob.doneFast();
-
-        // BSON object for 'feedback' field shows scores from historical executions of the plan.
-        BSONObjBuilder feedbackBob(planBob.subobjStart("feedback"));
-        if (i == 0U) {
-            feedbackBob.append("nfeedback", int(entry->feedback.size()));
-            BSONArrayBuilder scoresBob(feedbackBob.subarrayStart("scores"));
-            for (size_t i = 0; i < entry->feedback.size(); ++i) {
-                BSONObjBuilder scoreBob(scoresBob.subobjStart());
-                scoreBob.append("score", entry->feedback[i]->score);
+        // Some information for each candidate plan can only be provided if the plan cache entry
+        // contains debug info.
+        if (entry->debugInfo) {
+            const auto& debugInfo = *entry->debugInfo;
+            {
+                // 'reason' is comprised of score and initial stats provided by the multi-planner.
+                BSONObjBuilder reasonBob(planBob.subobjStart("reason"));
+                reasonBob.append("score", debugInfo.decision->scores[i]);
+                {
+                    BSONObjBuilder statsBob(reasonBob.subobjStart("stats"));
+                    PlanStageStats* stats = debugInfo.decision->stats[i].get();
+                    if (stats) {
+                        Explain::statsToBSON(*stats, &statsBob);
+                    }
+                }
+            }
+
+            // BSON object for 'feedback' field shows scores from historical executions of the plan.
+            BSONObjBuilder feedbackBob(planBob.subobjStart("feedback"));
+            if (i == 0U) {
+                feedbackBob.append("nfeedback", int(debugInfo.feedback.size()));
+                BSONArrayBuilder scoresBob(feedbackBob.subarrayStart("scores"));
+                for (auto&& feedback : debugInfo.feedback) {
+                    BSONObjBuilder scoreBob(scoresBob.subobjStart());
+                    scoreBob.append("score", feedback->score);
+                }
             }
-            scoresBob.doneFast();
         }
-        feedbackBob.doneFast();
 
         planBob.append("filterSet", entry->plannerData[i]->indexFilterApplied);
     }
@@ -447,6 +461,8 @@ Status PlanCacheListPlans::list(OperationContext* opCtx,
     // Append the time the entry was inserted into the plan cache.
     bob->append("timeOfCreation", entry->timeOfCreation);
 
+    bob->append("works", static_cast<long long>(entry->decisionWorks));
+    bob->append("estimatedSizeBytes", static_cast<long long>(entry->estimatedEntrySizeBytes));
     return Status::OK();
 }
 
diff --git a/src/mongo/db/exec/subplan.cpp b/src/mongo/db/exec/subplan.cpp
index 207e32b1ea8..391337f9d78 100644
--- a/src/mongo/db/exec/subplan.cpp
+++ b/src/mongo/db/exec/subplan.cpp
@@ -299,8 +299,11 @@ Status SubplanStage::choosePlanForSubqueries(PlanYieldPolicy* yieldPolicy) {
 
         if (branchResult->cachedSolution.get()) {
             // We can get the index tags we need out of the cache.
-            Status tagStatus = tagOrChildAccordingToCache(
-                cacheData.get(), branchResult->cachedSolution->plannerData[0], orChild, _indexMap);
+            Status tagStatus =
+                tagOrChildAccordingToCache(cacheData.get(),
+                                           branchResult->cachedSolution->plannerData.get(),
+                                           orChild,
+                                           _indexMap);
             if (!tagStatus.isOK()) {
                 return tagStatus;
             }
diff --git a/src/mongo/db/query/plan_cache.cpp b/src/mongo/db/query/plan_cache.cpp
index 61f894846c4..117a6479eed 100644
--- a/src/mongo/db/query/plan_cache.cpp
+++ b/src/mongo/db/query/plan_cache.cpp
@@ -371,34 +371,8 @@ bool PlanCache::shouldCacheQuery(const CanonicalQuery& query) {
     return true;
 }
 
-//
-// CachedSolution
-//
-
-CachedSolution::CachedSolution(const PlanCacheKey& key, const PlanCacheEntry& entry)
-    : plannerData(entry.plannerData.size()),
-      key(key),
-      query(entry.query.getOwned()),
-      sort(entry.sort.getOwned()),
-      projection(entry.projection.getOwned()),
-      collation(entry.collation.getOwned()),
-      decisionWorks(entry.decision->stats[0]->common.works) {
-    // CachedSolution should not having any references into
-    // cache entry. All relevant data should be cloned/copied.
-    for (size_t i = 0; i < entry.plannerData.size(); ++i) {
-        verify(entry.plannerData[i]);
-        plannerData[i] = entry.plannerData[i]->clone();
-    }
-}
-
-CachedSolution::~CachedSolution() {
-    for (std::vector<SolutionCacheData*>::const_iterator i = plannerData.begin();
-         i != plannerData.end();
-         ++i) {
-        SolutionCacheData* scd = *i;
-        delete scd;
-    }
-}
+CachedSolution::CachedSolution(const PlanCacheEntry& entry)
+    : plannerData(entry.plannerData[0]->clone()), decisionWorks(entry.decisionWorks) {}
 
 //
 // PlanCacheEntry
@@ -411,10 +385,11 @@ std::unique_ptr<PlanCacheEntry> PlanCacheEntry::create(
     Date_t timeOfCreation) {
     invariant(decision);
 
-    // The caller of this constructor is responsible for ensuring
-    // that the QuerySolution 's' has valid cacheData. If there's no
-    // data to cache you shouldn't be trying to construct a PlanCacheEntry.
+    const size_t decisionWorks = decision->stats[0]->common.works;
 
+    // The caller of this constructor is responsible for ensuring that the QuerySolution has valid
+    // cacheData. If there's no data to cache you shouldn't be trying to construct a PlanCacheEntry.
+    //
     // Copy the solution's cache data into the plan cache entry.
     std::vector<std::unique_ptr<const SolutionCacheData>> solutionCacheData(solutions.size());
     for (size_t i = 0; i < solutions.size(); ++i) {
@@ -423,53 +398,57 @@ std::unique_ptr<PlanCacheEntry> PlanCacheEntry::create(
             std::unique_ptr<const SolutionCacheData>(solutions[i]->cacheData->clone());
     }
 
-    const QueryRequest& qr = query.getQueryRequest();
-    BSONObjBuilder projBuilder;
-    for (auto elem : qr.getProj()) {
-        if (elem.fieldName()[0] == '$') {
-            continue;
+    // If the cumulative size of the plan caches is estimated to remain within a predefined
+    // threshold, then include additional debug info which is not strictly necessary for the plan
+    // cache to be functional. Once the cumulative plan cache size exceeds this threshold, omit this
+    // debug info as a heuristic to prevent plan cache memory consumption from growing too large.
+    const bool includeDebugInfo = planCacheTotalSizeEstimateBytes.get() <
+        internalQueryCacheMaxSizeBytesBeforeStripDebugInfo.load();
+
+    std::unique_ptr<DebugInfo> debugInfo;
+    if (includeDebugInfo) {
+        // Strip projections on $-prefixed fields, as these are added by internal callers of the
+        // system and are not considered part of the user projection.
+        const QueryRequest& qr = query.getQueryRequest();
+        BSONObjBuilder projBuilder;
+        for (auto elem : qr.getProj()) {
+            if (elem.fieldName()[0] == '$') {
+                continue;
+            }
+            projBuilder.append(elem);
         }
-        projBuilder.append(elem);
+
+        CreatedFromQuery createdFromQuery{
+            qr.getFilter(),
+            qr.getSort(),
+            projBuilder.obj(),
+            query.getCollator() ? query.getCollator()->getSpec().toBSON() : BSONObj()};
+        debugInfo =
+            stdx::make_unique<DebugInfo>(std::move(createdFromQuery),
+                                         std::move(decision),
+                                         std::vector<std::unique_ptr<PlanCacheEntryFeedback>>{});
     }
 
     return std::unique_ptr<PlanCacheEntry>(new PlanCacheEntry(
-        std::move(solutionCacheData),
-        qr.getFilter(),
-        qr.getSort(),
-        projBuilder.obj(),
-        query.getCollator() ? query.getCollator()->getSpec().toBSON() : BSONObj(),
-        timeOfCreation,
-        std::move(decision),
-        {}));
+        std::move(solutionCacheData), timeOfCreation, decisionWorks, std::move(debugInfo)));
 }
 
 PlanCacheEntry::PlanCacheEntry(std::vector<std::unique_ptr<const SolutionCacheData>> plannerData,
-                               const BSONObj& query,
-                               const BSONObj& sort,
-                               const BSONObj& projection,
-                               const BSONObj& collation,
-                               const Date_t timeOfCreation,
-                               std::unique_ptr<const PlanRankingDecision> decision,
-                               std::vector<PlanCacheEntryFeedback*> feedback)
+                               Date_t timeOfCreation,
+                               size_t decisionWorks,
+                               std::unique_ptr<DebugInfo> debugInfo)
     : plannerData(std::move(plannerData)),
-      query(query),
-      sort(sort),
-      projection(projection),
-      collation(collation),
       timeOfCreation(timeOfCreation),
-      decision(std::move(decision)),
-      feedback(std::move(feedback)),
-      _entireObjectSize(_estimateObjectSizeInBytes()) {
+      decisionWorks(decisionWorks),
+      debugInfo(std::move(debugInfo)),
+      estimatedEntrySizeBytes(_estimateObjectSizeInBytes()) {
     // Account for the object in the global metric for estimating the server's total plan cache
     // memory consumption.
-    planCacheTotalSizeEstimateBytes.increment(_entireObjectSize);
+    planCacheTotalSizeEstimateBytes.increment(estimatedEntrySizeBytes);
 }
 
 PlanCacheEntry::~PlanCacheEntry() {
-    for (size_t i = 0; i < feedback.size(); ++i) {
-        delete feedback[i];
-    }
-    planCacheTotalSizeEstimateBytes.decrement(_entireObjectSize);
+    planCacheTotalSizeEstimateBytes.decrement(estimatedEntrySizeBytes);
 }
 
 PlanCacheEntry* PlanCacheEntry::clone() const {
@@ -478,55 +457,80 @@ PlanCacheEntry* PlanCacheEntry::clone() const {
         invariant(plannerData[i]);
         solutionCacheData[i] = std::unique_ptr<const SolutionCacheData>(plannerData[i]->clone());
     }
-    auto decisionPtr = std::unique_ptr<PlanRankingDecision>(decision->clone());
-    PlanCacheEntry* entry = new PlanCacheEntry(std::move(solutionCacheData),
-                                               query,
-                                               sort,
-                                               projection,
-                                               collation,
-                                               timeOfCreation,
-                                               std::move(decisionPtr),
-                                               {});
 
-    // Copy performance stats.
-    for (size_t i = 0; i < feedback.size(); ++i) {
-        PlanCacheEntryFeedback* fb = new PlanCacheEntryFeedback();
-        fb->stats.reset(feedback[i]->stats->clone());
-        fb->score = feedback[i]->score;
-        entry->feedback.push_back(fb);
+    return new PlanCacheEntry(std::move(solutionCacheData),
+                              timeOfCreation,
+                              decisionWorks,
+                              debugInfo ? debugInfo->clone() : nullptr);
+}
+
+uint64_t PlanCacheEntry::CreatedFromQuery::estimateObjectSizeInBytes() const {
+    uint64_t size = 0;
+    size += filter.objsize();
+    size += sort.objsize();
+    size += projection.objsize();
+    size += collation.objsize();
+    return size;
+}
+
+PlanCacheEntry::DebugInfo::DebugInfo(CreatedFromQuery createdFromQuery,
+                                     std::unique_ptr<const PlanRankingDecision> decision,
+                                     std::vector<std::unique_ptr<PlanCacheEntryFeedback>> feedback)
+    : createdFromQuery(std::move(createdFromQuery)),
+      decision(std::move(decision)),
+      feedback(std::move(feedback)) {
+    invariant(this->decision);
+}
+
+uint64_t PlanCacheEntry::DebugInfo::estimateObjectSizeInBytes() const {
+    uint64_t size = sizeof(DebugInfo);
+    size += createdFromQuery.estimateObjectSizeInBytes();
+    size += decision->estimateObjectSizeInBytes();
+    size += container_size_helper::estimateObjectSizeInBytes(
+        feedback,
+        [](const auto& feedbackEntry) { return feedbackEntry->estimateObjectSizeInBytes(); },
+        true);
+    return size;
+}
+
+std::unique_ptr<PlanCacheEntry::DebugInfo> PlanCacheEntry::DebugInfo::clone() const {
+    std::vector<std::unique_ptr<PlanCacheEntryFeedback>> clonedFeedback;
+    for (auto&& feedbackEntry : feedback) {
+        clonedFeedback.push_back(feedbackEntry->clone());
     }
-    return entry;
+    return stdx::make_unique<DebugInfo>(
+        createdFromQuery, decision->clone(), std::move(clonedFeedback));
 }
 
 uint64_t PlanCacheEntry::_estimateObjectSizeInBytes() const {
-    return  // Add the size of each entry in 'plannerData' vector.
-        container_size_helper::estimateObjectSizeInBytes(
-            plannerData,
-            [](const auto& cacheData) { return cacheData->estimateObjectSizeInBytes(); },
-            true) +
-        // Add the size of each entry in 'feedback' vector.
-        container_size_helper::estimateObjectSizeInBytes(
-            feedback,
-            [](const auto& feedbackEntry) { return feedbackEntry->estimateObjectSizeInBytes(); },
-            true) +
-        // Add the entire size of 'decision' object.
-        (decision ? decision->estimateObjectSizeInBytes() : 0) +
-        // Add the size of all the owned BSON objects.
-        query.objsize() + sort.objsize() + projection.objsize() + collation.objsize() +
-        // Add size of the object.
-        sizeof(*this);
-}
-
-std::string PlanCacheEntry::toString() const {
-    return str::stream() << "(query: " << query.toString() << ";sort: " << sort.toString()
-                         << ";projection: " << projection.toString()
-                         << ";collation: " << collation.toString()
-                         << ";solutions: " << plannerData.size()
-                         << ";timeOfCreation: " << timeOfCreation.toString() << ")";
-}
-
-std::string CachedSolution::toString() const {
-    return str::stream() << "key: " << key << '\n';
+    uint64_t size = sizeof(PlanCacheEntry);
+    size += container_size_helper::estimateObjectSizeInBytes(
+        plannerData,
+        [](const auto& cacheData) { return cacheData->estimateObjectSizeInBytes(); },
+        true);
+
+    if (debugInfo) {
+        size += debugInfo->estimateObjectSizeInBytes();
+    }
+
+    return size;
+}
+
+std::string PlanCacheEntry::CreatedFromQuery::debugString() const {
+    return str::stream() << "query: " << filter.toString() << "; sort: " << sort.toString()
+                         << "; projection: " << projection.toString()
+                         << "; collation: " << collation.toString();
+}
+
+std::string PlanCacheEntry::debugString() const {
+    StringBuilder builder;
+    builder << "(";
+    if (debugInfo) {
+        builder << debugInfo->createdFromQuery.debugString();
+    }
+    builder << "; works: " << decisionWorks;
+    builder << "; timeOfCreation: " << timeOfCreation.toString() << ")";
+    return builder.str();
 }
 
 //
@@ -594,9 +598,9 @@ std::string PlanCacheIndexTree::toString(int indents) const {
 // SolutionCacheData
 //
 
-SolutionCacheData* SolutionCacheData::clone() const {
-    SolutionCacheData* other = new SolutionCacheData();
-    if (NULL != this->tree.get()) {
+std::unique_ptr<SolutionCacheData> SolutionCacheData::clone() const {
+    auto other = std::make_unique<SolutionCacheData>();
+    if (nullptr != this->tree.get()) {
         // 'tree' could be NULL if the cached solution
         // is a collection scan.
         other->tree.reset(this->tree->clone());
@@ -803,7 +807,7 @@ Status PlanCache::add(const CanonicalQuery& query,
 
     if (NULL != evictedEntry.get()) {
         LOG(1) << _ns << ": plan cache maximum size exceeded - "
-               << "removed least recently used entry " << redact(evictedEntry->toString());
+               << "removed least recently used entry " << redact(evictedEntry->debugString());
     }
 
     return Status::OK();
@@ -821,7 +825,7 @@ Status PlanCache::get(const CanonicalQuery& query, CachedSolution** crOut) const
     }
     invariant(entry);
 
-    *crOut = new CachedSolution(key, *entry);
+    *crOut = new CachedSolution(*entry);
 
     return Status::OK();
 }
@@ -841,9 +845,15 @@ Status PlanCache::feedback(const CanonicalQuery& cq, PlanCacheEntryFeedback* fee
     }
     invariant(entry);
 
+    // If no debug info is present in the cache entry, then this is a no-op.
+    if (!entry->debugInfo) {
+        return Status::OK();
+    }
+
     // We store up to a constant number of feedback entries.
-    if (entry->feedback.size() < static_cast<size_t>(internalQueryCacheFeedbacksStored.load())) {
-        entry->feedback.push_back(autoFeedback.release());
+    auto& feedbackArr = entry->debugInfo->feedback;
+    if (feedbackArr.size() < static_cast<size_t>(internalQueryCacheFeedbacksStored.load())) {
+        feedbackArr.push_back(std::move(autoFeedback));
     }
 
     return Status::OK();
diff --git a/src/mongo/db/query/plan_cache.h b/src/mongo/db/query/plan_cache.h
index e1e529d198f..b57fa2015b6 100644
--- a/src/mongo/db/query/plan_cache.h
+++ b/src/mongo/db/query/plan_cache.h
@@ -39,6 +39,7 @@
 #include "mongo/db/query/index_tag.h"
 #include "mongo/db/query/lru_key_value.h"
 #include "mongo/db/query/plan_cache_indexability.h"
+#include "mongo/db/query/plan_ranking_decision.h"
 #include "mongo/db/query/query_planner_params.h"
 #include "mongo/platform/atomic_word.h"
 #include "mongo/stdx/mutex.h"
@@ -49,7 +50,6 @@ namespace mongo {
 // A PlanCacheKey is a string-ified version of a query's predicate/projection/sort.
 typedef std::string PlanCacheKey;
 
-struct PlanRankingDecision;
 struct QuerySolution;
 struct QuerySolutionNode;
 
@@ -58,16 +58,23 @@ struct QuerySolutionNode;
  * feedback is available to anyone who retrieves that query in the future.
  */
 struct PlanCacheEntryFeedback {
+    uint64_t estimateObjectSizeInBytes() const {
+        return stats->estimateObjectSizeInBytes() + sizeof(*this);
+    }
+
+    std::unique_ptr<PlanCacheEntryFeedback> clone() const {
+        auto clonedFeedback = stdx::make_unique<PlanCacheEntryFeedback>();
+        clonedFeedback->stats.reset(stats->clone());
+        clonedFeedback->score = score;
+        return clonedFeedback;
+    }
+
     // How well did the cached plan perform?
     std::unique_ptr<PlanStageStats> stats;
 
     // The "goodness" score produced by the plan ranker
     // corresponding to 'stats'.
     double score;
-
-    uint64_t estimateObjectSizeInBytes() const {
-        return stats->estimateObjectSizeInBytes() + sizeof(*this);
-    }
 };
 
 // TODO: Replace with opaque type.
@@ -185,8 +192,7 @@ struct SolutionCacheData {
           wholeIXSolnDir(1),
           indexFilterApplied(false) {}
 
-    // Make a deep copy.
-    SolutionCacheData* clone() const;
+    std::unique_ptr<SolutionCacheData> clone() const;
 
     // For debugging.
     std::string toString() const;
@@ -234,28 +240,12 @@ private:
     MONGO_DISALLOW_COPYING(CachedSolution);
 
 public:
-    CachedSolution(const PlanCacheKey& key, const PlanCacheEntry& entry);
-    ~CachedSolution();
+    CachedSolution(const PlanCacheEntry& entry);
 
-    // Owned here.
-    std::vector<SolutionCacheData*> plannerData;
-
-    // Key used to provide feedback on the entry.
-    PlanCacheKey key;
-
-    // For debugging.
-    std::string toString() const;
+    // Information that can be used by the QueryPlanner to reconstitute the complete execution plan.
+    std::unique_ptr<SolutionCacheData> plannerData;
 
-    // We are extracting just enough information from the canonical
-    // query. We could clone the canonical query but the following
-    // items are all that is displayed to the user.
-    BSONObj query;
-    BSONObj sort;
-    BSONObj projection;
-    BSONObj collation;
-
-    // The number of work cycles taken to decide on a winning plan when the plan was first
-    // cached.
+    // The number of work cycles taken to decide on a winning plan when the plan was first cached.
     size_t decisionWorks;
 };
 
@@ -269,6 +259,54 @@ private:
 
 public:
     /**
+     * A description of the query from which a 'PlanCacheEntry' was created.
+     */
+    struct CreatedFromQuery {
+        /**
+         * Returns an estimate of the size of this object, including the memory allocated elsewhere
+         * that it owns, in bytes.
+         */
+        uint64_t estimateObjectSizeInBytes() const;
+
+        std::string debugString() const;
+
+        BSONObj filter;
+        BSONObj sort;
+        BSONObj projection;
+        BSONObj collation;
+    };
+
+    /**
+     * Per-plan cache entry information that is used strictly as debug information (e.g. is intended
+     * for display by the 'planCacheListPlans' command). In order to save memory, this information
+     * is sometimes discarded instead of kept in the plan cache entry. Therefore, this information
+     * may not be used for any purpose outside displaying debug info, such as recovering a plan from
+     * the cache or determining whether or not the cache entry is active.
+     */
+    struct DebugInfo {
+        DebugInfo(CreatedFromQuery createdFromQuery,
+                  std::unique_ptr<const PlanRankingDecision> decision,
+                  std::vector<std::unique_ptr<PlanCacheEntryFeedback>> feedback);
+
+        /**
+         * Returns an estimate of the size of this object, including the memory allocated elsewhere
+         * that it owns, in bytes.
+         */
+        uint64_t estimateObjectSizeInBytes() const;
+
+        std::unique_ptr<DebugInfo> clone() const;
+
+        CreatedFromQuery createdFromQuery;
+
+        // Information that went into picking the winning plan and also why the other plans lost.
+        // Never nullptr.
+        std::unique_ptr<const PlanRankingDecision> decision;
+
+        // Scores from uses of this cache entry.
+        std::vector<std::unique_ptr<PlanCacheEntryFeedback>> feedback;
+    };
+
+    /**
      * Create a new PlanCacheEntry.
      * Grabs any planner-specific data required from the solutions.
      */
@@ -285,39 +323,34 @@ public:
      */
     PlanCacheEntry* clone() const;
 
-    // For debugging.
-    std::string toString() const;
+    std::string debugString() const;
 
+    // Data provided to the planner to allow it to recreate the solution this entry represents. In
+    // order to return it from the cache for consumption by the 'QueryPlanner', a deep copy is made
+    // and returned inside 'CachedSolution'.
     //
-    // Planner data
-    //
-
-    // Data provided to the planner to allow it to recreate the solutions this entry
-    // represents. Each SolutionCacheData is fully owned here, so in order to return
-    // it from the cache a deep copy is made and returned inside CachedSolution.
+    // The first element of the vector is the cache data associated with the winning plan. The
+    // remaining elements correspond to the rejected plans, sorted by descending score.
     const std::vector<std::unique_ptr<const SolutionCacheData>> plannerData;
 
-    // TODO: Do we really want to just hold a copy of the CanonicalQuery?  For now we just
-    // extract the data we need.
-    //
-    // Used by the plan cache commands to display an example query
-    // of the appropriate shape.
-    const BSONObj query;
-    const BSONObj sort;
-    const BSONObj projection;
-    const BSONObj collation;
     const Date_t timeOfCreation;
 
-    //
-    // Performance stats
-    //
+    // The number of work taken to select the winning plan when this plan cache entry was first
+    // created.
+    const size_t decisionWorks;
 
-    // Information that went into picking the winning plan and also why the other plans lost.
-    const std::unique_ptr<const PlanRankingDecision> decision;
+    // Optional debug info containing detailed statistics. Includes a description of the query which
+    // resulted in this plan cache's creation as well as runtime stats from the multi-planner trial
+    // period that resulted in this cache entry.
+    //
+    // Once the estimated cumulative size of the mongod's plan caches exceeds a threshold, this
+    // debug info is omitted from new plan cache entries.
+    std::unique_ptr<DebugInfo> debugInfo;
 
-    // Annotations from cached runs.  The CachedPlanStage provides these stats about its
-    // runs when they complete.
-    std::vector<PlanCacheEntryFeedback*> feedback;
+    // An estimate of the size in bytes of this plan cache entry. This is the "deep size",
+    // calculated by recursively incorporating the size of owned objects, the objects that they in
+    // turn own, and so on.
+    const uint64_t estimatedEntrySizeBytes;
 
     /**
      * Tracks the approximate cumulative size of the plan cache entries across all the collections.
@@ -329,19 +362,11 @@ private:
      * All arguments constructor.
      */
     PlanCacheEntry(std::vector<std::unique_ptr<const SolutionCacheData>> plannerData,
-                   const BSONObj& query,
-                   const BSONObj& sort,
-                   const BSONObj& projection,
-                   const BSONObj& collation,
                    Date_t timeOfCreation,
-                   std::unique_ptr<const PlanRankingDecision> decision,
-                   std::vector<PlanCacheEntryFeedback*> feedback);
+                   size_t decisionWorks,
+                   std::unique_ptr<DebugInfo> debugInfo);
 
     uint64_t _estimateObjectSizeInBytes() const;
-
-    // The total runtime size of the current object in bytes. This is the deep size, obtained by
-    // recursively following references to all owned objects.
-    const uint64_t _entireObjectSize;
 };
 
 /**
diff --git a/src/mongo/db/query/plan_cache_test.cpp b/src/mongo/db/query/plan_cache_test.cpp
index e47dae50c16..5893e25f665 100644
--- a/src/mongo/db/query/plan_cache_test.cpp
+++ b/src/mongo/db/query/plan_cache_test.cpp
@@ -724,12 +724,12 @@ protected:
         // Create a CachedSolution the long way..
         // QuerySolution -> PlanCacheEntry -> CachedSolution
         QuerySolution qs;
-        qs.cacheData.reset(soln.cacheData->clone());
+        qs.cacheData = soln.cacheData->clone();
         std::vector<QuerySolution*> solutions;
         solutions.push_back(&qs);
 
         auto entry = PlanCacheEntry::create(solutions, createDecision(1U), *scopedCq, Date_t());
-        CachedSolution cachedSoln(ck, *entry);
+        CachedSolution cachedSoln(*entry);
 
         QuerySolution* out;
         Status s = QueryPlanner::planFromCache(*scopedCq, params, cachedSoln, &out);
diff --git a/src/mongo/db/query/plan_ranker.h b/src/mongo/db/query/plan_ranker.h
index 198315cae9b..a15a9843743 100644
--- a/src/mongo/db/query/plan_ranker.h
+++ b/src/mongo/db/query/plan_ranker.h
@@ -38,13 +38,13 @@
 #include "mongo/db/exec/plan_stage.h"
 #include "mongo/db/exec/plan_stats.h"
 #include "mongo/db/exec/working_set.h"
+#include "mongo/db/query/plan_ranking_decision.h"
 #include "mongo/db/query/query_solution.h"
 #include "mongo/util/container_size_helper.h"
 
 namespace mongo {
 
 struct CandidatePlan;
-struct PlanRankingDecision;
 
 /**
  * Ranks 2 or more plans.
@@ -85,61 +85,4 @@ struct CandidatePlan {
     bool failed;
 };
 
-/**
- * Information about why a plan was picked to be the best.  Data here is placed into the cache
- * and used to compare expected performance with actual.
- */
-struct PlanRankingDecision {
-    PlanRankingDecision() {}
-
-    /**
-     * Make a deep copy.
-     */
-    PlanRankingDecision* clone() const {
-        PlanRankingDecision* decision = new PlanRankingDecision();
-        for (size_t i = 0; i < stats.size(); ++i) {
-            PlanStageStats* s = stats[i].get();
-            invariant(s);
-            decision->stats.push_back(std::unique_ptr<PlanStageStats>{s->clone()});
-        }
-        decision->scores = scores;
-        decision->candidateOrder = candidateOrder;
-        return decision;
-    }
-
-    uint64_t estimateObjectSizeInBytes() const {
-        return  // Add size of each element in 'stats' vector.
-            container_size_helper::estimateObjectSizeInBytes(
-                stats, [](const auto& stat) { return stat->estimateObjectSizeInBytes(); }, true) +
-            // Add size of each element in 'candidateOrder' vector.
-            container_size_helper::estimateObjectSizeInBytes(candidateOrder) +
-            // Add size of each element in 'scores' vector.
-            container_size_helper::estimateObjectSizeInBytes(scores) +
-            // Add size of the object.
-            sizeof(*this);
-    }
-
-    // Stats of all plans sorted in descending order by score.
-    // Owned by us.
-    std::vector<std::unique_ptr<PlanStageStats>> stats;
-
-    // The "goodness" score corresponding to 'stats'.
-    // Sorted in descending order.
-    std::vector<double> scores;
-
-    // Ordering of original plans in descending of score.
-    // Filled in by PlanRanker::pickBestPlan(candidates, ...)
-    // so that candidates[candidateOrder[0]] refers to the best plan
-    // with corresponding cores[0] and stats[0]. Runner-up would be
-    // candidates[candidateOrder[1]] followed by
-    // candidates[candidateOrder[2]], ...
-    std::vector<size_t> candidateOrder;
-
-    // Whether two plans tied for the win.
-    //
-    // Reading this flag is the only reliable way for callers to determine if there was a tie,
-    // because the scores kept inside the PlanRankingDecision do not incorporate the EOF bonus.
-    bool tieForBest = false;
-};
-
 }  // namespace mongo
diff --git a/src/mongo/db/query/plan_ranking_decision.h b/src/mongo/db/query/plan_ranking_decision.h
new file mode 100644
index 00000000000..8736b37a938
--- /dev/null
+++ b/src/mongo/db/query/plan_ranking_decision.h
@@ -0,0 +1,90 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    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
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    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 Server Side 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.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/plan_stats.h"
+#include "mongo/util/container_size_helper.h"
+
+namespace mongo {
+/**
+ * Information about why a plan was picked to be the best. Data here is placed into the cache and
+ * used to compare expected performance with actual.
+ */
+struct PlanRankingDecision {
+    /**
+     * Make a deep copy.
+     */
+    std::unique_ptr<PlanRankingDecision> clone() const {
+        auto decision = std::make_unique<PlanRankingDecision>();
+        for (size_t i = 0; i < stats.size(); ++i) {
+            PlanStageStats* s = stats[i].get();
+            invariant(s);
+            decision->stats.push_back(std::unique_ptr<PlanStageStats>{s->clone()});
+        }
+        decision->scores = scores;
+        decision->candidateOrder = candidateOrder;
+        return decision;
+    }
+
+    uint64_t estimateObjectSizeInBytes() const {
+        return  // Add size of each element in 'stats' vector.
+            container_size_helper::estimateObjectSizeInBytes(
+                stats, [](const auto& stat) { return stat->estimateObjectSizeInBytes(); }, true) +
+            // Add size of each element in 'candidateOrder' vector.
+            container_size_helper::estimateObjectSizeInBytes(candidateOrder) +
+            // Add size of each element in 'scores' vector.
+            container_size_helper::estimateObjectSizeInBytes(scores) +
+            // Add size of the object.
+            sizeof(*this);
+    }
+
+    // Stats of all plans sorted in descending order by score.
+    // Owned by us.
+    std::vector<std::unique_ptr<PlanStageStats>> stats;
+
+    // The "goodness" score corresponding to 'stats'.
+    // Sorted in descending order.
+    std::vector<double> scores;
+
+    // Ordering of original plans in descending of score.
+    // Filled in by PlanRanker::pickBestPlan(candidates, ...)
+    // so that candidates[candidateOrder[0]] refers to the best plan
+    // with corresponding cores[0] and stats[0]. Runner-up would be
+    // candidates[candidateOrder[1]] followed by
+    // candidates[candidateOrder[2]], ...
+    std::vector<size_t> candidateOrder;
+
+    // Whether two plans tied for the win.
+    //
+    // Reading this flag is the only reliable way for callers to determine if there was a tie,
+    // because the scores kept inside the PlanRankingDecision do not incorporate the EOF bonus.
+    bool tieForBest = false;
+};
+}  // namespace mongo
diff --git a/src/mongo/db/query/query_knobs.cpp b/src/mongo/db/query/query_knobs.cpp
index 9e4a9aa3931..0d5d6f4c27a 100644
--- a/src/mongo/db/query/query_knobs.cpp
+++ b/src/mongo/db/query/query_knobs.cpp
@@ -46,6 +46,18 @@ MONGO_EXPORT_SERVER_PARAMETER(internalQueryCacheSize, int, 5000);
 
 MONGO_EXPORT_SERVER_PARAMETER(internalQueryCacheFeedbacksStored, int, 20);
 
+MONGO_EXPORT_SERVER_PARAMETER(internalQueryCacheMaxSizeBytesBeforeStripDebugInfo,
+                              long long,
+                              512 * 1024 * 1024)
+    ->withValidator([](const long long& newVal) {
+        if (newVal < 0) {
+            return Status(
+                ErrorCodes::Error(4036100),
+                "internalQueryCacheMaxSizeBytesBeforeStripDebugInfo must be non-negative");
+        }
+        return Status::OK();
+    });
+
 MONGO_EXPORT_SERVER_PARAMETER(internalQueryCacheEvictionRatio, double, 10.0);
 
 MONGO_EXPORT_SERVER_PARAMETER(internalQueryPlannerMaxIndexedSolutions, int, 64);
diff --git a/src/mongo/db/query/query_knobs.h b/src/mongo/db/query/query_knobs.h
index ed4304de4ae..08597794fbf 100644
--- a/src/mongo/db/query/query_knobs.h
+++ b/src/mongo/db/query/query_knobs.h
@@ -63,13 +63,20 @@ extern AtomicBool internalQueryPlannerEnableHashIntersection;
 // plan cache
 //
 
-// How many entries in the cache?
+// The maximum number of entries allowed in a given collection's plan cache.
 extern AtomicInt32 internalQueryCacheSize;
 
 // How many feedback entries do we collect before possibly evicting from the cache based on bad
 // performance?
 extern AtomicInt32 internalQueryCacheFeedbacksStored;
 
+// Limits the amount of debug info stored across all plan caches in the system. Once the estimate of
+// the number of bytes used across all plan caches exceeds this threshold, then debug info is not
+// stored alongside new cache entries, in order to limit plan cache memory consumption. If plan
+// cache entries are freed and the estimate once again dips below this threshold, then new cache
+// entries will once again have debug info associated with them.
+extern AtomicInt64 internalQueryCacheMaxSizeBytesBeforeStripDebugInfo;
+
 // How many times more works must we perform in order to justify plan cache eviction
 // and replanning?
 extern AtomicDouble internalQueryCacheEvictionRatio;
diff --git a/src/mongo/db/query/query_planner.cpp b/src/mongo/db/query/query_planner.cpp
index 9de0aec1c73..d317a4d6b6f 100644
--- a/src/mongo/db/query/query_planner.cpp
+++ b/src/mongo/db/query/query_planner.cpp
@@ -58,8 +58,8 @@
 
 namespace mongo {
 
-using std::unique_ptr;
 using std::numeric_limits;
+using std::unique_ptr;
 
 namespace dps = ::mongo::dotted_path_support;
 
@@ -441,14 +441,14 @@ Status QueryPlanner::planFromCache(const CanonicalQuery& query,
                                    const QueryPlannerParams& params,
                                    const CachedSolution& cachedSoln,
                                    QuerySolution** out) {
-    invariant(!cachedSoln.plannerData.empty());
+    invariant(cachedSoln.plannerData);
     invariant(out);
 
     // A query not suitable for caching should not have made its way into the cache.
     invariant(PlanCache::shouldCacheQuery(query));
 
     // Look up winning solution in cached solution's array.
-    const SolutionCacheData& winnerCacheData = *cachedSoln.plannerData[0];
+    const auto& winnerCacheData = *cachedSoln.plannerData;
 
     if (SolutionCacheData::WHOLE_IXSCAN_SOLN == winnerCacheData.solnType) {
         // The solution can be constructed by a scan over the entire index.