SERVER-51200 CollectionPtr in RequiresCollectionStage point to instance owned by AutoGetCollection

RequiresCollectionStage now holds a pointer to CollectionPtr owned by an AutoGetCollection.
When we save and restore the executor a new CollectionPtr pointer needs to be assigned.

Plan executors can no longer be created with temporary CollectionPtr instances and their interface have been changed to take pointers to avoid binding to rvalues.

RequiresCollectionStage no longer loads collections from the catalog and will be in sync with the owning AutoGetCollection.

80 files changed, 494 insertions, 361 deletions

diff --git a/src/mongo/db/catalog/capped_utils.cpp b/src/mongo/db/catalog/capped_utils.cpp
index 260fcf446a5..c635d7cf5e9 100644
--- a/src/mongo/db/catalog/capped_utils.cpp
+++ b/src/mongo/db/catalog/capped_utils.cpp
@@ -121,7 +121,7 @@ void cloneCollectionAsCapped(OperationContext* opCtx,
                              const NamespaceString& toNss,
                              long long size,
                              bool temp) {
-    const CollectionPtr& fromCollection =
+    CollectionPtr fromCollection =
         CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, fromNss);
     if (!fromCollection) {
         uassert(ErrorCodes::CommandNotSupportedOnView,
@@ -160,7 +160,7 @@ void cloneCollectionAsCapped(OperationContext* opCtx,
         uassertStatusOK(createCollection(opCtx, toNss.db().toString(), cmd.done()));
     }
 
-    const CollectionPtr& toCollection =
+    CollectionPtr toCollection =
         CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, toNss);
     invariant(toCollection);  // we created above
 
@@ -176,7 +176,7 @@ void cloneCollectionAsCapped(OperationContext* opCtx,
     auto exec =
         InternalPlanner::collectionScan(opCtx,
                                         fromNss.ns(),
-                                        fromCollection,
+                                        &fromCollection,
                                         PlanYieldPolicy::YieldPolicy::WRITE_CONFLICT_RETRY_ONLY,
                                         InternalPlanner::FORWARD);
 
diff --git a/src/mongo/db/catalog/collection.cpp b/src/mongo/db/catalog/collection.cpp
index 1341599ab65..806112ae28b 100644
--- a/src/mongo/db/catalog/collection.cpp
+++ b/src/mongo/db/catalog/collection.cpp
@@ -69,25 +69,23 @@ bool CappedInsertNotifier::isDead() {
 
 // We can't reference the catalog from this library as it would create a cyclic library dependency.
 // Setup a weak dependency using a std::function that is installed by the catalog lib
-std::function<CollectionPtr(OperationContext*, CollectionUUID, uint64_t)>& _catalogLookup() {
-    static std::function<CollectionPtr(OperationContext*, CollectionUUID, uint64_t)> func;
+std::function<CollectionPtr(OperationContext*, CollectionUUID)>& _catalogLookup() {
+    static std::function<CollectionPtr(OperationContext*, CollectionUUID)> func;
     return func;
 }
 
 void CollectionPtr::installCatalogLookupImpl(
-    std::function<CollectionPtr(OperationContext*, CollectionUUID, uint64_t)> impl) {
+    std::function<CollectionPtr(OperationContext*, CollectionUUID)> impl) {
     _catalogLookup() = std::move(impl);
 }
 
 CollectionPtr CollectionPtr::null;
 
 CollectionPtr::CollectionPtr() : _collection(nullptr), _opCtx(nullptr) {}
-CollectionPtr::CollectionPtr(OperationContext* opCtx,
-                             const Collection* collection,
-                             uint64_t catalogEpoch)
-    : _collection(collection), _opCtx(opCtx), _catalogEpoch(catalogEpoch) {}
+CollectionPtr::CollectionPtr(OperationContext* opCtx, const Collection* collection)
+    : _collection(collection), _opCtx(opCtx) {}
 CollectionPtr::CollectionPtr(const Collection* collection, NoYieldTag)
-    : CollectionPtr(nullptr, collection, 0) {}
+    : CollectionPtr(nullptr, collection) {}
 CollectionPtr::CollectionPtr(Collection* collection) : CollectionPtr(collection, NoYieldTag{}) {}
 CollectionPtr::CollectionPtr(const std::shared_ptr<const Collection>& collection)
     : CollectionPtr(collection.get(), NoYieldTag{}) {}
@@ -96,7 +94,7 @@ CollectionPtr::~CollectionPtr() {}
 CollectionPtr& CollectionPtr::operator=(CollectionPtr&&) = default;
 
 CollectionPtr CollectionPtr::detached() const {
-    return CollectionPtr(_opCtx, _collection, _catalogEpoch);
+    return CollectionPtr(_opCtx, _collection);
 }
 
 bool CollectionPtr::_canYield() const {
@@ -110,7 +108,6 @@ void CollectionPtr::yield() const {
     // Yield if we are yieldable and have a valid collection
     if (_canYield() && _collection) {
         _uuid = _collection->uuid();
-        _ns = _collection->ns();
         _collection = nullptr;
     }
 }
@@ -119,8 +116,8 @@ void CollectionPtr::restore() const {
     if (_canYield() && _uuid) {
         // We may only do yield restore when we were holding locks that was yielded so we need to
         // refresh from the catalog to make sure we have a valid collection pointer.
-        auto coll = _catalogLookup()(_opCtx, *_uuid, _catalogEpoch);
-        if (coll && coll->ns() == _ns) {
+        auto coll = _catalogLookup()(_opCtx, *_uuid);
+        if (coll) {
             _collection = coll.get();
         }
         _uuid.reset();
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index aba2fc835b0..58e4f5d75cb 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -603,7 +603,7 @@ public:
 
     // Creates a Yieldable CollectionPtr that reloads the Collection pointer from the catalog when
     // restoring from yield
-    CollectionPtr(OperationContext* opCtx, const Collection* collection, uint64_t catalogEpoch);
+    CollectionPtr(OperationContext* opCtx, const Collection* collection);
 
     // Creates non-yieldable CollectionPtr, performing yield/restore will be a no-op.
     struct NoYieldTag {};
@@ -648,7 +648,7 @@ public:
     void restore() const override;
 
     static void installCatalogLookupImpl(
-        std::function<CollectionPtr(OperationContext*, CollectionUUID, uint64_t)> impl);
+        std::function<CollectionPtr(OperationContext*, CollectionUUID)> impl);
 
     friend std::ostream& operator<<(std::ostream& os, const CollectionPtr& coll);
 
@@ -659,9 +659,7 @@ private:
     // yield/restore to require a non-const instance when it otherwise could be const.
     mutable const Collection* _collection;
     mutable OptionalCollectionUUID _uuid;
-    mutable NamespaceString _ns;
     OperationContext* _opCtx;
-    uint64_t _catalogEpoch;
 };
 
 inline std::ostream& operator<<(std::ostream& os, const CollectionPtr& coll) {
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 6b528a97c43..cbfca442750 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -157,14 +157,10 @@ const OperationContext::Decoration<UncommittedWritableCollections>
 
 struct installCatalogLookupFn {
     installCatalogLookupFn() {
-        CollectionPtr::installCatalogLookupImpl(
-            [](OperationContext* opCtx, CollectionUUID uuid, uint64_t catalogEpoch) {
-                const auto& catalog = CollectionCatalog::get(opCtx);
-                if (catalog.getEpoch() != catalogEpoch)
-                    return CollectionPtr();
-
-                return catalog.lookupCollectionByUUID(opCtx, uuid);
-            });
+        CollectionPtr::installCatalogLookupImpl([](OperationContext* opCtx, CollectionUUID uuid) {
+            const auto& catalog = CollectionCatalog::get(opCtx);
+            return catalog.lookupCollectionByUUID(opCtx, uuid);
+        });
     }
 } inst;
 
@@ -223,7 +219,7 @@ CollectionCatalog::iterator::value_type CollectionCatalog::iterator::operator*()
         return CollectionPtr();
     }
 
-    return {_opCtx, _mapIter->second.get(), _catalog->getEpoch()};
+    return {_opCtx, _mapIter->second.get()};
 }
 
 Collection* CollectionCatalog::iterator::getWritableCollection(OperationContext* opCtx,
@@ -471,14 +467,12 @@ CollectionPtr CollectionCatalog::lookupCollectionByUUID(OperationContext* opCtx,
     }
 
     if (auto coll = UncommittedCollections::getForTxn(opCtx, uuid)) {
-        return {opCtx, coll.get(), getEpoch()};
+        return {opCtx, coll.get()};
     }
 
     stdx::lock_guard<Latch> lock(_catalogLock);
     auto coll = _lookupCollectionByUUID(lock, uuid);
-
-    return (coll && coll->isCommitted()) ? CollectionPtr(opCtx, coll.get(), getEpoch())
-                                         : CollectionPtr();
+    return (coll && coll->isCommitted()) ? CollectionPtr(opCtx, coll.get()) : CollectionPtr();
 }
 
 void CollectionCatalog::makeCollectionVisible(CollectionUUID uuid) {
@@ -569,13 +563,13 @@ CollectionPtr CollectionCatalog::lookupCollectionByNamespace(OperationContext* o
     }
 
     if (auto coll = UncommittedCollections::getForTxn(opCtx, nss)) {
-        return {opCtx, coll.get(), getEpoch()};
+        return {opCtx, coll.get()};
     }
 
     stdx::lock_guard<Latch> lock(_catalogLock);
     auto it = _collections.find(nss);
     auto coll = (it == _collections.end() ? nullptr : it->second);
-    return (coll && coll->isCommitted()) ? CollectionPtr(opCtx, coll.get(), getEpoch()) : nullptr;
+    return (coll && coll->isCommitted()) ? CollectionPtr(opCtx, coll.get()) : nullptr;
 }
 
 boost::optional<NamespaceString> CollectionCatalog::lookupNSSByUUID(OperationContext* opCtx,
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 55e82bd900f..b5dff5774c0 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -1280,7 +1280,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> CollectionImpl::makePlanExe
     auto direction = isForward ? InternalPlanner::FORWARD : InternalPlanner::BACKWARD;
     return InternalPlanner::collectionScan(opCtx,
                                            yieldableCollection->ns().ns(),
-                                           yieldableCollection,
+                                           &yieldableCollection,
                                            yieldPolicy,
                                            direction,
                                            resumeAfterRecordId);
diff --git a/src/mongo/db/catalog_raii.cpp b/src/mongo/db/catalog_raii.cpp
index e154d8d2425..f3b9e5aa039 100644
--- a/src/mongo/db/catalog_raii.cpp
+++ b/src/mongo/db/catalog_raii.cpp
@@ -157,27 +157,21 @@ Collection* AutoGetCollection::getWritableCollection(CollectionCatalog::Lifetime
         class WritableCollectionReset : public RecoveryUnit::Change {
         public:
             WritableCollectionReset(AutoGetCollection& autoColl,
-                                    const CollectionPtr& rollbackCollection,
-                                    uint64_t catalogEpoch)
-                : _autoColl(autoColl),
-                  _rollbackCollection(rollbackCollection.get()),
-                  _catalogEpoch(catalogEpoch) {}
+                                    const CollectionPtr& rollbackCollection)
+                : _autoColl(autoColl), _rollbackCollection(rollbackCollection.get()) {}
             void commit(boost::optional<Timestamp> commitTime) final {
                 // Restore coll to a yieldable collection
-                _autoColl._coll = {
-                    _autoColl.getOperationContext(), _autoColl._coll.get(), _catalogEpoch};
+                _autoColl._coll = {_autoColl.getOperationContext(), _autoColl._coll.get()};
                 _autoColl._writableColl = nullptr;
             }
             void rollback() final {
-                _autoColl._coll = {
-                    _autoColl.getOperationContext(), _rollbackCollection, _catalogEpoch};
+                _autoColl._coll = {_autoColl.getOperationContext(), _rollbackCollection};
                 _autoColl._writableColl = nullptr;
             }
 
         private:
             AutoGetCollection& _autoColl;
             const Collection* _rollbackCollection;
-            uint64_t _catalogEpoch;
         };
 
         auto& catalog = CollectionCatalog::get(_opCtx);
@@ -185,7 +179,7 @@ Collection* AutoGetCollection::getWritableCollection(CollectionCatalog::Lifetime
             catalog.lookupCollectionByNamespaceForMetadataWrite(_opCtx, mode, _resolvedNss);
         if (mode == CollectionCatalog::LifetimeMode::kManagedInWriteUnitOfWork) {
             _opCtx->recoveryUnit()->registerChange(
-                std::make_unique<WritableCollectionReset>(*this, _coll, catalog.getEpoch()));
+                std::make_unique<WritableCollectionReset>(*this, _coll));
         }
 
         // Set to writable collection. We are no longer yieldable.
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index 427a9f2da96..ed880886f40 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -190,7 +190,7 @@ public:
             opCtx, request.getCollation().value_or(BSONObj()), nss);
 
         auto statusWithPlanExecutor =
-            getExecutorCount(expCtx, collection, request, true /*explain*/, nss);
+            getExecutorCount(expCtx, &collection, request, true /*explain*/, nss);
         if (!statusWithPlanExecutor.isOK()) {
             return statusWithPlanExecutor.getStatus();
         }
@@ -252,7 +252,7 @@ public:
         auto statusWithPlanExecutor =
             getExecutorCount(makeExpressionContextForGetExecutor(
                                  opCtx, request.getCollation().value_or(BSONObj()), nss),
-                             collection,
+                             &collection,
                              request,
                              false /*explain*/,
                              nss);
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 57c21d78ebc..b608cb8622a 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -526,7 +526,7 @@ public:
                 return 1;
             }
             exec = InternalPlanner::collectionScan(
-                opCtx, ns, collection.getCollection(), PlanYieldPolicy::YieldPolicy::NO_YIELD);
+                opCtx, ns, &collection.getCollection(), PlanYieldPolicy::YieldPolicy::NO_YIELD);
         } else if (min.isEmpty() || max.isEmpty()) {
             errmsg = "only one of min or max specified";
             return false;
@@ -551,7 +551,7 @@ public:
             max = Helpers::toKeyFormat(kp.extendRangeBound(max, false));
 
             exec = InternalPlanner::indexScan(opCtx,
-                                              collection.getCollection(),
+                                              &collection.getCollection(),
                                               idx,
                                               min,
                                               max,
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index 5a8aafa3f3a..bfe0fc17cb1 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -304,7 +304,7 @@ public:
             const CollectionPtr& coll = ctx->getCollection();
 
             auto exec = uassertStatusOK(getExecutor(opCtx,
-                                                    coll,
+                                                    &coll,
                                                     std::move(cq),
                                                     PlanYieldPolicy::YieldPolicy::YIELD_MANUAL,
                                                     QueryPlannerParams::NO_TABLE_SCAN));
@@ -389,7 +389,7 @@ public:
                     }
 
                     // Now that we have the lock again, we can restore the PlanExecutor.
-                    exec->restoreState(&coll);
+                    exec->restoreState(&ctx->getCollection());
                 }
             } catch (DBException& exception) {
                 exception.addContext("Executor error during filemd5 command");
diff --git a/src/mongo/db/commands/dbhash.cpp b/src/mongo/db/commands/dbhash.cpp
index 6ef685dd420..bae3b34eb19 100644
--- a/src/mongo/db/commands/dbhash.cpp
+++ b/src/mongo/db/commands/dbhash.cpp
@@ -350,7 +350,7 @@ private:
         std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec;
         if (desc) {
             exec = InternalPlanner::indexScan(opCtx,
-                                              collection,
+                                              &collection,
                                               desc,
                                               BSONObj(),
                                               BSONObj(),
@@ -360,7 +360,7 @@ private:
                                               InternalPlanner::IXSCAN_FETCH);
         } else if (collection->isCapped()) {
             exec = InternalPlanner::collectionScan(
-                opCtx, nss.ns(), collection, PlanYieldPolicy::YieldPolicy::NO_YIELD);
+                opCtx, nss.ns(), &collection, PlanYieldPolicy::YieldPolicy::NO_YIELD);
         } else {
             LOGV2(20455,
                   "Can't find _id index for namespace: {namespace}",
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index 226d3b52183..0d00569097c 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -177,7 +177,7 @@ public:
         const auto& collection = ctx->getCollection();
 
         auto executor = uassertStatusOK(
-            getExecutorDistinct(collection, QueryPlannerParams::DEFAULT, &parsedDistinct));
+            getExecutorDistinct(&collection, QueryPlannerParams::DEFAULT, &parsedDistinct));
 
         auto bodyBuilder = result->getBodyBuilder();
         Explain::explainStages(executor.get(), collection, verbosity, BSONObj(), &bodyBuilder);
@@ -238,7 +238,7 @@ public:
         const auto& collection = ctx->getCollection();
 
         auto executor =
-            getExecutorDistinct(collection, QueryPlannerParams::DEFAULT, &parsedDistinct);
+            getExecutorDistinct(&collection, QueryPlannerParams::DEFAULT, &parsedDistinct);
         uassertStatusOK(executor.getStatus());
 
         {
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 4fe85f09807..877e54d6c1d 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -292,7 +292,7 @@ public:
             CollectionShardingState::get(opCtx, nsString)->checkShardVersionOrThrow(opCtx);
 
             const auto exec = uassertStatusOK(
-                getExecutorDelete(opDebug, collection.getCollection(), &parsedDelete, verbosity));
+                getExecutorDelete(opDebug, &collection.getCollection(), &parsedDelete, verbosity));
 
             auto bodyBuilder = result->getBodyBuilder();
             Explain::explainStages(
@@ -316,7 +316,7 @@ public:
             CollectionShardingState::get(opCtx, nsString)->checkShardVersionOrThrow(opCtx);
 
             const auto exec = uassertStatusOK(
-                getExecutorUpdate(opDebug, collection.getCollection(), &parsedUpdate, verbosity));
+                getExecutorUpdate(opDebug, &collection.getCollection(), &parsedUpdate, verbosity));
 
             auto bodyBuilder = result->getBodyBuilder();
             Explain::explainStages(
@@ -476,7 +476,7 @@ public:
         checkIfTransactionOnCappedColl(collection.getCollection(), inTransaction);
 
         const auto exec = uassertStatusOK(getExecutorDelete(
-            opDebug, collection.getCollection(), &parsedDelete, boost::none /* verbosity */));
+            opDebug, &collection.getCollection(), &parsedDelete, boost::none /* verbosity */));
 
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
@@ -563,7 +563,7 @@ public:
         checkIfTransactionOnCappedColl(collection, inTransaction);
 
         const auto exec = uassertStatusOK(
-            getExecutorUpdate(opDebug, collection, parsedUpdate, boost::none /* verbosity */));
+            getExecutorUpdate(opDebug, &collection, parsedUpdate, boost::none /* verbosity */));
 
         {
             stdx::lock_guard<Client> lk(*opCtx->getClient());
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index bf992594df1..96d612b60e5 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -294,7 +294,7 @@ public:
             // Get the execution plan for the query.
             bool permitYield = true;
             auto exec =
-                uassertStatusOK(getExecutorFind(opCtx, collection, std::move(cq), permitYield));
+                uassertStatusOK(getExecutorFind(opCtx, &collection, std::move(cq), permitYield));
 
             auto bodyBuilder = result->getBodyBuilder();
             // Got the execution tree. Explain it.
@@ -423,7 +423,7 @@ public:
             // Get the execution plan for the query.
             bool permitYield = true;
             auto exec =
-                uassertStatusOK(getExecutorFind(opCtx, collection, std::move(cq), permitYield));
+                uassertStatusOK(getExecutorFind(opCtx, &collection, std::move(cq), permitYield));
 
             {
                 stdx::lock_guard<Client> lk(*opCtx->getClient());
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index 54be8b56b81..64edcbfff86 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -516,7 +516,7 @@ public:
                 opCtx->recoveryUnit()->setReadOnce(true);
             }
             exec->reattachToOperationContext(opCtx);
-            exec->restoreState(nullptr);
+            exec->restoreState(readLock ? &readLock->getCollection() : nullptr);
 
             auto planSummary = exec->getPlanExplainer().getPlanSummary();
             {
@@ -584,10 +584,10 @@ public:
             // of this operation's CurOp to signal that we've hit this point and then spin until the
             // failpoint is released.
             std::function<void()> saveAndRestoreStateWithReadLockReacquisition =
-                [exec, dropAndReacquireReadLock]() {
+                [exec, dropAndReacquireReadLock, &readLock]() {
                     exec->saveState();
                     dropAndReacquireReadLock();
-                    exec->restoreState(nullptr);
+                    exec->restoreState(&readLock->getCollection());
                 };
 
             waitWithPinnedCursorDuringGetMoreBatch.execute([&](const BSONObj& data) {
diff --git a/src/mongo/db/commands/haystack.cpp b/src/mongo/db/commands/haystack.cpp
index b17c97d22f9..92fc8f7a2a2 100644
--- a/src/mongo/db/commands/haystack.cpp
+++ b/src/mongo/db/commands/haystack.cpp
@@ -157,7 +157,7 @@ public:
 
 
                 auto exec = InternalPlanner::indexScan(opCtx,
-                                                       collection,
+                                                       &collection,
                                                        ham->_descriptor,
                                                        key,
                                                        key,
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index d8485d09937..bd93f374fac 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -373,7 +373,7 @@ public:
                 uassertStatusOK(plan_executor_factory::make(expCtx,
                                                             std::move(ws),
                                                             std::move(root),
-                                                            nullptr,
+                                                            &CollectionPtr::null,
                                                             PlanYieldPolicy::YieldPolicy::NO_YIELD,
                                                             cursorNss));
 
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 94d4eb1e51d..1b1386e1902 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -186,7 +186,7 @@ public:
                 uassertStatusOK(plan_executor_factory::make(expCtx,
                                                             std::move(ws),
                                                             std::move(root),
-                                                            nullptr,
+                                                            &CollectionPtr::null,
                                                             PlanYieldPolicy::YieldPolicy::NO_YIELD,
                                                             nss));
 
diff --git a/src/mongo/db/commands/test_commands.cpp b/src/mongo/db/commands/test_commands.cpp
index a5844835ccb..c3813857782 100644
--- a/src/mongo/db/commands/test_commands.cpp
+++ b/src/mongo/db/commands/test_commands.cpp
@@ -162,7 +162,7 @@ public:
             // end.
             auto exec = InternalPlanner::collectionScan(opCtx,
                                                         fullNs.ns(),
-                                                        collection.getCollection(),
+                                                        &collection.getCollection(),
                                                         PlanYieldPolicy::YieldPolicy::NO_YIELD,
                                                         InternalPlanner::BACKWARD);
 
diff --git a/src/mongo/db/commands/write_commands/write_commands.cpp b/src/mongo/db/commands/write_commands/write_commands.cpp
index a016aef5de2..fca0e4f3de4 100644
--- a/src/mongo/db/commands/write_commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands/write_commands.cpp
@@ -461,8 +461,10 @@ private:
             // info is more accurate.
             AutoGetCollection collection(opCtx, _batch.getNamespace(), MODE_IX);
 
-            auto exec = uassertStatusOK(getExecutorUpdate(
-                &CurOp::get(opCtx)->debug(), collection.getCollection(), &parsedUpdate, verbosity));
+            auto exec = uassertStatusOK(getExecutorUpdate(&CurOp::get(opCtx)->debug(),
+                                                          &collection.getCollection(),
+                                                          &parsedUpdate,
+                                                          verbosity));
             auto bodyBuilder = result->getBodyBuilder();
             Explain::explainStages(
                 exec.get(), collection.getCollection(), verbosity, BSONObj(), &bodyBuilder);
@@ -556,8 +558,10 @@ private:
             AutoGetCollection collection(opCtx, _batch.getNamespace(), MODE_IX);
 
             // Explain the plan tree.
-            auto exec = uassertStatusOK(getExecutorDelete(
-                &CurOp::get(opCtx)->debug(), collection.getCollection(), &parsedDelete, verbosity));
+            auto exec = uassertStatusOK(getExecutorDelete(&CurOp::get(opCtx)->debug(),
+                                                          &collection.getCollection(),
+                                                          &parsedDelete,
+                                                          verbosity));
             auto bodyBuilder = result->getBodyBuilder();
             Explain::explainStages(
                 exec.get(), collection.getCollection(), verbosity, BSONObj(), &bodyBuilder);
diff --git a/src/mongo/db/db_raii_test.cpp b/src/mongo/db/db_raii_test.cpp
index 48f946ed895..5c633195f58 100644
--- a/src/mongo/db/db_raii_test.cpp
+++ b/src/mongo/db/db_raii_test.cpp
@@ -89,7 +89,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> makeTailableQueryPlan(
     std::unique_ptr<CanonicalQuery> cq = std::move(statusWithCQ.getValue());
 
     bool permitYield = true;
-    auto swExec = getExecutorFind(opCtx, collection, std::move(cq), permitYield);
+    auto swExec = getExecutorFind(opCtx, &collection, std::move(cq), permitYield);
     ASSERT_OK(swExec.getStatus());
     return std::move(swExec.getValue());
 }
diff --git a/src/mongo/db/dbhelpers.cpp b/src/mongo/db/dbhelpers.cpp
index 21e602052ea..09993f84475 100644
--- a/src/mongo/db/dbhelpers.cpp
+++ b/src/mongo/db/dbhelpers.cpp
@@ -109,7 +109,7 @@ RecordId Helpers::findOne(OperationContext* opCtx,
 
     size_t options = requireIndex ? QueryPlannerParams::NO_TABLE_SCAN : QueryPlannerParams::DEFAULT;
     auto exec = uassertStatusOK(getExecutor(
-        opCtx, collection, std::move(cq), PlanYieldPolicy::YieldPolicy::NO_YIELD, options));
+        opCtx, &collection, std::move(cq), PlanYieldPolicy::YieldPolicy::NO_YIELD, options));
 
     PlanExecutor::ExecState state;
     BSONObj obj;
@@ -188,7 +188,7 @@ bool Helpers::getSingleton(OperationContext* opCtx, const char* ns, BSONObj& res
     const auto& collection = getCollectionForRead(opCtx, NamespaceString(ns), autoColl, autoOplog);
 
     auto exec = InternalPlanner::collectionScan(
-        opCtx, ns, collection, PlanYieldPolicy::YieldPolicy::NO_YIELD);
+        opCtx, ns, &collection, PlanYieldPolicy::YieldPolicy::NO_YIELD);
     PlanExecutor::ExecState state = exec->getNext(&result, nullptr);
 
     CurOp::get(opCtx)->done();
@@ -210,7 +210,7 @@ bool Helpers::getLast(OperationContext* opCtx, const char* ns, BSONObj& result)
     const auto& collection = getCollectionForRead(opCtx, NamespaceString(ns), autoColl, autoOplog);
 
     auto exec = InternalPlanner::collectionScan(
-        opCtx, ns, collection, PlanYieldPolicy::YieldPolicy::NO_YIELD, InternalPlanner::BACKWARD);
+        opCtx, ns, &collection, PlanYieldPolicy::YieldPolicy::NO_YIELD, InternalPlanner::BACKWARD);
     PlanExecutor::ExecState state = exec->getNext(&result, nullptr);
 
     // Non-yielding collection scans from InternalPlanner will never error.
diff --git a/src/mongo/db/exec/delete.cpp b/src/mongo/db/exec/delete.cpp
index 73de1abdbc5..faca86675bd 100644
--- a/src/mongo/db/exec/delete.cpp
+++ b/src/mongo/db/exec/delete.cpp
@@ -219,7 +219,7 @@ PlanStage::StageState DeleteStage::doWork(WorkingSetID* out) {
     // they are created, and a WriteUnitOfWork is a transaction, make sure to restore the state
     // outside of the WriteUnitOfWork.
     try {
-        child()->restoreState();
+        child()->restoreState(&collection());
     } catch (const WriteConflictException&) {
         // Note we don't need to retry anything in this case since the delete already was committed.
         // However, we still need to return the deleted document (if it was requested).
diff --git a/src/mongo/db/exec/plan_stage.cpp b/src/mongo/db/exec/plan_stage.cpp
index 9df5674dc97..2787d63cf47 100644
--- a/src/mongo/db/exec/plan_stage.cpp
+++ b/src/mongo/db/exec/plan_stage.cpp
@@ -46,13 +46,13 @@ void PlanStage::saveState() {
     doSaveState();
 }
 
-void PlanStage::restoreState() {
+void PlanStage::restoreState(const RestoreContext& context) {
     ++_commonStats.unyields;
     for (auto&& child : _children) {
-        child->restoreState();
+        child->restoreState(context);
     }
 
-    doRestoreState();
+    doRestoreState(context);
 }
 
 void PlanStage::detachFromOperationContext() {
diff --git a/src/mongo/db/exec/plan_stage.h b/src/mongo/db/exec/plan_stage.h
index 69c048ec8f7..ce60a82d9b4 100644
--- a/src/mongo/db/exec/plan_stage.h
+++ b/src/mongo/db/exec/plan_stage.h
@@ -36,6 +36,7 @@
 #include "mongo/db/exec/scoped_timer.h"
 #include "mongo/db/exec/working_set.h"
 #include "mongo/db/pipeline/expression_context.h"
+#include "mongo/db/query/restore_context.h"
 
 namespace mongo {
 
@@ -255,11 +256,15 @@ public:
      *
      * Propagates to all children, then calls doRestoreState().
      *
+     * RestoreContext is a context containing external state needed by plan stages to be able to
+     * restore into a valid state. The RequiresCollectionStage requires a valid CollectionPtr for
+     * example.
+     *
      * Throws a UserException on failure to restore due to a conflicting event such as a
      * collection drop. May throw a WriteConflictException, in which case the caller may choose to
      * retry.
      */
-    void restoreState();
+    void restoreState(const RestoreContext& context);
 
     /**
      * Detaches from the OperationContext and releases any storage-engine state.
@@ -366,7 +371,7 @@ protected:
     /**
      * Restores any stage-specific saved state and prepares to handle calls to work().
      */
-    virtual void doRestoreState() {}
+    virtual void doRestoreState(const RestoreContext& context) {}
 
     /**
      * Does stage-specific detaching.
diff --git a/src/mongo/db/exec/queued_data_stage_test.cpp b/src/mongo/db/exec/queued_data_stage_test.cpp
index 264e57b387d..03c0c4e7cd5 100644
--- a/src/mongo/db/exec/queued_data_stage_test.cpp
+++ b/src/mongo/db/exec/queued_data_stage_test.cpp
@@ -112,7 +112,7 @@ TEST_F(QueuedDataStageTest, ValidateStats) {
     ASSERT_EQUALS(stats->yields, 1U);
 
     // unyields
-    mock->restoreState();
+    mock->restoreState({});
     ASSERT_EQUALS(stats->unyields, 1U);
 
 
diff --git a/src/mongo/db/exec/requires_collection_stage.cpp b/src/mongo/db/exec/requires_collection_stage.cpp
index c4b726d08cb..f0345d7b311 100644
--- a/src/mongo/db/exec/requires_collection_stage.cpp
+++ b/src/mongo/db/exec/requires_collection_stage.cpp
@@ -35,40 +35,63 @@ namespace mongo {
 
 void RequiresCollectionStage::doSaveState() {
     doSaveStateRequiresCollection();
-
-    // A stage may not access storage while in a saved state.
-    _collection = CollectionPtr();
 }
 
-void RequiresCollectionStage::doRestoreState() {
-    invariant(!_collection);
-
+void RequiresCollectionStage::doRestoreState(const RestoreContext& context) {
     // We should be holding a lock associated with the name of the collection prior to yielding,
     // even if the collection was renamed during yield.
     dassert(opCtx()->lockState()->isCollectionLockedForMode(_nss, MODE_IS));
 
-    const CollectionCatalog& catalog = CollectionCatalog::get(opCtx());
-    auto newNss = catalog.lookupNSSByUUID(opCtx(), _collectionUUID);
-    uassert(ErrorCodes::QueryPlanKilled,
-            str::stream() << "collection dropped. UUID " << _collectionUUID,
-            newNss);
+    auto collectionDropped = [this]() {
+        uasserted(ErrorCodes::QueryPlanKilled,
+                  str::stream() << "collection dropped. UUID " << _collectionUUID);
+    };
 
-    // TODO SERVER-31695: Allow queries to survive collection rename, rather than throwing here when
-    // a rename has happened during yield.
-    uassert(ErrorCodes::QueryPlanKilled,
-            str::stream() << "collection renamed from '" << _nss << "' to '" << *newNss
-                          << "'. UUID " << _collectionUUID,
-            *newNss == _nss);
-
-    // At this point we know that the collection name has not changed, and therefore we have
-    // restored locks on the correct name. It is now safe to restore the Collection pointer. The
-    // collection must exist, since we already successfully looked up the namespace string by UUID
-    // under the correct lock manager locks.
-    // TODO SERVER-51115: We can't have every instance of RequiresCollectionStage do a catalog
-    // lookup with lock free reads. If we have multiple instances within a single executor they
-    // might get different pointers.
-    _collection = catalog.lookupCollectionByUUID(opCtx(), _collectionUUID);
-    invariant(_collection);
+    auto collectionRenamed = [this](const NamespaceString& newNss) {
+        uasserted(ErrorCodes::QueryPlanKilled,
+                  str::stream() << "collection renamed from '" << _nss << "' to '" << newNss
+                                << "'. UUID " << _collectionUUID);
+    };
+
+    if (context.type() == RestoreContext::RestoreType::kExternal) {
+        // RequiresCollectionStage requires a collection to be provided in restore. However, it may
+        // be null in case the collection got dropped or renamed.
+        auto collPtr = context.collection();
+        invariant(collPtr);
+        _collection = collPtr;
+
+        // If we restore externally and get a null Collection we need to figure out if this was a
+        // drop or rename. The external lookup could have been done for UUID or namespace.
+        const auto& coll = *collPtr;
+
+        // If collection exists uuid does not match assume lookup was over namespace and treat this
+        // as a drop.
+        if (coll && coll->uuid() != _collectionUUID) {
+            collectionDropped();
+        }
+
+        // If we didn't get a valid collection but can still find the UUID in the catalog then we
+        // treat this as a rename.
+        if (!coll) {
+            const CollectionCatalog& catalog = CollectionCatalog::get(opCtx());
+            auto newNss = catalog.lookupNSSByUUID(opCtx(), _collectionUUID);
+            if (newNss && *newNss != _nss) {
+                collectionRenamed(*newNss);
+            }
+        }
+    }
+
+    const auto& coll = *_collection;
+
+    if (!coll) {
+        collectionDropped();
+    }
+
+    // TODO SERVER-31695: Allow queries to survive collection rename, rather than throwing here
+    // when a rename has happened during yield.
+    if (const auto& newNss = coll->ns(); newNss != _nss) {
+        collectionRenamed(newNss);
+    }
 
     uassert(ErrorCodes::QueryPlanKilled,
             str::stream() << "The catalog was closed and reopened",
diff --git a/src/mongo/db/exec/requires_collection_stage.h b/src/mongo/db/exec/requires_collection_stage.h
index bbe6ea76d97..4e3484abf53 100644
--- a/src/mongo/db/exec/requires_collection_stage.h
+++ b/src/mongo/db/exec/requires_collection_stage.h
@@ -53,19 +53,17 @@ public:
                             ExpressionContext* expCtx,
                             const CollectionPtr& coll)
         : PlanStage(stageType, expCtx),
-          _collection(coll.detached()),
-          _collectionUUID(_collection->uuid()),
+          _collection(&coll),
+          _collectionUUID(coll->uuid()),
           _catalogEpoch(getCatalogEpoch()),
-          _nss(_collection->ns()) {
-        invariant(_collection);
-    }
+          _nss(coll->ns()) {}
 
     virtual ~RequiresCollectionStage() = default;
 
 protected:
     void doSaveState() final;
 
-    void doRestoreState() final;
+    void doRestoreState(const RestoreContext& context) final;
 
     /**
      * Performs yield preparation specific to a stage which subclasses from RequiresCollectionStage.
@@ -78,7 +76,7 @@ protected:
     virtual void doRestoreStateRequiresCollection() = 0;
 
     const CollectionPtr& collection() const {
-        return _collection;
+        return *_collection;
     }
 
     UUID uuid() const {
@@ -91,7 +89,11 @@ private:
         return CollectionCatalog::get(opCtx()).getEpoch();
     }
 
-    CollectionPtr _collection;
+    // Pointer to a CollectionPtr that is stored at a high level in a AutoGetCollection or other
+    // helper. It needs to stay valid until the PlanExecutor saves its state. To avoid this pointer
+    // from dangling it needs to be reset when doRestoreState() is called and it is reset to a
+    // different CollectionPtr.
+    const CollectionPtr* _collection;
     const UUID _collectionUUID;
     const uint64_t _catalogEpoch;
 
diff --git a/src/mongo/db/exec/sbe_cmd.cpp b/src/mongo/db/exec/sbe_cmd.cpp
index 1e46cec6d4f..cd60b97b22c 100644
--- a/src/mongo/db/exec/sbe_cmd.cpp
+++ b/src/mongo/db/exec/sbe_cmd.cpp
@@ -82,8 +82,12 @@ public:
         data.resultSlot = resultSlot;
         data.recordIdSlot = recordIdSlot;
 
-        exec = uassertStatusOK(plan_executor_factory::make(
-            opCtx, nullptr, {std::move(root), std::move(data)}, nullptr, nss, nullptr));
+        exec = uassertStatusOK(plan_executor_factory::make(opCtx,
+                                                           nullptr,
+                                                           {std::move(root), std::move(data)},
+                                                           &CollectionPtr::null,
+                                                           nss,
+                                                           nullptr));
 
         for (long long objCount = 0; objCount < batchSize; objCount++) {
             BSONObj next;
diff --git a/src/mongo/db/exec/stagedebug_cmd.cpp b/src/mongo/db/exec/stagedebug_cmd.cpp
index 856d2d527c4..8443f33a6f2 100644
--- a/src/mongo/db/exec/stagedebug_cmd.cpp
+++ b/src/mongo/db/exec/stagedebug_cmd.cpp
@@ -183,7 +183,7 @@ public:
             plan_executor_factory::make(expCtx,
                                         std::move(ws),
                                         std::move(rootFetch),
-                                        collection,
+                                        &collection,
                                         PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
         fassert(28536, statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
diff --git a/src/mongo/db/exec/update_stage.cpp b/src/mongo/db/exec/update_stage.cpp
index 5cecfb72314..a86f03e707b 100644
--- a/src/mongo/db/exec/update_stage.cpp
+++ b/src/mongo/db/exec/update_stage.cpp
@@ -479,7 +479,7 @@ PlanStage::StageState UpdateStage::doWork(WorkingSetID* out) {
         // As restoreState may restore (recreate) cursors, make sure to restore the
         // state outside of the WritUnitOfWork.
         try {
-            child()->restoreState();
+            child()->restoreState(&collection());
         } catch (const WriteConflictException&) {
             // Note we don't need to retry updating anything in this case since the update
             // already was committed. However, we still need to return the updated document
diff --git a/src/mongo/db/ops/delete.cpp b/src/mongo/db/ops/delete.cpp
index a546a57ea76..5bc4f83b19b 100644
--- a/src/mongo/db/ops/delete.cpp
+++ b/src/mongo/db/ops/delete.cpp
@@ -58,7 +58,7 @@ long long deleteObjects(OperationContext* opCtx,
     uassertStatusOK(parsedDelete.parseRequest());
 
     auto exec = uassertStatusOK(getExecutorDelete(
-        &CurOp::get(opCtx)->debug(), collection, &parsedDelete, boost::none /* verbosity */));
+        &CurOp::get(opCtx)->debug(), &collection, &parsedDelete, boost::none /* verbosity */));
 
     return exec->executeDelete();
 }
diff --git a/src/mongo/db/ops/update.cpp b/src/mongo/db/ops/update.cpp
index 899dbac9713..455afeadbf9 100644
--- a/src/mongo/db/ops/update.cpp
+++ b/src/mongo/db/ops/update.cpp
@@ -93,7 +93,7 @@ UpdateResult update(OperationContext* opCtx, Database* db, const UpdateRequest&
 
     OpDebug* const nullOpDebug = nullptr;
     auto exec = uassertStatusOK(
-        getExecutorUpdate(nullOpDebug, collection, &parsedUpdate, boost::none /* verbosity */));
+        getExecutorUpdate(nullOpDebug, &collection, &parsedUpdate, boost::none /* verbosity */));
 
     return exec->executeUpdate();
 }
diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp
index 981f1a542a4..22d6c9e29cb 100644
--- a/src/mongo/db/ops/write_ops_exec.cpp
+++ b/src/mongo/db/ops/write_ops_exec.cpp
@@ -666,7 +666,7 @@ static SingleWriteResult performSingleUpdateOp(OperationContext* opCtx,
     assertCanWrite_inlock(opCtx, ns);
 
     auto exec = uassertStatusOK(getExecutorUpdate(
-        &curOp.debug(), collection->getCollection(), &parsedUpdate, boost::none /* verbosity */));
+        &curOp.debug(), &collection->getCollection(), &parsedUpdate, boost::none /* verbosity */));
 
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
@@ -905,7 +905,7 @@ static SingleWriteResult performSingleDeleteOp(OperationContext* opCtx,
         &hangWithLockDuringBatchRemove, opCtx, "hangWithLockDuringBatchRemove");
 
     auto exec = uassertStatusOK(getExecutorDelete(
-        &curOp.debug(), collection.getCollection(), &parsedDelete, boost::none /* verbosity */));
+        &curOp.debug(), &collection.getCollection(), &parsedDelete, boost::none /* verbosity */));
 
     {
         stdx::lock_guard<Client> lk(*opCtx->getClient());
diff --git a/src/mongo/db/pipeline/document_source_check_resume_token_test.cpp b/src/mongo/db/pipeline/document_source_check_resume_token_test.cpp
index 71f01ec190c..fbea1ea1076 100644
--- a/src/mongo/db/pipeline/document_source_check_resume_token_test.cpp
+++ b/src/mongo/db/pipeline/document_source_check_resume_token_test.cpp
@@ -159,7 +159,7 @@ private:
 class DocumentSourceChangeStreamMock : public DocumentSourceMock {
 public:
     DocumentSourceChangeStreamMock(const boost::intrusive_ptr<ExpressionContextForTest>& expCtx)
-        : DocumentSourceMock({}, expCtx) {
+        : DocumentSourceMock({}, expCtx), _collectionPtr(&_collection) {
         _filterExpr = BSON("ns" << kTestNs);
         _filter = MatchExpressionParser::parseAndNormalize(_filterExpr, pExpCtx);
         _params.assertMinTsHasNotFallenOffOplog = true;
@@ -199,7 +199,7 @@ protected:
         // If this is the first call to doGetNext, we must create the COLLSCAN.
         if (!_collScan) {
             _collScan = std::make_unique<CollectionScan>(
-                pExpCtx.get(), &_collection, _params, &_ws, _filter.get());
+                pExpCtx.get(), _collectionPtr, _params, &_ws, _filter.get());
             // The first call to doWork will create the cursor and return NEED_TIME. But it won't
             // actually scan any of the documents that are present in the mock cursor queue.
             ASSERT_EQ(_collScan->doWork(nullptr), PlanStage::NEED_TIME);
@@ -239,6 +239,7 @@ private:
     }
 
     ChangeStreamOplogCollectionMock _collection;
+    CollectionPtr _collectionPtr;
     std::unique_ptr<CollectionScan> _collScan;
     CollectionScanParams _params;
 
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index c31a36ff29f..a2a383ed0fe 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -170,7 +170,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createRandomCursorEx
     }
 
     auto exec = plan_executor_factory::make(
-        expCtx, std::move(ws), std::move(root), coll, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+        expCtx, std::move(ws), std::move(root), &coll, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
 
     // For sharded collections, the root of the plan tree is a TrialStage that may have chosen
     // either a random-sampling cursor trial plan or a COLLSCAN backup plan. We can only optimize
@@ -252,7 +252,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
         // example, if we have a document {a: [1,2]} and group by "a" a DISTINCT_SCAN on an "a"
         // index would produce one result for '1' and another for '2', which would be incorrect.
         auto distinctExecutor = getExecutorDistinct(
-            collection, plannerOpts | QueryPlannerParams::STRICT_DISTINCT_ONLY, &parsedDistinct);
+            &collection, plannerOpts | QueryPlannerParams::STRICT_DISTINCT_ONLY, &parsedDistinct);
         if (!distinctExecutor.isOK()) {
             return distinctExecutor.getStatus().withContext(
                 "Unable to use distinct scan to optimize $group stage");
@@ -266,7 +266,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
 
     bool permitYield = true;
     return getExecutorFind(
-        expCtx->opCtx, collection, std::move(cq.getValue()), permitYield, plannerOpts);
+        expCtx->opCtx, &collection, std::move(cq.getValue()), permitYield, plannerOpts);
 }
 
 /**
diff --git a/src/mongo/db/pipeline/plan_executor_pipeline.h b/src/mongo/db/pipeline/plan_executor_pipeline.h
index 2bd7ae78041..af3513315ed 100644
--- a/src/mongo/db/pipeline/plan_executor_pipeline.h
+++ b/src/mongo/db/pipeline/plan_executor_pipeline.h
@@ -63,7 +63,7 @@ public:
     // underlying data access plan is saved/restored internally in between DocumentSourceCursor
     // batches, or when the underlying PlanStage tree yields.
     void saveState() override {}
-    void restoreState(const Yieldable* yieldable) override {}
+    void restoreState(const RestoreContext&) override {}
 
     void detachFromOperationContext() override {
         _pipeline->detachFromOperationContext();
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index b980c9a9ef6..eee9ce240cd 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -633,7 +633,7 @@ bool runQuery(OperationContext* opCtx,
     constexpr auto verbosity = ExplainOptions::Verbosity::kExecAllPlans;
     expCtx->explain = qr.isExplain() ? boost::make_optional(verbosity) : boost::none;
     auto exec =
-        uassertStatusOK(getExecutorLegacyFind(opCtx, collection.getCollection(), std::move(cq)));
+        uassertStatusOK(getExecutorLegacyFind(opCtx, &collection.getCollection(), std::move(cq)));
 
     // If it's actually an explain, do the explain and return rather than falling through
     // to the normal query execution loop.
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index d5e64f2fd05..a44eba9ef71 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -958,13 +958,13 @@ private:
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getClassicExecutor(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
     auto ws = std::make_unique<WorkingSet>();
     ClassicPrepareExecutionHelper helper{
-        opCtx, collection, ws.get(), canonicalQuery.get(), nullptr, plannerOptions};
+        opCtx, *collection, ws.get(), canonicalQuery.get(), nullptr, plannerOptions};
     auto executionResult = helper.prepare();
     if (!executionResult.isOK()) {
         return executionResult.getStatus();
@@ -1055,7 +1055,7 @@ std::unique_ptr<PlanYieldPolicySBE> makeSbeYieldPolicy(
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExecutor(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> cq,
     PlanYieldPolicy::YieldPolicy requestedYieldPolicy,
     size_t plannerOptions) {
@@ -1063,7 +1063,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
     auto nss = cq->nss();
     auto yieldPolicy = makeSbeYieldPolicy(opCtx, requestedYieldPolicy, nss);
     SlotBasedPrepareExecutionHelper helper{
-        opCtx, collection, cq.get(), yieldPolicy.get(), plannerOptions};
+        opCtx, *collection, cq.get(), yieldPolicy.get(), plannerOptions};
     auto executionResult = helper.prepare();
     if (!executionResult.isOK()) {
         return executionResult.getStatus();
@@ -1074,7 +1074,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
     auto&& solutions = result->solutions();
 
     if (auto planner = makeRuntimePlannerIfNeeded(opCtx,
-                                                  collection,
+                                                  *collection,
                                                   cq.get(),
                                                   solutions.size(),
                                                   result->decisionWorks(),
@@ -1104,7 +1104,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
@@ -1123,7 +1123,7 @@ namespace {
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> _getExecutorFind(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
@@ -1138,7 +1138,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> _getExecutorFin
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     bool permitYield,
     size_t plannerOptions) {
@@ -1151,7 +1151,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorLegacyFind(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery) {
     return _getExecutorFind(opCtx,
                             collection,
@@ -1210,9 +1210,10 @@ StatusWith<std::unique_ptr<projection_ast::Projection>> makeProjection(const BSO
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDelete(
     OpDebug* opDebug,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     ParsedDelete* parsedDelete,
     boost::optional<ExplainOptions::Verbosity> verbosity) {
+    const auto& collection = *coll;
     auto expCtx = parsedDelete->expCtx();
     OperationContext* opCtx = expCtx->opCtx;
     const DeleteRequest* request = parsedDelete->getRequest();
@@ -1257,8 +1258,12 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDele
                     "Collection does not exist. Using EOF stage",
                     "namespace"_attr = nss.ns(),
                     "query"_attr = redact(request->getQuery()));
-        return plan_executor_factory::make(
-            expCtx, std::move(ws), std::make_unique<EOFStage>(expCtx.get()), nullptr, policy, nss);
+        return plan_executor_factory::make(expCtx,
+                                           std::move(ws),
+                                           std::make_unique<EOFStage>(expCtx.get()),
+                                           &CollectionPtr::null,
+                                           policy,
+                                           nss);
     }
 
     if (!parsedDelete->hasParsedQuery()) {
@@ -1297,7 +1302,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDele
                                                   collection,
                                                   idHackStage.release());
                 return plan_executor_factory::make(
-                    expCtx, std::move(ws), std::move(root), collection, policy);
+                    expCtx, std::move(ws), std::move(root), &collection, policy);
             }
         }
 
@@ -1356,7 +1361,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDele
     return plan_executor_factory::make(std::move(cq),
                                        std::move(ws),
                                        std::move(root),
-                                       collection,
+                                       &collection,
                                        policy,
                                        NamespaceString(),
                                        std::move(querySolution));
@@ -1368,9 +1373,11 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDele
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorUpdate(
     OpDebug* opDebug,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     ParsedUpdate* parsedUpdate,
     boost::optional<ExplainOptions::Verbosity> verbosity) {
+    const auto& collection = *coll;
+
     auto expCtx = parsedUpdate->expCtx();
     OperationContext* opCtx = expCtx->opCtx;
 
@@ -1425,8 +1432,12 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorUpda
                     "Collection does not exist. Using EOF stage",
                     "namespace"_attr = nss.ns(),
                     "query"_attr = redact(request->getQuery()));
-        return plan_executor_factory::make(
-            expCtx, std::move(ws), std::make_unique<EOFStage>(expCtx.get()), nullptr, policy, nss);
+        return plan_executor_factory::make(expCtx,
+                                           std::move(ws),
+                                           std::make_unique<EOFStage>(expCtx.get()),
+                                           &CollectionPtr::null,
+                                           policy,
+                                           nss);
     }
 
     // Pass index information to the update driver, so that it can determine for us whether the
@@ -1454,7 +1465,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorUpda
                 // Working set 'ws' is discarded. InternalPlanner::updateWithIdHack() makes its own
                 // WorkingSet.
                 return InternalPlanner::updateWithIdHack(opCtx,
-                                                         collection,
+                                                         &collection,
                                                          updateStageParams,
                                                          descriptor,
                                                          unparsedQuery["_id"].wrap(),
@@ -1521,7 +1532,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorUpda
     return plan_executor_factory::make(std::move(cq),
                                        std::move(ws),
                                        std::move(root),
-                                       collection,
+                                       &collection,
                                        policy,
                                        NamespaceString(),
                                        std::move(querySolution));
@@ -1666,10 +1677,12 @@ bool getDistinctNodeIndex(const std::vector<IndexEntry>& indices,
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCount(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     const CountCommand& request,
     bool explain,
     const NamespaceString& nss) {
+    const auto& collection = *coll;
+
     OperationContext* opCtx = expCtx->opCtx;
     std::unique_ptr<WorkingSet> ws = std::make_unique<WorkingSet>();
 
@@ -1708,7 +1721,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCoun
         std::unique_ptr<PlanStage> root = std::make_unique<CountStage>(
             expCtx.get(), collection, limit, skip, ws.get(), new EOFStage(expCtx.get()));
         return plan_executor_factory::make(
-            expCtx, std::move(ws), std::move(root), nullptr, yieldPolicy, nss);
+            expCtx, std::move(ws), std::move(root), &CollectionPtr::null, yieldPolicy, nss);
     }
 
     // If the query is empty, then we can determine the count by just asking the collection
@@ -1724,7 +1737,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCoun
         std::unique_ptr<PlanStage> root =
             std::make_unique<RecordStoreFastCountStage>(expCtx.get(), collection, skip, limit);
         return plan_executor_factory::make(
-            expCtx, std::move(ws), std::move(root), nullptr, yieldPolicy, nss);
+            expCtx, std::move(ws), std::move(root), &CollectionPtr::null, yieldPolicy, nss);
     }
 
     size_t plannerOptions = QueryPlannerParams::IS_COUNT;
@@ -1751,7 +1764,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCoun
     return plan_executor_factory::make(std::move(cq),
                                        std::move(ws),
                                        std::move(root),
-                                       collection,
+                                       coll,
                                        yieldPolicy,
                                        NamespaceString(),
                                        std::move(querySolution));
@@ -2017,10 +2030,12 @@ QueryPlannerParams fillOutPlannerParamsForDistinct(OperationContext* opCtx,
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorForSimpleDistinct(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     const QueryPlannerParams& plannerParams,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     ParsedDistinct* parsedDistinct) {
+    const auto& collection = *coll;
+
     invariant(parsedDistinct->getQuery());
     auto collator = parsedDistinct->getQuery()->getCollator();
 
@@ -2071,7 +2086,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorForS
     return plan_executor_factory::make(parsedDistinct->releaseQuery(),
                                        std::move(ws),
                                        std::move(root),
-                                       collection,
+                                       coll,
                                        yieldPolicy,
                                        NamespaceString(),
                                        std::move(soln));
@@ -2091,11 +2106,13 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorForS
 // 'strictDistinctOnly' parameter.
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
 getExecutorDistinctFromIndexSolutions(OperationContext* opCtx,
-                                      const CollectionPtr& collection,
+                                      const CollectionPtr* coll,
                                       std::vector<std::unique_ptr<QuerySolution>> solutions,
                                       PlanYieldPolicy::YieldPolicy yieldPolicy,
                                       ParsedDistinct* parsedDistinct,
                                       bool strictDistinctOnly) {
+    const auto& collection = *coll;
+
     // We look for a solution that has an ixscan we can turn into a distinctixscan
     for (size_t i = 0; i < solutions.size(); ++i) {
         if (turnIxscanIntoDistinctIxscan(
@@ -2117,7 +2134,7 @@ getExecutorDistinctFromIndexSolutions(OperationContext* opCtx,
             return plan_executor_factory::make(parsedDistinct->releaseQuery(),
                                                std::move(ws),
                                                std::move(root),
-                                               collection,
+                                               coll,
                                                yieldPolicy,
                                                NamespaceString(),
                                                std::move(currentSolution));
@@ -2133,10 +2150,12 @@ getExecutorDistinctFromIndexSolutions(OperationContext* opCtx,
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorWithoutProjection(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     const CanonicalQuery* cq,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions) {
+    const auto& collection = *coll;
+
     auto qr = std::make_unique<QueryRequest>(cq->getQueryRequest());
     qr->setProj(BSONObj());
 
@@ -2150,12 +2169,14 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorWith
                                      MatchExpressionParser::kAllowAllSpecialFeatures);
 
     return getExecutor(
-        opCtx, collection, std::move(cqWithoutProjection.getValue()), yieldPolicy, plannerOptions);
+        opCtx, coll, std::move(cqWithoutProjection.getValue()), yieldPolicy, plannerOptions);
 }
 }  // namespace
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDistinct(
-    const CollectionPtr& collection, size_t plannerOptions, ParsedDistinct* parsedDistinct) {
+    const CollectionPtr* coll, size_t plannerOptions, ParsedDistinct* parsedDistinct) {
+    const auto& collection = *coll;
+
     auto expCtx = parsedDistinct->getQuery()->getExpCtx();
     OperationContext* opCtx = expCtx->opCtx;
     const auto yieldPolicy = opCtx->inMultiDocumentTransaction()
@@ -2167,7 +2188,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
         return plan_executor_factory::make(parsedDistinct->releaseQuery(),
                                            std::make_unique<WorkingSet>(),
                                            std::make_unique<EOFStage>(expCtx.get()),
-                                           collection,
+                                           coll,
                                            yieldPolicy);
     }
 
@@ -2198,7 +2219,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
             // the fields in the projection. That's definitely not possible in this case, so we
             // dispense with the projection.
             return getExecutorWithoutProjection(
-                opCtx, collection, parsedDistinct->getQuery(), yieldPolicy, plannerOptions);
+                opCtx, coll, parsedDistinct->getQuery(), yieldPolicy, plannerOptions);
         }
     }
 
@@ -2207,7 +2228,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
     //
 
     auto executorWithStatus =
-        getExecutorForSimpleDistinct(opCtx, collection, plannerParams, yieldPolicy, parsedDistinct);
+        getExecutorForSimpleDistinct(opCtx, coll, plannerParams, yieldPolicy, parsedDistinct);
     if (!executorWithStatus.isOK() || executorWithStatus.getValue()) {
         // We either got a DISTINCT plan or a fatal error.
         return executorWithStatus;
@@ -2223,7 +2244,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
             return {nullptr};
         } else {
             return getExecutor(
-                opCtx, collection, parsedDistinct->releaseQuery(), yieldPolicy, plannerOptions);
+                opCtx, coll, parsedDistinct->releaseQuery(), yieldPolicy, plannerOptions);
         }
     }
     auto solutions = std::move(statusWithSolutions.getValue());
@@ -2234,7 +2255,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
     // executor will still need deduplication.
     executorWithStatus = getExecutorDistinctFromIndexSolutions(
         opCtx,
-        collection,
+        coll,
         std::move(solutions),
         yieldPolicy,
         parsedDistinct,
@@ -2249,7 +2270,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
         // possible to cover the fields in the projection. That's definitely not possible in this
         // case, so we dispense with the projection.
         return getExecutorWithoutProjection(
-            opCtx, collection, parsedDistinct->getQuery(), yieldPolicy, plannerOptions);
+            opCtx, coll, parsedDistinct->getQuery(), yieldPolicy, plannerOptions);
     } else {
         // We did not find a solution that we could convert to DISTINCT_SCAN, and the
         // STRICT_DISTINCT_ONLY prohibits us from using any other kind of plan, so we return
diff --git a/src/mongo/db/query/get_executor.h b/src/mongo/db/query/get_executor.h
index 1cdb332dcdc..21637b03b88 100644
--- a/src/mongo/db/query/get_executor.h
+++ b/src/mongo/db/query/get_executor.h
@@ -120,7 +120,7 @@ bool shouldWaitForOplogVisibility(OperationContext* opCtx,
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     size_t plannerOptions = 0);
@@ -137,7 +137,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     bool permitYield = false,
     size_t plannerOptions = QueryPlannerParams::DEFAULT);
@@ -147,7 +147,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorLegacyFind(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery);
 
 /**
@@ -204,7 +204,7 @@ bool turnIxscanIntoDistinctIxscan(QuerySolution* soln,
  * distinct.
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDistinct(
-    const CollectionPtr& collection, size_t plannerOptions, ParsedDistinct* parsedDistinct);
+    const CollectionPtr* collection, size_t plannerOptions, ParsedDistinct* parsedDistinct);
 
 /*
  * Get a PlanExecutor for a query executing as part of a count command.
@@ -215,7 +215,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDist
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCount(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     const CountCommand& request,
     bool explain,
     const NamespaceString& nss);
@@ -241,7 +241,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCoun
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDelete(
     OpDebug* opDebug,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     ParsedDelete* parsedDelete,
     boost::optional<ExplainOptions::Verbosity> verbosity);
 
@@ -267,7 +267,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDele
  */
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorUpdate(
     OpDebug* opDebug,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     ParsedUpdate* parsedUpdate,
     boost::optional<ExplainOptions::Verbosity> verbosity);
 }  // namespace mongo
diff --git a/src/mongo/db/query/internal_plans.cpp b/src/mongo/db/query/internal_plans.cpp
index f7c6beeefac..612cfe0e02f 100644
--- a/src/mongo/db/query/internal_plans.cpp
+++ b/src/mongo/db/query/internal_plans.cpp
@@ -51,10 +51,12 @@ namespace mongo {
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::collectionScan(
     OperationContext* opCtx,
     StringData ns,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     const Direction direction,
     boost::optional<RecordId> resumeAfterRecordId) {
+    const auto& collection = *coll;
+
     std::unique_ptr<WorkingSet> ws = std::make_unique<WorkingSet>();
 
     auto expCtx = make_intrusive<ExpressionContext>(
@@ -63,42 +65,47 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::collection
     if (!collection) {
         auto eof = std::make_unique<EOFStage>(expCtx.get());
         // Takes ownership of 'ws' and 'eof'.
-        auto statusWithPlanExecutor = plan_executor_factory::make(
-            expCtx, std::move(ws), std::move(eof), nullptr, yieldPolicy, NamespaceString(ns));
+        auto statusWithPlanExecutor = plan_executor_factory::make(expCtx,
+                                                                  std::move(ws),
+                                                                  std::move(eof),
+                                                                  &CollectionPtr::null,
+                                                                  yieldPolicy,
+                                                                  NamespaceString(ns));
         invariant(statusWithPlanExecutor.isOK());
         return std::move(statusWithPlanExecutor.getValue());
     }
 
     invariant(ns == collection->ns().ns());
 
-    auto cs = _collectionScan(expCtx, ws.get(), collection, direction, resumeAfterRecordId);
+    auto cs = _collectionScan(expCtx, ws.get(), &collection, direction, resumeAfterRecordId);
 
     // Takes ownership of 'ws' and 'cs'.
     auto statusWithPlanExecutor =
-        plan_executor_factory::make(expCtx, std::move(ws), std::move(cs), collection, yieldPolicy);
+        plan_executor_factory::make(expCtx, std::move(ws), std::move(cs), &collection, yieldPolicy);
     invariant(statusWithPlanExecutor.isOK());
     return std::move(statusWithPlanExecutor.getValue());
 }
 
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWithCollectionScan(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     std::unique_ptr<DeleteStageParams> params,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     Direction direction) {
+    const auto& collection = *coll;
     invariant(collection);
     auto ws = std::make_unique<WorkingSet>();
 
     auto expCtx = make_intrusive<ExpressionContext>(
         opCtx, std::unique_ptr<CollatorInterface>(nullptr), collection->ns());
 
-    auto root = _collectionScan(expCtx, ws.get(), collection, direction);
+    auto root = _collectionScan(expCtx, ws.get(), &collection, direction);
 
     root = std::make_unique<DeleteStage>(
         expCtx.get(), std::move(params), ws.get(), collection, root.release());
 
     auto executor = plan_executor_factory::make(
-        expCtx, std::move(ws), std::move(root), collection, yieldPolicy);
+        expCtx, std::move(ws), std::move(root), &collection, yieldPolicy);
     invariant(executor.getStatus());
     return std::move(executor.getValue());
 }
@@ -106,7 +113,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWith
 
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::indexScan(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     const IndexDescriptor* descriptor,
     const BSONObj& startKey,
     const BSONObj& endKey,
@@ -114,6 +121,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::indexScan(
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     Direction direction,
     int options) {
+    const auto& collection = *coll;
     auto ws = std::make_unique<WorkingSet>();
 
     auto expCtx = make_intrusive<ExpressionContext>(
@@ -121,7 +129,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::indexScan(
 
     std::unique_ptr<PlanStage> root = _indexScan(expCtx,
                                                  ws.get(),
-                                                 collection,
+                                                 &collection,
                                                  descriptor,
                                                  startKey,
                                                  endKey,
@@ -130,14 +138,14 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::indexScan(
                                                  options);
 
     auto executor = plan_executor_factory::make(
-        expCtx, std::move(ws), std::move(root), collection, yieldPolicy);
+        expCtx, std::move(ws), std::move(root), &collection, yieldPolicy);
     invariant(executor.getStatus());
     return std::move(executor.getValue());
 }
 
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWithIndexScan(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     std::unique_ptr<DeleteStageParams> params,
     const IndexDescriptor* descriptor,
     const BSONObj& startKey,
@@ -145,6 +153,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWith
     BoundInclusion boundInclusion,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     Direction direction) {
+    const auto& collection = *coll;
     invariant(collection);
     auto ws = std::make_unique<WorkingSet>();
 
@@ -153,7 +162,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWith
 
     std::unique_ptr<PlanStage> root = _indexScan(expCtx,
                                                  ws.get(),
-                                                 collection,
+                                                 &collection,
                                                  descriptor,
                                                  startKey,
                                                  endKey,
@@ -165,18 +174,19 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::deleteWith
         expCtx.get(), std::move(params), ws.get(), collection, root.release());
 
     auto executor = plan_executor_factory::make(
-        expCtx, std::move(ws), std::move(root), collection, yieldPolicy);
+        expCtx, std::move(ws), std::move(root), &collection, yieldPolicy);
     invariant(executor.getStatus());
     return std::move(executor.getValue());
 }
 
 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::updateWithIdHack(
     OperationContext* opCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     const UpdateStageParams& params,
     const IndexDescriptor* descriptor,
     const BSONObj& key,
     PlanYieldPolicy::YieldPolicy yieldPolicy) {
+    const auto& collection = *coll;
     invariant(collection);
     auto ws = std::make_unique<WorkingSet>();
 
@@ -193,7 +203,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::updateWith
                                 expCtx.get(), params, ws.get(), collection, idHackStage.release()));
 
     auto executor = plan_executor_factory::make(
-        expCtx, std::move(ws), std::move(root), collection, yieldPolicy);
+        expCtx, std::move(ws), std::move(root), &collection, yieldPolicy);
     invariant(executor.getStatus());
     return std::move(executor.getValue());
 }
@@ -201,9 +211,10 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::updateWith
 std::unique_ptr<PlanStage> InternalPlanner::_collectionScan(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     WorkingSet* ws,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     Direction direction,
     boost::optional<RecordId> resumeAfterRecordId) {
+    const auto& collection = *coll;
     invariant(collection);
 
     CollectionScanParams params;
@@ -223,13 +234,14 @@ std::unique_ptr<PlanStage> InternalPlanner::_collectionScan(
 std::unique_ptr<PlanStage> InternalPlanner::_indexScan(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     WorkingSet* ws,
-    const CollectionPtr& collection,
+    const CollectionPtr* coll,
     const IndexDescriptor* descriptor,
     const BSONObj& startKey,
     const BSONObj& endKey,
     BoundInclusion boundInclusion,
     Direction direction,
     int options) {
+    const auto& collection = *coll;
     invariant(collection);
     invariant(descriptor);
 
diff --git a/src/mongo/db/query/internal_plans.h b/src/mongo/db/query/internal_plans.h
index c7507243369..7baf271db97 100644
--- a/src/mongo/db/query/internal_plans.h
+++ b/src/mongo/db/query/internal_plans.h
@@ -73,7 +73,7 @@ public:
     static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> collectionScan(
         OperationContext* opCtx,
         StringData ns,
-        const CollectionPtr& collection,
+        const CollectionPtr* collection,
         PlanYieldPolicy::YieldPolicy yieldPolicy,
         const Direction direction = FORWARD,
         boost::optional<RecordId> resumeAfterRecordId = boost::none);
@@ -83,7 +83,7 @@ public:
      */
     static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> deleteWithCollectionScan(
         OperationContext* opCtx,
-        const CollectionPtr& collection,
+        const CollectionPtr* collection,
         std::unique_ptr<DeleteStageParams> params,
         PlanYieldPolicy::YieldPolicy yieldPolicy,
         Direction direction = FORWARD);
@@ -93,7 +93,7 @@ public:
      */
     static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> indexScan(
         OperationContext* opCtx,
-        const CollectionPtr& collection,
+        const CollectionPtr* collection,
         const IndexDescriptor* descriptor,
         const BSONObj& startKey,
         const BSONObj& endKey,
@@ -107,7 +107,7 @@ public:
      */
     static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> deleteWithIndexScan(
         OperationContext* opCtx,
-        const CollectionPtr& collection,
+        const CollectionPtr* collection,
         std::unique_ptr<DeleteStageParams> params,
         const IndexDescriptor* descriptor,
         const BSONObj& startKey,
@@ -121,7 +121,7 @@ public:
      */
     static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> updateWithIdHack(
         OperationContext* opCtx,
-        const CollectionPtr& collection,
+        const CollectionPtr* collection,
         const UpdateStageParams& params,
         const IndexDescriptor* descriptor,
         const BSONObj& key,
@@ -136,7 +136,7 @@ private:
     static std::unique_ptr<PlanStage> _collectionScan(
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
         WorkingSet* ws,
-        const CollectionPtr& collection,
+        const CollectionPtr* collection,
         Direction direction,
         boost::optional<RecordId> resumeAfterRecordId = boost::none);
 
@@ -148,7 +148,7 @@ private:
     static std::unique_ptr<PlanStage> _indexScan(
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
         WorkingSet* ws,
-        const CollectionPtr& collection,
+        const CollectionPtr* collection,
         const IndexDescriptor* descriptor,
         const BSONObj& startKey,
         const BSONObj& endKey,
diff --git a/src/mongo/db/query/plan_executor.h b/src/mongo/db/query/plan_executor.h
index 0370c46b5da..84bfc8181cd 100644
--- a/src/mongo/db/query/plan_executor.h
+++ b/src/mongo/db/query/plan_executor.h
@@ -37,6 +37,7 @@
 #include "mongo/db/query/plan_explainer.h"
 #include "mongo/db/query/plan_summary_stats.h"
 #include "mongo/db/query/plan_yield_policy.h"
+#include "mongo/db/query/restore_context.h"
 
 namespace mongo {
 
@@ -180,6 +181,10 @@ public:
      * Restores the state saved by a saveState() call. When this method returns successfully, the
      * execution tree can once again be executed via work().
      *
+     * RestoreContext is a context containing external state needed by plan stages to be able to
+     * restore into a valid state. The RequiresCollectionStage requires a valid CollectionPtr for
+     * example.
+     *
      * Throws a UserException if the state cannot be successfully restored (e.g. a collection was
      * dropped or the position of a capped cursor was lost during a yield). If restore fails, it is
      * only safe to call dispose(), detachFromOperationContext(), or the destructor.
@@ -188,7 +193,7 @@ public:
      * WriteConflictException is encountered. If the time limit is exceeded during this retry
      * process, throws ErrorCodes::MaxTimeMSExpired.
      */
-    virtual void restoreState(const Yieldable* yieldable) = 0;
+    virtual void restoreState(const RestoreContext& context) = 0;
 
     /**
      * Detaches from the OperationContext and releases any storage-engine state.
diff --git a/src/mongo/db/query/plan_executor_factory.cpp b/src/mongo/db/query/plan_executor_factory.cpp
index 93ee590d1dc..ff15d3741cf 100644
--- a/src/mongo/db/query/plan_executor_factory.cpp
+++ b/src/mongo/db/query/plan_executor_factory.cpp
@@ -45,7 +45,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     std::unique_ptr<CanonicalQuery> cq,
     std::unique_ptr<WorkingSet> ws,
     std::unique_ptr<PlanStage> rt,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     NamespaceString nss,
     std::unique_ptr<QuerySolution> qs) {
@@ -65,7 +65,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     std::unique_ptr<WorkingSet> ws,
     std::unique_ptr<PlanStage> rt,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     NamespaceString nss,
     std::unique_ptr<QuerySolution> qs) {
@@ -87,10 +87,10 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     std::unique_ptr<QuerySolution> qs,
     std::unique_ptr<CanonicalQuery> cq,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     NamespaceString nss,
     PlanYieldPolicy::YieldPolicy yieldPolicy) {
-
+    dassert(collection);
     try {
         auto execImpl = new PlanExecutorImpl(opCtx,
                                              std::move(ws),
@@ -98,7 +98,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
                                              std::move(qs),
                                              std::move(cq),
                                              expCtx,
-                                             collection,
+                                             *collection,
                                              std::move(nss),
                                              yieldPolicy);
         PlanExecutor::Deleter planDeleter(opCtx);
@@ -113,10 +113,10 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     OperationContext* opCtx,
     std::unique_ptr<CanonicalQuery> cq,
     std::pair<std::unique_ptr<sbe::PlanStage>, stage_builder::PlanStageData> root,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     NamespaceString nss,
     std::unique_ptr<PlanYieldPolicySBE> yieldPolicy) {
-
+    dassert(collection);
     auto&& [rootStage, data] = root;
 
     LOGV2_DEBUG(4822860,
@@ -130,7 +130,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     auto exec = new PlanExecutorSBE(opCtx,
                                     std::move(cq),
                                     std::move(root),
-                                    collection,
+                                    *collection,
                                     std::move(nss),
                                     false,
                                     boost::none,
@@ -142,11 +142,11 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     OperationContext* opCtx,
     std::unique_ptr<CanonicalQuery> cq,
     std::pair<std::unique_ptr<sbe::PlanStage>, stage_builder::PlanStageData> root,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     NamespaceString nss,
     std::queue<std::pair<BSONObj, boost::optional<RecordId>>> stash,
     std::unique_ptr<PlanYieldPolicySBE> yieldPolicy) {
-
+    dassert(collection);
     auto&& [rootStage, data] = root;
 
     LOGV2_DEBUG(4822861,
@@ -158,7 +158,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     auto exec = new PlanExecutorSBE(opCtx,
                                     std::move(cq),
                                     std::move(root),
-                                    collection,
+                                    *collection,
                                     std::move(nss),
                                     true,
                                     stash,
diff --git a/src/mongo/db/query/plan_executor_factory.h b/src/mongo/db/query/plan_executor_factory.h
index 207e3065e20..f78c8f36fad 100644
--- a/src/mongo/db/query/plan_executor_factory.h
+++ b/src/mongo/db/query/plan_executor_factory.h
@@ -65,7 +65,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     std::unique_ptr<CanonicalQuery> cq,
     std::unique_ptr<WorkingSet> ws,
     std::unique_ptr<PlanStage> rt,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     NamespaceString nss = NamespaceString(),
     std::unique_ptr<QuerySolution> qs = nullptr);
@@ -81,7 +81,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     std::unique_ptr<WorkingSet> ws,
     std::unique_ptr<PlanStage> rt,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
     NamespaceString nss = NamespaceString(),
     std::unique_ptr<QuerySolution> qs = nullptr);
@@ -93,7 +93,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     std::unique_ptr<QuerySolution> qs,
     std::unique_ptr<CanonicalQuery> cq,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     NamespaceString nss,
     PlanYieldPolicy::YieldPolicy yieldPolicy);
 
@@ -105,7 +105,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     OperationContext* opCtx,
     std::unique_ptr<CanonicalQuery> cq,
     std::pair<std::unique_ptr<sbe::PlanStage>, stage_builder::PlanStageData> root,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     NamespaceString nss,
     std::unique_ptr<PlanYieldPolicySBE> yieldPolicy);
 
@@ -118,7 +118,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
     OperationContext* opCtx,
     std::unique_ptr<CanonicalQuery> cq,
     std::pair<std::unique_ptr<sbe::PlanStage>, stage_builder::PlanStageData> root,
-    const CollectionPtr& collection,
+    const CollectionPtr* collection,
     NamespaceString nss,
     std::queue<std::pair<BSONObj, boost::optional<RecordId>>> stash,
     std::unique_ptr<PlanYieldPolicySBE> yieldPolicy);
diff --git a/src/mongo/db/query/plan_executor_impl.cpp b/src/mongo/db/query/plan_executor_impl.cpp
index bb32e0aa566..d571cced21b 100644
--- a/src/mongo/db/query/plan_executor_impl.cpp
+++ b/src/mongo/db/query/plan_executor_impl.cpp
@@ -232,9 +232,9 @@ void PlanExecutorImpl::saveState() {
     _currentState = kSaved;
 }
 
-void PlanExecutorImpl::restoreState(const Yieldable* yieldable) {
+void PlanExecutorImpl::restoreState(const RestoreContext& context) {
     try {
-        restoreStateWithoutRetrying(yieldable);
+        restoreStateWithoutRetrying(context, context.collection());
     } catch (const WriteConflictException&) {
         if (!_yieldPolicy->canAutoYield())
             throw;
@@ -244,12 +244,13 @@ void PlanExecutorImpl::restoreState(const Yieldable* yieldable) {
     }
 }
 
-void PlanExecutorImpl::restoreStateWithoutRetrying(const Yieldable* yieldable) {
+void PlanExecutorImpl::restoreStateWithoutRetrying(const RestoreContext& context,
+                                                   const Yieldable* yieldable) {
     invariant(_currentState == kSaved);
 
     _yieldPolicy->setYieldable(yieldable);
     if (!isMarkedAsKilled()) {
-        _root->restoreState();
+        _root->restoreState(context);
     }
 
     _currentState = kUsable;
diff --git a/src/mongo/db/query/plan_executor_impl.h b/src/mongo/db/query/plan_executor_impl.h
index 61eae2bf06f..d67935464e0 100644
--- a/src/mongo/db/query/plan_executor_impl.h
+++ b/src/mongo/db/query/plan_executor_impl.h
@@ -68,7 +68,7 @@ public:
     const NamespaceString& nss() const final;
     OperationContext* getOpCtx() const final;
     void saveState() final;
-    void restoreState(const Yieldable* yieldable) final;
+    void restoreState(const RestoreContext& context) final;
     void detachFromOperationContext() final;
     void reattachToOperationContext(OperationContext* opCtx) final;
     ExecState getNextDocument(Document* objOut, RecordId* dlOut) final;
@@ -94,7 +94,7 @@ public:
      *
      * This is only public for PlanYieldPolicy. DO NOT CALL ANYWHERE ELSE.
      */
-    void restoreStateWithoutRetrying(const Yieldable* yieldable);
+    void restoreStateWithoutRetrying(const RestoreContext& context, const Yieldable* yieldable);
 
     /**
      * Return a pointer to this executor's MultiPlanStage, or nullptr if it does not have one.
diff --git a/src/mongo/db/query/plan_executor_sbe.cpp b/src/mongo/db/query/plan_executor_sbe.cpp
index 9e282c40da6..16a3ee3865f 100644
--- a/src/mongo/db/query/plan_executor_sbe.cpp
+++ b/src/mongo/db/query/plan_executor_sbe.cpp
@@ -108,7 +108,7 @@ void PlanExecutorSBE::saveState() {
     _root->saveState();
 }
 
-void PlanExecutorSBE::restoreState(const Yieldable* yieldable) {
+void PlanExecutorSBE::restoreState(const RestoreContext& context) {
     invariant(_root);
     _root->restoreState();
 }
diff --git a/src/mongo/db/query/plan_executor_sbe.h b/src/mongo/db/query/plan_executor_sbe.h
index 44cb51000c1..8f402336a31 100644
--- a/src/mongo/db/query/plan_executor_sbe.h
+++ b/src/mongo/db/query/plan_executor_sbe.h
@@ -62,7 +62,7 @@ public:
     }
 
     void saveState();
-    void restoreState(const Yieldable* yieldable);
+    void restoreState(const RestoreContext& context);
 
     void detachFromOperationContext();
     void reattachToOperationContext(OperationContext* opCtx);
diff --git a/src/mongo/db/query/plan_yield_policy_impl.cpp b/src/mongo/db/query/plan_yield_policy_impl.cpp
index a88378d58f3..e5492c2939d 100644
--- a/src/mongo/db/query/plan_yield_policy_impl.cpp
+++ b/src/mongo/db/query/plan_yield_policy_impl.cpp
@@ -77,7 +77,8 @@ Status PlanYieldPolicyImpl::yield(OperationContext* opCtx, std::function<void()>
                 _yieldAllLocks(opCtx, yieldable, whileYieldingFn, _planYielding->nss());
             }
 
-            _planYielding->restoreStateWithoutRetrying(yieldable);
+            _planYielding->restoreStateWithoutRetrying(
+                {RestoreContext::RestoreType::kYield, nullptr}, yieldable);
             return Status::OK();
         } catch (const WriteConflictException&) {
             CurOp::get(opCtx)->debug().additiveMetrics.incrementWriteConflicts(1);
@@ -118,6 +119,11 @@ void PlanYieldPolicyImpl::_yieldAllLocks(OperationContext* opCtx,
         opCtx->checkForInterrupt();  // throws
     }
 
+    ON_BLOCK_EXIT([yieldable]() {
+        if (yieldable)
+            yieldable->restore();
+    });
+
     if (!unlocked) {
         // Nothing was unlocked, just return, yielding is pointless.
         return;
@@ -138,10 +144,6 @@ void PlanYieldPolicyImpl::_yieldAllLocks(OperationContext* opCtx,
 
     locker->restoreLockState(opCtx, snapshot);
 
-    if (yieldable) {
-        yieldable->restore();
-    }
-
     // After yielding and reacquiring locks, the preconditions that were used to select our
     // ReadSource initially need to be checked again. Queries hold an AutoGetCollectionForRead RAII
     // lock for their lifetime, which may select a ReadSource based on state (e.g. replication
diff --git a/src/mongo/db/query/restore_context.h b/src/mongo/db/query/restore_context.h
new file mode 100644
index 00000000000..6c700158a68
--- /dev/null
+++ b/src/mongo/db/query/restore_context.h
@@ -0,0 +1,64 @@
+/**
+ *    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
+
+namespace mongo {
+class CollectionPtr;
+
+/**
+ * Context about outside environment when restoring a PlanExecutor.
+ *
+ * Contains reference to a CollectionPtr owned by an AutoGetCollection lock helper to be used by the
+ * RequiresCollectionStage plan stage.
+ */
+class RestoreContext {
+public:
+    enum class RestoreType {
+        kExternal,  // Restore on the PlanExecutor by an external call
+        kYield      // Internal restore after yield
+    };
+
+    RestoreContext() = default;
+    /* implicit */ RestoreContext(const CollectionPtr* coll) : _collection(coll) {}
+    /* implicit */ RestoreContext(RestoreType type, const CollectionPtr* coll)
+        : _type(type), _collection(coll) {}
+
+    RestoreType type() const {
+        return _type;
+    }
+    const CollectionPtr* collection() const {
+        return _collection;
+    }
+
+private:
+    RestoreType _type = RestoreType::kExternal;
+    const CollectionPtr* _collection;
+};
+}  // namespace mongo
diff --git a/src/mongo/db/repl/dbcheck.cpp b/src/mongo/db/repl/dbcheck.cpp
index 34127bb5e95..6dfae59d3ba 100644
--- a/src/mongo/db/repl/dbcheck.cpp
+++ b/src/mongo/db/repl/dbcheck.cpp
@@ -189,7 +189,7 @@ DbCheckHasher::DbCheckHasher(OperationContext* opCtx,
 
     // Set up a simple index scan on that.
     _exec = InternalPlanner::indexScan(opCtx,
-                                       collection,
+                                       &collection,
                                        desc,
                                        start.obj(),
                                        end.obj(),
diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp
index 571b5abedfe..f721e567688 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.cpp
+++ b/src/mongo/db/repl/idempotency_test_fixture.cpp
@@ -328,7 +328,7 @@ std::string IdempotencyTest::computeDataHash(const CollectionPtr& collection) {
     auto desc = collection->getIndexCatalog()->findIdIndex(_opCtx.get());
     ASSERT_TRUE(desc);
     auto exec = InternalPlanner::indexScan(_opCtx.get(),
-                                           collection,
+                                           &collection,
                                            desc,
                                            BSONObj(),
                                            BSONObj(),
diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
index 72f996976c2..566ddefaa8c 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
@@ -323,7 +323,7 @@ CollectionReader::CollectionReader(OperationContext* opCtx, const NamespaceStrin
     : _collToScan(opCtx, nss),
       _exec(InternalPlanner::collectionScan(opCtx,
                                             nss.ns(),
-                                            _collToScan.getCollection(),
+                                            &_collToScan.getCollection(),
                                             PlanYieldPolicy::YieldPolicy::NO_YIELD,
                                             InternalPlanner::FORWARD)) {}
 
diff --git a/src/mongo/db/repl/oplog_interface_local.cpp b/src/mongo/db/repl/oplog_interface_local.cpp
index a0b9e89a404..77ee3f4cea0 100644
--- a/src/mongo/db/repl/oplog_interface_local.cpp
+++ b/src/mongo/db/repl/oplog_interface_local.cpp
@@ -58,13 +58,11 @@ private:
 OplogIteratorLocal::OplogIteratorLocal(OperationContext* opCtx)
     : _oplogRead(opCtx, OplogAccessMode::kRead),
       _ctx(opCtx, NamespaceString::kRsOplogNamespace.ns()),
-      _exec(
-          InternalPlanner::collectionScan(opCtx,
-                                          NamespaceString::kRsOplogNamespace.ns(),
-                                          CollectionCatalog::get(opCtx).lookupCollectionByNamespace(
-                                              opCtx, NamespaceString::kRsOplogNamespace),
-                                          PlanYieldPolicy::YieldPolicy::NO_YIELD,
-                                          InternalPlanner::BACKWARD)) {}
+      _exec(InternalPlanner::collectionScan(opCtx,
+                                            NamespaceString::kRsOplogNamespace.ns(),
+                                            &_oplogRead.getCollection(),
+                                            PlanYieldPolicy::YieldPolicy::NO_YIELD,
+                                            InternalPlanner::BACKWARD)) {}
 
 StatusWith<OplogInterface::Iterator::Value> OplogIteratorLocal::next() {
     BSONObj obj;
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 0689bab6384..ba48a3f4e2c 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -1033,7 +1033,7 @@ void dropCollection(OperationContext* opCtx,
         // Performs a collection scan and writes all documents in the collection to disk
         // in order to keep an archive of items that were rolled back.
         auto exec = InternalPlanner::collectionScan(
-            opCtx, nss.toString(), collection, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+            opCtx, nss.toString(), &collection, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
         PlanExecutor::ExecState execState;
         try {
             BSONObj curObj;
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 926cc60efcb..23fa7f05a08 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -687,12 +687,12 @@ StatusWith<std::vector<BSONObj>> _findOrDeleteDocuments(
                 planExecutor = isFind
                     ? InternalPlanner::collectionScan(opCtx,
                                                       nsOrUUID.toString(),
-                                                      collection,
+                                                      &collection,
                                                       PlanYieldPolicy::YieldPolicy::NO_YIELD,
                                                       direction)
                     : InternalPlanner::deleteWithCollectionScan(
                           opCtx,
-                          collection,
+                          &collection,
                           makeDeleteStageParamsForDeleteDocuments(),
                           PlanYieldPolicy::YieldPolicy::NO_YIELD,
                           direction);
@@ -728,7 +728,7 @@ StatusWith<std::vector<BSONObj>> _findOrDeleteDocuments(
                 }
                 planExecutor = isFind
                     ? InternalPlanner::indexScan(opCtx,
-                                                 collection,
+                                                 &collection,
                                                  indexDescriptor,
                                                  bounds.first,
                                                  bounds.second,
@@ -738,7 +738,7 @@ StatusWith<std::vector<BSONObj>> _findOrDeleteDocuments(
                                                  InternalPlanner::IXSCAN_FETCH)
                     : InternalPlanner::deleteWithIndexScan(
                           opCtx,
-                          collection,
+                          &collection,
                           makeDeleteStageParamsForDeleteDocuments(),
                           indexDescriptor,
                           bounds.first,
@@ -929,7 +929,7 @@ Status _updateWithQuery(OperationContext* opCtx,
         }
 
         auto planExecutorResult = mongo::getExecutorUpdate(
-            nullptr, collection, &parsedUpdate, boost::none /* verbosity */);
+            nullptr, &collection, &parsedUpdate, boost::none /* verbosity */);
         if (!planExecutorResult.isOK()) {
             return planExecutorResult.getStatus();
         }
@@ -1003,7 +1003,7 @@ Status StorageInterfaceImpl::upsertById(OperationContext* opCtx,
         UpdateStageParams updateStageParams(
             parsedUpdate.getRequest(), parsedUpdate.getDriver(), nullptr);
         auto planExecutor = InternalPlanner::updateWithIdHack(opCtx,
-                                                              collection,
+                                                              &collection,
                                                               updateStageParams,
                                                               descriptor,
                                                               idKey.wrap(""),
@@ -1082,7 +1082,7 @@ Status StorageInterfaceImpl::deleteByFilter(OperationContext* opCtx,
         const auto& collection = *collectionResult.getValue();
 
         auto planExecutorResult = mongo::getExecutorDelete(
-            nullptr, collection, &parsedDelete, boost::none /* verbosity */);
+            nullptr, &collection, &parsedDelete, boost::none /* verbosity */);
         if (!planExecutorResult.isOK()) {
             return planExecutorResult.getStatus();
         }
@@ -1109,7 +1109,7 @@ boost::optional<BSONObj> StorageInterfaceImpl::findOplogEntryLessThanOrEqualToTi
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec =
         InternalPlanner::collectionScan(opCtx,
                                         NamespaceString::kRsOplogNamespace.ns(),
-                                        oplog,
+                                        &oplog,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD,
                                         InternalPlanner::BACKWARD);
 
diff --git a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
index 85b91e96903..c3fb43073f7 100644
--- a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
+++ b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp
@@ -824,7 +824,7 @@ MigrationChunkClonerSourceLegacy::_getIndexScanExecutor(OperationContext* opCtx,
     // We can afford to yield here because any change to the base data that we might miss is already
     // being queued and will migrate in the 'transferMods' stage.
     return InternalPlanner::indexScan(opCtx,
-                                      collection,
+                                      &collection,
                                       idx,
                                       min,
                                       max,
diff --git a/src/mongo/db/s/range_deletion_util.cpp b/src/mongo/db/s/range_deletion_util.cpp
index 2a7f7cf5891..8dd7347d481 100644
--- a/src/mongo/db/s/range_deletion_util.cpp
+++ b/src/mongo/db/s/range_deletion_util.cpp
@@ -182,7 +182,7 @@ StatusWith<int> deleteNextBatch(OperationContext* opCtx,
     }
 
     auto exec = InternalPlanner::deleteWithIndexScan(opCtx,
-                                                     collection,
+                                                     &collection,
                                                      std::move(deleteStageParams),
                                                      descriptor,
                                                      min,
diff --git a/src/mongo/db/s/split_chunk.cpp b/src/mongo/db/s/split_chunk.cpp
index 4efdc2e8656..718aa93c20e 100644
--- a/src/mongo/db/s/split_chunk.cpp
+++ b/src/mongo/db/s/split_chunk.cpp
@@ -66,7 +66,7 @@ bool checkIfSingleDoc(OperationContext* opCtx,
     BSONObj newmax = Helpers::toKeyFormat(kp.extendRangeBound(chunk->getMax(), true));
 
     auto exec = InternalPlanner::indexScan(opCtx,
-                                           collection,
+                                           &collection,
                                            idx,
                                            newmin,
                                            newmax,
diff --git a/src/mongo/db/s/split_vector.cpp b/src/mongo/db/s/split_vector.cpp
index 7879027e91c..57d4743e93d 100644
--- a/src/mongo/db/s/split_vector.cpp
+++ b/src/mongo/db/s/split_vector.cpp
@@ -163,7 +163,7 @@ std::vector<BSONObj> splitVector(OperationContext* opCtx,
         long long numChunks = 0;
 
         auto exec = InternalPlanner::indexScan(opCtx,
-                                               collection.getCollection(),
+                                               &collection.getCollection(),
                                                idx,
                                                minKey,
                                                maxKey,
@@ -181,7 +181,7 @@ std::vector<BSONObj> splitVector(OperationContext* opCtx,
         BSONObj maxKeyInChunk;
         {
             auto exec = InternalPlanner::indexScan(opCtx,
-                                                   collection.getCollection(),
+                                                   &collection.getCollection(),
                                                    idx,
                                                    maxKey,
                                                    minKey,
@@ -298,7 +298,7 @@ std::vector<BSONObj> splitVector(OperationContext* opCtx,
                   "keyCount"_attr = keyCount);
 
             exec = InternalPlanner::indexScan(opCtx,
-                                              collection.getCollection(),
+                                              &collection.getCollection(),
                                               idx,
                                               minKey,
                                               maxKey,
diff --git a/src/mongo/db/transaction_history_iterator.cpp b/src/mongo/db/transaction_history_iterator.cpp
index fae77f37d7b..c4bd0c0b2f4 100644
--- a/src/mongo/db/transaction_history_iterator.cpp
+++ b/src/mongo/db/transaction_history_iterator.cpp
@@ -88,7 +88,7 @@ BSONObj findOneOplogEntry(OperationContext* opCtx,
         Date_t::max());
 
     auto exec = uassertStatusOK(
-        getExecutorFind(opCtx, oplogRead.getCollection(), std::move(cq), permitYield));
+        getExecutorFind(opCtx, &oplogRead.getCollection(), std::move(cq), permitYield));
 
     PlanExecutor::ExecState getNextResult;
     try {
diff --git a/src/mongo/db/ttl.cpp b/src/mongo/db/ttl.cpp
index dec5e29350a..0de561e8115 100644
--- a/src/mongo/db/ttl.cpp
+++ b/src/mongo/db/ttl.cpp
@@ -372,7 +372,7 @@ private:
 
         auto exec =
             InternalPlanner::deleteWithIndexScan(opCtx,
-                                                 collection.getCollection(),
+                                                 &collection.getCollection(),
                                                  std::move(params),
                                                  desc,
                                                  startKey,
diff --git a/src/mongo/dbtests/cursor_manager_test.cpp b/src/mongo/dbtests/cursor_manager_test.cpp
index 1465e0428b2..a5065dc13f7 100644
--- a/src/mongo/dbtests/cursor_manager_test.cpp
+++ b/src/mongo/dbtests/cursor_manager_test.cpp
@@ -79,7 +79,7 @@ public:
             plan_executor_factory::make(expCtx,
                                         std::move(workingSet),
                                         std::move(queuedDataStage),
-                                        nullptr,
+                                        &CollectionPtr::null,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD,
                                         kTestNss));
     }
diff --git a/src/mongo/dbtests/deferred_writer.cpp b/src/mongo/dbtests/deferred_writer.cpp
index 07ae48ebd2c..d45d9bedaa2 100644
--- a/src/mongo/dbtests/deferred_writer.cpp
+++ b/src/mongo/dbtests/deferred_writer.cpp
@@ -119,7 +119,7 @@ public:
 
         auto plan = InternalPlanner::collectionScan(_opCtx.get(),
                                                     kTestNamespace.ns(),
-                                                    agc.getCollection(),
+                                                    &agc.getCollection(),
                                                     PlanYieldPolicy::YieldPolicy::NO_YIELD);
 
         std::vector<BSONObj> result;
diff --git a/src/mongo/dbtests/documentsourcetests.cpp b/src/mongo/dbtests/documentsourcetests.cpp
index 71e5ac4927d..cb49a3c3c63 100644
--- a/src/mongo/dbtests/documentsourcetests.cpp
+++ b/src/mongo/dbtests/documentsourcetests.cpp
@@ -90,6 +90,7 @@ protected:
         _source.reset();
 
         dbtests::WriteContextForTests ctx(opCtx(), nss.ns());
+        _coll = ctx.getCollection();
 
         auto qr = std::make_unique<QueryRequest>(nss);
         if (hint) {
@@ -97,15 +98,12 @@ protected:
         }
         auto cq = uassertStatusOK(CanonicalQuery::canonicalize(opCtx(), std::move(qr)));
 
-        auto exec = uassertStatusOK(getExecutor(opCtx(),
-                                                ctx.getCollection(),
-                                                std::move(cq),
-                                                PlanYieldPolicy::YieldPolicy::NO_YIELD,
-                                                0));
+        auto exec = uassertStatusOK(
+            getExecutor(opCtx(), &_coll, std::move(cq), PlanYieldPolicy::YieldPolicy::NO_YIELD, 0));
 
         exec->saveState();
         _source = DocumentSourceCursor::create(
-            ctx.getCollection(), std::move(exec), _ctx, DocumentSourceCursor::CursorType::kRegular);
+            _coll, std::move(exec), _ctx, DocumentSourceCursor::CursorType::kRegular);
     }
 
     intrusive_ptr<ExpressionContextForTest> ctx() {
@@ -134,6 +132,7 @@ private:
     // It is important that these are ordered to ensure correct destruction order.
     intrusive_ptr<ExpressionContextForTest> _ctx;
     intrusive_ptr<DocumentSourceCursor> _source;
+    CollectionPtr _coll;
 };
 
 /** Create a DocumentSourceCursor. */
@@ -318,7 +317,7 @@ TEST_F(DocumentSourceCursorTest, TailableAwaitDataCursorShouldErrorAfterTimeout)
         uassertStatusOK(plan_executor_factory::make(std::move(canonicalQuery),
                                                     std::move(workingSet),
                                                     std::move(collectionScan),
-                                                    readLock.getCollection(),
+                                                    &readLock.getCollection(),
                                                     PlanYieldPolicy::YieldPolicy::ALWAYS_TIME_OUT));
 
     // Make a DocumentSourceCursor.
@@ -359,7 +358,7 @@ TEST_F(DocumentSourceCursorTest, NonAwaitDataCursorShouldErrorAfterTimeout) {
         uassertStatusOK(plan_executor_factory::make(std::move(canonicalQuery),
                                                     std::move(workingSet),
                                                     std::move(collectionScan),
-                                                    readLock.getCollection(),
+                                                    &readLock.getCollection(),
                                                     PlanYieldPolicy::YieldPolicy::ALWAYS_TIME_OUT));
 
     // Make a DocumentSourceCursor.
@@ -409,7 +408,7 @@ TEST_F(DocumentSourceCursorTest, TailableAwaitDataCursorShouldErrorAfterBeingKil
         plan_executor_factory::make(std::move(canonicalQuery),
                                     std::move(workingSet),
                                     std::move(collectionScan),
-                                    readLock.getCollection(),
+                                    &readLock.getCollection(),
                                     PlanYieldPolicy::YieldPolicy::ALWAYS_MARK_KILLED));
 
     // Make a DocumentSourceCursor.
@@ -449,7 +448,7 @@ TEST_F(DocumentSourceCursorTest, NormalCursorShouldErrorAfterBeingKilled) {
         plan_executor_factory::make(std::move(canonicalQuery),
                                     std::move(workingSet),
                                     std::move(collectionScan),
-                                    readLock.getCollection(),
+                                    &readLock.getCollection(),
                                     PlanYieldPolicy::YieldPolicy::ALWAYS_MARK_KILLED));
 
     // Make a DocumentSourceCursor.
diff --git a/src/mongo/dbtests/plan_executor_invalidation_test.cpp b/src/mongo/dbtests/plan_executor_invalidation_test.cpp
index 8a5369ba36c..7d52881cc51 100644
--- a/src/mongo/dbtests/plan_executor_invalidation_test.cpp
+++ b/src/mongo/dbtests/plan_executor_invalidation_test.cpp
@@ -87,12 +87,12 @@ public:
         std::unique_ptr<CanonicalQuery> cq = std::move(statusWithCQ.getValue());
 
         // Takes ownership of 'ws', 'scan', and 'cq'.
-        auto statusWithPlanExecutor = plan_executor_factory::make(
-            std::move(cq),
-            std::move(ws),
-            std::move(scan),
-            CollectionCatalog::get(&_opCtx).lookupCollectionByNamespace(&_opCtx, nss),
-            PlanYieldPolicy::YieldPolicy::YIELD_MANUAL);
+        auto statusWithPlanExecutor =
+            plan_executor_factory::make(std::move(cq),
+                                        std::move(ws),
+                                        std::move(scan),
+                                        &collection(),
+                                        PlanYieldPolicy::YieldPolicy::YIELD_MANUAL);
 
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         return std::move(statusWithPlanExecutor.getValue());
@@ -105,7 +105,7 @@ public:
             &_opCtx, keyPattern, _makeMinimalIndexSpec(keyPattern));
         ASSERT(indexDescriptor);
         return InternalPlanner::indexScan(&_opCtx,
-                                          collection(),
+                                          &collection(),
                                           indexDescriptor,
                                           startKey,
                                           endKey,
@@ -117,8 +117,9 @@ public:
         return 50;
     }
 
-    CollectionPtr collection() const {
-        return CollectionCatalog::get(&_opCtx).lookupCollectionByNamespace(&_opCtx, nss);
+    const CollectionPtr& collection() const {
+        _coll = CollectionCatalog::get(&_opCtx).lookupCollectionByNamespace(&_opCtx, nss);
+        return _coll;
     }
 
     void truncateCollection() const {
@@ -135,6 +136,7 @@ public:
     OperationContext& _opCtx = *_opCtxPtr;
     unique_ptr<dbtests::WriteContextForTests> _ctx;
     DBDirectClient _client;
+    mutable CollectionPtr _coll;
 
     boost::intrusive_ptr<ExpressionContext> _expCtx;
 
@@ -162,7 +164,7 @@ TEST_F(PlanExecutorInvalidationTest, ExecutorToleratesDeletedDocumentsDuringYiel
     _client.remove(nss.ns(), BSON("foo" << 10));
     _client.remove(nss.ns(), BSON("foo" << 11));
 
-    exec->restoreState(nullptr);
+    exec->restoreState(&collection());
 
     // Make sure that the PlanExecutor moved forward over the deleted data.  We don't see foo==10 or
     // foo==11.
@@ -189,7 +191,7 @@ TEST_F(PlanExecutorInvalidationTest, PlanExecutorThrowsOnRestoreWhenCollectionIs
     // Drop a collection that's not ours.
     _client.dropCollection("unittests.someboguscollection");
 
-    exec->restoreState(nullptr);
+    exec->restoreState(&collection());
 
     ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, nullptr));
     ASSERT_EQUALS(10, obj["foo"].numberInt());
@@ -198,7 +200,7 @@ TEST_F(PlanExecutorInvalidationTest, PlanExecutorThrowsOnRestoreWhenCollectionIs
 
     _client.dropCollection(nss.ns());
 
-    ASSERT_THROWS_CODE(exec->restoreState(nullptr), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(exec->restoreState(&collection()), DBException, ErrorCodes::QueryPlanKilled);
 }
 
 TEST_F(PlanExecutorInvalidationTest, CollScanExecutorDoesNotDieWhenAllIndicesDropped) {
@@ -215,7 +217,7 @@ TEST_F(PlanExecutorInvalidationTest, CollScanExecutorDoesNotDieWhenAllIndicesDro
 
     exec->saveState();
     _client.dropIndexes(nss.ns());
-    exec->restoreState(nullptr);
+    exec->restoreState(&collection());
 
     // Read the rest of the collection.
     for (int i = 10; i < N(); ++i) {
@@ -238,7 +240,7 @@ TEST_F(PlanExecutorInvalidationTest, CollScanExecutorDoesNotDieWhenOneIndexDropp
 
     exec->saveState();
     _client.dropIndex(nss.ns(), BSON("foo" << 1));
-    exec->restoreState(nullptr);
+    exec->restoreState(&collection());
 
     // Read the rest of the collection.
     for (int i = 10; i < N(); ++i) {
@@ -268,7 +270,7 @@ TEST_F(PlanExecutorInvalidationTest, IxscanExecutorDiesWhenAllIndexesDropped) {
     _client.dropIndexes(nss.ns());
 
     // Restoring the executor should throw.
-    ASSERT_THROWS_CODE(exec->restoreState(nullptr), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(exec->restoreState(&collection()), DBException, ErrorCodes::QueryPlanKilled);
 }
 
 TEST_F(PlanExecutorInvalidationTest, IxscanExecutorDiesWhenIndexBeingScannedIsDropped) {
@@ -289,7 +291,7 @@ TEST_F(PlanExecutorInvalidationTest, IxscanExecutorDiesWhenIndexBeingScannedIsDr
     _client.dropIndex(nss.ns(), keyPattern);
 
     // Restoring the executor should throw.
-    ASSERT_THROWS_CODE(exec->restoreState(nullptr), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(exec->restoreState(&collection()), DBException, ErrorCodes::QueryPlanKilled);
 }
 
 TEST_F(PlanExecutorInvalidationTest, IxscanExecutorSurvivesWhenUnrelatedIndexIsDropped) {
@@ -311,7 +313,7 @@ TEST_F(PlanExecutorInvalidationTest, IxscanExecutorSurvivesWhenUnrelatedIndexIsD
     // state.
     exec->saveState();
     _client.dropIndex(nss.ns(), keyPatternBar);
-    exec->restoreState(nullptr);
+    exec->restoreState(&collection());
 
     // Scan the rest of the index.
     for (int i = 10; i < N(); ++i) {
@@ -337,7 +339,7 @@ TEST_F(PlanExecutorInvalidationTest, ExecutorThrowsOnRestoreWhenDatabaseIsDroppe
     _ctx.reset();
     _client.dropDatabase("somesillydb");
     _ctx.reset(new dbtests::WriteContextForTests(&_opCtx, nss.ns()));
-    exec->restoreState(nullptr);
+    exec->restoreState(&collection());
 
     ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, nullptr));
     ASSERT_EQUALS(10, obj["foo"].numberInt());
@@ -348,7 +350,7 @@ TEST_F(PlanExecutorInvalidationTest, ExecutorThrowsOnRestoreWhenDatabaseIsDroppe
     _ctx.reset();
     _client.dropDatabase("unittests");
     _ctx.reset(new dbtests::WriteContextForTests(&_opCtx, nss.ns()));
-    ASSERT_THROWS_CODE(exec->restoreState(nullptr), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(exec->restoreState(&collection()), DBException, ErrorCodes::QueryPlanKilled);
 }
 
 // TODO SERVER-31695: Allow PlanExecutors to remain valid after collection rename.
@@ -371,7 +373,7 @@ TEST_F(PlanExecutorInvalidationTest, CollScanDiesOnCollectionRenameWithinDatabas
                                                            << "dropTarget" << true),
                                    info));
 
-    ASSERT_THROWS_CODE(exec->restoreState(nullptr), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(exec->restoreState(&collection()), DBException, ErrorCodes::QueryPlanKilled);
 }
 
 // TODO SERVER-31695: Allow PlanExecutors to remain valid after collection rename.
@@ -397,7 +399,7 @@ TEST_F(PlanExecutorInvalidationTest, IxscanDiesOnCollectionRenameWithinDatabase)
                                                            << "dropTarget" << true),
                                    info));
 
-    ASSERT_THROWS_CODE(exec->restoreState(nullptr), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(exec->restoreState(&collection()), DBException, ErrorCodes::QueryPlanKilled);
 }
 
 TEST_F(PlanExecutorInvalidationTest, IxscanDiesWhenTruncateCollectionDropsAllIndices) {
@@ -417,7 +419,7 @@ TEST_F(PlanExecutorInvalidationTest, IxscanDiesWhenTruncateCollectionDropsAllInd
     // expected error code.
     exec->saveState();
     truncateCollection();
-    ASSERT_THROWS_CODE(exec->restoreState(nullptr), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(exec->restoreState(&collection()), DBException, ErrorCodes::QueryPlanKilled);
 }
 
 TEST_F(PlanExecutorInvalidationTest, CollScanExecutorSurvivesCollectionTruncate) {
@@ -434,7 +436,7 @@ TEST_F(PlanExecutorInvalidationTest, CollScanExecutorSurvivesCollectionTruncate)
     // successfully.
     exec->saveState();
     truncateCollection();
-    exec->restoreState(nullptr);
+    exec->restoreState(&collection());
 
     // Since all documents in the collection have been deleted, the PlanExecutor should issue EOF.
     ASSERT_EQUALS(PlanExecutor::IS_EOF, exec->getNext(&obj, nullptr));
diff --git a/src/mongo/dbtests/query_plan_executor.cpp b/src/mongo/dbtests/query_plan_executor.cpp
index 491755db37e..83fbe89282b 100644
--- a/src/mongo/dbtests/query_plan_executor.cpp
+++ b/src/mongo/dbtests/query_plan_executor.cpp
@@ -120,7 +120,7 @@ public:
 
         // Hand the plan off to the executor.
         auto statusWithPlanExecutor = plan_executor_factory::make(
-            std::move(cq), std::move(ws), std::move(root), coll, yieldPolicy);
+            std::move(cq), std::move(ws), std::move(root), &coll, yieldPolicy);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         return std::move(statusWithPlanExecutor.getValue());
     }
@@ -136,10 +136,8 @@ public:
      * Returns a PlanExecutor capable of executing an index scan
      * over the specified index with the specified bounds.
      */
-    unique_ptr<PlanExecutor, PlanExecutor::Deleter> makeIndexScanExec(Database* db,
-                                                                      BSONObj& indexSpec,
-                                                                      int start,
-                                                                      int end) {
+    unique_ptr<PlanExecutor, PlanExecutor::Deleter> makeIndexScanExec(
+        Database* db, const CollectionPtr& coll, BSONObj& indexSpec, int start, int end) {
         // Build the index scan stage.
         auto descriptor = getIndex(db, indexSpec);
         IndexScanParams ixparams(&_opCtx, descriptor);
@@ -148,10 +146,6 @@ public:
         ixparams.bounds.endKey = BSON("" << end);
         ixparams.bounds.boundInclusion = BoundInclusion::kIncludeBothStartAndEndKeys;
 
-        const CollectionPtr& coll =
-            CollectionCatalog::get(&_opCtx).lookupCollectionByNamespace(&_opCtx, nss);
-
-
         unique_ptr<WorkingSet> ws(new WorkingSet());
         auto ixscan = std::make_unique<IndexScan>(_expCtx.get(), coll, ixparams, ws.get(), nullptr);
         unique_ptr<PlanStage> root =
@@ -168,7 +162,7 @@ public:
             plan_executor_factory::make(std::move(cq),
                                         std::move(ws),
                                         std::move(root),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::YIELD_MANUAL);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         return std::move(statusWithPlanExecutor.getValue());
@@ -214,7 +208,7 @@ TEST_F(PlanExecutorTest, DropIndexScanAgg) {
     // Create an "inner" plan executor and register it with the cursor manager so that it can
     // get notified when the collection is dropped.
     unique_ptr<PlanExecutor, PlanExecutor::Deleter> innerExec(
-        makeIndexScanExec(ctx.db(), indexSpec, 7, 10));
+        makeIndexScanExec(ctx.db(), collection, indexSpec, 7, 10));
 
     // Wrap the "inner" plan executor in a DocumentSourceCursor and add it as the first source
     // in the pipeline.
@@ -311,12 +305,14 @@ TEST_F(PlanExecutorTest, ShouldReportErrorIfKilledDuringYield) {
 
 class PlanExecutorSnapshotTest : public PlanExecutorTest {
 protected:
-    void setupCollection() {
+    CollectionPtr setupCollection() {
         insert(BSON("_id" << 1 << "a" << 1));
         insert(BSON("_id" << 2 << "a" << 2 << "payload"
                           << "x"));
         insert(BSON("_id" << 3 << "a" << 3));
         insert(BSON("_id" << 4 << "a" << 4));
+
+        return CollectionCatalog::get(&_opCtx).lookupCollectionByNamespace(&_opCtx, nss);
     }
 
     /**
@@ -384,12 +380,12 @@ TEST_F(PlanExecutorSnapshotTest, SnapshotControl) {
  */
 TEST_F(PlanExecutorSnapshotTest, SnapshotTest) {
     dbtests::WriteContextForTests ctx(&_opCtx, nss.ns());
-    setupCollection();
+    auto coll = setupCollection();
     BSONObj indexSpec = BSON("_id" << 1);
     addIndex(indexSpec);
 
     BSONObj filterObj = fromjson("{a: {$gte: 2}}");
-    auto exec = makeIndexScanExec(ctx.db(), indexSpec, 2, 5);
+    auto exec = makeIndexScanExec(ctx.db(), coll, indexSpec, 2, 5);
 
     BSONObj objOut;
     ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&objOut, nullptr));
diff --git a/src/mongo/dbtests/query_stage_and.cpp b/src/mongo/dbtests/query_stage_and.cpp
index 4a1bfa748ce..064f2a59236 100644
--- a/src/mongo/dbtests/query_stage_and.cpp
+++ b/src/mongo/dbtests/query_stage_and.cpp
@@ -226,7 +226,7 @@ public:
             }
         }
         size_t memUsageAfter = ah->getMemUsage();
-        ah->restoreState();
+        ah->restoreState(&coll);
 
         // The deleted result should still be buffered inside the AND_HASH stage, so there should be
         // no change in memory consumption.
@@ -319,7 +319,7 @@ public:
         size_t memUsageAfter = ah->getMemUsage();
         ASSERT_EQUALS(memUsageBefore, memUsageAfter);
 
-        ah->restoreState();
+        ah->restoreState(&coll);
 
         // We expect that the deleted document doers not appear in our result set.
         int count = 0;
@@ -963,7 +963,7 @@ public:
         // very first insert, which should be the very first thing in data. Delete it.
         ah->saveState();
         remove(coll->docFor(&_opCtx, *data.begin()).value());
-        ah->restoreState();
+        ah->restoreState(&coll);
 
         auto it = data.begin();
 
@@ -995,7 +995,7 @@ public:
         // Remove a result that's coming up.
         ah->saveState();
         remove(coll->docFor(&_opCtx, *it).value());
-        ah->restoreState();
+        ah->restoreState(&coll);
 
         // Get all results aside from the two we deleted.
         while (!ah->isEOF()) {
diff --git a/src/mongo/dbtests/query_stage_cached_plan.cpp b/src/mongo/dbtests/query_stage_cached_plan.cpp
index 8bf13d8f2c6..4625bd6ebf9 100644
--- a/src/mongo/dbtests/query_stage_cached_plan.cpp
+++ b/src/mongo/dbtests/query_stage_cached_plan.cpp
@@ -476,7 +476,9 @@ TEST_F(QueryStageCachedPlan, ThrowsOnYieldRecoveryWhenIndexIsDroppedBeforePlanSe
     readLock.reset();
     dropIndex(keyPattern);
     readLock.emplace(&_opCtx, nss);
-    ASSERT_THROWS_CODE(cachedPlanStage.restoreState(), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(cachedPlanStage.restoreState(&readLock->getCollection()),
+                       DBException,
+                       ErrorCodes::QueryPlanKilled);
 }
 
 TEST_F(QueryStageCachedPlan, DoesNotThrowOnYieldRecoveryWhenIndexIsDroppedAferPlanSelection) {
@@ -521,7 +523,7 @@ TEST_F(QueryStageCachedPlan, DoesNotThrowOnYieldRecoveryWhenIndexIsDroppedAferPl
     readLock.reset();
     dropIndex(keyPattern);
     readLock.emplace(&_opCtx, nss);
-    cachedPlanStage.restoreState();
+    cachedPlanStage.restoreState(&readLock->getCollection());
 }
 
 }  // namespace QueryStageCachedPlan
diff --git a/src/mongo/dbtests/query_stage_collscan.cpp b/src/mongo/dbtests/query_stage_collscan.cpp
index 53aa2e78483..4b09cd82750 100644
--- a/src/mongo/dbtests/query_stage_collscan.cpp
+++ b/src/mongo/dbtests/query_stage_collscan.cpp
@@ -107,7 +107,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(ps),
-                                        collection.getCollection(),
+                                        &collection.getCollection(),
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -200,7 +200,7 @@ TEST_F(QueryStageCollectionScanTest, QueryStageCollscanObjectsInOrderForward) {
         plan_executor_factory::make(_expCtx,
                                     std::move(ws),
                                     std::move(ps),
-                                    collection.getCollection(),
+                                    &collection.getCollection(),
                                     PlanYieldPolicy::YieldPolicy::NO_YIELD);
     ASSERT_OK(statusWithPlanExecutor.getStatus());
     auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -232,7 +232,7 @@ TEST_F(QueryStageCollectionScanTest, QueryStageCollscanObjectsInOrderBackward) {
         plan_executor_factory::make(_expCtx,
                                     std::move(ws),
                                     std::move(ps),
-                                    collection.getCollection(),
+                                    &collection.getCollection(),
                                     PlanYieldPolicy::YieldPolicy::NO_YIELD);
     ASSERT_OK(statusWithPlanExecutor.getStatus());
     auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -281,7 +281,7 @@ TEST_F(QueryStageCollectionScanTest, QueryStageCollscanDeleteUpcomingObject) {
     // Remove recordIds[count].
     scan->saveState();
     remove(coll->docFor(&_opCtx, recordIds[count]).value());
-    scan->restoreState();
+    scan->restoreState(&coll);
 
     // Skip over recordIds[count].
     ++count;
@@ -334,7 +334,7 @@ TEST_F(QueryStageCollectionScanTest, QueryStageCollscanDeleteUpcomingObjectBackw
     // Remove recordIds[count].
     scan->saveState();
     remove(coll->docFor(&_opCtx, recordIds[count]).value());
-    scan->restoreState();
+    scan->restoreState(&coll);
 
     // Skip over recordIds[count].
     ++count;
@@ -388,7 +388,7 @@ TEST_F(QueryStageCollectionScanTest, QueryTestCollscanResumeAfterRecordIdSeekSuc
         plan_executor_factory::make(_expCtx,
                                     std::move(ws),
                                     std::move(ps),
-                                    collection.getCollection(),
+                                    &collection.getCollection(),
                                     PlanYieldPolicy::YieldPolicy::NO_YIELD);
     ASSERT_OK(statusWithPlanExecutor.getStatus());
     auto exec = std::move(statusWithPlanExecutor.getValue());
diff --git a/src/mongo/dbtests/query_stage_count.cpp b/src/mongo/dbtests/query_stage_count.cpp
index a43346e849b..008d7fbff8f 100644
--- a/src/mongo/dbtests/query_stage_count.cpp
+++ b/src/mongo/dbtests/query_stage_count.cpp
@@ -71,14 +71,15 @@ public:
         WriteUnitOfWork wunit(&_opCtx);
 
         _ctx.db()->dropCollection(&_opCtx, nss()).transitional_ignore();
-        _coll = _ctx.db()->createCollection(&_opCtx, nss());
+        auto coll = _ctx.db()->createCollection(&_opCtx, nss());
 
-        _coll->getIndexCatalog()
+        coll->getIndexCatalog()
             ->createIndexOnEmptyCollection(&_opCtx,
                                            BSON("key" << BSON("x" << 1) << "name"
                                                       << "x_1"
                                                       << "v" << 1))
             .status_with_transitional_ignore();
+        _coll = coll;
 
         for (int i = 0; i < kDocuments; i++) {
             insert(BSON(GENOID << "x" << i));
@@ -190,14 +191,14 @@ public:
             }
 
             // Resume from yield.
-            countStage.restoreState();
+            countStage.restoreState(&_coll);
         }
 
         return static_cast<const CountStats*>(countStage.getSpecificStats());
     }
 
     IndexScan* createIndexScan(MatchExpression* expr, WorkingSet* ws) {
-        IndexCatalog* catalog = _coll->getIndexCatalog();
+        const IndexCatalog* catalog = _coll->getIndexCatalog();
         std::vector<const IndexDescriptor*> indexes;
         catalog->findIndexesByKeyPattern(&_opCtx, BSON("x" << 1), false, &indexes);
         ASSERT_EQ(indexes.size(), 1U);
@@ -237,7 +238,7 @@ protected:
     Lock::DBLock _dbLock;
     OldClientContext _ctx;
     boost::intrusive_ptr<ExpressionContext> _expCtx;
-    Collection* _coll;
+    CollectionPtr _coll;
 };
 
 class QueryStageCountNoChangeDuringYield : public CountStageTest {
diff --git a/src/mongo/dbtests/query_stage_count_scan.cpp b/src/mongo/dbtests/query_stage_count_scan.cpp
index cf6f4005405..91f0d847ad4 100644
--- a/src/mongo/dbtests/query_stage_count_scan.cpp
+++ b/src/mongo/dbtests/query_stage_count_scan.cpp
@@ -336,7 +336,8 @@ public:
         static_cast<PlanStage*>(&count)->saveState();
 
         // Recover from yield
-        static_cast<PlanStage*>(&count)->restoreState();
+        auto coll = getCollection();
+        static_cast<PlanStage*>(&count)->restoreState(&coll);
 
         // finish counting
         while (PlanStage::IS_EOF != countState) {
@@ -391,7 +392,8 @@ public:
         remove(BSON("a" << GTE << 5));
 
         // Recover from yield
-        static_cast<PlanStage*>(&count)->restoreState();
+        auto coll = getCollection();
+        static_cast<PlanStage*>(&count)->restoreState(&coll);
 
         // finish counting
         while (PlanStage::IS_EOF != countState) {
@@ -449,7 +451,8 @@ public:
         insert(BSON("a" << 6.5));
 
         // Recover from yield
-        static_cast<PlanStage*>(&count)->restoreState();
+        auto coll = getCollection();
+        static_cast<PlanStage*>(&count)->restoreState(&coll);
 
         // finish counting
         while (PlanStage::IS_EOF != countState) {
@@ -569,7 +572,8 @@ public:
         remove(BSON("a" << 1 << "b" << 5));
 
         // Recover from yield
-        static_cast<PlanStage*>(&count)->restoreState();
+        auto coll = getCollection();
+        static_cast<PlanStage*>(&count)->restoreState(&coll);
 
         // finish counting
         while (PlanStage::IS_EOF != countState) {
diff --git a/src/mongo/dbtests/query_stage_delete.cpp b/src/mongo/dbtests/query_stage_delete.cpp
index 7bdd12bb81e..03b7834e079 100644
--- a/src/mongo/dbtests/query_stage_delete.cpp
+++ b/src/mongo/dbtests/query_stage_delete.cpp
@@ -172,7 +172,7 @@ public:
         BSONObj targetDoc = coll->docFor(&_opCtx, recordIds[targetDocIndex]).value();
         ASSERT(!targetDoc.isEmpty());
         remove(targetDoc);
-        static_cast<PlanStage*>(&deleteStage)->restoreState();
+        static_cast<PlanStage*>(&deleteStage)->restoreState(&coll);
 
         // Remove the rest.
         while (!deleteStage.isEOF()) {
diff --git a/src/mongo/dbtests/query_stage_ixscan.cpp b/src/mongo/dbtests/query_stage_ixscan.cpp
index d74a212b865..4aa7462de72 100644
--- a/src/mongo/dbtests/query_stage_ixscan.cpp
+++ b/src/mongo/dbtests/query_stage_ixscan.cpp
@@ -59,6 +59,7 @@ public:
 
         _ctx.db()->dropCollection(&_opCtx, nss()).transitional_ignore();
         _coll = _ctx.db()->createCollection(&_opCtx, nss());
+        _collPtr = _coll;
 
         ASSERT_OK(_coll->getIndexCatalog()->createIndexOnEmptyCollection(
             &_opCtx,
@@ -108,7 +109,7 @@ public:
 
         // This child stage gets owned and freed by the caller.
         MatchExpression* filter = nullptr;
-        return new IndexScan(_expCtx.get(), _coll, params, &_ws, filter);
+        return new IndexScan(_expCtx.get(), _collPtr, params, &_ws, filter);
     }
 
     IndexScan* createIndexScan(BSONObj startKey,
@@ -149,6 +150,7 @@ protected:
     Lock::DBLock _dbLock;
     OldClientContext _ctx;
     Collection* _coll;
+    CollectionPtr _collPtr;
 
     WorkingSet _ws;
 
@@ -201,7 +203,7 @@ public:
         static_cast<PlanStage*>(ixscan.get())->saveState();
         insert(fromjson("{_id: 4, x: 10}"));
         insert(fromjson("{_id: 5, x: 11}"));
-        static_cast<PlanStage*>(ixscan.get())->restoreState();
+        static_cast<PlanStage*>(ixscan.get())->restoreState(&_collPtr);
 
         member = getNext(ixscan.get());
         ASSERT_EQ(WorkingSetMember::RID_AND_IDX, member->getState());
@@ -234,7 +236,7 @@ public:
         // Save state and insert an indexed doc.
         static_cast<PlanStage*>(ixscan.get())->saveState();
         insert(fromjson("{_id: 4, x: 7}"));
-        static_cast<PlanStage*>(ixscan.get())->restoreState();
+        static_cast<PlanStage*>(ixscan.get())->restoreState(&_collPtr);
 
         member = getNext(ixscan.get());
         ASSERT_EQ(WorkingSetMember::RID_AND_IDX, member->getState());
@@ -267,7 +269,7 @@ public:
         // Save state and insert an indexed doc.
         static_cast<PlanStage*>(ixscan.get())->saveState();
         insert(fromjson("{_id: 4, x: 10}"));
-        static_cast<PlanStage*>(ixscan.get())->restoreState();
+        static_cast<PlanStage*>(ixscan.get())->restoreState(&_collPtr);
 
         // Ensure that we're EOF and we don't erroneously return {'': 12}.
         WorkingSetID id;
@@ -301,7 +303,7 @@ public:
         static_cast<PlanStage*>(ixscan.get())->saveState();
         insert(fromjson("{_id: 4, x: 6}"));
         insert(fromjson("{_id: 5, x: 9}"));
-        static_cast<PlanStage*>(ixscan.get())->restoreState();
+        static_cast<PlanStage*>(ixscan.get())->restoreState(&_collPtr);
 
         // Ensure that we don't erroneously return {'': 9} or {'':3}.
         member = getNext(ixscan.get());
diff --git a/src/mongo/dbtests/query_stage_merge_sort.cpp b/src/mongo/dbtests/query_stage_merge_sort.cpp
index 4b1c76f1ef6..1bb4df5d87e 100644
--- a/src/mongo/dbtests/query_stage_merge_sort.cpp
+++ b/src/mongo/dbtests/query_stage_merge_sort.cpp
@@ -190,7 +190,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -259,7 +259,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -328,7 +328,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -403,7 +403,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -474,7 +474,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -532,7 +532,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -615,7 +615,7 @@ public:
         // stage, and therefore should still be returned.
         ms->saveState();
         remove(BSON(std::string(1u, 'a' + count) << 1));
-        ms->restoreState();
+        ms->restoreState(&coll);
 
         // Make sure recordIds[11] is returned as expected. We expect the corresponding working set
         // member to remain in RID_AND_IDX state. It should have been marked as "suspicious", since
@@ -740,7 +740,7 @@ public:
         // Update doc {a: 5} such that the postimage will no longer match the query.
         ms->saveState();
         update(BSON("a" << 5), BSON("$set" << BSON("a" << 15)));
-        ms->restoreState();
+        ms->restoreState(&coll);
 
         // Invalidated doc {a: 5} should still get returned. We expect an RID_AND_OBJ working set
         // member with an owned BSONObj.
@@ -825,7 +825,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -900,7 +900,7 @@ public:
             make_unique<FetchStage>(_expCtx.get(), ws.get(), std::move(idxScan), nullptr, coll));
 
         auto statusWithPlanExecutor = plan_executor_factory::make(
-            _expCtx, std::move(ws), std::move(ms), coll, PlanYieldPolicy::YieldPolicy::NO_YIELD);
+            _expCtx, std::move(ws), std::move(ms), &coll, PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
 
diff --git a/src/mongo/dbtests/query_stage_multiplan.cpp b/src/mongo/dbtests/query_stage_multiplan.cpp
index 6765130f36c..2ac93c0e04d 100644
--- a/src/mongo/dbtests/query_stage_multiplan.cpp
+++ b/src/mongo/dbtests/query_stage_multiplan.cpp
@@ -260,7 +260,7 @@ TEST_F(QueryStageMultiPlanTest, MPSCollectionScanVsHighlySelectiveIXScan) {
         plan_executor_factory::make(std::move(cq),
                                     std::move(sharedWs),
                                     std::move(mps),
-                                    coll,
+                                    &coll,
                                     PlanYieldPolicy::YieldPolicy::NO_YIELD);
     ASSERT_OK(statusWithPlanExecutor.getStatus());
     auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -500,7 +500,7 @@ TEST_F(QueryStageMultiPlanTest, MPSExplainAllPlans) {
         uassertStatusOK(plan_executor_factory::make(_expCtx,
                                                     std::move(ws),
                                                     std::move(mps),
-                                                    ctx.getCollection(),
+                                                    &ctx.getCollection(),
                                                     PlanYieldPolicy::YieldPolicy::NO_YIELD));
 
     auto execImpl = dynamic_cast<PlanExecutorImpl*>(exec.get());
@@ -553,7 +553,7 @@ TEST_F(QueryStageMultiPlanTest, MPSSummaryStats) {
     qr->setFilter(BSON("foo" << BSON("$gte" << 0)));
     auto cq = uassertStatusOK(CanonicalQuery::canonicalize(opCtx(), std::move(qr)));
     auto exec = uassertStatusOK(
-        getExecutor(opCtx(), coll, std::move(cq), PlanYieldPolicy::YieldPolicy::NO_YIELD, 0));
+        getExecutor(opCtx(), &coll, std::move(cq), PlanYieldPolicy::YieldPolicy::NO_YIELD, 0));
 
     auto execImpl = dynamic_cast<PlanExecutorImpl*>(exec.get());
     ASSERT(execImpl);
diff --git a/src/mongo/dbtests/query_stage_sort.cpp b/src/mongo/dbtests/query_stage_sort.cpp
index 16640fb9e5e..c35511ab247 100644
--- a/src/mongo/dbtests/query_stage_sort.cpp
+++ b/src/mongo/dbtests/query_stage_sort.cpp
@@ -133,7 +133,7 @@ public:
         // The PlanExecutor will be automatically registered on construction due to the auto
         // yield policy, so it can receive invalidations when we remove documents later.
         auto statusWithPlanExecutor = plan_executor_factory::make(
-            _expCtx, std::move(ws), std::move(ss), coll, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
+            _expCtx, std::move(ws), std::move(ss), &coll, PlanYieldPolicy::YieldPolicy::YIELD_AUTO);
         invariant(statusWithPlanExecutor.isOK());
         return std::move(statusWithPlanExecutor.getValue());
     }
@@ -179,7 +179,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
@@ -610,7 +610,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(fetchStage),
-                                        coll,
+                                        &coll,
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         auto exec = std::move(statusWithPlanExecutor.getValue());
 
diff --git a/src/mongo/dbtests/query_stage_subplan.cpp b/src/mongo/dbtests/query_stage_subplan.cpp
index facd2746bbe..8609b3f8af7 100644
--- a/src/mongo/dbtests/query_stage_subplan.cpp
+++ b/src/mongo/dbtests/query_stage_subplan.cpp
@@ -538,8 +538,8 @@ TEST_F(QueryStageSubplanTest, ShouldReportErrorIfExceedsTimeLimitDuringPlanning)
 
     // Create the SubplanStage.
     WorkingSet workingSet;
-    SubplanStage subplanStage(
-        _expCtx.get(), ctx.getCollection(), &workingSet, params, canonicalQuery.get());
+    auto coll = ctx.getCollection();
+    SubplanStage subplanStage(_expCtx.get(), coll, &workingSet, params, canonicalQuery.get());
 
     AlwaysTimeOutYieldPolicy alwaysTimeOutPolicy(serviceContext()->getFastClockSource());
     ASSERT_EQ(ErrorCodes::ExceededTimeLimit, subplanStage.pickBestPlan(&alwaysTimeOutPolicy));
@@ -563,8 +563,8 @@ TEST_F(QueryStageSubplanTest, ShouldReportErrorIfKilledDuringPlanning) {
 
     // Create the SubplanStage.
     WorkingSet workingSet;
-    SubplanStage subplanStage(
-        _expCtx.get(), ctx.getCollection(), &workingSet, params, canonicalQuery.get());
+    auto coll = ctx.getCollection();
+    SubplanStage subplanStage(_expCtx.get(), coll, &workingSet, params, canonicalQuery.get());
 
     AlwaysPlanKilledYieldPolicy alwaysPlanKilledYieldPolicy(serviceContext()->getFastClockSource());
     ASSERT_EQ(ErrorCodes::QueryPlanKilled, subplanStage.pickBestPlan(&alwaysPlanKilledYieldPolicy));
@@ -611,11 +611,13 @@ TEST_F(QueryStageSubplanTest, ShouldThrowOnRestoreIfIndexDroppedBeforePlanSelect
     // Attempt to restore state. This should throw due the index drop. As a future improvement, we
     // may wish to make the subplan stage tolerate drops of indices it is not using.
     collLock.emplace(opCtx(), nss);
-    ASSERT_THROWS_CODE(subplanStage.restoreState(), DBException, ErrorCodes::QueryPlanKilled);
+    ASSERT_THROWS_CODE(subplanStage.restoreState(&collLock->getCollection()),
+                       DBException,
+                       ErrorCodes::QueryPlanKilled);
 }
 
 TEST_F(QueryStageSubplanTest, ShouldNotThrowOnRestoreIfIndexDroppedAfterPlanSelection) {
-    CollectionPtr collection = nullptr;
+    CollectionPtr collection;
     {
         dbtests::WriteContextForTests ctx{opCtx(), nss.ns()};
         addIndex(BSON("p1" << 1 << "opt1" << 1));
@@ -658,7 +660,7 @@ TEST_F(QueryStageSubplanTest, ShouldNotThrowOnRestoreIfIndexDroppedAfterPlanSele
     // Restoring state should succeed, since the plan selected by pickBestPlan() does not use the
     // index {irrelevant: 1}.
     collLock.emplace(opCtx(), nss);
-    subplanStage.restoreState();
+    subplanStage.restoreState(&collLock->getCollection());
 }
 
 }  // namespace
diff --git a/src/mongo/dbtests/query_stage_tests.cpp b/src/mongo/dbtests/query_stage_tests.cpp
index 50948a6cf32..4690bc67c57 100644
--- a/src/mongo/dbtests/query_stage_tests.cpp
+++ b/src/mongo/dbtests/query_stage_tests.cpp
@@ -96,7 +96,7 @@ public:
             plan_executor_factory::make(_expCtx,
                                         std::move(ws),
                                         std::move(ix),
-                                        ctx.getCollection(),
+                                        &ctx.getCollection(),
                                         PlanYieldPolicy::YieldPolicy::NO_YIELD);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         auto exec = std::move(statusWithPlanExecutor.getValue());
diff --git a/src/mongo/dbtests/query_stage_update.cpp b/src/mongo/dbtests/query_stage_update.cpp
index 084012ca61c..93fb705f8ea 100644
--- a/src/mongo/dbtests/query_stage_update.cpp
+++ b/src/mongo/dbtests/query_stage_update.cpp
@@ -275,7 +275,7 @@ public:
             CurOp& curOp = *CurOp::get(_opCtx);
             OpDebug* opDebug = &curOp.debug();
             UpdateDriver driver(_expCtx);
-            const CollectionPtr& coll =
+            CollectionPtr coll =
                 CollectionCatalog::get(&_opCtx).lookupCollectionByNamespace(&_opCtx, nss);
             ASSERT(coll);
 
@@ -335,7 +335,7 @@ public:
             BSONObj targetDoc = coll->docFor(&_opCtx, recordIds[targetDocIndex]).value();
             ASSERT(!targetDoc.isEmpty());
             remove(targetDoc);
-            static_cast<PlanStage*>(updateStage.get())->restoreState();
+            static_cast<PlanStage*>(updateStage.get())->restoreState(&coll);
 
             // Do the remaining updates.
             while (!updateStage->isEOF()) {