SERVER-68254 Create a function to instantiate collections and index entries from earlier points in time

14 files changed, 879 insertions, 36 deletions

diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 097b115864f..ad129695d7c 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -218,6 +218,14 @@ public:
 
     virtual void init(OperationContext* opCtx) {}
 
+    /**
+     * Initializes a collection representative at the provided read timestamp using the shared state
+     * from an already existing, later collection.
+     */
+    virtual void initFromExisting(OperationContext* opCtx,
+                                  std::shared_ptr<Collection> collection,
+                                  Timestamp readTimestamp) {}
+
     virtual bool isCommitted() const {
         return true;
     }
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 1c4fd93cffb..29bb7c6093c 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -38,6 +38,7 @@
 #include "mongo/db/multitenancy_gen.h"
 #include "mongo/db/server_feature_flags_gen.h"
 #include "mongo/db/server_options.h"
+#include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/storage/snapshot_helper.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
@@ -61,6 +62,22 @@ std::shared_ptr<CollectionCatalog> batchedCatalogWriteInstance;
 const OperationContext::Decoration<std::shared_ptr<const CollectionCatalog>> stashedCatalog =
     OperationContext::declareDecoration<std::shared_ptr<const CollectionCatalog>>();
 
+/**
+ * Returns true if the collection is compatible with the read timestamp.
+ */
+bool isCollectionCompatible(std::shared_ptr<Collection> coll, Timestamp readTimestamp) {
+    if (!coll) {
+        return false;
+    }
+
+    boost::optional<Timestamp> minValidSnapshot = coll->getMinimumValidSnapshot();
+    if (!minValidSnapshot) {
+        // Collection is valid in all snapshots.
+        return true;
+    }
+    return readTimestamp >= *minValidSnapshot;
+}
+
 }  // namespace
 
 class IgnoreExternalViewChangesForDatabase {
@@ -659,6 +676,91 @@ Status CollectionCatalog::reloadViews(OperationContext* opCtx, const DatabaseNam
     return status;
 }
 
+std::shared_ptr<Collection> CollectionCatalog::openCollection(OperationContext* opCtx,
+                                                              const DurableCatalog::Entry& entry,
+                                                              Timestamp readTimestamp) const {
+    if (!feature_flags::gPointInTimeCatalogLookups.isEnabledAndIgnoreFCV()) {
+        return nullptr;
+    }
+
+    auto metadata = DurableCatalog::get(opCtx)->getMetaData(opCtx, entry.catalogId);
+    if (!metadata) {
+        // Treat the collection as non-existent at a point-in-time the same as not-existent at
+        // latest.
+        return nullptr;
+    }
+
+    const auto& collectionOptions = metadata->options;
+
+    // Check if the collection already exists in the catalog and if it's compatible with the read
+    // timestamp.
+    std::shared_ptr<Collection> latestColl = _lookupCollectionByUUID(*collectionOptions.uuid);
+    if (isCollectionCompatible(latestColl, readTimestamp)) {
+        return latestColl;
+    }
+
+    // Check if the collection is drop pending, not expired, and compatible with the read timestamp.
+    std::shared_ptr<Collection> dropPendingColl = [&]() -> std::shared_ptr<Collection> {
+        auto dropPendingIt = _dropPendingCollection.find(entry.ident);
+        if (dropPendingIt == _dropPendingCollection.end()) {
+            return nullptr;
+        }
+
+        return dropPendingIt->second.lock();
+    }();
+
+    if (isCollectionCompatible(dropPendingColl, readTimestamp)) {
+        return dropPendingColl;
+    }
+
+    // Neither the latest collection or drop pending collection exist, or were compatible with the
+    // read timestamp. We'll need to instantiate a new Collection instance.
+    if (latestColl || dropPendingColl) {
+        // If the latest or drop pending collection exists, instantiate a new collection using
+        // their shared state.
+        LOGV2_DEBUG(6825400,
+                    1,
+                    "Instantiating a collection using shared state",
+                    logAttrs(entry.nss),
+                    "ident"_attr = entry.ident,
+                    "md"_attr = metadata->toBSON(),
+                    "timestamp"_attr = readTimestamp);
+
+        std::shared_ptr<Collection> collToReturn = Collection::Factory::get(opCtx)->make(
+            opCtx, entry.nss, entry.catalogId, metadata, /*rs=*/nullptr);
+        collToReturn->initFromExisting(
+            opCtx, latestColl ? latestColl : dropPendingColl, readTimestamp);
+        return collToReturn;
+    }
+
+    // Instantiate a new collection without any shared state.
+    LOGV2_DEBUG(6825401,
+                1,
+                "Instantiating a new collection",
+                logAttrs(entry.nss),
+                "ident"_attr = entry.ident,
+                "md"_attr = metadata->toBSON(),
+                "timestamp"_attr = readTimestamp);
+
+    std::unique_ptr<RecordStore> rs =
+        opCtx->getServiceContext()->getStorageEngine()->getEngine()->getRecordStore(
+            opCtx, entry.nss, entry.ident, collectionOptions);
+    std::shared_ptr<Collection> collToReturn = Collection::Factory::get(opCtx)->make(
+        opCtx, entry.nss, entry.catalogId, metadata, std::move(rs));
+    collToReturn->init(opCtx);
+    return collToReturn;
+}
+
+std::shared_ptr<IndexCatalogEntry> CollectionCatalog::findDropPendingIndex(
+    const std::string& ident) const {
+    auto it = _dropPendingIndex.find(ident);
+    if (it == _dropPendingIndex.end()) {
+        return nullptr;
+    }
+
+    return it->second.lock();
+}
+
 void CollectionCatalog::onCreateCollection(OperationContext* opCtx,
                                            std::shared_ptr<Collection> coll) const {
     invariant(coll);
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index f38f85c5841..e5d2845a527 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -38,6 +38,8 @@
 #include "mongo/db/database_name.h"
 #include "mongo/db/profile_filter.h"
 #include "mongo/db/service_context.h"
+// TODO SERVER-68265: remove include.
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/views/view.h"
 #include "mongo/stdx/unordered_map.h"
 #include "mongo/util/uuid.h"
@@ -206,6 +208,28 @@ public:
     Status reloadViews(OperationContext* opCtx, const DatabaseName& dbName) const;
 
     /**
+     * Returns the collection instance representative of 'entry' at the provided read timestamp.
+     *
+     * TODO SERVER-68571:
+     * - If the data files have already been removed, return nullptr.
+     * - Update the drop pending map ident token when not initialized from shared state.
+     *
+     * TODO SERVER-68265:
+     * - Use NamespaceString instead of DurableCatalog::Entry.
+     * - Remove DurableCatalog dependency.
+     *
+     * Returns nullptr when reading from a point-in-time where the collection did not exist.
+     */
+    std::shared_ptr<Collection> openCollection(OperationContext* opCtx,
+                                               const DurableCatalog::Entry& entry,
+                                               Timestamp readTimestamp) const;
+
+    /**
+     * Returns a shared_ptr to a drop pending index if it's found and not expired.
+     */
+    std::shared_ptr<IndexCatalogEntry> findDropPendingIndex(const std::string& ident) const;
+
+    /**
      * Handles committing a collection to the catalog within a WriteUnitOfWork.
      *
      * Must be called within a WriteUnitOfWork.
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 1d7632709ec..43a089fea2b 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -34,8 +34,11 @@
 #include "mongo/db/catalog/catalog_test_fixture.h"
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/collection_mock.h"
+#include "mongo/db/catalog_raii.h"
 #include "mongo/db/concurrency/lock_manager_defs.h"
+#include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/service_context_d_test_fixture.h"
+#include "mongo/idl/server_parameter_test_util.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 
@@ -857,5 +860,514 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
     }
 }
 
+/**
+ * RAII type for operating at a timestamp. Will remove any timestamping when the object destructs.
+ */
+class OneOffRead {
+public:
+    OneOffRead(OperationContext* opCtx, const Timestamp& ts) : _opCtx(opCtx) {
+        _opCtx->recoveryUnit()->abandonSnapshot();
+        if (ts.isNull()) {
+            _opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
+        } else {
+            _opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kProvided, ts);
+        }
+    }
+
+    ~OneOffRead() {
+        _opCtx->recoveryUnit()->abandonSnapshot();
+        _opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
+    }
+
+private:
+    OperationContext* _opCtx;
+};
+
+class CollectionCatalogTimestampTest : public ServiceContextMongoDTest {
+public:
+    // Disable table logging. When table logging is enabled, timestamps are discarded by WiredTiger.
+    CollectionCatalogTimestampTest()
+        : ServiceContextMongoDTest(Options{}.forceDisableTableLogging()) {}
+
+    void setUp() override {
+        ServiceContextMongoDTest::setUp();
+        opCtx = makeOperationContext();
+    }
+
+    void createCollection(OperationContext* opCtx,
+                          const NamespaceString& nss,
+                          Timestamp timestamp) {
+        opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
+        opCtx->recoveryUnit()->abandonSnapshot();
+
+        if (!opCtx->recoveryUnit()->getCommitTimestamp().isNull()) {
+            opCtx->recoveryUnit()->clearCommitTimestamp();
+        }
+        opCtx->recoveryUnit()->setCommitTimestamp(timestamp);
+
+        AutoGetDb databaseWriteGuard(opCtx, nss.dbName(), MODE_IX);
+        auto db = databaseWriteGuard.ensureDbExists(opCtx);
+        ASSERT(db);
+
+        Lock::CollectionLock lk(opCtx, nss, MODE_IX);
+
+        WriteUnitOfWork wuow(opCtx);
+        CollectionOptions options;
+        options.uuid.emplace(UUID::gen());
+
+        auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
+        std::pair<RecordId, std::unique_ptr<RecordStore>> catalogIdRecordStorePair =
+            uassertStatusOK(storageEngine->getCatalog()->createCollection(
+                opCtx, nss, options, /*allocateDefaultSpace=*/true));
+        auto& catalogId = catalogIdRecordStorePair.first;
+        std::shared_ptr<Collection> ownedCollection = Collection::Factory::get(opCtx)->make(
+            opCtx, nss, catalogId, options, std::move(catalogIdRecordStorePair.second));
+        ownedCollection->init(opCtx);
+        ownedCollection->setCommitted(false);
+
+        CollectionCatalog::get(opCtx)->onCreateCollection(opCtx, std::move(ownedCollection));
+
+        wuow.commit();
+    }
+
+    void dropCollection(OperationContext* opCtx, const NamespaceString& nss, Timestamp timestamp) {
+        opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
+        opCtx->recoveryUnit()->abandonSnapshot();
+
+        if (!opCtx->recoveryUnit()->getCommitTimestamp().isNull()) {
+            opCtx->recoveryUnit()->clearCommitTimestamp();
+        }
+        opCtx->recoveryUnit()->setCommitTimestamp(timestamp);
+
+        Lock::DBLock dbLk(opCtx, nss.db(), MODE_IX);
+        Lock::CollectionLock collLk(opCtx, nss, MODE_X);
+        CollectionWriter collection(opCtx, nss);
+
+        WriteUnitOfWork wuow(opCtx);
+        CollectionCatalog::get(opCtx)->dropCollection(
+            opCtx, collection.getWritableCollection(opCtx), /*isDropPending=*/true);
+        wuow.commit();
+    }
+
+    void createIndex(OperationContext* opCtx,
+                     const NamespaceString& nss,
+                     BSONObj indexSpec,
+                     Timestamp timestamp) {
+        opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
+        opCtx->recoveryUnit()->abandonSnapshot();
+
+        if (!opCtx->recoveryUnit()->getCommitTimestamp().isNull()) {
+            opCtx->recoveryUnit()->clearCommitTimestamp();
+        }
+        opCtx->recoveryUnit()->setCommitTimestamp(timestamp);
+
+        AutoGetCollection autoColl(opCtx, nss, MODE_X);
+
+        WriteUnitOfWork wuow(opCtx);
+        CollectionWriter collection(opCtx, nss);
+
+        IndexBuildsCoordinator::get(opCtx)->createIndexesOnEmptyCollection(
+            opCtx, collection, {indexSpec}, /*fromMigrate=*/false);
+
+        wuow.commit();
+    }
+
+    void dropIndex(OperationContext* opCtx,
+                   const NamespaceString& nss,
+                   const std::string& indexName,
+                   Timestamp timestamp) {
+        opCtx->recoveryUnit()->setTimestampReadSource(RecoveryUnit::ReadSource::kNoTimestamp);
+        opCtx->recoveryUnit()->abandonSnapshot();
+
+        if (!opCtx->recoveryUnit()->getCommitTimestamp().isNull()) {
+            opCtx->recoveryUnit()->clearCommitTimestamp();
+        }
+        opCtx->recoveryUnit()->setCommitTimestamp(timestamp);
+
+        AutoGetCollection autoColl(opCtx, nss, MODE_X);
+
+        WriteUnitOfWork wuow(opCtx);
+        CollectionWriter collection(opCtx, nss);
+
+        Collection* writableCollection = collection.getWritableCollection(opCtx);
+
+        IndexCatalog* indexCatalog = writableCollection->getIndexCatalog();
+        auto indexDescriptor =
+            indexCatalog->findIndexByName(opCtx, indexName, IndexCatalog::InclusionPolicy::kReady);
+        ASSERT_OK(indexCatalog->dropIndex(opCtx, writableCollection, indexDescriptor));
+
+        wuow.commit();
+    }
+
+    DurableCatalog::Entry getEntry(OperationContext* opCtx, const NamespaceString& nss) {
+        AutoGetCollection autoColl(opCtx, nss, LockMode::MODE_IS);
+        auto durableCatalog = DurableCatalog::get(opCtx);
+        return durableCatalog->getEntry(autoColl->getCatalogId());
+    }
+
+    std::shared_ptr<Collection> openCollection(OperationContext* opCtx,
+                                               const DurableCatalog::Entry& entry,
+                                               Timestamp readTimestamp) {
+        Lock::GlobalLock globalLock(opCtx, MODE_IS);
+        auto coll = CollectionCatalog::get(opCtx)->openCollection(opCtx, entry, readTimestamp);
+        ASSERT(coll);
+        return coll;
+    }
+
+protected:
+    ServiceContext::UniqueOperationContext opCtx;
+};
+
+TEST_F(CollectionCatalogTimestampTest, MinimumValidSnapshot) {
+    const NamespaceString nss("a.b");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+    const Timestamp createXIndexTs = Timestamp(20, 20);
+    const Timestamp createYIndexTs = Timestamp(30, 30);
+    const Timestamp dropIndexTs = Timestamp(40, 40);
+
+    createCollection(opCtx.get(), nss, createCollectionTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "x_1"
+                         << "key" << BSON("x" << 1)),
+                createXIndexTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "y_1"
+                         << "key" << BSON("y" << 1)),
+                createYIndexTs);
+
+    auto coll = CollectionCatalog::get(opCtx.get())->lookupCollectionByNamespace(opCtx.get(), nss);
+    ASSERT(coll);
+    ASSERT_EQ(coll->getMinimumVisibleSnapshot(), createCollectionTs);
+    ASSERT_EQ(coll->getMinimumValidSnapshot(), createYIndexTs);
+
+    const IndexDescriptor* desc = coll->getIndexCatalog()->findIndexByName(opCtx.get(), "x_1");
+    const IndexCatalogEntry* entry = coll->getIndexCatalog()->getEntry(desc);
+    ASSERT_EQ(entry->getMinimumVisibleSnapshot(), createXIndexTs);
+
+    desc = coll->getIndexCatalog()->findIndexByName(opCtx.get(), "y_1");
+    entry = coll->getIndexCatalog()->getEntry(desc);
+    ASSERT_EQ(entry->getMinimumVisibleSnapshot(), createYIndexTs);
+
+    dropIndex(opCtx.get(), nss, "x_1", dropIndexTs);
+    dropIndex(opCtx.get(), nss, "y_1", dropIndexTs);
+
+    // Fetch the latest collection instance without the indexes.
+    coll = CollectionCatalog::get(opCtx.get())->lookupCollectionByNamespace(opCtx.get(), nss);
+    ASSERT(coll);
+    ASSERT_EQ(coll->getMinimumVisibleSnapshot(), createCollectionTs);
+    ASSERT_EQ(coll->getMinimumValidSnapshot(), dropIndexTs);
+}
+
+TEST_F(CollectionCatalogTimestampTest, OpenCollectionBeforeCreateTimestamp) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString nss("a.b");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+
+    createCollection(opCtx.get(), nss, createCollectionTs);
+    DurableCatalog::Entry entry = getEntry(opCtx.get(), nss);
+
+    // Try to open the collection before it was created.
+    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(), entry, readTimestamp);
+    ASSERT(!coll);
+}
+
+TEST_F(CollectionCatalogTimestampTest, OpenEarlierCollection) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString nss("a.b");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+    const Timestamp createIndexTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), nss, createCollectionTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "x_1"
+                         << "key" << BSON("x" << 1)),
+                createIndexTs);
+
+    DurableCatalog::Entry entry = getEntry(opCtx.get(), nss);
+
+    // Open an instance of the collection before the index was created.
+    const Timestamp readTimestamp(15, 15);
+    OneOffRead oor(opCtx.get(), readTimestamp);
+    std::shared_ptr<Collection> coll = openCollection(opCtx.get(), entry, readTimestamp);
+    ASSERT(coll);
+    ASSERT_EQ(0, coll->getIndexCatalog()->numIndexesTotal(opCtx.get()));
+
+    // Verify that the CollectionCatalog returns the latest collection with the index present.
+    auto latestColl = CollectionCatalog::get(opCtx.get())
+                          ->lookupCollectionByNamespaceForRead(opCtx.get(), entry.nss);
+    ASSERT(latestColl);
+    ASSERT_EQ(1, latestColl->getIndexCatalog()->numIndexesTotal(opCtx.get()));
+
+    // Ensure the idents are shared between the collection instances.
+    ASSERT_NE(coll, latestColl);
+    ASSERT_EQ(coll->getSharedIdent(), latestColl->getSharedIdent());
+}
+
+TEST_F(CollectionCatalogTimestampTest, OpenEarlierCollectionWithIndex) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString nss("a.b");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+    const Timestamp createXIndexTs = Timestamp(20, 20);
+    const Timestamp createYIndexTs = Timestamp(30, 30);
+
+    createCollection(opCtx.get(), nss, createCollectionTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "x_1"
+                         << "key" << BSON("x" << 1)),
+                createXIndexTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "y_1"
+                         << "key" << BSON("y" << 1)),
+                createYIndexTs);
+
+    DurableCatalog::Entry entry = getEntry(opCtx.get(), nss);
+
+    // Open an instance of the collection when only one of the two indexes were present.
+    const Timestamp readTimestamp(25, 25);
+    OneOffRead oor(opCtx.get(), readTimestamp);
+    std::shared_ptr<Collection> coll = openCollection(opCtx.get(), entry, readTimestamp);
+    ASSERT(coll);
+    ASSERT_EQ(1, coll->getIndexCatalog()->numIndexesTotal(opCtx.get()));
+
+    // Verify that the CollectionCatalog returns the latest collection.
+    auto latestColl = CollectionCatalog::get(opCtx.get())
+                          ->lookupCollectionByNamespaceForRead(opCtx.get(), entry.nss);
+    ASSERT(latestColl);
+    ASSERT_EQ(2, latestColl->getIndexCatalog()->numIndexesTotal(opCtx.get()));
+
+    // Ensure the idents are shared between the collection and index instances.
+    ASSERT_NE(coll, latestColl);
+    ASSERT_EQ(coll->getSharedIdent(), latestColl->getSharedIdent());
+
+    auto indexDescPast = coll->getIndexCatalog()->findIndexByName(opCtx.get(), "x_1");
+    auto indexDescLatest = latestColl->getIndexCatalog()->findIndexByName(opCtx.get(), "x_1");
+    ASSERT_BSONOBJ_EQ(indexDescPast->infoObj(), indexDescLatest->infoObj());
+    ASSERT_EQ(coll->getIndexCatalog()->getEntryShared(indexDescPast),
+              latestColl->getIndexCatalog()->getEntryShared(indexDescLatest));
+}
+
+TEST_F(CollectionCatalogTimestampTest, OpenLatestCollectionWithIndex) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString nss("a.b");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+    const Timestamp createXIndexTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), nss, createCollectionTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "x_1"
+                         << "key" << BSON("x" << 1)),
+                createXIndexTs);
+
+    DurableCatalog::Entry entry = getEntry(opCtx.get(), nss);
+
+    // Setting the read timestamp to the last DDL operation on the collection returns the latest
+    // collection.
+    const Timestamp readTimestamp(20, 20);
+    OneOffRead oor(opCtx.get(), readTimestamp);
+    std::shared_ptr<Collection> coll = openCollection(opCtx.get(), entry, readTimestamp);
+    ASSERT(coll);
+
+    // Verify that the CollectionCatalog returns the latest collection.
+    auto currentColl = CollectionCatalog::get(opCtx.get())
+                           ->lookupCollectionByNamespaceForRead(opCtx.get(), entry.nss);
+    ASSERT_EQ(coll, currentColl);
+
+    // Ensure the idents are shared between the collection and index instances.
+    ASSERT_EQ(coll->getSharedIdent(), currentColl->getSharedIdent());
+
+    auto indexDesc = coll->getIndexCatalog()->findIndexByName(opCtx.get(), "x_1");
+    auto indexDescCurrent = currentColl->getIndexCatalog()->findIndexByName(opCtx.get(), "x_1");
+    ASSERT_BSONOBJ_EQ(indexDesc->infoObj(), indexDescCurrent->infoObj());
+    ASSERT_EQ(coll->getIndexCatalog()->getEntryShared(indexDesc),
+              currentColl->getIndexCatalog()->getEntryShared(indexDescCurrent));
+}
+
+TEST_F(CollectionCatalogTimestampTest, OpenEarlierCollectionWithDropPendingIndex) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString nss("a.b");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+    const Timestamp createIndexTs = Timestamp(20, 20);
+    const Timestamp dropIndexTs = Timestamp(30, 30);
+
+    createCollection(opCtx.get(), nss, createCollectionTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "x_1"
+                         << "key" << BSON("x" << 1)),
+                createIndexTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "y_1"
+                         << "key" << BSON("y" << 1)),
+                createIndexTs);
+
+    DurableCatalog::Entry entry = getEntry(opCtx.get(), nss);
+
+    // Maintain a shared_ptr to "x_1", so it's not expired in drop pending map, but not for "y_1".
+    std::shared_ptr<const IndexCatalogEntry> index;
+    {
+        auto latestColl = CollectionCatalog::get(opCtx.get())
+                              ->lookupCollectionByNamespaceForRead(opCtx.get(), entry.nss);
+        auto desc = latestColl->getIndexCatalog()->findIndexByName(opCtx.get(), "x_1");
+        index = latestColl->getIndexCatalog()->getEntryShared(desc);
+    }
+
+    dropIndex(opCtx.get(), nss, "x_1", dropIndexTs);
+    dropIndex(opCtx.get(), nss, "y_1", dropIndexTs);
+
+    // Open the collection while both indexes were present.
+    const Timestamp readTimestamp(20, 20);
+    OneOffRead oor(opCtx.get(), readTimestamp);
+    std::shared_ptr<Collection> coll = openCollection(opCtx.get(), entry, readTimestamp);
+    ASSERT(coll);
+
+    // Collection is not shared from the latest instance.
+    auto latestColl = CollectionCatalog::get(opCtx.get())
+                          ->lookupCollectionByNamespaceForRead(opCtx.get(), entry.nss);
+    ASSERT_NE(coll, latestColl);
+
+    auto indexDescX = coll->getIndexCatalog()->findIndexByName(opCtx.get(), "x_1");
+    auto indexDescY = coll->getIndexCatalog()->findIndexByName(opCtx.get(), "y_1");
+
+    auto indexEntryX = coll->getIndexCatalog()->getEntryShared(indexDescX);
+    auto indexEntryY = coll->getIndexCatalog()->getEntryShared(indexDescY);
+
+    // Check use_count(). 2 in the unit test, 1 in the opened collection.
+    ASSERT_EQ(3, indexEntryX.use_count());
+
+    // Check use_count(). 1 in the unit test, 1 in the opened collection.
+    ASSERT_EQ(2, indexEntryY.use_count());
+
+    // Verify that "x_1" was retrieved from the drop pending map for the opened collection.
+    ASSERT_EQ(index, indexEntryX);
+}
+
+TEST_F(CollectionCatalogTimestampTest, OpenEarlierAlreadyDropPendingCollection) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString firstNss("a.b");
+    const NamespaceString secondNss("c.d");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+    const Timestamp dropCollectionTs = Timestamp(30, 30);
+
+    createCollection(opCtx.get(), firstNss, createCollectionTs);
+    createCollection(opCtx.get(), secondNss, createCollectionTs);
+
+    DurableCatalog::Entry firstEntry = getEntry(opCtx.get(), firstNss);
+    DurableCatalog::Entry secondEntry = getEntry(opCtx.get(), secondNss);
+
+    // Maintain a shared_ptr to "a.b" so that it isn't expired in the drop pending map after we drop
+    // the collections.
+    std::shared_ptr<const Collection> coll =
+        CollectionCatalog::get(opCtx.get())
+            ->lookupCollectionByNamespaceForRead(opCtx.get(), firstNss);
+    ASSERT(coll);
+
+    // Make the collections drop pending.
+    dropCollection(opCtx.get(), firstNss, dropCollectionTs);
+    dropCollection(opCtx.get(), secondNss, dropCollectionTs);
+
+    // Set the read timestamp to be before the drop timestamp.
+    const Timestamp readTimestamp(20, 20);
+    OneOffRead oor(opCtx.get(), readTimestamp);
+
+    {
+        // Open "a.b", which is not expired in the drop pending map.
+        std::shared_ptr<Collection> openedColl =
+            openCollection(opCtx.get(), firstEntry, readTimestamp);
+        ASSERT(openedColl);
+        ASSERT_EQ(coll, openedColl);
+
+        // Check use_count(). 2 in the unit test.
+        ASSERT_EQ(2, openedColl.use_count());
+    }
+
+    {
+        // Open "c.d" which is expired in the drop pending map.
+        std::shared_ptr<Collection> openedColl =
+            openCollection(opCtx.get(), secondEntry, readTimestamp);
+        ASSERT(openedColl);
+        ASSERT_NE(coll, openedColl);
+
+        // Check use_count(). 1 in the unit test.
+        ASSERT_EQ(1, openedColl.use_count());
+    }
+}
+
+TEST_F(CollectionCatalogTimestampTest, OpenNewCollectionUsingDropPendingCollectionSharedState) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString nss("a.b");
+    const Timestamp createCollectionTs = Timestamp(10, 10);
+    const Timestamp createIndexTs = Timestamp(20, 20);
+    const Timestamp dropCollectionTs = Timestamp(30, 30);
+
+    createCollection(opCtx.get(), nss, createCollectionTs);
+    createIndex(opCtx.get(),
+                nss,
+                BSON("v" << 2 << "name"
+                         << "x_1"
+                         << "key" << BSON("x" << 1)),
+                createIndexTs);
+
+    DurableCatalog::Entry entry = getEntry(opCtx.get(), nss);
+
+    // Maintain a shared_ptr to "a.b" so that it isn't expired in the drop pending map after we drop
+    // it.
+    std::shared_ptr<const Collection> coll =
+        CollectionCatalog::get(opCtx.get())->lookupCollectionByNamespaceForRead(opCtx.get(), nss);
+    ASSERT(coll);
+    ASSERT_EQ(*coll->getMinimumValidSnapshot(), createIndexTs);
+
+    // Make the collection drop pending.
+    dropCollection(opCtx.get(), nss, dropCollectionTs);
+
+    // Open the collection before the index was created. The drop pending collection is incompatible
+    // as it has an index entry. But we can still use the drop pending collections shared state to
+    // instantiate a new collection.
+    const Timestamp readTimestamp(10, 10);
+    OneOffRead oor(opCtx.get(), readTimestamp);
+
+    std::shared_ptr<Collection> openedColl = openCollection(opCtx.get(), entry, readTimestamp);
+    ASSERT(openedColl);
+    ASSERT_NE(coll, openedColl);
+
+    // Check use_count(). 1 in the unit test.
+    ASSERT_EQ(1, openedColl.use_count());
+
+    // Ensure the idents are shared between the opened collection and the drop pending collection.
+    ASSERT_EQ(coll->getSharedIdent(), openedColl->getSharedIdent());
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 1f37bb989a0..d61a693869f 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -366,10 +366,107 @@ void CollectionImpl::init(OperationContext* opCtx) {
     const auto& collectionOptions = _metadata->options;
 
     _shared->_collator = parseCollation(opCtx, _ns, collectionOptions.collation);
+
+    _initCommon(opCtx);
+
+    if (collectionOptions.clusteredIndex) {
+        if (collectionOptions.expireAfterSeconds) {
+            // If this collection has been newly created, we need to register with the TTL cache at
+            // commit time, otherwise it is startup and we can register immediately.
+            auto svcCtx = opCtx->getClient()->getServiceContext();
+            auto uuid = *collectionOptions.uuid;
+            if (opCtx->lockState()->inAWriteUnitOfWork()) {
+                opCtx->recoveryUnit()->onCommit([svcCtx, uuid](auto ts) {
+                    TTLCollectionCache::get(svcCtx).registerTTLInfo(
+                        uuid, TTLCollectionCache::Info{TTLCollectionCache::ClusteredId{}});
+                });
+            } else {
+                TTLCollectionCache::get(svcCtx).registerTTLInfo(
+                    uuid, TTLCollectionCache::Info{TTLCollectionCache::ClusteredId{}});
+            }
+        }
+    }
+
+    uassertStatusOK(getIndexCatalog()->init(opCtx, this));
+    _initialized = true;
+}
+
+void CollectionImpl::initFromExisting(OperationContext* opCtx,
+                                      std::shared_ptr<Collection> collection,
+                                      Timestamp readTimestamp) {
+    LOGV2_DEBUG(
+        6825402, 1, "Initializing collection using shared state", logAttrs(collection->ns()));
+
+    // We are per definition committed if we initialize from an existing collection.
+    _cachedCommitted = true;
+
+    // Use the shared state from the existing collection.
+    _shared = static_cast<CollectionImpl*>(collection.get())->_shared;
+
+    // When initializing a collection from an earlier point-in-time, we don't know when the last DDL
+    // operation took place at that point-in-time. We conservatively set the minimum valid snapshot
+    // to the read point-in-time.
+    _minVisibleSnapshot = readTimestamp;
+    _minValidSnapshot = readTimestamp;
+
+    _initCommon(opCtx);
+
+    // Determine which indexes from the existing collection can be shared with this newly
+    // initialized collection. The remaining indexes will be initialized by the IndexCatalog.
+    IndexCatalogEntryContainer preexistingIndexes;
+    for (const auto& index : _metadata->indexes) {
+        // First check the index catalog of the existing collection for the index entry.
+        std::shared_ptr<IndexCatalogEntry> entry = [&]() -> std::shared_ptr<IndexCatalogEntry> {
+            auto desc =
+                collection->getIndexCatalog()->findIndexByName(opCtx, index.nameStringData());
+            if (!desc)
+                return nullptr;
+
+            auto entry = collection->getIndexCatalog()->getEntryShared(desc);
+            if (!entry->getMinimumVisibleSnapshot())
+                // Index is valid in all snapshots.
+                return entry;
+            return readTimestamp < *entry->getMinimumVisibleSnapshot() ? nullptr : entry;
+        }();
+
+        if (entry) {
+            preexistingIndexes.add(std::move(entry));
+            continue;
+        }
+
+        // Next check the CollectionCatalog for a compatible drop pending index.
+        entry = [&]() -> std::shared_ptr<IndexCatalogEntry> {
+            const std::string ident = DurableCatalog::get(opCtx)->getIndexIdent(
+                opCtx, _catalogId, index.nameStringData());
+            auto entry = CollectionCatalog::get(opCtx)->findDropPendingIndex(ident);
+            if (!entry)
+                return nullptr;
+
+            if (!entry->getMinimumVisibleSnapshot())
+                // Index is valid in all snapshots.
+                return entry;
+            return readTimestamp < *entry->getMinimumVisibleSnapshot() ? nullptr : entry;
+        }();
+
+        if (entry) {
+            preexistingIndexes.add(std::move(entry));
+            continue;
+        }
+    }
+
+    uassertStatusOK(
+        getIndexCatalog()->initFromExisting(opCtx, this, preexistingIndexes, readTimestamp));
+    _initialized = true;
+}
+
+void CollectionImpl::_initCommon(OperationContext* opCtx) {
+    invariant(!_initialized);
+
+    const auto& collectionOptions = _metadata->options;
     auto validatorDoc = collectionOptions.validator.getOwned();
 
     // Enforce that the validator can be used on this namespace.
-    uassertStatusOK(checkValidatorCanBeUsedOnNs(validatorDoc, ns(), _uuid));
+    uassertStatusOK(checkValidatorCanBeUsedOnNs(validatorDoc, _ns, _uuid));
 
     // Make sure validationAction and validationLevel are allowed on this collection
     uassertStatusOK(checkValidationOptionsCanBeUsed(
@@ -399,27 +496,6 @@ void CollectionImpl::init(OperationContext* opCtx) {
                               logAttrs(_ns),
                               "validatorStatus"_attr = _validator.getStatus());
     }
-
-    if (collectionOptions.clusteredIndex) {
-        if (collectionOptions.expireAfterSeconds) {
-            // If this collection has been newly created, we need to register with the TTL cache at
-            // commit time, otherwise it is startup and we can register immediately.
-            auto svcCtx = opCtx->getClient()->getServiceContext();
-            auto uuid = *collectionOptions.uuid;
-            if (opCtx->lockState()->inAWriteUnitOfWork()) {
-                opCtx->recoveryUnit()->onCommit([svcCtx, uuid](auto ts) {
-                    TTLCollectionCache::get(svcCtx).registerTTLInfo(
-                        uuid, TTLCollectionCache::Info{TTLCollectionCache::ClusteredId{}});
-                });
-            } else {
-                TTLCollectionCache::get(svcCtx).registerTTLInfo(
-                    uuid, TTLCollectionCache::Info{TTLCollectionCache::ClusteredId{}});
-            }
-        }
-    }
-
-    getIndexCatalog()->init(opCtx, this).transitional_ignore();
-    _initialized = true;
 }
 
 bool CollectionImpl::isInitialized() const {
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index cf7cfaaf805..4300a207635 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -73,6 +73,9 @@ public:
     SharedCollectionDecorations* getSharedDecorations() const final;
 
     void init(OperationContext* opCtx) final;
+    void initFromExisting(OperationContext* opCtx,
+                          std::shared_ptr<Collection> collection,
+                          Timestamp readTimestamp) final;
     bool isInitialized() const final;
     bool isCommitted() const final;
     void setCommitted(bool val) final;
@@ -445,6 +448,11 @@ private:
     void _writeMetadata(OperationContext* opCtx, Func func);
 
     /**
+     * Helper for init() and initFromExisting() to initialize common state.
+     */
+    void _initCommon(OperationContext* opCtx);
+
+    /**
      * Holder of shared state between CollectionImpl clones
      */
     struct SharedState {
diff --git a/src/mongo/db/catalog/index_catalog.h b/src/mongo/db/catalog/index_catalog.h
index 625a7bc56cc..e02733eaace 100644
--- a/src/mongo/db/catalog/index_catalog.h
+++ b/src/mongo/db/catalog/index_catalog.h
@@ -217,6 +217,17 @@ public:
     // Must be called before used.
     virtual Status init(OperationContext* opCtx, Collection* collection) = 0;
 
+    /**
+     * Must be called before used.
+     *
+     * When initializing an index that exists in 'preexistingIndexes', the IndexCatalogEntry will be
+     * taken from there instead of initializing a new IndexCatalogEntry.
+     */
+    virtual Status initFromExisting(OperationContext* opCtx,
+                                    Collection* collection,
+                                    const IndexCatalogEntryContainer& preexistingIndexes,
+                                    boost::optional<Timestamp> readTimestamp) = 0;
+
     // ---- accessors -----
 
     virtual bool haveAnyIndexes() const = 0;
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 3b0bc930e25..59f7924b27e 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -46,6 +46,7 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/field_ref.h"
 #include "mongo/db/fts/fts_spec.h"
+#include "mongo/db/global_settings.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index/s2_access_method.h"
@@ -184,6 +185,20 @@ std::unique_ptr<IndexCatalog> IndexCatalogImpl::clone() const {
 }
 
 Status IndexCatalogImpl::init(OperationContext* opCtx, Collection* collection) {
+    return _init(opCtx, collection, /*preexistingIndexes=*/{}, /*readTimestamp=*/boost::none);
+};
+
+Status IndexCatalogImpl::initFromExisting(OperationContext* opCtx,
+                                          Collection* collection,
+                                          const IndexCatalogEntryContainer& preexistingIndexes,
+                                          boost::optional<Timestamp> readTimestamp) {
+    return _init(opCtx, collection, preexistingIndexes, readTimestamp);
+};
+
+Status IndexCatalogImpl::_init(OperationContext* opCtx,
+                               Collection* collection,
+                               const IndexCatalogEntryContainer& preexistingIndexes,
+                               boost::optional<Timestamp> readTimestamp) {
     vector<string> indexNames;
     collection->getAllIndexes(&indexNames);
     const bool replSetMemberInStandaloneMode =
@@ -200,6 +215,32 @@ Status IndexCatalogImpl::init(OperationContext* opCtx, Collection* collection) {
         BSONObj spec = collection->getIndexSpec(indexName).getOwned();
         BSONObj keyPattern = spec.getObjectField("key");
 
+        if (readTimestamp) {
+            auto preexistingIt = std::find_if(
+                preexistingIndexes.begin(),
+                preexistingIndexes.end(),
+                [&](const std::shared_ptr<IndexCatalogEntry>& preexistingEntry) {
+                    return spec.woCompare(preexistingEntry->descriptor()->infoObj()) == 0;
+                });
+            if (preexistingIt != preexistingIndexes.end()) {
+                std::shared_ptr<IndexCatalogEntry> existingEntry = *preexistingIt;
+                LOGV2_DEBUG(6825404,
+                            1,
+                            "Adding preexisting index entry",
+                            logAttrs(collection->ns()),
+                            "index"_attr = existingEntry->descriptor()->infoObj());
+
+                if (existingEntry->isReady(opCtx)) {
+                    _readyIndexes.add(std::move(existingEntry));
+                } else if (existingEntry->isFrozen()) {
+                    _frozenIndexes.add(std::move(existingEntry));
+                } else {
+                    _buildingIndexes.add(std::move(existingEntry));
+                }
+                continue;
+            }
+        }
+
         auto descriptor = std::make_unique<IndexDescriptor>(_getAccessMethodName(keyPattern), spec);
 
         // TTL indexes with NaN 'expireAfterSeconds' cause problems in multiversion settings.
@@ -257,16 +298,27 @@ Status IndexCatalogImpl::init(OperationContext* opCtx, Collection* collection) {
                 createIndexEntry(opCtx, collection, std::move(descriptor), flags);
             fassert(17340, entry->isReady(opCtx));
 
-            // When initializing indexes from disk, we conservatively set the minimumVisibleSnapshot
-            // to non _id indexes to the recovery timestamp. The _id index is left visible. It's
-            // assumed if the collection is visible, it's _id is valid to be used.
-            if (recoveryTs && !entry->descriptor()->isIdIndex()) {
+            if (readTimestamp) {
+                // When initializing indexes from an earlier point-in-time, we conservatively set
+                // the minimum valid snapshot to the read point-in-time as we don't know when the
+                // last DDL operation took place at that point-in-time.
+                entry->setMinimumVisibleSnapshot(*readTimestamp);
+            } else if (recoveryTs && !entry->descriptor()->isIdIndex()) {
+                // When initializing indexes from disk, we conservatively set the
+                // minimumVisibleSnapshot to non _id indexes to the recovery timestamp. The _id
+                // index is left visible. It's assumed if the collection is visible, it's _id is
+                // valid to be used.
                 entry->setMinimumVisibleSnapshot(recoveryTs.value());
             }
         }
     }
 
-    CollectionQueryInfo::get(collection).init(opCtx, collection);
+    if (!readTimestamp) {
+        // Only do this when we're not initializing an earlier collection from the shared state of
+        // an existing collection.
+        CollectionQueryInfo::get(collection).init(opCtx, collection);
+    }
+
     return Status::OK();
 }
 
@@ -349,15 +401,21 @@ Status IndexCatalogImpl::_isNonIDIndexAndNotAllowedToBuild(OperationContext* opC
                                                            const BSONObj& spec) const {
     const BSONObj key = spec.getObjectField("key");
     invariant(!key.isEmpty());
-    if (!IndexDescriptor::isIdIndexPattern(key)) {
-        // Check whether the replica set member's config has {buildIndexes:false} set, which means
-        // we are not allowed to build non-_id indexes on this server.
-        if (!repl::ReplicationCoordinator::get(opCtx)->buildsIndexes()) {
-            // We return an IndexAlreadyExists error so that the caller can catch it and silently
-            // skip building it.
-            return Status(ErrorCodes::IndexAlreadyExists,
-                          "this replica set member's 'buildIndexes' setting is set to false");
-        }
+    if (IndexDescriptor::isIdIndexPattern(key)) {
+        return Status::OK();
+    }
+
+    if (!getGlobalReplSettings().usingReplSets()) {
+        return Status::OK();
+    }
+
+    // Check whether the replica set member's config has {buildIndexes:false} set, which means
+    // we are not allowed to build non-_id indexes on this server.
+    if (!repl::ReplicationCoordinator::get(opCtx)->buildsIndexes()) {
+        // We return an IndexAlreadyExists error so that the caller can catch it and silently
+        // skip building it.
+        return Status(ErrorCodes::IndexAlreadyExists,
+                      "this replica set member's 'buildIndexes' setting is set to false");
     }
 
     return Status::OK();
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index 655c43e6858..4576dddaa3c 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -67,6 +67,17 @@ public:
     // must be called before used
     Status init(OperationContext* opCtx, Collection* collection) override;
 
+    /**
+     * Must be called before used.
+     *
+     * When initializing an index that exists in 'preexistingIndexes', the IndexCatalogEntry will be
+     * taken from there instead of initializing a new IndexCatalogEntry.
+     */
+    Status initFromExisting(OperationContext* opCtx,
+                            Collection* collection,
+                            const IndexCatalogEntryContainer& preexistingIndexes,
+                            boost::optional<Timestamp> readTimestamp) override;
+
     // ---- accessors -----
 
     bool haveAnyIndexes() const override;
@@ -300,6 +311,17 @@ private:
     static const BSONObj _idObj;  // { _id : 1 }
 
     /**
+     * Helper for init() and initFromExisting().
+     *
+     * Passing boost::none for 'preexistingIndexes' indicates that the IndexCatalog is not being
+     * initialized at an earlier point-in-time.
+     */
+    Status _init(OperationContext* opCtx,
+                 Collection* collection,
+                 const IndexCatalogEntryContainer& preexistingIndexes,
+                 boost::optional<Timestamp> readTimestamp);
+
+    /**
      * In addition to IndexNames::findPluginName, validates that it is a known index type.
      * If all you need is to check for a certain type, just use IndexNames::findPluginName.
      *
diff --git a/src/mongo/db/service_context_d_test_fixture.cpp b/src/mongo/db/service_context_d_test_fixture.cpp
index caef9950914..cad9b4dfa62 100644
--- a/src/mongo/db/service_context_d_test_fixture.cpp
+++ b/src/mongo/db/service_context_d_test_fixture.cpp
@@ -65,6 +65,10 @@ ServiceContextMongoDTest::ServiceContextMongoDTest(Options options)
     : _journalListener(std::move(options._journalListener)),
       _tempDir("service_context_d_test_fixture") {
 
+    if (options._forceDisableTableLogging) {
+        storageGlobalParams.forceDisableTableLogging = true;
+    }
+
     if (options._useReplSettings) {
         repl::ReplSettings replSettings;
         replSettings.setOplogSizeBytes(10 * 1024 * 1024);
@@ -174,6 +178,8 @@ ServiceContextMongoDTest::~ServiceContextMongoDTest() {
     std::swap(storageGlobalParams.repair, _stashedStorageParams.repair);
     std::swap(serverGlobalParams.enableMajorityReadConcern,
               _stashedServerParams.enableMajorityReadConcern);
+
+    storageGlobalParams.reset();
 }
 
 void ServiceContextMongoDTest::tearDown() {
diff --git a/src/mongo/db/service_context_d_test_fixture.h b/src/mongo/db/service_context_d_test_fixture.h
index 3973a344ffd..763038eb95e 100644
--- a/src/mongo/db/service_context_d_test_fixture.h
+++ b/src/mongo/db/service_context_d_test_fixture.h
@@ -87,6 +87,11 @@ protected:
             return std::move(*this);
         }
 
+        Options forceDisableTableLogging() {
+            _forceDisableTableLogging = true;
+            return std::move(*this);
+        }
+
     private:
         std::string _engine = "wiredTiger";
         // We use ephemeral instances by default to advise Storage Engines (in particular
@@ -99,6 +104,7 @@ protected:
         Milliseconds _autoAdvancingMockClockIncrement{0};
         std::unique_ptr<TickSource> _mockTickSource;
         std::unique_ptr<JournalListener> _journalListener;
+        bool _forceDisableTableLogging = false;
 
         friend class ServiceContextMongoDTest;
     };
diff --git a/src/mongo/db/storage/storage_options.cpp b/src/mongo/db/storage/storage_options.cpp
index 5df2299e0ab..f31217b89ae 100644
--- a/src/mongo/db/storage/storage_options.cpp
+++ b/src/mongo/db/storage/storage_options.cpp
@@ -58,6 +58,7 @@ void StorageGlobalParams::reset() {
     allowOplogTruncation = true;
     disableLockFreeReads = false;
     checkpointDelaySecs = 0;
+    forceDisableTableLogging = false;
 }
 
 StorageGlobalParams storageGlobalParams;
diff --git a/src/mongo/db/storage/storage_options.h b/src/mongo/db/storage/storage_options.h
index c10b71a25aa..386e13efde5 100644
--- a/src/mongo/db/storage/storage_options.h
+++ b/src/mongo/db/storage/storage_options.h
@@ -132,6 +132,9 @@ struct StorageGlobalParams {
     // Delay in seconds between triggering the next checkpoint after the completion of the previous
     // one. A value of 0 indicates that checkpointing will be skipped.
     size_t checkpointDelaySecs;
+
+    // Test-only option. Disables table logging.
+    bool forceDisableTableLogging = false;
 };
 
 extern StorageGlobalParams storageGlobalParams;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
index f6087cb0d98..a5a6986bf7c 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
@@ -795,6 +795,12 @@ void WiredTigerUtil::notifyStartupComplete() {
 }
 
 bool WiredTigerUtil::useTableLogging(const NamespaceString& nss) {
+    if (storageGlobalParams.forceDisableTableLogging) {
+        invariant(TestingProctor::instance().isEnabled());
+        LOGV2(6825405, "Table logging disabled", logAttrs(nss));
+        return false;
+    }
+
     // We only turn off logging in the case of:
     // 1) Replication is enabled (the typical deployment), or
     // 2) We're running as a standalone with recoverFromOplogAsStandalone=true