SERVER-50984 Add CollectionPtr to replace usage of const Collection*

It implements a yieldable interface that is used to re-load the
Collection pointer from the catalog after a yield that released locks.
With lock-free reads and copy-on-write on Collection instances releasing
locks without notifying an AutoGetCollection at a higher level may cause
its pointers to dangle if a MODE_X writer installs a new Collection
instance in the catalog.

CollectionPtr should be passed by const reference so a yield can notify
all the way up.

9 files changed, 40 insertions, 43 deletions

diff --git a/src/mongo/db/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index ba6151d341a..a651b387ce4 100644
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -142,7 +142,7 @@ void DocumentSourceCursor::loadBatch() {
                             ->checkCanServeReadsFor(pExpCtx->opCtx, _exec->nss(), true));
     }
 
-    _exec->restoreState();
+    _exec->restoreState(autoColl ? &autoColl->getCollection() : nullptr);
 
     try {
         ON_BLOCK_EXIT([this] { recordPlanSummaryStats(); });
@@ -285,7 +285,7 @@ DocumentSourceCursor::~DocumentSourceCursor() {
 }
 
 DocumentSourceCursor::DocumentSourceCursor(
-    const Collection* collection,
+    const CollectionPtr& collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const intrusive_ptr<ExpressionContext>& pCtx,
     CursorType cursorType,
@@ -316,7 +316,7 @@ DocumentSourceCursor::DocumentSourceCursor(
 }
 
 intrusive_ptr<DocumentSourceCursor> DocumentSourceCursor::create(
-    const Collection* collection,
+    const CollectionPtr& collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const intrusive_ptr<ExpressionContext>& pExpCtx,
     CursorType cursorType,
diff --git a/src/mongo/db/pipeline/document_source_cursor.h b/src/mongo/db/pipeline/document_source_cursor.h
index a7794df2e2c..4ee177593c5 100644
--- a/src/mongo/db/pipeline/document_source_cursor.h
+++ b/src/mongo/db/pipeline/document_source_cursor.h
@@ -63,7 +63,7 @@ public:
      * $cursor stage can return a sequence of empty documents for the caller to count.
      */
     static boost::intrusive_ptr<DocumentSourceCursor> create(
-        const Collection* collection,
+        const CollectionPtr& collection,
         std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
         const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
         CursorType cursorType,
@@ -112,7 +112,7 @@ public:
     }
 
 protected:
-    DocumentSourceCursor(const Collection* collection,
+    DocumentSourceCursor(const CollectionPtr& collection,
                          std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                          const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
                          CursorType cursorType,
diff --git a/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp b/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
index 4277bd26424..8e6885567d0 100644
--- a/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
@@ -51,7 +51,7 @@
 namespace mongo {
 
 boost::intrusive_ptr<DocumentSourceGeoNearCursor> DocumentSourceGeoNearCursor::create(
-    const Collection* collection,
+    const CollectionPtr& collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     FieldPath distanceField,
@@ -66,7 +66,7 @@ boost::intrusive_ptr<DocumentSourceGeoNearCursor> DocumentSourceGeoNearCursor::c
 }
 
 DocumentSourceGeoNearCursor::DocumentSourceGeoNearCursor(
-    const Collection* collection,
+    const CollectionPtr& collection,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     FieldPath distanceField,
diff --git a/src/mongo/db/pipeline/document_source_geo_near_cursor.h b/src/mongo/db/pipeline/document_source_geo_near_cursor.h
index f8d3b483914..b3ddf9a834b 100644
--- a/src/mongo/db/pipeline/document_source_geo_near_cursor.h
+++ b/src/mongo/db/pipeline/document_source_geo_near_cursor.h
@@ -60,7 +60,7 @@ public:
      * nonnegative.
      */
     static boost::intrusive_ptr<DocumentSourceGeoNearCursor> create(
-        const Collection*,
+        const CollectionPtr&,
         std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>,
         const boost::intrusive_ptr<ExpressionContext>&,
         FieldPath distanceField,
@@ -70,7 +70,7 @@ public:
     const char* getSourceName() const final;
 
 private:
-    DocumentSourceGeoNearCursor(const Collection*,
+    DocumentSourceGeoNearCursor(const CollectionPtr&,
                                 std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>,
                                 const boost::intrusive_ptr<ExpressionContext>&,
                                 FieldPath distanceField,
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index c7b8bd273ec..6d0406f7495 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -99,7 +99,7 @@ namespace {
  * storage engine support for random cursors.
  */
 StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createRandomCursorExecutor(
-    const Collection* coll,
+    const CollectionPtr& coll,
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     long long sampleSize,
     long long numRecords,
@@ -187,7 +187,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> createRandomCursorEx
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExecutor(
     const intrusive_ptr<ExpressionContext>& expCtx,
-    const Collection* collection,
+    const CollectionPtr& collection,
     const NamespaceString& nss,
     BSONObj queryObj,
     BSONObj projectionObj,
@@ -273,7 +273,8 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
  *
  * The 'collection' is required to exist. Throws if no usable 2d or 2dsphere index could be found.
  */
-StringData extractGeoNearFieldFromIndexes(OperationContext* opCtx, const Collection* collection) {
+StringData extractGeoNearFieldFromIndexes(OperationContext* opCtx,
+                                          const CollectionPtr& collection) {
     invariant(collection);
 
     std::vector<const IndexDescriptor*> idxs;
@@ -313,7 +314,7 @@ StringData extractGeoNearFieldFromIndexes(OperationContext* opCtx, const Collect
 }  // namespace
 
 std::pair<PipelineD::AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-PipelineD::buildInnerQueryExecutor(const Collection* collection,
+PipelineD::buildInnerQueryExecutor(const CollectionPtr& collection,
                                    const NamespaceString& nss,
                                    const AggregationRequest* aggRequest,
                                    Pipeline* pipeline) {
@@ -348,7 +349,7 @@ PipelineD::buildInnerQueryExecutor(const Collection* collection,
                     ? DocumentSourceCursor::CursorType::kEmptyDocuments
                     : DocumentSourceCursor::CursorType::kRegular;
                 auto attachExecutorCallback =
-                    [cursorType](const Collection* collection,
+                    [cursorType](const CollectionPtr& collection,
                                  std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                                  Pipeline* pipeline) {
                         auto cursor = DocumentSourceCursor::create(
@@ -372,7 +373,7 @@ PipelineD::buildInnerQueryExecutor(const Collection* collection,
 }
 
 void PipelineD::attachInnerQueryExecutorToPipeline(
-    const Collection* collection,
+    const CollectionPtr& collection,
     PipelineD::AttachExecutorCallback attachExecutorCallback,
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
     Pipeline* pipeline) {
@@ -384,7 +385,7 @@ void PipelineD::attachInnerQueryExecutorToPipeline(
     }
 }
 
-void PipelineD::buildAndAttachInnerQueryExecutorToPipeline(const Collection* collection,
+void PipelineD::buildAndAttachInnerQueryExecutorToPipeline(const CollectionPtr& collection,
                                                            const NamespaceString& nss,
                                                            const AggregationRequest* aggRequest,
                                                            Pipeline* pipeline) {
@@ -483,7 +484,7 @@ auto buildProjectionForPushdown(const DepsTracker& deps, Pipeline* pipeline) {
 }  // namespace
 
 std::pair<PipelineD::AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-PipelineD::buildInnerQueryExecutorGeneric(const Collection* collection,
+PipelineD::buildInnerQueryExecutorGeneric(const CollectionPtr& collection,
                                           const NamespaceString& nss,
                                           const AggregationRequest* aggRequest,
                                           Pipeline* pipeline) {
@@ -561,7 +562,7 @@ PipelineD::buildInnerQueryExecutorGeneric(const Collection* collection,
         (pipeline->peekFront() && pipeline->peekFront()->constraints().isChangeStreamStage());
 
     auto attachExecutorCallback =
-        [cursorType, trackOplogTS](const Collection* collection,
+        [cursorType, trackOplogTS](const CollectionPtr& collection,
                                    std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                                    Pipeline* pipeline) {
             auto cursor = DocumentSourceCursor::create(
@@ -572,7 +573,7 @@ PipelineD::buildInnerQueryExecutorGeneric(const Collection* collection,
 }
 
 std::pair<PipelineD::AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-PipelineD::buildInnerQueryExecutorGeoNear(const Collection* collection,
+PipelineD::buildInnerQueryExecutorGeoNear(const CollectionPtr& collection,
                                           const NamespaceString& nss,
                                           const AggregationRequest* aggRequest,
                                           Pipeline* pipeline) {
@@ -616,7 +617,7 @@ PipelineD::buildInnerQueryExecutorGeoNear(const Collection* collection,
                                    locationField = geoNearStage->getLocationField(),
                                    distanceMultiplier =
                                        geoNearStage->getDistanceMultiplier().value_or(1.0)](
-                                      const Collection* collection,
+                                      const CollectionPtr& collection,
                                       std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
                                       Pipeline* pipeline) {
         auto cursor = DocumentSourceGeoNearCursor::create(collection,
@@ -634,7 +635,7 @@ PipelineD::buildInnerQueryExecutorGeoNear(const Collection* collection,
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prepareExecutor(
     const intrusive_ptr<ExpressionContext>& expCtx,
-    const Collection* collection,
+    const CollectionPtr& collection,
     const NamespaceString& nss,
     Pipeline* pipeline,
     const boost::intrusive_ptr<DocumentSourceSort>& sortStage,
diff --git a/src/mongo/db/pipeline/pipeline_d.h b/src/mongo/db/pipeline/pipeline_d.h
index ec1baacf6fd..96a2b83c57c 100644
--- a/src/mongo/db/pipeline/pipeline_d.h
+++ b/src/mongo/db/pipeline/pipeline_d.h
@@ -43,6 +43,7 @@
 
 namespace mongo {
 class Collection;
+class CollectionPtr;
 class DocumentSourceCursor;
 class DocumentSourceMatch;
 class DocumentSourceSort;
@@ -67,7 +68,7 @@ public:
      * the new stage to the pipeline.
      */
     using AttachExecutorCallback = std::function<void(
-        const Collection*, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>, Pipeline*)>;
+        const CollectionPtr&, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>, Pipeline*)>;
 
     /**
      * This method looks for early pipeline stages that can be folded into the underlying
@@ -88,7 +89,7 @@ public:
      * 'nullptr'.
      */
     static std::pair<AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-    buildInnerQueryExecutor(const Collection* collection,
+    buildInnerQueryExecutor(const CollectionPtr& collection,
                             const NamespaceString& nss,
                             const AggregationRequest* aggRequest,
                             Pipeline* pipeline);
@@ -101,7 +102,7 @@ public:
      * 'nullptr'.
      */
     static void attachInnerQueryExecutorToPipeline(
-        const Collection* collection,
+        const CollectionPtr& collection,
         AttachExecutorCallback attachExecutorCallback,
         std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
         Pipeline* pipeline);
@@ -112,7 +113,7 @@ public:
      * used when the executor attachment phase doesn't need to be deferred and the $cursor stage
      * can be created right after buiding the executor.
      */
-    static void buildAndAttachInnerQueryExecutorToPipeline(const Collection* collection,
+    static void buildAndAttachInnerQueryExecutorToPipeline(const CollectionPtr& collection,
                                                            const NamespaceString& nss,
                                                            const AggregationRequest* aggRequest,
                                                            Pipeline* pipeline);
@@ -130,7 +131,7 @@ public:
      */
     static std::unique_ptr<CollatorInterface> resolveCollator(OperationContext* opCtx,
                                                               BSONObj userCollation,
-                                                              const Collection* collection) {
+                                                              const CollectionPtr& collection) {
         if (!userCollation.isEmpty()) {
             return uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext())
                                        ->makeFromBSON(userCollation));
@@ -149,7 +150,7 @@ private:
      * the 'pipeline'.
      */
     static std::pair<AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-    buildInnerQueryExecutorGeneric(const Collection* collection,
+    buildInnerQueryExecutorGeneric(const CollectionPtr& collection,
                                    const NamespaceString& nss,
                                    const AggregationRequest* aggRequest,
                                    Pipeline* pipeline);
@@ -160,7 +161,7 @@ private:
      * not exist, as the $geoNearCursor requires a 2d or 2dsphere index.
      */
     static std::pair<AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
-    buildInnerQueryExecutorGeoNear(const Collection* collection,
+    buildInnerQueryExecutorGeoNear(const CollectionPtr& collection,
                                    const NamespaceString& nss,
                                    const AggregationRequest* aggRequest,
                                    Pipeline* pipeline);
@@ -179,7 +180,7 @@ private:
      */
     static StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> prepareExecutor(
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
-        const Collection* collection,
+        const CollectionPtr& collection,
         const NamespaceString& nss,
         Pipeline* pipeline,
         const boost::intrusive_ptr<DocumentSourceSort>& sortStage,
diff --git a/src/mongo/db/pipeline/plan_executor_pipeline.h b/src/mongo/db/pipeline/plan_executor_pipeline.h
index 54a66979bfc..72d61471720 100644
--- a/src/mongo/db/pipeline/plan_executor_pipeline.h
+++ b/src/mongo/db/pipeline/plan_executor_pipeline.h
@@ -62,7 +62,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() override {}
+    void restoreState(const Yieldable* yieldable) override {}
 
     void detachFromOperationContext() override {
         _pipeline->detachFromOperationContext();
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
index 35a3c11acad..7e795c181a0 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
@@ -156,9 +156,8 @@ std::vector<Document> CommonMongodProcessInterface::getIndexStats(OperationConte
                                                                   const NamespaceString& ns,
                                                                   StringData host,
                                                                   bool addShardName) {
-    AutoGetCollectionForReadCommand autoColl(opCtx, ns);
+    AutoGetCollectionForReadCommand collection(opCtx, ns);
 
-    const Collection* collection = autoColl.getCollection();
     std::vector<Document> indexStats;
     if (!collection) {
         LOGV2_DEBUG(23881,
@@ -227,15 +226,13 @@ Status CommonMongodProcessInterface::appendRecordCount(OperationContext* opCtx,
 Status CommonMongodProcessInterface::appendQueryExecStats(OperationContext* opCtx,
                                                           const NamespaceString& nss,
                                                           BSONObjBuilder* builder) const {
-    AutoGetCollectionForReadCommand autoColl(opCtx, nss);
+    AutoGetCollectionForReadCommand collection(opCtx, nss);
 
-    if (!autoColl.getDb()) {
+    if (!collection.getDb()) {
         return {ErrorCodes::NamespaceNotFound,
                 str::stream() << "Database [" << nss.db().toString() << "] not found."};
     }
 
-    const Collection* collection = autoColl.getCollection();
-
     if (!collection) {
         return {ErrorCodes::NamespaceNotFound,
                 str::stream() << "Collection [" << nss.toString() << "] not found."};
@@ -261,12 +258,11 @@ Status CommonMongodProcessInterface::appendQueryExecStats(OperationContext* opCt
 
 BSONObj CommonMongodProcessInterface::getCollectionOptions(OperationContext* opCtx,
                                                            const NamespaceString& nss) {
-    AutoGetCollectionForReadCommand autoColl(opCtx, nss);
+    AutoGetCollectionForReadCommand collection(opCtx, nss);
     BSONObj collectionOptions = {};
-    if (!autoColl.getDb()) {
+    if (!collection.getDb()) {
         return collectionOptions;
     }
-    const Collection* collection = autoColl.getCollection();
     if (!collection) {
         return collectionOptions;
     }
@@ -412,12 +408,11 @@ std::vector<BSONObj> CommonMongodProcessInterface::getMatchingPlanCacheEntryStat
         return !matchExp ? true : matchExp->matchesBSON(obj);
     };
 
-    AutoGetCollection autoColl(opCtx, nss, MODE_IS);
-    const auto collection = autoColl.getCollection();
+    AutoGetCollection collection(opCtx, nss, MODE_IS);
     uassert(
         50933, str::stream() << "collection '" << nss.toString() << "' does not exist", collection);
 
-    const auto planCache = CollectionQueryInfo::get(collection).getPlanCache();
+    const auto planCache = CollectionQueryInfo::get(collection.getCollection()).getPlanCache();
     invariant(planCache);
 
     return planCache->getMatchingStats(serializer, predicate);
diff --git a/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp b/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp
index 537d89749d1..5d085033f96 100644
--- a/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/non_shardsvr_process_interface.cpp
@@ -116,7 +116,7 @@ void NonShardServerProcessInterface::createIndexesOnEmptyCollection(
                     str::stream() << "The database is in the process of being dropped " << ns.db(),
                     autoColl.getDb() && !autoColl.getDb()->isDropPending(opCtx));
 
-            auto collection = autoColl.getCollection();
+            const auto& collection = autoColl.getCollection();
             uassert(ErrorCodes::NamespaceNotFound,
                     str::stream() << "Failed to create indexes for aggregation because collection "
                                      "does not exist: "