Revert "SERVER-65974 Replace HistoricalIdentTracker with checkpoint cursors"r7.0.10-rc0r7.0.10

Revert "SERVER-76534 Fix reporting the namespace and UUID in a backup cursor when the changes have not yet been checkpointed"

GitOrigin-RevId: 40ecdae4c3f6a19abc2bdaf48b9040a83261e54d

21 files changed, 974 insertions, 139 deletions

diff --git a/jstests/hot_backups/backup_restore_metadata.js b/jstests/hot_backups/backup_restore_metadata.js
deleted file mode 100644
index bfd50f4b791..00000000000
--- a/jstests/hot_backups/backup_restore_metadata.js
+++ /dev/null
@@ -1,65 +0,0 @@
-/**
- * Tests that the ident metadata is pulled from the catalog using a checkpoint cursor, if it exists.
- * Otherwise we fallback to retrieving the ident metadata from the latest catalog as a best guess.
- *
- * @tags: [
- *   requires_persistence,
- *   requires_replication,
- * ]
- */
-(function() {
-"use strict";
-
-const rst = ReplSetTest({
-    nodes: 1,
-    nodeOptions: {
-        setParameter:
-            // Control checkpoints.
-            {"failpoint.pauseCheckpointThread": tojson({mode: "alwaysOn"})}
-    }
-});
-rst.startSet();
-rst.initiate();
-
-let primary = rst.getPrimary();
-let primaryDB = primary.getDB("test");
-
-// Take the initial checkpoint.
-assert.commandWorked(primaryDB.adminCommand({fsync: 1}));
-
-// Create "a" and ensure it is in the checkpoint.
-assert.commandWorked(primaryDB.createCollection("a"));
-assert.commandWorked(primaryDB.adminCommand({fsync: 1}));
-
-// Create "b", which will not be part of the checkpoint used by $backupCursor.
-assert.commandWorked(primaryDB.createCollection("b"));
-
-// Rename "a" to "c". The backup cursor will still report the namespace as "a".
-assert.commandWorked(primaryDB.adminCommand({renameCollection: "test.a", to: "test.c"}));
-
-let namespacesFound = {};
-let backupCursor = primary.getDB("admin").aggregate([{$backupCursor: {}}]);
-while (backupCursor.hasNext()) {
-    let doc = backupCursor.next();
-    jsTestLog(doc);
-
-    if (namespacesFound.hasOwnProperty(doc.ns)) {
-        namespacesFound[doc.ns] += 1;
-    } else {
-        namespacesFound[doc.ns] = 1;
-    }
-}
-
-// Two entries for these namespaces. One for the collection, and one for the _id index.
-assert.eq(2, namespacesFound["test.a"]);
-assert.eq(2, namespacesFound["test.b"]);
-
-// The rename was not part of the checkpoint.
-assert.eq(undefined, namespacesFound["test.c"]);
-
-backupCursor.close();
-
-assert.commandWorked(
-    primary.adminCommand({configureFailPoint: "pauseCheckpointThread", mode: "off"}));
-rst.stopSet();
-})();
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 9ecd00bcd9e..ba0c529469e 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -405,6 +405,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/capped_snapshots',
         '$BUILD_DIR/mongo/db/storage/durable_catalog_impl',
         '$BUILD_DIR/mongo/db/storage/execution_context',
+        '$BUILD_DIR/mongo/db/storage/historical_ident_tracker',
         '$BUILD_DIR/mongo/db/storage/key_string',
         '$BUILD_DIR/mongo/db/storage/record_store_base',
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 732c0c66880..5e7f720783c 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -67,6 +67,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/top.h"
 #include "mongo/db/storage/durable_catalog.h"
+#include "mongo/db/storage/historical_ident_tracker.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
@@ -564,6 +565,16 @@ Status DatabaseImpl::_finishDropCollection(OperationContext* opCtx,
             opCtx, collection->ns(), collection->getCatalogId(), sharedIdent);
         if (!status.isOK())
             return status;
+
+        opCtx->recoveryUnit()->onCommit(
+            [nss, uuid, ident = sharedIdent->getIdent()](OperationContext* opCtx,
+                                                         boost::optional<Timestamp> commitTime) {
+                if (!commitTime) {
+                    return;
+                }
+
+                HistoricalIdentTracker::get(opCtx).recordDrop(ident, nss, uuid, commitTime.value());
+            });
     }
 
     CollectionCatalog::get(opCtx)->dropCollection(
@@ -615,6 +626,30 @@ Status DatabaseImpl::renameCollection(OperationContext* opCtx,
         return status;
 
     CollectionCatalog::get(opCtx)->onCollectionRename(opCtx, writableCollection, fromNss);
+
+    opCtx->recoveryUnit()->onCommit([fromNss,
+                                     writableCollection](OperationContext* opCtx,
+                                                         boost::optional<Timestamp> commitTime) {
+        if (!commitTime) {
+            return;
+        }
+
+        HistoricalIdentTracker::get(opCtx).recordRename(
+            writableCollection->getSharedIdent()->getIdent(),
+            fromNss,
+            writableCollection->uuid(),
+            commitTime.value());
+
+        const auto readyIndexes = writableCollection->getIndexCatalog()->getAllReadyEntriesShared();
+        for (const auto& readyIndex : readyIndexes) {
+            HistoricalIdentTracker::get(opCtx).recordRename(
+                readyIndex->getIdent(), fromNss, writableCollection->uuid(), commitTime.value());
+        }
+
+        // Ban reading from this collection on committed reads on snapshots before now.
+        writableCollection->setMinimumVisibleSnapshot(commitTime.value());
+    });
+
     return status;
 }
 
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index cafada4b0ff..0064cb2c6a4 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -70,6 +70,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/execution_context.h"
+#include "mongo/db/storage/historical_ident_tracker.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
@@ -1385,7 +1386,12 @@ public:
           _entry(std::move(entry)),
           _collectionDecorations(collectionDecorations) {}
 
-    void commit(OperationContext* opCtx, boost::optional<Timestamp>) final {
+    void commit(OperationContext* opCtx, boost::optional<Timestamp> commitTime) final {
+        if (commitTime) {
+            HistoricalIdentTracker::get(opCtx).recordDrop(
+                _entry->getIdent(), _nss, _uuid, commitTime.value());
+        }
+
         _entry->setDropped();
     }
 
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 1bd46f3484b..dcb7643d0a1 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -564,6 +564,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/serverless/serverless_lock',
         '$BUILD_DIR/mongo/db/session/kill_sessions_local',
         '$BUILD_DIR/mongo/db/session/session_catalog_mongod',
+        '$BUILD_DIR/mongo/db/storage/historical_ident_tracker',
         '$BUILD_DIR/mongo/util/namespace_string_database_name_util',
         'drop_pending_collection_reaper',
     ],
diff --git a/src/mongo/db/repl/rollback_impl.cpp b/src/mongo/db/repl/rollback_impl.cpp
index 7192aefa5c8..f4aadb526df 100644
--- a/src/mongo/db/repl/rollback_impl.cpp
+++ b/src/mongo/db/repl/rollback_impl.cpp
@@ -64,6 +64,7 @@
 #include "mongo/db/session/kill_sessions_local.h"
 #include "mongo/db/session/session_catalog_mongod.h"
 #include "mongo/db/session/session_txn_record_gen.h"
+#include "mongo/db/storage/historical_ident_tracker.h"
 #include "mongo/db/storage/remove_saver.h"
 #include "mongo/db/transaction/transaction_history_iterator.h"
 #include "mongo/logv2/log.h"
@@ -602,6 +603,9 @@ void RollbackImpl::_runPhaseFromAbortToReconstructPreparedTxns(
     _rollbackStats.stableTimestamp = stableTimestamp;
     _listener->onRecoverToStableTimestamp(stableTimestamp);
 
+    // Rollback historical ident entries.
+    HistoricalIdentTracker::get(opCtx).rollbackTo(stableTimestamp);
+
     // Log the total number of insert and update operations that have been rolled back as a
     // result of recovering to the stable timestamp.
     auto getCommandCount = [&](StringData key) {
diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index 359744acd4d..ff9cb2cc946 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -432,6 +432,16 @@ env.Library(
 )
 
 env.Library(
+    target='historical_ident_tracker',
+    source=[
+        'historical_ident_tracker.cpp',
+    ],
+    LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/server_base',
+    ],
+)
+
+env.Library(
     target="write_unit_of_work",
     source=[
         "write_unit_of_work.cpp",
@@ -549,6 +559,7 @@ env.CppUnitTest(
         'external_record_store_test.cpp',
         'disk_space_monitor_test.cpp',
         'flow_control_test.cpp',
+        'historical_ident_tracker_test.cpp',
         'index_entry_comparison_test.cpp',
         'key_string_test.cpp',
         'kv/durable_catalog_test.cpp',
@@ -584,6 +595,7 @@ env.CppUnitTest(
         'disk_space_monitor',
         'flow_control',
         'flow_control_parameters',
+        'historical_ident_tracker',
         'key_string',
         'kv/kv_drop_pending_ident_reaper',
         'record_store_base',
@@ -671,6 +683,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/server_base',
+        'historical_ident_tracker',
         'storage_options',
     ],
 )
@@ -700,6 +713,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/storage_repair_observer',
         '$BUILD_DIR/mongo/db/vector_clock',
         'backup_block',
+        'historical_ident_tracker',
         'storage_control',
         'storage_util',
         'two_phase_index_build_knobs_idl',
diff --git a/src/mongo/db/storage/backup_block.cpp b/src/mongo/db/storage/backup_block.cpp
index 79450c173e8..357290caeb3 100644
--- a/src/mongo/db/storage/backup_block.cpp
+++ b/src/mongo/db/storage/backup_block.cpp
@@ -35,6 +35,7 @@
 #include "mongo/base/string_data.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/storage/durable_catalog.h"
+#include "mongo/db/storage/historical_ident_tracker.h"
 #include "mongo/db/storage/storage_options.h"
 
 namespace mongo {
@@ -49,19 +50,17 @@ const std::set<std::string> kRequiredMDBFiles = {"_mdb_catalog.wt", "sizeStorer.
 }  // namespace
 
 BackupBlock::BackupBlock(OperationContext* opCtx,
-                         boost::optional<NamespaceString> nss,
-                         boost::optional<UUID> uuid,
                          std::string filePath,
+                         const IdentToNamespaceAndUUIDMap& identToNamespaceAndUUIDMap,
                          boost::optional<Timestamp> checkpointTimestamp,
                          std::uint64_t offset,
                          std::uint64_t length,
                          std::uint64_t fileSize)
-    : _filePath(filePath),
-      _offset(offset),
-      _length(length),
-      _fileSize(fileSize),
-      _nss(nss),
-      _uuid(uuid) {}
+    : _filePath(filePath), _offset(offset), _length(length), _fileSize(fileSize) {
+    boost::filesystem::path path(filePath);
+    _filenameStem = path.stem().string();
+    _initialize(opCtx, identToNamespaceAndUUIDMap, checkpointTimestamp);
+}
 
 bool BackupBlock::isRequired() const {
     // Extract the filename from the path.
@@ -90,21 +89,56 @@ bool BackupBlock::isRequired() const {
         return true;
     }
 
-    if (!_nss) {
-        return false;
-    }
-
     // Check if collection resides in an internal database (admin, local, or config).
-    if (_nss->isOnInternalDb()) {
+    if (_nss.isOnInternalDb()) {
         return true;
     }
 
     // Check if collection is 'system.views'.
-    if (_nss->isSystemDotViews()) {
+    if (_nss.isSystemDotViews()) {
         return true;
     }
 
     return false;
 }
 
+void BackupBlock::_setNamespaceString(const NamespaceString& nss) {
+    // Remove "system.buckets." from time-series collection namespaces since it is an internal
+    // detail that is not intended to be visible externally.
+    if (nss.isTimeseriesBucketsCollection()) {
+        _nss = nss.getTimeseriesViewNamespace();
+        return;
+    }
+
+    _nss = nss;
+}
+
+void BackupBlock::_initialize(OperationContext* opCtx,
+                              const IdentToNamespaceAndUUIDMap& identToNamespaceAndUUIDMap,
+                              boost::optional<Timestamp> checkpointTimestamp) {
+    if (!opCtx) {
+        return;
+    }
+
+    // Fetch the latest values for the ident.
+    auto it = identToNamespaceAndUUIDMap.find(_filenameStem);
+    if (it != identToNamespaceAndUUIDMap.end()) {
+        _uuid = it->second.second;
+        _setNamespaceString(it->second.first);
+    }
+
+    if (!checkpointTimestamp) {
+        return;
+    }
+
+    // Check if the ident had a different value at the checkpoint timestamp. If so, we want to use
+    // that instead as that will be the ident's value when restoring from the backup.
+    boost::optional<std::pair<NamespaceString, UUID>> historicalEntry =
+        HistoricalIdentTracker::get(opCtx).lookup(_filenameStem, checkpointTimestamp.value());
+    if (historicalEntry) {
+        _uuid = historicalEntry->second;
+        _setNamespaceString(historicalEntry->first);
+    }
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/storage/backup_block.h b/src/mongo/db/storage/backup_block.h
index 09c7519ac60..fd96b8e5e2e 100644
--- a/src/mongo/db/storage/backup_block.h
+++ b/src/mongo/db/storage/backup_block.h
@@ -53,10 +53,12 @@ namespace mongo {
  */
 class BackupBlock final {
 public:
+    using IdentToNamespaceAndUUIDMap =
+        stdx::unordered_map<std::string, std::pair<NamespaceString, UUID>>;
+
     explicit BackupBlock(OperationContext* opCtx,
-                         boost::optional<NamespaceString> nss,
-                         boost::optional<UUID> uuid,
                          std::string filePath,
+                         const IdentToNamespaceAndUUIDMap& identToNamespaceAndUUIDMap,
                          boost::optional<Timestamp> checkpointTimestamp,
                          std::uint64_t offset = 0,
                          std::uint64_t length = 0,
@@ -68,8 +70,8 @@ public:
         return _filePath;
     }
 
-    boost::optional<NamespaceString> ns() const {
-        return _nss;
+    std::string ns() const {
+        return _nss.toString();
     }
 
     std::uint64_t offset() const {
@@ -94,12 +96,27 @@ public:
     bool isRequired() const;
 
 private:
+    /**
+     * Sets '_nss' and '_uuid' that is representative of the ident at the checkpoint timestamp for:
+     * - collections
+     * - indexes, to the NSS/UUID of their respective collection
+     *
+     * The 'checkpointTimestamp' will be boost::none if the backup is being taken on a standalone
+     * node.
+     * A null opCtx is ignored. A null opCtx is exercised by FCBIS unit tests.
+     */
+    void _initialize(OperationContext* opCtx,
+                     const IdentToNamespaceAndUUIDMap& identToNamespaceAndUUIDMap,
+                     boost::optional<Timestamp> checkpointTimestamp);
+    void _setNamespaceString(const NamespaceString& nss);
+
     const std::string _filePath;
     const std::uint64_t _offset;
     const std::uint64_t _length;
     const std::uint64_t _fileSize;
 
-    boost::optional<NamespaceString> _nss;
+    std::string _filenameStem;
+    NamespaceString _nss;
     boost::optional<UUID> _uuid;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/storage/devnull/devnull_kv_engine.cpp b/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
index b9a68faaecf..e0877397d84 100644
--- a/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
+++ b/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
@@ -290,9 +290,8 @@ public:
 
 DevNullKVEngine::DevNullKVEngine() {
     _mockBackupBlocks.push_back(BackupBlock(/*opCtx=*/nullptr,
-                                            /*nss=*/boost::none,
-                                            /*uuid=*/boost::none,
                                             "filename.wt",
+                                            /*identToNamespaceAndUUIDMap=*/{},
                                             /*checkpointTimestamp=*/boost::none));
 }
 
@@ -340,9 +339,6 @@ public:
         return BSONObj();
     }
 
-    void setCatalogEntries(const stdx::unordered_map<std::string, std::pair<NamespaceString, UUID>>&
-                               identsToNsAndUUID) {}
-
     StatusWith<std::deque<BackupBlock>> getNextBatch(OperationContext* opCtx,
                                                      const std::size_t batchSize) {
         if (_exhaustCursor) {
diff --git a/src/mongo/db/storage/durable_catalog.h b/src/mongo/db/storage/durable_catalog.h
index 7f94a7fb4dd..f05c2f3329c 100644
--- a/src/mongo/db/storage/durable_catalog.h
+++ b/src/mongo/db/storage/durable_catalog.h
@@ -136,13 +136,6 @@ public:
         OperationContext* opCtx, const RecordId& catalogId) const = 0;
 
     /**
-     * Parses the passed in catalog entry object.
-     */
-    virtual DurableCatalogEntry getParsedCatalogEntry(OperationContext* opCtx,
-                                                      const RecordId& catalogId,
-                                                      const BSONObj& obj) const = 0;
-
-    /**
      * Like 'getParsedCatalogEntry' above but only extracts the metadata component.
      */
     virtual std::shared_ptr<BSONCollectionCatalogEntry::MetaData> getMetaData(
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index 1c5c24548cc..d89760f68d3 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -441,16 +441,6 @@ boost::optional<DurableCatalogEntry> DurableCatalogImpl::getParsedCatalogEntry(
                                _parseMetaData(obj["md"])};
 }
 
-DurableCatalogEntry DurableCatalogImpl::getParsedCatalogEntry(OperationContext* opCtx,
-                                                              const RecordId& catalogId,
-                                                              const BSONObj& obj) const {
-    BSONElement idxIdent = obj["idxIdent"];
-    return DurableCatalogEntry{catalogId,
-                               obj["ident"].String(),
-                               idxIdent.eoo() ? BSONObj() : idxIdent.Obj().getOwned(),
-                               _parseMetaData(obj["md"])};
-}
-
 std::shared_ptr<BSONCollectionCatalogEntry::MetaData> DurableCatalogImpl::getMetaData(
     OperationContext* opCtx, const RecordId& catalogId) const {
     BSONObj obj = _findEntry(opCtx, catalogId);
diff --git a/src/mongo/db/storage/durable_catalog_impl.h b/src/mongo/db/storage/durable_catalog_impl.h
index 6c3805f880e..f85a02de188 100644
--- a/src/mongo/db/storage/durable_catalog_impl.h
+++ b/src/mongo/db/storage/durable_catalog_impl.h
@@ -88,10 +88,6 @@ public:
     boost::optional<DurableCatalogEntry> getParsedCatalogEntry(
         OperationContext* opCtx, const RecordId& catalogId) const override;
 
-    DurableCatalogEntry getParsedCatalogEntry(OperationContext* opCtx,
-                                              const RecordId& catalogId,
-                                              const BSONObj& obj) const override;
-
     std::shared_ptr<BSONCollectionCatalogEntry::MetaData> getMetaData(
         OperationContext* opCtx, const RecordId& catalogId) const;
     void putMetaData(OperationContext* opCtx,
diff --git a/src/mongo/db/storage/historical_ident_tracker.cpp b/src/mongo/db/storage/historical_ident_tracker.cpp
new file mode 100644
index 00000000000..34a1a72b2a7
--- /dev/null
+++ b/src/mongo/db/storage/historical_ident_tracker.cpp
@@ -0,0 +1,218 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+
+#include "mongo/db/storage/historical_ident_tracker.h"
+#include "mongo/logv2/log.h"
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
+
+
+namespace mongo {
+
+namespace {
+
+const auto getHistoricalIdentTracker = ServiceContext::declareDecoration<HistoricalIdentTracker>();
+
+}  // namespace
+
+HistoricalIdentTracker& HistoricalIdentTracker::get(ServiceContext* svcCtx) {
+    return getHistoricalIdentTracker(svcCtx);
+}
+
+HistoricalIdentTracker& HistoricalIdentTracker::get(OperationContext* opCtx) {
+    return get(opCtx->getServiceContext());
+}
+
+boost::optional<std::pair<NamespaceString, UUID>> HistoricalIdentTracker::lookup(
+    const std::string& ident, Timestamp timestamp) const {
+    stdx::lock_guard<Latch> lk(_mutex);
+
+    auto mapIt = _historicalIdents.find(ident);
+    if (mapIt == _historicalIdents.end()) {
+        // No historical entries for this ident.
+        return boost::none;
+    }
+
+    for (auto listIt = mapIt->second.begin(); listIt != mapIt->second.end(); listIt++) {
+        if (timestamp >= listIt->start && timestamp <= listIt->end) {
+            // Found the historical entry for the requested timestamp.
+            return std::make_pair(listIt->nss, listIt->uuid);
+        }
+    }
+
+    // No historical entry for the requested timestamp was found.
+    return boost::none;
+}
+
+void HistoricalIdentTracker::pinAtTimestamp(Timestamp timestamp) {
+    stdx::lock_guard<Latch> lk(_mutex);
+    _pinnedTimestamp = timestamp;
+}
+
+void HistoricalIdentTracker::unpin() {
+    stdx::lock_guard<Latch> lk(_mutex);
+    _pinnedTimestamp = Timestamp::min();
+}
+
+void HistoricalIdentTracker::removeEntriesOlderThan(Timestamp timestamp) {
+    Timestamp removeOlderThan =
+        _pinnedTimestamp.isNull() ? timestamp : std::min(timestamp, _pinnedTimestamp);
+
+    LOGV2_DEBUG(
+        6321801, 2, "Removing historical entries older than", "timestamp"_attr = removeOlderThan);
+
+    std::vector<std::string> keysToRemove;
+    stdx::lock_guard<Latch> lk(_mutex);
+    for (auto mapIt = _historicalIdents.begin(); mapIt != _historicalIdents.end(); mapIt++) {
+
+        auto listIt = mapIt->second.begin();
+        while (listIt != mapIt->second.end()) {
+            if (listIt->end < removeOlderThan) {
+                // This historical entry needs to be a removed, but we'll do a ranged delete later.
+                LOGV2_DEBUG(6321802,
+                            2,
+                            "Removing historical entry",
+                            "ident"_attr = mapIt->first,
+                            "nss"_attr = listIt->nss,
+                            "uuid"_attr = listIt->uuid,
+                            "start"_attr = listIt->start,
+                            "end"_attr = listIt->end);
+                listIt++;
+                continue;
+            }
+
+            // We need to keep this and any following historical entries. We can do a ranged delete
+            // now for what we don't need.
+            mapIt->second.erase(mapIt->second.begin(), listIt);
+            break;
+        }
+
+        if (listIt == mapIt->second.end()) {
+            // All of the historical entries need to be deleted for this ident. We'll erase the map
+            // entry outside of the loop to avoid iterator invalidation.
+            keysToRemove.push_back(mapIt->first);
+        }
+    }
+
+    for (const auto& keyToRemove : keysToRemove) {
+        _historicalIdents.erase(keyToRemove);
+    }
+}
+
+void HistoricalIdentTracker::rollbackTo(Timestamp timestamp) {
+    Timestamp rollbackTo =
+        _pinnedTimestamp.isNull() ? timestamp : std::max(timestamp, _pinnedTimestamp);
+
+    LOGV2_DEBUG(6321803, 2, "Rolling back historical entries to", "timestamp"_attr = rollbackTo);
+
+    std::vector<std::string> keysToRemove;
+    stdx::lock_guard<Latch> lk(_mutex);
+    for (auto mapIt = _historicalIdents.begin(); mapIt != _historicalIdents.end(); mapIt++) {
+
+        auto listIt = mapIt->second.begin();
+        while (listIt != mapIt->second.end()) {
+            if (listIt->end < rollbackTo) {
+                // This historical entry needs to be kept.
+                listIt++;
+                continue;
+            }
+
+            LOGV2_DEBUG(6321804,
+                        2,
+                        "Removing historical entries at and beyond",
+                        "ident"_attr = mapIt->first,
+                        "nss"_attr = listIt->nss,
+                        "uuid"_attr = listIt->uuid,
+                        "start"_attr = listIt->start,
+                        "end"_attr = listIt->end);
+
+            // We need to remove this and any following historical entries. We can do a ranged
+            // delete now for what we don't need.
+            mapIt->second.erase(listIt, mapIt->second.end());
+
+            if (mapIt->second.empty()) {
+                // Everything was erased. The map entry will be erased outside of the loop to avoid
+                // iterator invalidation.
+                keysToRemove.push_back(mapIt->first);
+            }
+
+            break;
+        }
+    }
+
+    for (const auto& keyToRemove : keysToRemove) {
+        _historicalIdents.erase(keyToRemove);
+    }
+}
+
+void HistoricalIdentTracker::_addHistoricalIdent(const std::string& ident,
+                                                 const NamespaceString& nss,
+                                                 const UUID& uuid,
+                                                 Timestamp timestamp) {
+    if (timestamp.isNull()) {
+        // Standalone nodes don't use timestamps.
+        return;
+    }
+
+    HistoricalIdentEntry entry{nss, uuid, /*start=*/Timestamp::min(), /*end=*/timestamp - 1};
+
+    stdx::lock_guard<Latch> lk(_mutex);
+    auto it = _historicalIdents.find(ident);
+    if (it == _historicalIdents.end()) {
+        // There are no historical entries for this ident yet.
+        LOGV2_DEBUG(6321805,
+                    2,
+                    "Adding new historical entry",
+                    "ident"_attr = ident,
+                    "nss"_attr = entry.nss,
+                    "uuid"_attr = entry.uuid,
+                    "start"_attr = entry.start,
+                    "end"_attr = entry.end);
+        _historicalIdents.insert({ident, {std::move(entry)}});
+        return;
+    }
+
+    invariant(!it->second.empty());
+
+    // Update the start timestamp to be the last entry's end timestamp + 1.
+    entry.start = it->second.back().end + 1;
+
+    LOGV2_DEBUG(6321806,
+                2,
+                "Adding new historical entry",
+                "ident"_attr = ident,
+                "nss"_attr = entry.nss,
+                "uuid"_attr = entry.uuid,
+                "start"_attr = entry.start,
+                "end"_attr = entry.end);
+    it->second.push_back(std::move(entry));
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/storage/historical_ident_tracker.h b/src/mongo/db/storage/historical_ident_tracker.h
new file mode 100644
index 00000000000..fe14e83a04f
--- /dev/null
+++ b/src/mongo/db/storage/historical_ident_tracker.h
@@ -0,0 +1,145 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <list>
+#include <string>
+#include <unordered_map>
+
+#include "mongo/bson/timestamp.h"
+#include "mongo/db/namespace_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/service_context.h"
+#include "mongo/platform/mutex.h"
+#include "mongo/util/uuid.h"
+
+namespace mongo {
+
+/**
+ * Keeps track of historical ident information when a collection is renamed or dropped.
+ */
+class HistoricalIdentTracker final {
+public:
+    HistoricalIdentTracker(const HistoricalIdentTracker&) = delete;
+    HistoricalIdentTracker(HistoricalIdentTracker&&) = delete;
+
+    static HistoricalIdentTracker& get(ServiceContext* svcCtx);
+    static HistoricalIdentTracker& get(OperationContext* opCtx);
+
+    HistoricalIdentTracker() = default;
+    ~HistoricalIdentTracker() = default;
+
+    /**
+     * Returns the historical namespace and UUID for 'ident' at 'timestamp'. Returns boost::none if
+     * there was no historical namespace.
+     */
+    boost::optional<std::pair<NamespaceString, UUID>> lookup(const std::string& ident,
+                                                             Timestamp timestamp) const;
+
+    /**
+     * Pins the historical content to the given timestamp, preventing it from being removed.
+     *
+     * This is necessary for backup cursors, which need to report the namespace and UUID at the time
+     * of the checkpoint the backup is being taken on. When a backup cursor is open, it pins the
+     * checkpoint the backup is being taken on. Checkpoints can still be taken, which advances the
+     * last checkpoint timestamp and would remove historical content needed by the open backup
+     * cursor. This method prevents that from happening by pinning the content.
+     *
+     * The checkpoint timestamp of the backup can be earlier than the oldest timestamp, which
+     * prevents us from opening a snapshot at the checkpoint timestamp as history before the oldest
+     * timestamp is discarded.
+     */
+    void pinAtTimestamp(Timestamp timestamp);
+    void unpin();
+
+    /**
+     * Records the idents namespace and UUID before it was renamed.
+     */
+    void recordRename(const std::string& ident,
+                      const NamespaceString& oldNss,
+                      const UUID& uuid,
+                      Timestamp timestamp) {
+        _addHistoricalIdent(ident, oldNss, uuid, timestamp);
+    }
+
+    /**
+     * Records the idents namespace and UUID before it was dropped.
+     */
+    void recordDrop(const std::string& ident,
+                    const NamespaceString& nss,
+                    const UUID& uuid,
+                    Timestamp timestamp) {
+        _addHistoricalIdent(ident, nss, uuid, timestamp);
+    }
+
+    /**
+     * Removes historical content that is no longer necessary. This is anything older than the last
+     * checkpoint timestamp.
+     *
+     * If there's a pinned timestamp, min(timestamp, _pinnedTimestamp) is used.
+     */
+    void removeEntriesOlderThan(Timestamp timestamp);
+
+    /**
+     * Historical content added may not be stable yet and can be rolled back. When rollback to
+     * stable runs, we need to remove any historical content that is considered current.
+     *
+     * If there's a pinned timestamp, max(timestamp, _pinnedTimestamp) is used.
+     */
+    void rollbackTo(Timestamp timestamp);
+
+private:
+    /**
+     * Helper function for recordRename() and recordDrop().
+     *
+     * Appends a new historical entry with 'nss' and 'uuid' for 'ident' in '_historicalIdents'.
+     * Sets the 'end' timestamp to be 'timestamp - 1'.
+     * Sets the 'start' timestamp to the timestamp of the last entry + 1, or Timestamp::min() if
+     * there was no earlier entry.
+     */
+    void _addHistoricalIdent(const std::string& ident,
+                             const NamespaceString& nss,
+                             const UUID& uuid,
+                             Timestamp timestamp);
+
+    struct HistoricalIdentEntry {
+        const NamespaceString nss;
+        const UUID uuid;
+        Timestamp start;
+        Timestamp end;
+    };
+
+    // Protects all the member variables below.
+    mutable Mutex _mutex = MONGO_MAKE_LATCH("HistoricalIdentTracker::_mutex");
+    stdx::unordered_map<std::string, std::list<HistoricalIdentEntry>> _historicalIdents;
+    Timestamp _pinnedTimestamp;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/storage/historical_ident_tracker_test.cpp b/src/mongo/db/storage/historical_ident_tracker_test.cpp
new file mode 100644
index 00000000000..072c10b443e
--- /dev/null
+++ b/src/mongo/db/storage/historical_ident_tracker_test.cpp
@@ -0,0 +1,437 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/storage/historical_ident_tracker.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+
+TEST(HistoricalIdentTracker, RecordHistoricalIdents) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(10, 10));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.b"),
+                         UUID::gen(),
+                         Timestamp(20, 20));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.c"),
+                         UUID::gen(),
+                         Timestamp(21, 21));
+    tracker.recordDrop(ident,
+                       /*nss=*/NamespaceString::createNamespaceString_forTest("test.d"),
+                       UUID::gen(),
+                       Timestamp(25, 25));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(15, 15))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(21, 21))->first,
+              NamespaceString::createNamespaceString_forTest("test.d"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(24, 24))->first,
+              NamespaceString::createNamespaceString_forTest("test.d"));
+
+    ASSERT(!tracker.lookup(ident, Timestamp(25, 25)));
+    ASSERT(!tracker.lookup(ident, Timestamp::max()));
+}
+
+TEST(HistoricalIdentTracker, SkipRecordingNullTimestamps) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(1));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.b"),
+                         UUID::gen(),
+                         Timestamp(2));
+    tracker.recordDrop(ident,
+                       /*nss=*/NamespaceString::createNamespaceString_forTest("test.c"),
+                       UUID::gen(),
+                       Timestamp(3));
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(1)));
+    ASSERT(!tracker.lookup(ident, Timestamp(2)));
+    ASSERT(!tracker.lookup(ident, Timestamp(3)));
+    ASSERT(!tracker.lookup(ident, Timestamp::max()));
+}
+
+TEST(HistoricalIdentTracker, RemoveEntriesOlderThanSingle) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(50, 50));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(49, 49))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+
+    ASSERT(!tracker.lookup(ident, Timestamp(50, 50)));
+
+    tracker.removeEntriesOlderThan(Timestamp::min());
+    tracker.removeEntriesOlderThan(Timestamp(49, 49));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(49, 49))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+
+    tracker.removeEntriesOlderThan(Timestamp(50, 50));
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(49, 49)));
+    ASSERT(!tracker.lookup(ident, Timestamp(50, 50)));
+}
+
+TEST(HistoricalIdentTracker, RemoveEntriesOlderThanMultiple) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(10, 10));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.b"),
+                         UUID::gen(),
+                         Timestamp(20, 20));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.c"),
+                         UUID::gen(),
+                         Timestamp(21, 21));
+
+    tracker.removeEntriesOlderThan(Timestamp::min());
+    tracker.removeEntriesOlderThan(Timestamp(5, 5));
+    tracker.removeEntriesOlderThan(Timestamp(9, 9));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.removeEntriesOlderThan(Timestamp(15, 15));
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.removeEntriesOlderThan(Timestamp(21, 21));
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT(!tracker.lookup(ident, Timestamp(10, 10)));
+    ASSERT(!tracker.lookup(ident, Timestamp(19, 19)));
+    ASSERT(!tracker.lookup(ident, Timestamp(20, 20)));
+    ASSERT(!tracker.lookup(ident, Timestamp::max()));
+}
+
+TEST(HistoricalIdentTracker, RollbackToSingle) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(10, 10));
+
+    tracker.rollbackTo(Timestamp(10, 10));
+    tracker.rollbackTo(Timestamp::max());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+
+    tracker.rollbackTo(Timestamp(9, 9));
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT(!tracker.lookup(ident, Timestamp(10, 10)));
+    ASSERT(!tracker.lookup(ident, Timestamp::max()));
+}
+
+TEST(HistoricalIdentTracker, RollbackToMultiple) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(10, 10));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.b"),
+                         UUID::gen(),
+                         Timestamp(20, 20));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.c"),
+                         UUID::gen(),
+                         Timestamp(21, 21));
+
+    tracker.rollbackTo(Timestamp::max());
+    tracker.rollbackTo(Timestamp(22, 22));
+    tracker.rollbackTo(Timestamp(21, 21));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.rollbackTo(Timestamp(15, 15));
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT(!tracker.lookup(ident, Timestamp(10, 10)));
+    ASSERT(!tracker.lookup(ident, Timestamp(19, 19)));
+    ASSERT(!tracker.lookup(ident, Timestamp(20, 20)));
+
+    tracker.rollbackTo(Timestamp::min());
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT(!tracker.lookup(ident, Timestamp(10, 10)));
+    ASSERT(!tracker.lookup(ident, Timestamp(19, 19)));
+    ASSERT(!tracker.lookup(ident, Timestamp(20, 20)));
+    ASSERT(!tracker.lookup(ident, Timestamp::max()));
+}
+
+TEST(HistoricalIdentTracker, PinAndUnpinTimestamp) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(10, 10));
+
+    tracker.pinAtTimestamp(Timestamp(5, 5));
+    tracker.removeEntriesOlderThan(Timestamp::max());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+
+    tracker.pinAtTimestamp(Timestamp(9, 9));
+    tracker.removeEntriesOlderThan(Timestamp::max());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+
+    tracker.unpin();
+    tracker.removeEntriesOlderThan(Timestamp::max());
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT(!tracker.lookup(ident, Timestamp(10, 10)));
+    ASSERT(!tracker.lookup(ident, Timestamp::max()));
+}
+
+TEST(HistoricalIdentTracker, PinnedTimestampRemoveEntriesOlderThan) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(10, 10));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.b"),
+                         UUID::gen(),
+                         Timestamp(20, 20));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.c"),
+                         UUID::gen(),
+                         Timestamp(21, 21));
+
+    tracker.pinAtTimestamp(Timestamp(5, 5));
+    tracker.removeEntriesOlderThan(Timestamp::max());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.pinAtTimestamp(Timestamp(9, 9));
+    tracker.removeEntriesOlderThan(Timestamp(9, 9));
+    tracker.removeEntriesOlderThan(Timestamp(10, 10));
+    tracker.removeEntriesOlderThan(Timestamp::max());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.pinAtTimestamp(Timestamp(15, 15));
+    tracker.removeEntriesOlderThan(Timestamp::max());
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.pinAtTimestamp(Timestamp(21, 21));
+    tracker.removeEntriesOlderThan(Timestamp::max());
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT(!tracker.lookup(ident, Timestamp(10, 10)));
+    ASSERT(!tracker.lookup(ident, Timestamp(19, 19)));
+    ASSERT(!tracker.lookup(ident, Timestamp(20, 20)));
+}
+
+TEST(HistoricalIdentTracker, PinnedTimestampRollbackTo) {
+    HistoricalIdentTracker tracker;
+
+    const std::string ident = "ident";
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.a"),
+                         UUID::gen(),
+                         Timestamp(10, 10));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.b"),
+                         UUID::gen(),
+                         Timestamp(20, 20));
+    tracker.recordRename(ident,
+                         /*oldNss=*/NamespaceString::createNamespaceString_forTest("test.c"),
+                         UUID::gen(),
+                         Timestamp(21, 21));
+
+    tracker.pinAtTimestamp(Timestamp(30, 30));
+    tracker.rollbackTo(Timestamp::min());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.pinAtTimestamp(Timestamp(21, 21));
+    tracker.rollbackTo(Timestamp::min());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(20, 20))->first,
+              NamespaceString::createNamespaceString_forTest("test.c"));
+
+    tracker.pinAtTimestamp(Timestamp(20, 20));
+    tracker.rollbackTo(Timestamp::min());
+
+    ASSERT_EQ(tracker.lookup(ident, Timestamp::min())->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(9, 9))->first,
+              NamespaceString::createNamespaceString_forTest("test.a"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(10, 10))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT_EQ(tracker.lookup(ident, Timestamp(19, 19))->first,
+              NamespaceString::createNamespaceString_forTest("test.b"));
+    ASSERT(!tracker.lookup(ident, Timestamp(20, 20)));
+
+
+    tracker.pinAtTimestamp(Timestamp(5, 5));
+    tracker.rollbackTo(Timestamp::min());
+
+    ASSERT(!tracker.lookup(ident, Timestamp::min()));
+    ASSERT(!tracker.lookup(ident, Timestamp(9, 9)));
+    ASSERT(!tracker.lookup(ident, Timestamp(10, 10)));
+    ASSERT(!tracker.lookup(ident, Timestamp(19, 19)));
+    ASSERT(!tracker.lookup(ident, Timestamp(20, 20)));
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index d6e4638d472..00eac79661a 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -322,10 +322,6 @@ public:
 
         virtual ~StreamingCursor() = default;
 
-        virtual void setCatalogEntries(
-            const stdx::unordered_map<std::string, std::pair<NamespaceString, UUID>>&
-                identsToNsAndUUID) = 0;
-
         virtual StatusWith<std::deque<BackupBlock>> getNextBatch(OperationContext* opCtx,
                                                                  std::size_t batchSize) = 0;
 
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index fe693cefb7e..367791feb6a 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -48,6 +48,7 @@
 #include "mongo/db/storage/deferred_drop_record_store.h"
 #include "mongo/db/storage/durable_catalog_impl.h"
 #include "mongo/db/storage/durable_history_pin.h"
+#include "mongo/db/storage/historical_ident_tracker.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
 #include "mongo/db/storage/storage_repair_observer.h"
@@ -91,6 +92,12 @@ StorageEngineImpl::StorageEngineImpl(OperationContext* opCtx,
           [this](OperationContext* opCtx, Timestamp timestamp) {
               _onMinOfCheckpointAndOldestTimestampChanged(opCtx, timestamp);
           }),
+      _historicalIdentTimestampListener(
+          TimestampMonitor::TimestampType::kCheckpoint,
+          [serviceContext = opCtx->getServiceContext()](OperationContext* opCtx,
+                                                        Timestamp timestamp) {
+              HistoricalIdentTracker::get(opCtx).removeEntriesOlderThan(timestamp);
+          }),
       _collectionCatalogCleanupTimestampListener(
           TimestampMonitor::TimestampType::kOldest,
           [serviceContext = opCtx->getServiceContext()](OperationContext* opCtx,
@@ -890,6 +897,7 @@ void StorageEngineImpl::startTimestampMonitor() {
         _engine.get(), getGlobalServiceContext()->getPeriodicRunner());
 
     _timestampMonitor->addListener(&_minOfCheckpointAndOldestTimestampListener);
+    _timestampMonitor->addListener(&_historicalIdentTimestampListener);
     _timestampMonitor->addListener(&_collectionCatalogCleanupTimestampListener);
 }
 
diff --git a/src/mongo/db/storage/storage_engine_impl.h b/src/mongo/db/storage/storage_engine_impl.h
index dcebf9383b0..e6902baaa52 100644
--- a/src/mongo/db/storage/storage_engine_impl.h
+++ b/src/mongo/db/storage/storage_engine_impl.h
@@ -454,6 +454,10 @@ private:
     // Listener for min of checkpoint and oldest timestamp changes.
     TimestampMonitor::TimestampListener _minOfCheckpointAndOldestTimestampListener;
 
+    // Listener for checkpoint timestamp changes to remove historical ident entries older than the
+    // checkpoint timestamp.
+    TimestampMonitor::TimestampListener _historicalIdentTimestampListener;
+
     // Listener for cleanup of CollectionCatalog when oldest timestamp advances.
     TimestampMonitor::TimestampListener _collectionCatalogCleanupTimestampListener;
 
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index d3358fcb7ce..8f348a08724 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -1053,11 +1053,6 @@ public:
 
     ~StreamingCursorImpl() = default;
 
-    void setCatalogEntries(const stdx::unordered_map<std::string, std::pair<NamespaceString, UUID>>&
-                               identsToNsAndUUID) {
-        _identsToNsAndUUID = std::move(identsToNsAndUUID);
-    }
-
     StatusWith<std::deque<BackupBlock>> getNextBatch(OperationContext* opCtx,
                                                      const std::size_t batchSize) {
         int wtRet = 0;
@@ -1119,11 +1114,9 @@ public:
                 // to an entire file. Full backups cannot open an incremental cursor, even if they
                 // are the initial incremental backup.
                 const std::uint64_t length = options.incrementalBackup ? fileSize : 0;
-                auto nsAndUUID = _getNsAndUUID(filePath.stem().string());
                 backupBlocks.push_back(BackupBlock(opCtx,
-                                                   nsAndUUID.first,
-                                                   nsAndUUID.second,
                                                    filePath.string(),
+                                                   _wtBackup->identToNamespaceAndUUIDMap,
                                                    _checkpointTimestamp,
                                                    0 /* offset */,
                                                    length,
@@ -1139,15 +1132,6 @@ public:
     }
 
 private:
-    std::pair<boost::optional<NamespaceString>, boost::optional<UUID>> _getNsAndUUID(
-        const std::string& ident) const {
-        auto it = _identsToNsAndUUID.find(ident);
-        if (it == _identsToNsAndUUID.end()) {
-            return std::make_pair(boost::none, boost::none);
-        }
-        return it->second;
-    }
-
     Status _getNextIncrementalBatchForFile(OperationContext* opCtx,
                                            const char* filename,
                                            boost::filesystem::path filePath,
@@ -1201,11 +1185,9 @@ private:
                         "offset"_attr = offset,
                         "size"_attr = size,
                         "type"_attr = type);
-            auto nsAndUUID = _getNsAndUUID(filePath.stem().string());
             backupBlocks->push_back(BackupBlock(opCtx,
-                                                nsAndUUID.first,
-                                                nsAndUUID.second,
                                                 filePath.string(),
+                                                _wtBackup->identToNamespaceAndUUIDMap,
                                                 _checkpointTimestamp,
                                                 offset,
                                                 size,
@@ -1215,11 +1197,9 @@ private:
         // If the file is unchanged, push a BackupBlock with offset=0 and length=0. This allows us
         // to distinguish between an unchanged file and a deleted file in an incremental backup.
         if (fileUnchangedFlag) {
-            auto nsAndUUID = _getNsAndUUID(filePath.stem().string());
             backupBlocks->push_back(BackupBlock(opCtx,
-                                                nsAndUUID.first,
-                                                nsAndUUID.second,
                                                 filePath.string(),
+                                                _wtBackup->identToNamespaceAndUUIDMap,
                                                 _checkpointTimestamp,
                                                 0 /* offset */,
                                                 0 /* length */,
@@ -1242,7 +1222,6 @@ private:
 
     WT_SESSION* _session;
     std::string _path;
-    stdx::unordered_map<std::string, std::pair<NamespaceString, UUID>> _identsToNsAndUUID;
     boost::optional<Timestamp> _checkpointTimestamp;
     WiredTigerBackup* _wtBackup;  // '_wtBackup' is an out parameter.
 };
@@ -1302,10 +1281,34 @@ WiredTigerKVEngine::beginNonBlockingBackup(OperationContext* opCtx,
 
     invariant(_wtBackup.logFilePathsSeenByExtendBackupCursor.empty());
     invariant(_wtBackup.logFilePathsSeenByGetNextBatch.empty());
+    invariant(_wtBackup.identToNamespaceAndUUIDMap.empty());
+
+    // Fetching the catalog entries requires reading from the storage engine. During cache pressure,
+    // this read could be rolled back. In that case, we need to clear the map.
+    ScopeGuard clearGuard([&] { _wtBackup.identToNamespaceAndUUIDMap.clear(); });
+
+    {
+        Lock::GlobalLock lk(opCtx, MODE_IS);
+        DurableCatalog* catalog = DurableCatalog::get(opCtx);
+        std::vector<DurableCatalog::EntryIdentifier> catalogEntries =
+            catalog->getAllCatalogEntries(opCtx);
+        for (const DurableCatalog::EntryIdentifier& e : catalogEntries) {
+            // Populate the collection ident with its namespace and UUID.
+            UUID uuid = catalog->getMetaData(opCtx, e.catalogId)->options.uuid.value();
+            _wtBackup.identToNamespaceAndUUIDMap.emplace(e.ident, std::make_pair(e.nss, uuid));
+
+            // Populate the collection's index idents with the collection's namespace and UUID.
+            std::vector<std::string> idxIdents = catalog->getIndexIdents(opCtx, e.catalogId);
+            for (const std::string& idxIdent : idxIdents) {
+                _wtBackup.identToNamespaceAndUUIDMap.emplace(idxIdent, std::make_pair(e.nss, uuid));
+            }
+        }
+    }
 
     auto streamingCursor = std::make_unique<StreamingCursorImpl>(
         session, _path, checkpointTimestamp, options, &_wtBackup);
 
+    clearGuard.dismiss();
     pinOplogGuard.dismiss();
     _backupSession = std::move(sessionRaii);
     _wtBackup.cursor = cursor;
@@ -1326,6 +1329,7 @@ void WiredTigerKVEngine::endNonBlockingBackup(OperationContext* opCtx) {
     _wtBackup.dupCursor = nullptr;
     _wtBackup.logFilePathsSeenByExtendBackupCursor = {};
     _wtBackup.logFilePathsSeenByGetNextBatch = {};
+    _wtBackup.identToNamespaceAndUUIDMap = {};
 
     boost::filesystem::remove(getOngoingBackupPath());
 }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index 0cffd8772a1..6a1445c8e98 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -82,6 +82,7 @@ struct WiredTigerBackup {
     WT_CURSOR* dupCursor = nullptr;
     std::set<std::string> logFilePathsSeenByExtendBackupCursor;
     std::set<std::string> logFilePathsSeenByGetNextBatch;
+    BackupBlock::IdentToNamespaceAndUUIDMap identToNamespaceAndUUIDMap;
 
     // 'wtBackupCursorMutex' provides concurrency control between beginNonBlockingBackup(),
     // endNonBlockingBackup(), and getNextBatch() because we stream the output of the backup cursor.