SERVER-72641 Add establishConsistentCollection API

Unifies openCollection() and needsOpenCollection() on the CollectionCatalog.

Integrate into the AutoGet setup helpers and simplify their logic.

6 files changed, 318 insertions, 336 deletions

diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 75f4b02f4c2..130d460ef83 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -784,9 +784,24 @@ void CollectionCatalog::reloadViews(OperationContext* opCtx, const DatabaseName&
     PublishCatalogUpdates::ensureRegisteredWithRecoveryUnit(opCtx, uncommittedCatalogUpdates);
 }
 
-bool CollectionCatalog::needsOpenCollection(OperationContext* opCtx,
-                                            const NamespaceStringOrUUID& nsOrUUID,
-                                            boost::optional<Timestamp> readTimestamp) const {
+CollectionPtr CollectionCatalog::establishConsistentCollection(
+    OperationContext* opCtx,
+    const NamespaceStringOrUUID& nssOrUUID,
+    boost::optional<Timestamp> readTimestamp) const {
+    if (_needsOpenCollection(opCtx, nssOrUUID, readTimestamp)) {
+        return _openCollection(opCtx, nssOrUUID, readTimestamp);
+    }
+
+    return lookupCollectionByNamespaceOrUUID(opCtx, nssOrUUID);
+}
+
+bool CollectionCatalog::_needsOpenCollection(OperationContext* opCtx,
+                                             const NamespaceStringOrUUID& nsOrUUID,
+                                             boost::optional<Timestamp> readTimestamp) const {
+    if (!feature_flags::gPointInTimeCatalogLookups.isEnabledAndIgnoreFCV()) {
+        return false;
+    }
+
     if (readTimestamp) {
         auto coll = lookupCollectionByNamespaceOrUUID(opCtx, nsOrUUID);
         return !coll || *readTimestamp < coll->getMinimumValidSnapshot();
@@ -799,9 +814,9 @@ bool CollectionCatalog::needsOpenCollection(OperationContext* opCtx,
     }
 }
 
-CollectionPtr CollectionCatalog::openCollection(OperationContext* opCtx,
-                                                const NamespaceStringOrUUID& nssOrUUID,
-                                                boost::optional<Timestamp> readTimestamp) const {
+CollectionPtr CollectionCatalog::_openCollection(OperationContext* opCtx,
+                                                 const NamespaceStringOrUUID& nssOrUUID,
+                                                 boost::optional<Timestamp> readTimestamp) const {
     if (!feature_flags::gPointInTimeCatalogLookups.isEnabledAndIgnoreFCV()) {
         return CollectionPtr();
     }
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index d8c2d97be38..fd8c32c2b44 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -216,31 +216,35 @@ public:
     void reloadViews(OperationContext* opCtx, const DatabaseName& dbName) const;
 
     /**
-     * Returns true if catalog information about this namespace or UUID should be looked up from the
-     * durable catalog rather than using the in-memory state of the catalog.
+     * Establish a collection instance consistent with the opened storage snapshot.
      *
-     * This is true when either:
-     *  - The readTimestamp is prior to the minimum valid timestamp for the collection corresponding
-     *    to this namespace, or
-     *  - There's no read timestamp provided and this namespace has a pending DDL operation that has
-     *    not completed yet (which would imply that the latest version of the catalog may or may not
-     *    match the state of the durable catalog for this collection).
-     */
-    bool needsOpenCollection(OperationContext* opCtx,
-                             const NamespaceStringOrUUID& nsOrUUID,
-                             boost::optional<Timestamp> readTimestamp) const;
-
-    /**
      * Returns the collection pointer representative of 'nssOrUUID' at the provided read timestamp.
-     * If no timestamp is provided, returns instance of the latest collection. The returned
-     * collection instance is only valid while the storage snapshot is open and becomes invalidated
-     * when the snapshot is closed.
+     * If no timestamp is provided, returns instance of the latest collection. When called
+     * concurrently with a DDL operation the latest collection returned may be the instance being
+     * committed by the concurrent DDL operation.
      *
      * Returns nullptr when reading from a point-in-time where the collection did not exist.
+     *
+     * The returned collection instance is only valid while a reference to this catalog instance is
+     * held or stashed and as long as the storage snapshot remains open. Releasing catalog reference
+     * or closing the storage snapshot invalidates the instance.
+     *
+     * Future calls to lookupCollection, lookupNSS, lookupUUID on this namespace/UUID will return
+     * results consistent with the opened storage snapshot.
+     *
+     * Depending on the internal state of the CollectionCatalog a read from the durable catalog may
+     * be performed and this call may block on I/O. No mutex should be held while calling this
+     * function.
+     *
+     * Multikey state is not guaranteed to be consistent with the storage snapshot. It may indicate
+     * an index to be multikey where it is not multikey in the storage snapshot. However, it will
+     * never be wrong in the other direction.
+     *
+     * No collection level lock is required to call this function.
      */
-    CollectionPtr openCollection(OperationContext* opCtx,
-                                 const NamespaceStringOrUUID& nssOrUUID,
-                                 boost::optional<Timestamp> readTimestamp) const;
+    CollectionPtr establishConsistentCollection(OperationContext* opCtx,
+                                                const NamespaceStringOrUUID& nssOrUUID,
+                                                boost::optional<Timestamp> readTimestamp) const;
 
     /**
      * Returns a shared_ptr to a drop pending index if it's found and not expired.
@@ -807,6 +811,33 @@ private:
     void _markNamespaceForCatalogIdCleanupIfNeeded(const NamespaceString& nss,
                                                    const std::vector<TimestampedCatalogId>& ids);
 
+    /**
+     * Returns true if catalog information about this namespace or UUID should be looked up from the
+     * durable catalog rather than using the in-memory state of the catalog.
+     *
+     * This is true when either:
+     *  - The readTimestamp is prior to the minimum valid timestamp for the collection corresponding
+     *    to this namespace, or
+     *  - There's no read timestamp provided and this namespace has a pending DDL operation that has
+     *    not completed yet (which would imply that the latest version of the catalog may or may not
+     *    match the state of the durable catalog for this collection).
+     */
+    bool _needsOpenCollection(OperationContext* opCtx,
+                              const NamespaceStringOrUUID& nsOrUUID,
+                              boost::optional<Timestamp> readTimestamp) const;
+
+    /**
+     * Returns the collection pointer representative of 'nssOrUUID' at the provided read timestamp.
+     * If no timestamp is provided, returns instance of the latest collection. The returned
+     * collection instance is only valid while the storage snapshot is open and becomes invalidated
+     * when the snapshot is closed.
+     *
+     * Returns nullptr when reading from a point-in-time where the collection did not exist.
+     */
+    CollectionPtr _openCollection(OperationContext* opCtx,
+                                  const NamespaceStringOrUUID& nssOrUUID,
+                                  boost::optional<Timestamp> readTimestamp) const;
+
     // Helpers to perform openCollection at latest or at point-in-time on Namespace/UUID.
     CollectionPtr _openCollectionAtLatestByNamespace(OperationContext* opCtx,
                                                      const NamespaceString& nss) const;
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 4baf0274d15..bc2c31efe47 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -945,13 +945,13 @@ public:
         wuow.commit();
     }
 
-    void concurrentCreateCollectionAndOpenCollection(OperationContext* opCtx,
-                                                     const NamespaceString& nss,
-                                                     boost::optional<UUID> uuid,
-                                                     Timestamp timestamp,
-                                                     bool openSnapshotBeforeCommit,
-                                                     bool expectedExistence,
-                                                     int expectedNumIndexes) {
+    void concurrentCreateCollectionAndEstablishConsistentCollection(OperationContext* opCtx,
+                                                                    const NamespaceString& nss,
+                                                                    boost::optional<UUID> uuid,
+                                                                    Timestamp timestamp,
+                                                                    bool openSnapshotBeforeCommit,
+                                                                    bool expectedExistence,
+                                                                    int expectedNumIndexes) {
         NamespaceStringOrUUID readNssOrUUID = [&]() {
             if (uuid) {
                 return NamespaceStringOrUUID(nss.dbName(), *uuid);
@@ -959,7 +959,7 @@ public:
                 return NamespaceStringOrUUID(nss);
             }
         }();
-        _concurrentDDLOperationAndOpenCollection(
+        _concurrentDDLOperationAndEstablishConsistentCollection(
             opCtx,
             readNssOrUUID,
             timestamp,
@@ -969,25 +969,27 @@ public:
             expectedNumIndexes);
     }
 
-    void concurrentDropCollectionAndOpenCollection(OperationContext* opCtx,
-                                                   const NamespaceString& nss,
-                                                   const NamespaceStringOrUUID& readNssOrUUID,
-                                                   Timestamp timestamp,
-                                                   bool openSnapshotBeforeCommit,
-                                                   bool expectedExistence,
-                                                   int expectedNumIndexes) {
-        _concurrentDDLOperationAndOpenCollection(opCtx,
-                                                 readNssOrUUID,
-                                                 timestamp,
-                                                 [this, &nss, &timestamp](OperationContext* opCtx) {
-                                                     _dropCollection(opCtx, nss, timestamp);
-                                                 },
-                                                 openSnapshotBeforeCommit,
-                                                 expectedExistence,
-                                                 expectedNumIndexes);
-    }
-
-    void concurrentRenameCollectionAndOpenCollection(
+    void concurrentDropCollectionAndEstablishConsistentCollection(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const NamespaceStringOrUUID& readNssOrUUID,
+        Timestamp timestamp,
+        bool openSnapshotBeforeCommit,
+        bool expectedExistence,
+        int expectedNumIndexes) {
+        _concurrentDDLOperationAndEstablishConsistentCollection(
+            opCtx,
+            readNssOrUUID,
+            timestamp,
+            [this, &nss, &timestamp](OperationContext* opCtx) {
+                _dropCollection(opCtx, nss, timestamp);
+            },
+            openSnapshotBeforeCommit,
+            expectedExistence,
+            expectedNumIndexes);
+    }
+
+    void concurrentRenameCollectionAndEstablishConsistentCollection(
         OperationContext* opCtx,
         const NamespaceString& from,
         const NamespaceString& to,
@@ -997,7 +999,7 @@ public:
         bool expectedExistence,
         int expectedNumIndexes,
         std::function<void()> verifyStateCallback = {}) {
-        _concurrentDDLOperationAndOpenCollection(
+        _concurrentDDLOperationAndEstablishConsistentCollection(
             opCtx,
             lookupNssOrUUID,
             timestamp,
@@ -1010,42 +1012,46 @@ public:
             std::move(verifyStateCallback));
     }
 
-    void concurrentCreateIndexAndOpenCollection(OperationContext* opCtx,
-                                                const NamespaceString& nss,
-                                                const NamespaceStringOrUUID& readNssOrUUID,
-                                                BSONObj indexSpec,
-                                                Timestamp timestamp,
-                                                bool openSnapshotBeforeCommit,
-                                                bool expectedExistence,
-                                                int expectedNumIndexes) {
-        _concurrentDDLOperationAndOpenCollection(opCtx,
-                                                 readNssOrUUID,
-                                                 timestamp,
-                                                 [this, &nss, &indexSpec](OperationContext* opCtx) {
-                                                     _createIndex(opCtx, nss, indexSpec);
-                                                 },
-                                                 openSnapshotBeforeCommit,
-                                                 expectedExistence,
-                                                 expectedNumIndexes);
-    }
-
-    void concurrentDropIndexAndOpenCollection(OperationContext* opCtx,
-                                              const NamespaceString& nss,
-                                              const NamespaceStringOrUUID& readNssOrUUID,
-                                              const std::string& indexName,
-                                              Timestamp timestamp,
-                                              bool openSnapshotBeforeCommit,
-                                              bool expectedExistence,
-                                              int expectedNumIndexes) {
-        _concurrentDDLOperationAndOpenCollection(opCtx,
-                                                 readNssOrUUID,
-                                                 timestamp,
-                                                 [this, &nss, &indexName](OperationContext* opCtx) {
-                                                     _dropIndex(opCtx, nss, indexName);
-                                                 },
-                                                 openSnapshotBeforeCommit,
-                                                 expectedExistence,
-                                                 expectedNumIndexes);
+    void concurrentCreateIndexAndEstablishConsistentCollection(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const NamespaceStringOrUUID& readNssOrUUID,
+        BSONObj indexSpec,
+        Timestamp timestamp,
+        bool openSnapshotBeforeCommit,
+        bool expectedExistence,
+        int expectedNumIndexes) {
+        _concurrentDDLOperationAndEstablishConsistentCollection(
+            opCtx,
+            readNssOrUUID,
+            timestamp,
+            [this, &nss, &indexSpec](OperationContext* opCtx) {
+                _createIndex(opCtx, nss, indexSpec);
+            },
+            openSnapshotBeforeCommit,
+            expectedExistence,
+            expectedNumIndexes);
+    }
+
+    void concurrentDropIndexAndEstablishConsistentCollection(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        const NamespaceStringOrUUID& readNssOrUUID,
+        const std::string& indexName,
+        Timestamp timestamp,
+        bool openSnapshotBeforeCommit,
+        bool expectedExistence,
+        int expectedNumIndexes) {
+        _concurrentDDLOperationAndEstablishConsistentCollection(
+            opCtx,
+            readNssOrUUID,
+            timestamp,
+            [this, &nss, &indexName](OperationContext* opCtx) {
+                _dropIndex(opCtx, nss, indexName);
+            },
+            openSnapshotBeforeCommit,
+            expectedExistence,
+            expectedNumIndexes);
     }
 
 protected:
@@ -1181,14 +1187,15 @@ private:
      * updating the in-memory catalog.
      */
     template <typename Callable>
-    void _concurrentDDLOperationAndOpenCollection(OperationContext* opCtx,
-                                                  const NamespaceStringOrUUID& nssOrUUID,
-                                                  Timestamp timestamp,
-                                                  Callable&& ddlOperation,
-                                                  bool openSnapshotBeforeCommit,
-                                                  bool expectedExistence,
-                                                  int expectedNumIndexes,
-                                                  std::function<void()> verifyStateCallback = {}) {
+    void _concurrentDDLOperationAndEstablishConsistentCollection(
+        OperationContext* opCtx,
+        const NamespaceStringOrUUID& nssOrUUID,
+        Timestamp timestamp,
+        Callable&& ddlOperation,
+        bool openSnapshotBeforeCommit,
+        bool expectedExistence,
+        int expectedNumIndexes,
+        std::function<void()> verifyStateCallback = {}) {
         mongo::Mutex mutex;
         stdx::condition_variable cv;
         int numCalls = 0;
@@ -1244,8 +1251,8 @@ private:
         // Stash the catalog so we may perform multiple lookups that will be in sync with our
         // snapshot
         CollectionCatalog::stash(opCtx, CollectionCatalog::get(opCtx));
-        CollectionPtr coll =
-            CollectionCatalog::get(opCtx)->openCollection(opCtx, nssOrUUID, boost::none);
+        CollectionPtr coll = CollectionCatalog::get(opCtx)->establishConsistentCollection(
+            opCtx, nssOrUUID, boost::none);
 
         // Notify the thread that our openCollection lookup is done.
         {
@@ -1383,8 +1390,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenCollectionBeforeCreateTimestamp) {
     const Timestamp readTimestamp(5, 5);
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
     OneOffRead oor(opCtx.get(), readTimestamp);
-    auto coll =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto coll = CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
     ASSERT(!coll);
 
     // Lookups from the catalog should return the newly opened collection (in this case nullptr).
@@ -1414,8 +1421,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenEarlierCollection) {
     const Timestamp readTimestamp(15, 15);
     OneOffRead oor(opCtx.get(), readTimestamp);
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-    auto coll =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto coll = CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
     ASSERT(coll);
     ASSERT_EQ(0, coll->getIndexCatalog()->numIndexesTotal());
 
@@ -1462,8 +1469,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenEarlierCollectionWithIndex) {
     const Timestamp readTimestamp(25, 25);
     OneOffRead oor(opCtx.get(), readTimestamp);
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-    auto coll =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto coll = CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
     ASSERT(coll);
     ASSERT_EQ(1, coll->getIndexCatalog()->numIndexesTotal());
 
@@ -1509,8 +1516,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenLatestCollectionWithIndex) {
     const Timestamp readTimestamp(20, 20);
     OneOffRead oor(opCtx.get(), readTimestamp);
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-    auto coll =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto coll = CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
     ASSERT(coll);
 
     // Verify that the CollectionCatalog returns the latest collection.
@@ -1567,8 +1574,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenEarlierCollectionWithDropPendingIndex
     const Timestamp readTimestamp(20, 20);
     OneOffRead oor(opCtx.get(), readTimestamp);
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-    auto coll =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto coll = CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
     ASSERT(coll);
     ASSERT_EQ(coll->getIndexCatalog()->numIndexesReady(), 2);
 
@@ -1663,8 +1670,8 @@ TEST_F(CollectionCatalogTimestampTest,
     const Timestamp readTimestamp = tsBetweenDroppingYAndZ;
     OneOffRead oor(opCtx.get(), readTimestamp);
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-    auto coll =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto coll = CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
     ASSERT(coll);
     ASSERT_EQ(coll->getIndexCatalog()->numIndexesReady(), 2);
 
@@ -1717,7 +1724,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenEarlierAlreadyDropPendingCollection)
         // Open "a.b", which is not expired in the drop pending map.
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         auto openedColl = CollectionCatalog::get(opCtx.get())
-                              ->openCollection(opCtx.get(), firstNss, readTimestamp);
+                              ->establishConsistentCollection(opCtx.get(), firstNss, readTimestamp);
         ASSERT(openedColl);
         ASSERT_EQ(
             CollectionCatalog::get(opCtx.get())->lookupCollectionByNamespace(opCtx.get(), firstNss),
@@ -1740,8 +1747,9 @@ TEST_F(CollectionCatalogTimestampTest, OpenEarlierAlreadyDropPendingCollection)
         // Before openCollection, looking up the collection returns null.
         ASSERT(CollectionCatalog::get(opCtx.get())
                    ->lookupCollectionByNamespace(opCtx.get(), secondNss) == nullptr);
-        auto openedColl = CollectionCatalog::get(opCtx.get())
-                              ->openCollection(opCtx.get(), secondNss, readTimestamp);
+        auto openedColl =
+            CollectionCatalog::get(opCtx.get())
+                ->establishConsistentCollection(opCtx.get(), secondNss, readTimestamp);
         ASSERT(openedColl);
         ASSERT_EQ(CollectionCatalog::get(opCtx.get())
                       ->lookupCollectionByNamespace(opCtx.get(), secondNss),
@@ -1785,8 +1793,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionUsingDropPendingCollecti
     OneOffRead oor(opCtx.get(), readTimestamp);
 
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-    auto openedColl =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto openedColl = CollectionCatalog::get(opCtx.get())
+                          ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
     ASSERT(openedColl);
     ASSERT_NE(coll.get(), openedColl.get());
     // Ensure the idents are shared between the opened collection and the drop pending collection.
@@ -1832,7 +1840,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenExistingCollectionWithReaper) {
         OneOffRead oor(opCtx.get(), createCollectionTs);
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         auto openedColl = CollectionCatalog::get(opCtx.get())
-                              ->openCollection(opCtx.get(), nss, createCollectionTs);
+                              ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs);
         ASSERT(openedColl);
         ASSERT_EQ(coll->getSharedIdent(), openedColl->getSharedIdent());
 
@@ -1852,7 +1860,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenExistingCollectionWithReaper) {
         OneOffRead oor(opCtx.get(), createCollectionTs);
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         ASSERT(!CollectionCatalog::get(opCtx.get())
-                    ->openCollection(opCtx.get(), nss, createCollectionTs));
+                    ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs));
     }
 }
 
@@ -1878,7 +1886,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionWithReaper) {
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         auto openedColl = CollectionCatalog::get(opCtx.get())
-                              ->openCollection(opCtx.get(), nss, createCollectionTs);
+                              ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs);
         ASSERT(openedColl);
 
         ASSERT_EQ(1, storageEngine->getDropPendingIdents().size());
@@ -1906,7 +1914,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionWithReaper) {
         OneOffRead oor(opCtx.get(), createCollectionTs);
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         ASSERT(!CollectionCatalog::get(opCtx.get())
-                    ->openCollection(opCtx.get(), nss, createCollectionTs));
+                    ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs));
     }
 }
 
@@ -1955,8 +1963,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenExistingCollectionAndIndexesWithReape
         OneOffRead oor(opCtx.get(), createIndexTs);
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-        auto openedColl =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, createIndexTs);
+        auto openedColl = CollectionCatalog::get(opCtx.get())
+                              ->establishConsistentCollection(opCtx.get(), nss, createIndexTs);
         ASSERT(openedColl);
         ASSERT_EQ(openedColl->getSharedIdent(), coll->getSharedIdent());
         ASSERT_EQ(2, openedColl->getIndexCatalog()->numIndexesTotal());
@@ -1971,8 +1979,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenExistingCollectionAndIndexesWithReape
         OneOffRead oor(opCtx.get(), dropXIndexTs);
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IX);
-        auto openedColl =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, dropXIndexTs);
+        auto openedColl = CollectionCatalog::get(opCtx.get())
+                              ->establishConsistentCollection(opCtx.get(), nss, dropXIndexTs);
         ASSERT(openedColl);
         ASSERT_EQ(openedColl->getSharedIdent(), coll->getSharedIdent());
         ASSERT_EQ(1, openedColl->getIndexCatalog()->numIndexesTotal());
@@ -1994,7 +2002,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenExistingCollectionAndIndexesWithReape
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         auto openedColl = CollectionCatalog::get(opCtx.get())
-                              ->openCollection(opCtx.get(), nss, createCollectionTs);
+                              ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs);
         ASSERT(openedColl);
         ASSERT_EQ(openedColl->getSharedIdent(), coll->getSharedIdent());
         ASSERT_EQ(0, openedColl->getIndexCatalog()->numIndexesTotal());
@@ -2006,8 +2014,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenExistingCollectionAndIndexesWithReape
         OneOffRead oor(opCtx.get(), createIndexTs);
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-        ASSERT(
-            !CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, createIndexTs));
+        ASSERT(!CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, createIndexTs));
 
         ASSERT_EQ(2, storageEngine->getDropPendingIdents().size());
     }
@@ -2025,7 +2033,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenExistingCollectionAndIndexesWithReape
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         ASSERT(!CollectionCatalog::get(opCtx.get())
-                    ->openCollection(opCtx.get(), nss, createCollectionTs));
+                    ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs));
     }
 }
 
@@ -2068,8 +2076,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionAndIndexesWithReaper) {
         OneOffRead oor(opCtx.get(), createIndexTs);
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IX);
-        auto openedColl =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, createIndexTs);
+        auto openedColl = CollectionCatalog::get(opCtx.get())
+                              ->establishConsistentCollection(opCtx.get(), nss, createIndexTs);
         ASSERT(openedColl);
         ASSERT_EQ(2, openedColl->getIndexCatalog()->numIndexesTotal());
 
@@ -2083,8 +2091,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionAndIndexesWithReaper) {
         OneOffRead oor(opCtx.get(), dropXIndexTs);
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IX);
-        auto openedColl =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, dropXIndexTs);
+        auto openedColl = CollectionCatalog::get(opCtx.get())
+                              ->establishConsistentCollection(opCtx.get(), nss, dropXIndexTs);
         ASSERT(openedColl);
         ASSERT_EQ(1, openedColl->getIndexCatalog()->numIndexesTotal());
 
@@ -2104,7 +2112,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionAndIndexesWithReaper) {
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         auto openedColl = CollectionCatalog::get(opCtx.get())
-                              ->openCollection(opCtx.get(), nss, createCollectionTs);
+                              ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs);
         ASSERT(openedColl);
         ASSERT_EQ(0, openedColl->getIndexCatalog()->numIndexesTotal());
     }
@@ -2115,8 +2123,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionAndIndexesWithReaper) {
         OneOffRead oor(opCtx.get(), createIndexTs);
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-        ASSERT(
-            !CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, createIndexTs));
+        ASSERT(!CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, createIndexTs));
 
         ASSERT_EQ(2, storageEngine->getDropPendingIdents().size());
     }
@@ -2130,7 +2138,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionAndIndexesWithReaper) {
 
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
         ASSERT(!CollectionCatalog::get(opCtx.get())
-                    ->openCollection(opCtx.get(), nss, createCollectionTs));
+                    ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs));
     }
 }
 
@@ -2640,7 +2648,8 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
     {
         OneOffRead oor(opCtx.get(), Timestamp(1, 17));
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, Timestamp(1, 17));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 17));
 
         // Lookups before the inserted timestamp is still unknown
         ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 11)).result,
@@ -2666,7 +2675,8 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
         OneOffRead oor(opCtx.get(), Timestamp(1, 12));
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
 
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, Timestamp(1, 12));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 12));
 
         // We should now have extended the range from Timestamp(1, 17) to Timestamp(1, 12)
         ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 12)).result,
@@ -2693,7 +2703,8 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
     {
         OneOffRead oor(opCtx.get(), Timestamp(1, 25));
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, Timestamp(1, 25));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 25));
 
         // Check the entries, most didn't change
         ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
@@ -2720,7 +2731,8 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
     {
         OneOffRead oor(opCtx.get(), Timestamp(1, 25));
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, Timestamp(1, 26));
+        CollectionCatalog::get(opCtx.get())
+            ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 26));
 
         // We should not have re-written the existing entry at Timestamp(1, 26)
         ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
@@ -2772,7 +2784,8 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsertUnknown) {
               CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
 
     // Try to instantiate a non existing collection at this timestamp.
-    CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, Timestamp(1, 15));
+    CollectionCatalog::get(opCtx.get())
+        ->establishConsistentCollection(opCtx.get(), nss, Timestamp(1, 15));
 
     // Lookup should now be not existing
     ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
@@ -2802,8 +2815,8 @@ TEST_F(CollectionCatalogTimestampTest, CollectionLifetimeTiedToStorageTransactio
         OneOffRead oor(opCtx.get(), readTimestamp);
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
 
-        auto coll =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+        auto coll = CollectionCatalog::get(opCtx.get())
+                        ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
         ASSERT(coll);
 
         std::shared_ptr<const Collection> fetchedColl =
@@ -2821,8 +2834,8 @@ TEST_F(CollectionCatalogTimestampTest, CollectionLifetimeTiedToStorageTransactio
         OneOffRead oor(opCtx.get(), readTimestamp);
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
 
-        auto coll =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+        auto coll = CollectionCatalog::get(opCtx.get())
+                        ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
         ASSERT(coll);
 
         WriteUnitOfWork wuow(opCtx.get());
@@ -2845,8 +2858,8 @@ TEST_F(CollectionCatalogTimestampTest, CollectionLifetimeTiedToStorageTransactio
         OneOffRead oor(opCtx.get(), readTimestamp);
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
 
-        auto coll =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+        auto coll = CollectionCatalog::get(opCtx.get())
+                        ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
         ASSERT(coll);
 
         boost::optional<WriteUnitOfWork> wuow(opCtx.get());
@@ -2966,8 +2979,8 @@ DEATH_TEST_F(CollectionCatalogTimestampTest, OpenCollectionInWriteUnitOfWork, "i
     WriteUnitOfWork wuow(opCtx.get());
 
     Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
-    auto coll =
-        CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, readTimestamp);
+    auto coll = CollectionCatalog::get(opCtx.get())
+                    ->establishConsistentCollection(opCtx.get(), nss, readTimestamp);
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateCollectionAndOpenCollectionBeforeCommit) {
@@ -2979,7 +2992,7 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateCollectionAndOpenCollecti
 
     // When the snapshot is opened right before the create is committed to the durable catalog, the
     // collection instance should not exist yet.
-    concurrentCreateCollectionAndOpenCollection(
+    concurrentCreateCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, boost::none, createCollectionTs, true, false, 0);
 }
 
@@ -2992,7 +3005,7 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateCollectionAndOpenCollecti
 
     // When the snapshot is opened right after the create is committed to the durable catalog, the
     // collection instance should exist.
-    concurrentCreateCollectionAndOpenCollection(
+    concurrentCreateCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, boost::none, createCollectionTs, false, true, 0);
 }
 
@@ -3007,7 +3020,7 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right before the create is committed to the durable catalog, the
     // collection instance should not exist yet.
-    concurrentCreateCollectionAndOpenCollection(
+    concurrentCreateCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, uuid, createCollectionTs, true, false, 0);
 }
 
@@ -3022,7 +3035,7 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right after the create is committed to the durable catalog, the
     // collection instance should exist.
-    concurrentCreateCollectionAndOpenCollection(
+    concurrentCreateCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, uuid, createCollectionTs, false, true, 0);
 }
 
@@ -3038,7 +3051,7 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentDropCollectionAndOpenCollection
 
     // When the snapshot is opened right before the drop is committed to the durable catalog, the
     // collection instance should be returned.
-    concurrentDropCollectionAndOpenCollection(
+    concurrentDropCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, nss, dropCollectionTs, true, true, 0);
 }
 
@@ -3054,7 +3067,7 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentDropCollectionAndOpenCollection
 
     // When the snapshot is opened right after the drop is committed to the durable catalog, no
     // collection instance should be returned.
-    concurrentDropCollectionAndOpenCollection(
+    concurrentDropCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, nss, dropCollectionTs, false, false, 0);
 }
 
@@ -3074,7 +3087,7 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right before the drop is committed to the durable catalog, the
     // collection instance should be returned.
-    concurrentDropCollectionAndOpenCollection(
+    concurrentDropCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, uuidWithDbName, dropCollectionTs, true, true, 0);
 }
 
@@ -3093,7 +3106,7 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentDropCollectionAndOpenCollection
 
     // When the snapshot is opened right after the drop is committed to the durable catalog, no
     // collection instance should be returned.
-    concurrentDropCollectionAndOpenCollection(
+    concurrentDropCollectionAndEstablishConsistentCollection(
         opCtx.get(), nss, uuidWithDbName, dropCollectionTs, false, false, 0);
 }
 
@@ -3111,7 +3124,7 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right before the rename is committed to the durable catalog, and
     // the openCollection looks for the originalNss, the collection instance should be returned.
-    concurrentRenameCollectionAndOpenCollection(
+    concurrentRenameCollectionAndEstablishConsistentCollection(
         opCtx.get(), originalNss, newNss, originalNss, renameCollectionTs, true, true, 0);
 }
 
@@ -3132,7 +3145,7 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right after the rename is committed to the durable catalog, and
     // the openCollection looks for the originalNss, no collection instance should be returned.
-    concurrentRenameCollectionAndOpenCollection(
+    concurrentRenameCollectionAndEstablishConsistentCollection(
         opCtx.get(), originalNss, newNss, originalNss, renameCollectionTs, false, false, 0, [&]() {
             // Verify that we can find the Collection when we search by UUID when the setup occured
             // during concurrent rename (rename is not affecting UUID), even if we can't find it by
@@ -3164,7 +3177,7 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right before the rename is committed to the durable catalog, and
     // the openCollection looks for the newNss, no collection instance should be returned.
-    concurrentRenameCollectionAndOpenCollection(
+    concurrentRenameCollectionAndEstablishConsistentCollection(
         opCtx.get(), originalNss, newNss, newNss, renameCollectionTs, true, false, 0, [&]() {
             // Verify that we can find the Collection when we search by UUID when the setup occured
             // during concurrent rename (rename is not affecting UUID), even if we can't find it by
@@ -3193,7 +3206,7 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right after the rename is committed to the durable catalog, and
     // the openCollection looks for the newNss, the collection instance should be returned.
-    concurrentRenameCollectionAndOpenCollection(
+    concurrentRenameCollectionAndEstablishConsistentCollection(
         opCtx.get(), originalNss, newNss, newNss, renameCollectionTs, false, true, 0);
 }
 
@@ -3214,7 +3227,7 @@ TEST_F(CollectionCatalogTimestampTest,
     NamespaceStringOrUUID uuidWithDbName(originalNss.dbName(), uuid);
     // When the snapshot is opened right before the rename is committed to the durable catalog, and
     // the openCollection looks for the originalNss, the collection instance should be returned.
-    concurrentRenameCollectionAndOpenCollection(
+    concurrentRenameCollectionAndEstablishConsistentCollection(
         opCtx.get(), originalNss, newNss, uuidWithDbName, renameCollectionTs, true, true, 0, [&]() {
             // Verify that we cannot find the Collection when we search by the new namespace as
             // the rename was committed when we read.
@@ -3242,23 +3255,23 @@ TEST_F(CollectionCatalogTimestampTest,
 
     // When the snapshot is opened right after the rename is committed to the durable catalog, and
     // the openCollection looks for the originalNss, no collection instance should be returned.
-    concurrentRenameCollectionAndOpenCollection(opCtx.get(),
-                                                originalNss,
-                                                newNss,
-                                                uuidWithDbName,
-                                                renameCollectionTs,
-                                                false,
-                                                true,
-                                                0,
-                                                [&]() {
-                                                    // Verify that we cannot find the Collection
-                                                    // when we search by the original namespace as
-                                                    // the rename was committed when we read.
-                                                    auto coll = CollectionCatalog::get(opCtx.get())
-                                                                    ->lookupCollectionByNamespace(
-                                                                        opCtx.get(), originalNss);
-                                                    ASSERT(!coll);
-                                                });
+    concurrentRenameCollectionAndEstablishConsistentCollection(
+        opCtx.get(),
+        originalNss,
+        newNss,
+        uuidWithDbName,
+        renameCollectionTs,
+        false,
+        true,
+        0,
+        [&]() {
+            // Verify that we cannot find the Collection
+            // when we search by the original namespace as
+            // the rename was committed when we read.
+            auto coll = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), originalNss);
+            ASSERT(!coll);
+        });
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionBeforeCommit) {
@@ -3280,16 +3293,16 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionBef
 
     // When the snapshot is opened right before the second index create is committed to the durable
     // catalog, the collection instance should not have the second index.
-    concurrentCreateIndexAndOpenCollection(opCtx.get(),
-                                           nss,
-                                           nss,
-                                           BSON("v" << 2 << "name"
-                                                    << "y_1"
-                                                    << "key" << BSON("y" << 1)),
-                                           createYIndexTs,
-                                           true,
-                                           true,
-                                           1);
+    concurrentCreateIndexAndEstablishConsistentCollection(opCtx.get(),
+                                                          nss,
+                                                          nss,
+                                                          BSON("v" << 2 << "name"
+                                                                   << "y_1"
+                                                                   << "key" << BSON("y" << 1)),
+                                                          createYIndexTs,
+                                                          true,
+                                                          true,
+                                                          1);
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionAfterCommit) {
@@ -3311,16 +3324,16 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionAft
 
     // When the snapshot is opened right before the second index create is committed to the durable
     // catalog, the collection instance should have both indexes.
-    concurrentCreateIndexAndOpenCollection(opCtx.get(),
-                                           nss,
-                                           nss,
-                                           BSON("v" << 2 << "name"
-                                                    << "y_1"
-                                                    << "key" << BSON("y" << 1)),
-                                           createYIndexTs,
-                                           false,
-                                           true,
-                                           2);
+    concurrentCreateIndexAndEstablishConsistentCollection(opCtx.get(),
+                                                          nss,
+                                                          nss,
+                                                          BSON("v" << 2 << "name"
+                                                                   << "y_1"
+                                                                   << "key" << BSON("y" << 1)),
+                                                          createYIndexTs,
+                                                          false,
+                                                          true,
+                                                          2);
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionByUUIDBeforeCommit) {
@@ -3345,16 +3358,16 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionByU
 
     // When the snapshot is opened right before the second index create is committed to the durable
     // catalog, the collection instance should not have the second index.
-    concurrentCreateIndexAndOpenCollection(opCtx.get(),
-                                           nss,
-                                           uuidWithDbName,
-                                           BSON("v" << 2 << "name"
-                                                    << "y_1"
-                                                    << "key" << BSON("y" << 1)),
-                                           createYIndexTs,
-                                           true,
-                                           true,
-                                           1);
+    concurrentCreateIndexAndEstablishConsistentCollection(opCtx.get(),
+                                                          nss,
+                                                          uuidWithDbName,
+                                                          BSON("v" << 2 << "name"
+                                                                   << "y_1"
+                                                                   << "key" << BSON("y" << 1)),
+                                                          createYIndexTs,
+                                                          true,
+                                                          true,
+                                                          1);
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionByUUIDAfterCommit) {
@@ -3379,16 +3392,16 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionByU
 
     // When the snapshot is opened right before the second index create is committed to the durable
     // catalog, the collection instance should have both indexes.
-    concurrentCreateIndexAndOpenCollection(opCtx.get(),
-                                           nss,
-                                           uuidWithDbName,
-                                           BSON("v" << 2 << "name"
-                                                    << "y_1"
-                                                    << "key" << BSON("y" << 1)),
-                                           createYIndexTs,
-                                           false,
-                                           true,
-                                           2);
+    concurrentCreateIndexAndEstablishConsistentCollection(opCtx.get(),
+                                                          nss,
+                                                          uuidWithDbName,
+                                                          BSON("v" << 2 << "name"
+                                                                   << "y_1"
+                                                                   << "key" << BSON("y" << 1)),
+                                                          createYIndexTs,
+                                                          false,
+                                                          true,
+                                                          2);
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentDropIndexAndOpenCollectionBeforeCommit) {
@@ -3416,7 +3429,8 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentDropIndexAndOpenCollectionBefor
 
     // When the snapshot is opened right before the index drop is committed to the durable
     // catalog, the collection instance should not have the second index.
-    concurrentDropIndexAndOpenCollection(opCtx.get(), nss, nss, "y_1", dropIndexTs, true, true, 2);
+    concurrentDropIndexAndEstablishConsistentCollection(
+        opCtx.get(), nss, nss, "y_1", dropIndexTs, true, true, 2);
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentDropIndexAndOpenCollectionAfterCommit) {
@@ -3444,7 +3458,8 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentDropIndexAndOpenCollectionAfter
 
     // When the snapshot is opened right before the index drop is committed to the durable
     // catalog, the collection instance should not have the second index.
-    concurrentDropIndexAndOpenCollection(opCtx.get(), nss, nss, "y_1", dropIndexTs, false, true, 1);
+    concurrentDropIndexAndEstablishConsistentCollection(
+        opCtx.get(), nss, nss, "y_1", dropIndexTs, false, true, 1);
 }
 
 TEST_F(CollectionCatalogTimestampTest, ConcurrentDropIndexAndOpenCollectionByUUIDBeforeCommit) {
@@ -3475,7 +3490,7 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentDropIndexAndOpenCollectionByUUI
 
     // When the snapshot is opened right before the index drop is committed to the durable
     // catalog, the collection instance should not have the second index.
-    concurrentDropIndexAndOpenCollection(
+    concurrentDropIndexAndEstablishConsistentCollection(
         opCtx.get(), nss, uuidWithDbName, "y_1", dropIndexTs, true, true, 2);
 }
 
@@ -3507,7 +3522,7 @@ TEST_F(CollectionCatalogTimestampTest, ConcurrentDropIndexAndOpenCollectionByUUI
 
     // When the snapshot is opened right before the index drop is committed to the durable
     // catalog, the collection instance should not have the second index.
-    concurrentDropIndexAndOpenCollection(
+    concurrentDropIndexAndEstablishConsistentCollection(
         opCtx.get(), nss, uuidWithDbName, "y_1", dropIndexTs, false, true, 1);
 }
 
@@ -3535,7 +3550,7 @@ TEST_F(CollectionCatalogTimestampTest, OpenCollectionBetweenIndexBuildInProgress
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
 
         auto coll = CollectionCatalog::get(opCtx.get())
-                        ->openCollection(opCtx.get(), nss, createCollectionTs);
+                        ->establishConsistentCollection(opCtx.get(), nss, createCollectionTs);
         ASSERT(coll);
         ASSERT_EQ(coll->getIndexCatalog()->numIndexesReady(), 0);
 
@@ -3558,8 +3573,8 @@ TEST_F(CollectionCatalogTimestampTest, OpenCollectionBetweenIndexBuildInProgress
         OneOffRead oor(opCtx.get(), createIndexTs);
         Lock::GlobalLock globalLock(opCtx.get(), MODE_IS);
 
-        auto coll =
-            CollectionCatalog::get(opCtx.get())->openCollection(opCtx.get(), nss, createIndexTs);
+        auto coll = CollectionCatalog::get(opCtx.get())
+                        ->establishConsistentCollection(opCtx.get(), nss, createIndexTs);
         ASSERT(coll);
         ASSERT_EQ(coll->getIndexCatalog()->numIndexesReady(), 0);
 
diff --git a/src/mongo/db/catalog/collection_uuid_mismatch.cpp b/src/mongo/db/catalog/collection_uuid_mismatch.cpp
index a82c197b1fb..76d5f524770 100644
--- a/src/mongo/db/catalog/collection_uuid_mismatch.cpp
+++ b/src/mongo/db/catalog/collection_uuid_mismatch.cpp
@@ -45,7 +45,7 @@ void checkCollectionUUIDMismatch(OperationContext* opCtx,
 void checkCollectionUUIDMismatch(OperationContext* opCtx,
                                  const std::shared_ptr<const CollectionCatalog>& catalog,
                                  const NamespaceString& ns,
-                                 const Collection* coll,
+                                 const CollectionPtr& coll,
                                  const boost::optional<UUID>& uuid) {
     if (!uuid) {
         return;
diff --git a/src/mongo/db/catalog/collection_uuid_mismatch.h b/src/mongo/db/catalog/collection_uuid_mismatch.h
index 1ebbef31c51..bc4f3d4158c 100644
--- a/src/mongo/db/catalog/collection_uuid_mismatch.h
+++ b/src/mongo/db/catalog/collection_uuid_mismatch.h
@@ -46,7 +46,7 @@ void checkCollectionUUIDMismatch(OperationContext* opCtx,
 void checkCollectionUUIDMismatch(OperationContext* opCtx,
                                  const std::shared_ptr<const CollectionCatalog>& catalog,
                                  const NamespaceString& ns,
-                                 const Collection* coll,
+                                 const CollectionPtr& coll,
                                  const boost::optional<UUID>& uuid);
 
 }  // namespace mongo
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index b3e66797bfd..f9544eab063 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -127,22 +127,11 @@ void assertCollectionChangesCompatibleWithReadTimestamp(OperationContext* opCtx,
 }
 
 /**
- * Performs some sanity checks on the collection and database.
+ * Performs validation of special locking requirements for certain namespaces.
  */
-void verifyDbAndCollectionReadIntent(OperationContext* opCtx,
-                                     LockMode modeColl,
-                                     const NamespaceStringOrUUID& nsOrUUID,
-                                     const NamespaceString& resolvedNss,
-                                     CollectionPtr& coll,
-                                     Database* db) {
-    invariant(!nsOrUUID.uuid() || coll,
-              str::stream() << "Collection for " << resolvedNss.ns()
-                            << " disappeared after successfully resolving " << nsOrUUID.toString());
-
-    invariant(!nsOrUUID.uuid() || db,
-              str::stream() << "Database for " << resolvedNss.ns()
-                            << " disappeared after successfully resolving " << nsOrUUID.toString());
-
+void verifyNamespaceLockingRequirements(OperationContext* opCtx,
+                                        LockMode modeColl,
+                                        const NamespaceString& resolvedNss) {
     // In most cases we expect modifications for system.views to upgrade MODE_IX to MODE_X before
     // taking the lock. One exception is a query by UUID of system.views in a transaction. Usual
     // queries of system.views (by name, not UUID) within a transaction are rejected. However, if
@@ -474,32 +463,6 @@ void checkInvariantsForReadOptions(const NamespaceString& nss,
     }
 }
 
-/**
- * Returns a collection reference compatible with the specified 'readTimestamp'. Creates and places
- * a compatible PIT collection reference in the 'catalog' if needed and the collection exists at
- * that PIT.
- */
-CollectionPtr fetchOrCreatePITCollection(OperationContext* opCtx,
-                                         const CollectionCatalog* catalog,
-                                         const NamespaceString nss,
-                                         boost::optional<Timestamp> readTimestamp,
-                                         bool isPrimaryNs) {
-    if (catalog->needsOpenCollection(opCtx, nss, readTimestamp)) {
-        // Getting this far means that either the collection is not present in the catalog (it may
-        // have been dropped) or the read timestamp is earlier than the last DDL operation
-        // (minValidSnapshot). So try to create a valid past Collection instance for the
-        // readTimestamp.
-        (void)catalog->openCollection(opCtx, nss, readTimestamp);
-    }
-
-    // TODO SERVER-70846: openCollection does not set a yield handler on the CollectionPtr
-    // returned. Therefore, it is necessary to lookup the collection again via a different
-    // lookup fn to fetch a yielding-aware CollectionPtr from the catalog.
-    CollectionPtr coll = catalog->lookupCollectionByNamespace(opCtx, nss);
-    coll.makeYieldable(opCtx, LockedCollectionYieldRestore{opCtx, coll});
-    return coll;
-}
-
 }  // namespace
 
 AutoStatsTracker::AutoStatsTracker(
@@ -760,7 +723,6 @@ AutoGetCollectionForReadPITCatalog::AutoGetCollectionForReadPITCatalog(
         [&](const BSONObj& data) { sleepFor(Milliseconds(data["waitForMillis"].numberInt())); });
 
     auto catalog = CollectionCatalog::get(opCtx);
-    auto databaseHolder = DatabaseHolder::get(opCtx);
 
     _resolvedNss = catalog->resolveNamespaceStringOrUUID(opCtx, nsOrUUID);
 
@@ -780,14 +742,15 @@ AutoGetCollectionForReadPITCatalog::AutoGetCollectionForReadPITCatalog(
     const auto readSource = opCtx->recoveryUnit()->getTimestampReadSource();
     const auto readTimestamp = opCtx->recoveryUnit()->getPointInTimeReadTimestamp(opCtx);
 
-    // Check that the collections are all safe to use. First acquire collection from our catalog.
-    _coll = fetchOrCreatePITCollection(
-        opCtx, catalog.get(), _resolvedNss, readTimestamp, true /* isPrimaryNs */);
+    // Check that the collections are all safe to use. First acquire collection from our catalog
+    // compatible with the specified 'readTimestamp'. Creates and places a compatible PIT collection
+    // reference in the 'catalog' if needed and the collection exists at that PIT.
+    _coll = catalog->establishConsistentCollection(opCtx, nsOrUUID, readTimestamp);
+    _coll.makeYieldable(opCtx, LockedCollectionYieldRestore{opCtx, _coll});
 
     // Validate primary collection.
     checkCollectionUUIDMismatch(opCtx, _resolvedNss, _coll, options._expectedUUID);
-    verifyDbAndCollectionReadIntent(
-        opCtx, modeColl, nsOrUUID, _resolvedNss, _coll, _autoDb.getDb());
+    verifyNamespaceLockingRequirements(opCtx, modeColl, _resolvedNss);
 
     // Check secondary collections and verify they are valid for use.
     if (!secondaryNssOrUUIDs.empty()) {
@@ -800,18 +763,13 @@ AutoGetCollectionForReadPITCatalog::AutoGetCollectionForReadPITCatalog(
         if (!_secondaryNssIsAViewOrSharded) {
             // Ensure that the readTimestamp is compatible with the latest Collection instances or
             // create PIT instances in the 'catalog' (if the collections existed at that PIT).
-            for (const auto& secondaryNss : *resolvedSecondaryNamespaces) {
-                auto secondaryCollectionAtPIT = fetchOrCreatePITCollection(
-                    opCtx, catalog.get(), secondaryNss, readTimestamp, false /* isPrimaryNs */);
+            for (const auto& secondaryNssOrUUID : secondaryNssOrUUIDs) {
+                auto secondaryCollectionAtPIT = catalog->establishConsistentCollection(
+                    opCtx, secondaryNssOrUUID, readTimestamp);
                 if (secondaryCollectionAtPIT) {
                     invariant(secondaryCollectionAtPIT->ns().dbName() == _resolvedNss.dbName());
-                    verifyDbAndCollectionReadIntent(
-                        opCtx,
-                        MODE_IS,
-                        secondaryNss,
-                        secondaryNss,
-                        secondaryCollectionAtPIT,
-                        databaseHolder->getDb(opCtx, secondaryNss.dbName()));
+                    verifyNamespaceLockingRequirements(
+                        opCtx, MODE_IS, secondaryCollectionAtPIT->ns());
                 }
             }
         }
@@ -1197,42 +1155,7 @@ std::shared_ptr<const ViewDefinition> lookupView(
     return view;
 }
 
-const Collection* getCollectionFromCatalog(OperationContext* opCtx,
-                                           const std::shared_ptr<const CollectionCatalog>& catalog,
-                                           const NamespaceStringOrUUID& nsOrUUID,
-                                           const boost::optional<Timestamp>& readTimestamp) {
-    if (catalog->needsOpenCollection(opCtx, nsOrUUID, readTimestamp)) {
-        auto coll = catalog->openCollection(opCtx, nsOrUUID, readTimestamp).get();
-        if (coll) {
-            if (auto capSnap =
-                    CappedSnapshots::get(opCtx).getSnapshot(coll->getRecordStore()->getIdent())) {
-                // The only way openCollection can be required for a collection that uses capped
-                // snapshots (i.e. a collection that is unreplicated and capped) is:
-                //  * The present read operation is reading without a timestamp (since unreplicated
-                //    collections don't support timestamped reads), and
-                //  * When opening the storage snapshot (and thus when establishing the capped
-                //    snapshot), there was a DDL operation pending on the namespace or UUID
-                //    requested for this read (because this is the only time openCollection is
-                //    called for an untimestamped read).
-                //
-                // Because DDL operations require a collection X lock, there cannot have been any
-                // ongoing concurrent writes to the collection while establishing the capped
-                // snapshot. This means that if there was a capped snapshot, it should not have
-                // contained any uncommitted writes, and so the _lowestUncommittedRecord must be
-                // null.
-                invariant(!capSnap->hasUncommittedRecords());
-            }
-        }
-        return coll;
-    } else {
-        // It's safe to extract the raw pointer from the CollectionPtr returned by lookupCollection
-        // because it is owned by the catalog, which will remain valid for as long as we're using
-        // the collection returned here.
-        return catalog->lookupCollectionByNamespaceOrUUID(opCtx, nsOrUUID).get();
-    }
-}
-
-std::tuple<NamespaceString, const Collection*, std::shared_ptr<const ViewDefinition>>
+std::tuple<NamespaceString, CollectionPtr, std::shared_ptr<const ViewDefinition>>
 getCollectionForLockFreeRead(OperationContext* opCtx,
                              const std::shared_ptr<const CollectionCatalog>& catalog,
                              boost::optional<Timestamp> readTimestamp,
@@ -1250,8 +1173,11 @@ getCollectionForLockFreeRead(OperationContext* opCtx,
                 opCtx->getLogicalSessionId()->getId() == UUID::fromCDR(data["lsid"].uuid());
         });
 
-    auto coll = getCollectionFromCatalog(opCtx, catalog, nsOrUUID, readTimestamp);
 
+    // Returns a collection reference compatible with the specified 'readTimestamp'. Creates and
+    // places a compatible PIT collection reference in the 'catalog' if needed and the collection
+    // exists at that PIT.
+    CollectionPtr coll = catalog->establishConsistentCollection(opCtx, nsOrUUID, readTimestamp);
     // TODO (SERVER-71222): This is broken if the UUID doesn't exist in the latest catalog.
     //
     // Note: This call to resolveNamespaceStringOrUUID must happen after getCollectionFromCatalog
@@ -1260,7 +1186,9 @@ getCollectionForLockFreeRead(OperationContext* opCtx,
     const auto nss = catalog->resolveNamespaceStringOrUUID(opCtx, nsOrUUID);
     checkCollectionUUIDMismatch(opCtx, catalog, nss, coll, options._expectedUUID);
 
-    return {nss, coll, coll ? nullptr : lookupView(opCtx, catalog, nss, options._viewMode)};
+    std::shared_ptr<const ViewDefinition> viewDefinition =
+        coll ? nullptr : lookupView(opCtx, catalog, nss, options._viewMode);
+    return {nss, std::move(coll), std::move(viewDefinition)};
 }
 
 static const Lock::GlobalLockSkipOptions kLockFreeReadsGlobalLockOptions{[] {
@@ -1269,16 +1197,11 @@ static const Lock::GlobalLockSkipOptions kLockFreeReadsGlobalLockOptions{[] {
     return options;
 }()};
 
-CollectionPtr makeCollectionPtrForLockFreeReadSubOperation(OperationContext* opCtx,
-                                                           const Collection* coll) {
-    return {coll};
-}
-
 struct CatalogStateForNamespace {
     std::shared_ptr<const CollectionCatalog> catalog;
     bool isAnySecondaryNssShardedOrAView;
     NamespaceString resolvedNss;
-    const Collection* collection;
+    CollectionPtr collection;
     std::shared_ptr<const ViewDefinition> view;
 };
 
@@ -1300,7 +1223,7 @@ CatalogStateForNamespace acquireCatalogStateForNamespace(
         getCollectionForLockFreeRead(opCtx, catalog, readTimestamp, nsOrUUID, options);
 
     return CatalogStateForNamespace{
-        catalog, isAnySecondaryNssShardedOrAView, resolvedNss, collection, view};
+        catalog, isAnySecondaryNssShardedOrAView, resolvedNss, std::move(collection), view};
 }
 
 boost::optional<ShouldNotConflictWithSecondaryBatchApplicationBlock>
@@ -1339,7 +1262,7 @@ CollectionPtr::RestoreFn AutoGetCollectionForReadLockFreePITCatalog::_makeRestor
             _view = catalogStateForNamespace.view;
             _catalogStasher.stash(std::move(catalogStateForNamespace.catalog));
 
-            return catalogStateForNamespace.collection;
+            return catalogStateForNamespace.collection.get();
         } catch (const ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
             // Calls to CollectionCatalog::resolveNamespaceStringOrUUID (called from
             // acquireCatalogStateForNamespace) will result in a NamespaceNotFound error if the
@@ -1397,7 +1320,7 @@ AutoGetCollectionForReadLockFreePITCatalog::AutoGetCollectionForReadLockFreePITC
         if (_view) {
             _lockFreeReadsBlock.reset();
         }
-        _collectionPtr = makeCollectionPtrForLockFreeReadSubOperation(opCtx, collection);
+        _collectionPtr = std::move(collection);
     } else {
         auto catalogStateForNamespace =
             acquireCatalogStateForNamespace(opCtx,
@@ -1415,9 +1338,7 @@ AutoGetCollectionForReadLockFreePITCatalog::AutoGetCollectionForReadLockFreePITC
         _catalogStasher.stash(std::move(catalogStateForNamespace.catalog));
         _secondaryNssIsAViewOrSharded = catalogStateForNamespace.isAnySecondaryNssShardedOrAView;
 
-        _collectionPtr = CollectionPtr(
-
-            catalogStateForNamespace.collection);
+        _collectionPtr = std::move(catalogStateForNamespace.collection);
         _collectionPtr.makeYieldable(
             opCtx,
             _makeRestoreFromYieldFn(options,