SERVER-62010 Change DurableCatalog to store TenantNamespace

19 files changed, 257 insertions, 171 deletions

diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 8ed4e53ae99..d5e8917b383 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -1876,11 +1876,11 @@ Status CollectionImpl::rename(OperationContext* opCtx,
                               const TenantNamespace& tenantNs,
                               bool stayTemp) {
     auto metadata = std::make_shared<BSONCollectionCatalogEntry::MetaData>(*_metadata);
-    metadata->ns = tenantNs.getNss().ns();
+    metadata->tenantNs = tenantNs;
     if (!stayTemp)
         metadata->options.temp = false;
-    Status status = DurableCatalog::get(opCtx)->renameCollection(
-        opCtx, getCatalogId(), tenantNs.getNss(), *metadata);
+    Status status =
+        DurableCatalog::get(opCtx)->renameCollection(opCtx, getCatalogId(), tenantNs, *metadata);
     if (!status.isOK()) {
         return status;
     }
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index d7198f05c7f..e9855352037 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -752,11 +752,11 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
 
     // Create Collection object
     auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
+    TenantNamespace tenantNs(getActiveTenant(opCtx), nss);
     std::pair<RecordId, std::unique_ptr<RecordStore>> catalogIdRecordStorePair =
         uassertStatusOK(storageEngine->getCatalog()->createCollection(
-            opCtx, nss, optionsWithUUID, true /*allocateDefaultSpace*/));
+            opCtx, tenantNs, optionsWithUUID, true /*allocateDefaultSpace*/));
     auto catalogId = catalogIdRecordStorePair.first;
-    TenantNamespace tenantNs(getActiveTenant(opCtx), nss);
     std::shared_ptr<Collection> ownedCollection = Collection::Factory::get(opCtx)->make(
         opCtx, tenantNs, catalogId, optionsWithUUID, std::move(catalogIdRecordStorePair.second));
     auto collection = ownedCollection.get();
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index 94e09994b0c..420fb5cf892 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -79,7 +79,7 @@ IndexCatalogEntryImpl::IndexCatalogEntryImpl(OperationContext* const opCtx,
     _descriptor->_entry = this;
     _isReady = collection->isIndexReady(_descriptor->indexName());
 
-    auto nss = DurableCatalog::get(opCtx)->getEntry(_catalogId).nss;
+    auto nss = DurableCatalog::get(opCtx)->getEntry(_catalogId).tenantNs.getNss();
     const BSONObj& collation = _descriptor->collation();
     if (!collation.isEmpty()) {
         auto statusWithCollator =
@@ -352,7 +352,7 @@ std::shared_ptr<Ident> IndexCatalogEntryImpl::getSharedIdent() const {
 // ----
 
 NamespaceString IndexCatalogEntryImpl::getNSSFromCatalog(OperationContext* opCtx) const {
-    return DurableCatalog::get(opCtx)->getEntry(_catalogId).nss;
+    return DurableCatalog::get(opCtx)->getEntry(_catalogId).tenantNs.getNss();
 }
 
 bool IndexCatalogEntryImpl::isReadyInMySnapshot(OperationContext* opCtx) const {
diff --git a/src/mongo/db/index_builds_coordinator_mongod_test.cpp b/src/mongo/db/index_builds_coordinator_mongod_test.cpp
index 22be756d7c0..0590d344b7b 100644
--- a/src/mongo/db/index_builds_coordinator_mongod_test.cpp
+++ b/src/mongo/db/index_builds_coordinator_mongod_test.cpp
@@ -62,8 +62,8 @@ public:
     const NamespaceString _testBarNss = NamespaceString("test.bar");
     const UUID _othertestFooUUID = UUID::gen();
     const NamespaceString _othertestFooNss = NamespaceString("othertest.foo");
-    const std::string _tenantId{"tenant"};
-    const NamespaceString _testTenantFooNss{_tenantId + "_test.test"};
+    const TenantId _tenantId{OID::gen()};
+    const NamespaceString _testTenantFooNss{_tenantId.toString() + "_test.test"};
     const UUID _testFooTenantUUID = UUID::gen();
     const IndexBuildsCoordinator::IndexBuildOptions _indexBuildOptions = {
         CommitQuorumOptions(CommitQuorumOptions::kDisabled)};
@@ -388,7 +388,8 @@ TEST_F(IndexBuildsCoordinatorMongodTest, AbortBuildIndexDueToTenantMigration) {
 
     // This call may see the index build active and wait for it to be unregistered, or the index
     // build may already have been unregistered.
-    _indexBuildsCoord->abortTenantIndexBuilds(operationContext(), _tenantId, "tenant migration");
+    _indexBuildsCoord->abortTenantIndexBuilds(
+        operationContext(), _tenantId.toString(), "tenant migration");
 
     ASSERT_EQ(0, _indexBuildsCoord->getActiveIndexBuildCount(operationContext()));
 
diff --git a/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp b/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp
index 31f446146a1..030edd19a15 100644
--- a/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp
+++ b/src/mongo/db/repl/tenant_migration_shard_merge_util.cpp
@@ -134,6 +134,7 @@ void wiredTigerImportFromBackupCursor(OperationContext* opCtx,
             });
 
             // Create Collection object
+            TenantNamespace tenantNs(getActiveTenant(opCtx), nss);
             auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
             auto durableCatalog = storageEngine->getCatalog();
             ImportOptions importOptions(ImportOptions::ImportCollectionUUIDOption::kKeepOld);
@@ -142,7 +143,7 @@ void wiredTigerImportFromBackupCursor(OperationContext* opCtx,
             // TODO SERVER-62659 Ensure the correct tenantId is used when importing the collection.
             auto importResult = uassertStatusOK(
                 DurableCatalog::get(opCtx)->importCollection(opCtx,
-                                                             collectionMetadata.ns,
+                                                             tenantNs,
                                                              collectionMetadata.catalogObject,
                                                              storageMetadata.done(),
                                                              importOptions));
@@ -151,7 +152,6 @@ void wiredTigerImportFromBackupCursor(OperationContext* opCtx,
                 uassert(6114301, "Cannot import non-ready indexes", index.ready);
             }
 
-            TenantNamespace tenantNs(getActiveTenant(opCtx), nss);
             std::shared_ptr<Collection> ownedCollection = Collection::Factory::get(opCtx)->make(
                 opCtx, tenantNs, importResult.catalogId, md, std::move(importResult.rs));
             ownedCollection->init(opCtx);
diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index eaff1bff6bb..1003bffd2ec 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -113,6 +113,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/service_context',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/server_options',
     ],
 )
@@ -513,6 +514,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/collection',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/storage/storage_options'
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.cpp b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
index b5527947067..1266e07ac37 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
@@ -190,7 +190,7 @@ bool BSONCollectionCatalogEntry::MetaData::eraseIndex(StringData name) {
 
 BSONObj BSONCollectionCatalogEntry::MetaData::toBSON(bool hasExclusiveAccess) const {
     BSONObjBuilder b;
-    b.append("ns", ns);
+    b.append("ns", tenantNs.toString());
     b.append("options", options.toBSON());
     {
         BSONArrayBuilder arr(b.subarrayStart("indexes"));
@@ -238,7 +238,7 @@ BSONObj BSONCollectionCatalogEntry::MetaData::toBSON(bool hasExclusiveAccess) co
 }
 
 void BSONCollectionCatalogEntry::MetaData::parse(const BSONObj& obj) {
-    ns = obj.getStringField("ns").toString();
+    tenantNs = TenantNamespace::parseTenantNamespaceFromDisk(obj.getStringField("ns").toString());
 
     if (obj["options"].isABSONObj()) {
         options = uassertStatusOK(
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.h b/src/mongo/db/storage/bson_collection_catalog_entry.h
index 65f08aa3c95..cf9c907e491 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.h
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.h
@@ -34,6 +34,7 @@
 
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/index/multikey_paths.h"
+#include "mongo/db/tenant_namespace.h"
 
 namespace mongo {
 
@@ -147,7 +148,7 @@ public:
          */
         bool eraseIndex(StringData name);
 
-        std::string ns;
+        TenantNamespace tenantNs;
         CollectionOptions options;
         std::vector<IndexMetaData> indexes;
 
diff --git a/src/mongo/db/storage/durable_catalog.h b/src/mongo/db/storage/durable_catalog.h
index 25099729dbf..24c61455531 100644
--- a/src/mongo/db/storage/durable_catalog.h
+++ b/src/mongo/db/storage/durable_catalog.h
@@ -37,6 +37,7 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/bson_collection_catalog_entry.h"
 #include "mongo/db/storage/storage_engine.h"
+#include "mongo/db/tenant_namespace.h"
 
 namespace mongo {
 /**
@@ -57,11 +58,11 @@ public:
      */
     struct Entry {
         Entry() {}
-        Entry(RecordId catalogId, std::string ident, NamespaceString nss)
-            : catalogId(catalogId), ident(std::move(ident)), nss(std::move(nss)) {}
+        Entry(RecordId catalogId, std::string ident, TenantNamespace tenantNs)
+            : catalogId(catalogId), ident(std::move(ident)), tenantNs(std::move(tenantNs)) {}
         RecordId catalogId;
         std::string ident;
-        NamespaceString nss;
+        TenantNamespace tenantNs;
     };
 
     virtual ~DurableCatalog() {}
@@ -86,7 +87,7 @@ public:
         OperationContext* opCtx, RecordId id) const = 0;
 
     /**
-     * Updates the catalog entry for the collection 'nss' with the fields specified in 'md'. If
+     * Updates the catalog entry for the collection 'tenantNs' with the fields specified in 'md'. If
      * 'md.indexes' contains a new index entry, then this method generates a new index ident and
      * adds it to the catalog entry.
      */
@@ -132,7 +133,7 @@ public:
      */
     virtual StatusWith<std::pair<RecordId, std::unique_ptr<RecordStore>>> createCollection(
         OperationContext* opCtx,
-        const NamespaceString& nss,
+        const TenantNamespace& tenantNs,
         const CollectionOptions& options,
         bool allocateDefaultSpace) = 0;
 
@@ -147,7 +148,7 @@ public:
      * catalog entry and contain the following fields:
      * "md": A document representing the BSONCollectionCatalogEntry::MetaData of the collection.
      * "idxIdent": A document containing {<index_name>: <index_ident>} pairs for all indexes.
-     * "ns": Namespace of the collection being imported.
+     * "tenantNs": TenantNamespace of the collection being imported.
      * "ident": Ident of the collection file.
      *
      * On success, returns an ImportResult structure containing the RecordId which identifies the
@@ -165,14 +166,14 @@ public:
     };
 
     virtual StatusWith<ImportResult> importCollection(OperationContext* opCtx,
-                                                      const NamespaceString& nss,
+                                                      const TenantNamespace& tenantNs,
                                                       const BSONObj& metadata,
                                                       const BSONObj& storageMetadata,
                                                       const ImportOptions& importOptions) = 0;
 
     virtual Status renameCollection(OperationContext* opCtx,
                                     RecordId catalogId,
-                                    const NamespaceString& toNss,
+                                    const TenantNamespace& toTenantNs,
                                     BSONCollectionCatalogEntry::MetaData& md) = 0;
 
     /**
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index 7382b6301a9..fa5d38bb9f2 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/multitenancy.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/repl/replication_coordinator.h"
@@ -259,12 +260,12 @@ std::string DurableCatalogImpl::getFilesystemPathForDb(const std::string& dbName
     }
 }
 
-std::string DurableCatalogImpl::_newUniqueIdent(NamespaceString nss, const char* kind) {
+std::string DurableCatalogImpl::_newUniqueIdent(TenantNamespace tenantNs, const char* kind) {
     // If this changes to not put _rand at the end, _hasEntryCollidingWithRand will need fixing.
     stdx::lock_guard<Latch> lk(_randLock);
     StringBuilder buf;
     if (_directoryPerDb) {
-        buf << escapeDbName(nss.db()) << '/';
+        buf << escapeDbName(tenantNs.db()) << '/';
     }
     buf << kind;
     buf << (_directoryForIndexes ? '/' : '-');
@@ -286,7 +287,8 @@ void DurableCatalogImpl::init(OperationContext* opCtx) {
         // No rollback since this is just loading already committed data.
         auto ident = obj["ident"].String();
         auto ns = obj["ns"].String();
-        _catalogIdToEntryMap[record->id] = Entry(record->id, ident, NamespaceString(ns));
+        _catalogIdToEntryMap[record->id] =
+            Entry(record->id, ident, TenantNamespace::parseTenantNamespaceFromDisk(ns));
     }
 
     // In the unlikely event that we have used this _rand before generate a new one.
@@ -308,9 +310,9 @@ std::vector<DurableCatalog::Entry> DurableCatalogImpl::getAllCatalogEntries(
             continue;
         }
         auto ident = obj["ident"].String();
-        auto collName = obj["ns"].String();
+        auto ns = obj["ns"].String();
 
-        ret.emplace_back(record->id, ident, NamespaceString(collName));
+        ret.emplace_back(record->id, ident, TenantNamespace::parseTenantNamespaceFromDisk(ns));
     }
 
     return ret;
@@ -324,19 +326,19 @@ DurableCatalog::Entry DurableCatalogImpl::getEntry(RecordId catalogId) const {
 }
 
 StatusWith<DurableCatalog::Entry> DurableCatalogImpl::_addEntry(OperationContext* opCtx,
-                                                                NamespaceString nss,
+                                                                TenantNamespace tenantNs,
                                                                 const CollectionOptions& options) {
-    invariant(opCtx->lockState()->isDbLockedForMode(nss.db(), MODE_IX));
+    invariant(opCtx->lockState()->isDbLockedForMode(tenantNs.db(), MODE_IX));
 
-    auto ident = _newUniqueIdent(nss, "collection");
+    auto ident = _newUniqueIdent(tenantNs, "collection");
 
     BSONObj obj;
     {
         BSONObjBuilder b;
-        b.append("ns", nss.ns());
+        b.append("ns", tenantNs.toString());
         b.append("ident", ident);
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = nss.ns();
+        md.tenantNs = tenantNs;
         md.options = options;
 
         // (Generic FCV reference): TODO SERVER-60912: When kLastLTS is 6.0, remove this FCV-gated
@@ -361,21 +363,21 @@ StatusWith<DurableCatalog::Entry> DurableCatalogImpl::_addEntry(OperationContext
 
     stdx::lock_guard<Latch> lk(_catalogIdToEntryMapLock);
     invariant(_catalogIdToEntryMap.find(res.getValue()) == _catalogIdToEntryMap.end());
-    _catalogIdToEntryMap[res.getValue()] = {res.getValue(), ident, nss};
+    _catalogIdToEntryMap[res.getValue()] = {res.getValue(), ident, tenantNs};
     opCtx->recoveryUnit()->registerChange(std::make_unique<AddIdentChange>(this, res.getValue()));
 
     LOGV2_DEBUG(22207,
                 1,
-                "stored meta data for {nss_ns} @ {res_getValue}",
-                "nss_ns"_attr = nss.ns(),
+                "stored meta data for {tenantNs} @ {res_getValue}",
+                logAttrs(tenantNs),
                 "res_getValue"_attr = res.getValue());
-    return {{res.getValue(), ident, nss}};
+    return {{res.getValue(), ident, tenantNs}};
 }
 
 StatusWith<DurableCatalog::Entry> DurableCatalogImpl::_importEntry(OperationContext* opCtx,
-                                                                   NamespaceString nss,
+                                                                   TenantNamespace tenantNs,
                                                                    const BSONObj& metadata) {
-    invariant(opCtx->lockState()->isDbLockedForMode(nss.db(), MODE_IX));
+    invariant(opCtx->lockState()->isDbLockedForMode(tenantNs.db(), MODE_IX));
 
     auto ident = metadata["ident"].String();
     StatusWith<RecordId> res =
@@ -385,11 +387,12 @@ StatusWith<DurableCatalog::Entry> DurableCatalogImpl::_importEntry(OperationCont
 
     stdx::lock_guard<Latch> lk(_catalogIdToEntryMapLock);
     invariant(_catalogIdToEntryMap.find(res.getValue()) == _catalogIdToEntryMap.end());
-    _catalogIdToEntryMap[res.getValue()] = {res.getValue(), ident, nss};
+    _catalogIdToEntryMap[res.getValue()] = {res.getValue(), ident, tenantNs};
     opCtx->recoveryUnit()->registerChange(std::make_unique<AddIdentChange>(this, res.getValue()));
 
-    LOGV2_DEBUG(5095101, 1, "imported meta data", logAttrs(nss), "metadata"_attr = res.getValue());
-    return {{res.getValue(), ident, nss}};
+    LOGV2_DEBUG(
+        5095101, 1, "imported meta data", logAttrs(tenantNs), "metadata"_attr = res.getValue());
+    return {{res.getValue(), ident, tenantNs}};
 }
 
 std::string DurableCatalogImpl::getIndexIdent(OperationContext* opCtx,
@@ -430,7 +433,6 @@ std::shared_ptr<BSONCollectionCatalogEntry::MetaData> DurableCatalogImpl::getMet
 void DurableCatalogImpl::putMetaData(OperationContext* opCtx,
                                      RecordId catalogId,
                                      BSONCollectionCatalogEntry::MetaData& md) {
-    NamespaceString nss(md.ns);
     BSONObj obj = _findEntry(opCtx, catalogId);
 
     {
@@ -461,7 +463,7 @@ void DurableCatalogImpl::putMetaData(OperationContext* opCtx,
                 continue;
             }
             // missing, create new
-            newIdentMap.append(name, _newUniqueIdent(nss, "index"));
+            newIdentMap.append(name, _newUniqueIdent(md.tenantNs, "index"));
         }
         b.append("idxIdent", newIdentMap.obj());
 
@@ -470,7 +472,7 @@ void DurableCatalogImpl::putMetaData(OperationContext* opCtx,
         obj = b.obj();
     }
 
-    if (requiresTimestampForCatalogWrite(opCtx, nss)) {
+    if (requiresTimestampForCatalogWrite(opCtx, md.tenantNs.getNss())) {
         opCtx->recoveryUnit()->setMustBeTimestamped();
     }
 
@@ -481,13 +483,13 @@ void DurableCatalogImpl::putMetaData(OperationContext* opCtx,
 
 Status DurableCatalogImpl::_replaceEntry(OperationContext* opCtx,
                                          RecordId catalogId,
-                                         const NamespaceString& toNss,
+                                         const TenantNamespace& toTenantNs,
                                          BSONCollectionCatalogEntry::MetaData& md) {
     BSONObj old = _findEntry(opCtx, catalogId).getOwned();
     {
         BSONObjBuilder b;
 
-        b.append("ns", toNss.ns());
+        b.append("ns", toTenantNs.toString());
         b.append("md", md.toBSON());
 
         b.appendElementsUnique(old);
@@ -501,16 +503,16 @@ Status DurableCatalogImpl::_replaceEntry(OperationContext* opCtx,
     const auto it = _catalogIdToEntryMap.find(catalogId);
     invariant(it != _catalogIdToEntryMap.end());
 
-    NamespaceString fromName = it->second.nss;
-    it->second.nss = toNss;
+    TenantNamespace fromName = it->second.tenantNs;
+    it->second.tenantNs = toTenantNs;
     opCtx->recoveryUnit()->onRollback([this, catalogId, fromName]() {
         stdx::lock_guard<Latch> lk(_catalogIdToEntryMapLock);
         const auto it = _catalogIdToEntryMap.find(catalogId);
         invariant(it != _catalogIdToEntryMap.end());
-        it->second.nss = fromName;
+        it->second.tenantNs = fromName;
     });
 
-    if (requiresTimestampForCatalogWrite(opCtx, fromName)) {
+    if (requiresTimestampForCatalogWrite(opCtx, fromName.getNss())) {
         opCtx->recoveryUnit()->setMustBeTimestamped();
     }
 
@@ -531,8 +533,8 @@ Status DurableCatalogImpl::_removeEntry(OperationContext* opCtx, RecordId catalo
 
     LOGV2_DEBUG(22212,
                 1,
-                "deleting metadata for {it_second_nss} @ {catalogId}",
-                "it_second_nss"_attr = it->second.nss,
+                "deleting metadata for {it_second_namespace} @ {catalogId}",
+                "it_second_namespace"_attr = it->second.tenantNs,
                 "catalogId"_attr = catalogId);
     _rs->deleteRecord(opCtx, catalogId);
     _catalogIdToEntryMap.erase(it);
@@ -597,8 +599,11 @@ StatusWith<std::string> DurableCatalogImpl::newOrphanedIdent(OperationContext* o
     // The collection will be named local.orphan.xxxxx.
     std::string identNs = ident;
     std::replace(identNs.begin(), identNs.end(), '-', '_');
-    NamespaceString ns = NamespaceString(NamespaceString::kOrphanCollectionDb,
-                                         NamespaceString::kOrphanCollectionPrefix + identNs);
+    // TODO SERVER-62491 Use system tenantId.
+    TenantNamespace tenantNs =
+        TenantNamespace(boost::none,
+                        NamespaceString(NamespaceString::kOrphanCollectionDb,
+                                        NamespaceString::kOrphanCollectionPrefix + identNs));
 
     // Generate a new UUID for the orphaned collection.
     CollectionOptions optionsWithUUID;
@@ -606,10 +611,10 @@ StatusWith<std::string> DurableCatalogImpl::newOrphanedIdent(OperationContext* o
     BSONObj obj;
     {
         BSONObjBuilder b;
-        b.append("ns", ns.ns());
+        b.append("ns", tenantNs.toString());
         b.append("ident", ident);
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = ns.ns();
+        md.tenantNs = tenantNs;
         // Default options with newly generated UUID.
         md.options = optionsWithUUID;
         b.append("md", md.toBSON());
@@ -621,30 +626,30 @@ StatusWith<std::string> DurableCatalogImpl::newOrphanedIdent(OperationContext* o
 
     stdx::lock_guard<Latch> lk(_catalogIdToEntryMapLock);
     invariant(_catalogIdToEntryMap.find(res.getValue()) == _catalogIdToEntryMap.end());
-    _catalogIdToEntryMap[res.getValue()] = Entry(res.getValue(), ident, ns);
+    _catalogIdToEntryMap[res.getValue()] = Entry(res.getValue(), ident, tenantNs);
     opCtx->recoveryUnit()->registerChange(std::make_unique<AddIdentChange>(this, res.getValue()));
 
     LOGV2_DEBUG(22213,
                 1,
                 "stored meta data for orphaned collection {namespace} @ {res_getValue}",
-                logAttrs(ns),
+                logAttrs(tenantNs),
                 "res_getValue"_attr = res.getValue());
-    return {ns.ns()};
+    return {tenantNs.toString()};
 }
 
 StatusWith<std::pair<RecordId, std::unique_ptr<RecordStore>>> DurableCatalogImpl::createCollection(
     OperationContext* opCtx,
-    const NamespaceString& nss,
+    const TenantNamespace& tenantNs,
     const CollectionOptions& options,
     bool allocateDefaultSpace) {
-    invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_IX));
-    invariant(nss.coll().size() > 0);
+    invariant(opCtx->lockState()->isCollectionLockedForMode(tenantNs.getNss(), MODE_IX));
+    invariant(tenantNs.getNss().coll().size() > 0);
 
-    if (CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss)) {
+    if (CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, tenantNs.getNss())) {
         throw WriteConflictException();
     }
 
-    StatusWith<Entry> swEntry = _addEntry(opCtx, nss, options);
+    StatusWith<Entry> swEntry = _addEntry(opCtx, tenantNs, options);
     if (!swEntry.isOK())
         return swEntry.getStatus();
     Entry& entry = swEntry.getValue();
@@ -658,19 +663,19 @@ StatusWith<std::pair<RecordId, std::unique_ptr<RecordStore>>> DurableCatalogImpl
         }
         return KeyFormat::Long;
     }();
-    Status status =
-        _engine->getEngine()->createRecordStore(opCtx, nss.ns(), entry.ident, options, keyFormat);
+    Status status = _engine->getEngine()->createRecordStore(
+        opCtx, tenantNs.getNss().ns(), entry.ident, options, keyFormat);
     if (!status.isOK())
         return status;
 
     auto ru = opCtx->recoveryUnit();
-    UUID uuid = options.uuid.get();
-    opCtx->recoveryUnit()->onRollback([ru, catalog = this, nss, ident = entry.ident, uuid]() {
+    opCtx->recoveryUnit()->onRollback([ru, catalog = this, ident = entry.ident]() {
         // Intentionally ignoring failure
         catalog->_engine->getEngine()->dropIdent(ru, ident).ignore();
     });
 
-    auto rs = _engine->getEngine()->getRecordStore(opCtx, nss.ns(), entry.ident, options);
+    auto rs =
+        _engine->getEngine()->getRecordStore(opCtx, tenantNs.getNss().ns(), entry.ident, options);
     invariant(rs);
 
     return std::pair<RecordId, std::unique_ptr<RecordStore>>(entry.catalogId, std::move(rs));
@@ -696,16 +701,16 @@ Status DurableCatalogImpl::createIndex(OperationContext* opCtx,
 
 StatusWith<DurableCatalog::ImportResult> DurableCatalogImpl::importCollection(
     OperationContext* opCtx,
-    const NamespaceString& nss,
+    const TenantNamespace& tenantNs,
     const BSONObj& metadata,
     const BSONObj& storageMetadata,
     const ImportOptions& importOptions) {
-    invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_X));
-    invariant(nss.coll().size() > 0);
+    invariant(opCtx->lockState()->isCollectionLockedForMode(tenantNs.getNss(), MODE_X));
+    invariant(tenantNs.getNss().coll().size() > 0);
 
     uassert(ErrorCodes::NamespaceExists,
-            str::stream() << "Collection already exists. NS: " << nss,
-            !CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss));
+            str::stream() << "Collection already exists. NS: " << tenantNs.toString(),
+            !CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, tenantNs.getNss()));
 
     BSONCollectionCatalogEntry::MetaData md;
     const BSONElement mdElement = metadata["md"];
@@ -762,7 +767,7 @@ StatusWith<DurableCatalog::ImportResult> DurableCatalogImpl::importCollection(
         }
     }
 
-    StatusWith<Entry> swEntry = _importEntry(opCtx, nss, catalogEntry);
+    StatusWith<Entry> swEntry = _importEntry(opCtx, tenantNs, catalogEntry);
     if (!swEntry.isOK())
         return swEntry.getStatus();
     Entry& entry = swEntry.getValue();
@@ -788,7 +793,8 @@ StatusWith<DurableCatalog::ImportResult> DurableCatalogImpl::importCollection(
         }
     }
 
-    auto rs = _engine->getEngine()->getRecordStore(opCtx, nss.ns(), entry.ident, md.options);
+    auto rs = _engine->getEngine()->getRecordStore(
+        opCtx, tenantNs.getNss().ns(), entry.ident, md.options);
     invariant(rs);
 
     return DurableCatalog::ImportResult(entry.catalogId, std::move(rs), md.options.uuid.get());
@@ -796,9 +802,9 @@ StatusWith<DurableCatalog::ImportResult> DurableCatalogImpl::importCollection(
 
 Status DurableCatalogImpl::renameCollection(OperationContext* opCtx,
                                             RecordId catalogId,
-                                            const NamespaceString& toNss,
+                                            const TenantNamespace& toTenantNs,
                                             BSONCollectionCatalogEntry::MetaData& md) {
-    return _replaceEntry(opCtx, catalogId, toNss, md);
+    return _replaceEntry(opCtx, catalogId, toTenantNs, md);
 }
 
 Status DurableCatalogImpl::dropCollection(OperationContext* opCtx, RecordId catalogId) {
@@ -808,7 +814,7 @@ Status DurableCatalogImpl::dropCollection(OperationContext* opCtx, RecordId cata
         entry = _catalogIdToEntryMap[catalogId];
     }
 
-    invariant(opCtx->lockState()->isCollectionLockedForMode(entry.nss, MODE_X));
+    invariant(opCtx->lockState()->isCollectionLockedForMode(entry.tenantNs.getNss(), MODE_X));
     invariant(getTotalIndexCount(opCtx, catalogId) == 0);
 
     // Remove metadata from mdb_catalog
diff --git a/src/mongo/db/storage/durable_catalog_impl.h b/src/mongo/db/storage/durable_catalog_impl.h
index cf2520449bd..264130a8536 100644
--- a/src/mongo/db/storage/durable_catalog_impl.h
+++ b/src/mongo/db/storage/durable_catalog_impl.h
@@ -105,7 +105,7 @@ public:
 
     StatusWith<std::pair<RecordId, std::unique_ptr<RecordStore>>> createCollection(
         OperationContext* opCtx,
-        const NamespaceString& nss,
+        const TenantNamespace& tenantNs,
         const CollectionOptions& options,
         bool allocateDefaultSpace);
 
@@ -115,14 +115,14 @@ public:
                        const IndexDescriptor* spec);
 
     StatusWith<ImportResult> importCollection(OperationContext* opCtx,
-                                              const NamespaceString& nss,
+                                              const TenantNamespace& tenantNs,
                                               const BSONObj& metadata,
                                               const BSONObj& storageMetadata,
                                               const ImportOptions& importOptions) override;
 
     Status renameCollection(OperationContext* opCtx,
                             RecordId catalogId,
-                            const NamespaceString& toNss,
+                            const TenantNamespace& toTenantNs,
                             BSONCollectionCatalogEntry::MetaData& md);
 
     Status dropCollection(OperationContext* opCtx, RecordId catalogId);
@@ -156,23 +156,23 @@ private:
 
     BSONObj _findEntry(OperationContext* opCtx, RecordId catalogId) const;
     StatusWith<Entry> _addEntry(OperationContext* opCtx,
-                                NamespaceString nss,
+                                TenantNamespace tenantNs,
                                 const CollectionOptions& options);
     StatusWith<Entry> _importEntry(OperationContext* opCtx,
-                                   NamespaceString nss,
+                                   TenantNamespace tenantNs,
                                    const BSONObj& metadata);
     Status _replaceEntry(OperationContext* opCtx,
                          RecordId catalogId,
-                         const NamespaceString& toNss,
+                         const TenantNamespace& toTenantNs,
                          BSONCollectionCatalogEntry::MetaData& md);
     Status _removeEntry(OperationContext* opCtx, RecordId catalogId);
 
     /**
      * Generates a new unique identifier for a new "thing".
-     * @param nss - the containing namespace
+     * @param tenantNs - the containing tenant namespace
      * @param kind - what this "thing" is, likely collection or index
      */
-    std::string _newUniqueIdent(NamespaceString nss, const char* kind);
+    std::string _newUniqueIdent(TenantNamespace tenantNs, const char* kind);
 
     std::string _newInternalIdent(StringData identStem);
 
diff --git a/src/mongo/db/storage/kv/durable_catalog_test.cpp b/src/mongo/db/storage/kv/durable_catalog_test.cpp
index 2f9343d784f..d3d6d8a8de6 100644
--- a/src/mongo/db/storage/kv/durable_catalog_test.cpp
+++ b/src/mongo/db/storage/kv/durable_catalog_test.cpp
@@ -41,11 +41,14 @@
 #include "mongo/db/index/multikey_paths.h"
 #include "mongo/db/index_names.h"
 #include "mongo/db/multitenancy.h"
+#include "mongo/db/multitenancy_gen.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/service_context_test_fixture.h"
 #include "mongo/db/storage/devnull/devnull_kv_engine.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/storage_engine_impl.h"
+#include "mongo/db/tenant_id.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/db/timeseries/timeseries_options.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
@@ -65,12 +68,12 @@ public:
     void setUp() final {
         CatalogTestFixture::setUp();
 
-        _nss = NamespaceString("unittests.durable_catalog");
-        _collectionUUID = createCollection(_nss, CollectionOptions());
+        _tenantNs = TenantNamespace(boost::none, NamespaceString("unittests.durable_catalog"));
+        _collectionUUID = createCollection(_tenantNs, CollectionOptions());
     }
 
-    NamespaceString ns() {
-        return _nss;
+    TenantNamespace tenantNs() {
+        return _tenantNs;
     }
 
     DurableCatalog* getCatalog() {
@@ -87,23 +90,22 @@ public:
             operationContext(), *_collectionUUID, CollectionCatalog::LifetimeMode::kInplace);
     }
 
-    UUID createCollection(const NamespaceString& nss, CollectionOptions options) {
-        Lock::DBLock dbLk(operationContext(), nss.db(), MODE_IX);
-        Lock::CollectionLock collLk(operationContext(), nss, MODE_IX);
+    UUID createCollection(const TenantNamespace& tenantNs, CollectionOptions options) {
+        Lock::DBLock dbLk(operationContext(), tenantNs.getNss().db(), MODE_IX);
+        Lock::CollectionLock collLk(operationContext(), tenantNs.getNss(), MODE_IX);
 
         WriteUnitOfWork wuow(operationContext());
 
         const bool allocateDefaultSpace = true;
         options.uuid = UUID::gen();
 
-        auto swColl =
-            getCatalog()->createCollection(operationContext(), nss, options, allocateDefaultSpace);
+        auto swColl = getCatalog()->createCollection(
+            operationContext(), tenantNs, options, allocateDefaultSpace);
         ASSERT_OK(swColl.getStatus());
 
         std::pair<RecordId, std::unique_ptr<RecordStore>> coll = std::move(swColl.getValue());
         RecordId catalogId = coll.first;
 
-        TenantNamespace tenantNs(getActiveTenant(operationContext()), nss);
         std::shared_ptr<Collection> collection = std::make_shared<CollectionImpl>(
             operationContext(),
             tenantNs,
@@ -123,8 +125,8 @@ public:
     IndexCatalogEntry* createIndex(BSONObj keyPattern,
                                    std::string indexType = IndexNames::BTREE,
                                    bool twoPhase = false) {
-        Lock::DBLock dbLk(operationContext(), _nss.db(), MODE_IX);
-        Lock::CollectionLock collLk(operationContext(), _nss, MODE_X);
+        Lock::DBLock dbLk(operationContext(), _tenantNs.getNss().db(), MODE_IX);
+        Lock::CollectionLock collLk(operationContext(), _tenantNs.getNss(), MODE_X);
 
         std::string indexName = "idx" + std::to_string(_numIndexesCreated);
         // Make sure we have a valid IndexSpec for the type requested
@@ -167,15 +169,15 @@ public:
         ASSERT(match);
     }
 
-    StatusWith<DurableCatalog::ImportResult> importCollectionTest(const NamespaceString& nss,
+    StatusWith<DurableCatalog::ImportResult> importCollectionTest(const TenantNamespace& tenantNs,
                                                                   const BSONObj& metadata) {
-        Lock::DBLock dbLock(operationContext(), nss.db(), MODE_IX);
-        Lock::CollectionLock collLock(operationContext(), nss, MODE_X);
+        Lock::DBLock dbLock(operationContext(), tenantNs.getNss().db(), MODE_IX);
+        Lock::CollectionLock collLock(operationContext(), tenantNs.getNss(), MODE_X);
 
         WriteUnitOfWork wuow(operationContext());
         auto res = getCatalog()->importCollection(
             operationContext(),
-            nss,
+            tenantNs,
             metadata,
             BSON("storage"
                  << "metadata"),
@@ -203,7 +205,7 @@ private:
         return ss.str();
     }
 
-    NamespaceString _nss;
+    TenantNamespace _tenantNs;
 
     size_t _numIndexesCreated = 0;
 
@@ -519,16 +521,17 @@ DEATH_TEST_REGEX_F(DurableCatalogTest,
 TEST_F(DurableCatalogTest, ImportCollection) {
     // Import should fail if the namespace already exists.
     ASSERT_THROWS_CODE(
-        importCollectionTest(ns(), {}), AssertionException, ErrorCodes::NamespaceExists);
+        importCollectionTest(tenantNs(), {}), AssertionException, ErrorCodes::NamespaceExists);
 
-    const auto nss = NamespaceString("unittest.import");
+    const auto tenantNs = TenantNamespace(boost::none, NamespaceString("unittest.import"));
 
     // Import should fail with empty metadata.
-    ASSERT_THROWS_CODE(importCollectionTest(nss, {}), AssertionException, ErrorCodes::BadValue);
+    ASSERT_THROWS_CODE(
+        importCollectionTest(tenantNs, {}), AssertionException, ErrorCodes::BadValue);
 
     BSONCollectionCatalogEntry::MetaData md;
 
-    md.ns = nss.ns();
+    md.tenantNs = tenantNs;
 
     CollectionOptions optionsWithUUID;
     optionsWithUUID.uuid = UUID::gen();
@@ -548,27 +551,29 @@ TEST_F(DurableCatalogTest, ImportCollection) {
     // Import should fail with missing "md" field.
     ASSERT_THROWS_CODE(
         importCollectionTest(
-            nss, BSON("idxIdent" << idxIdentObj << "ns" << nss.ns() << "ident" << ident)),
+            tenantNs,
+            BSON("idxIdent" << idxIdentObj << "ns" << tenantNs.toString() << "ident" << ident)),
         AssertionException,
         ErrorCodes::BadValue);
 
     // Import should fail with missing "ident" field.
-    ASSERT_THROWS_CODE(
-        importCollectionTest(nss,
-                             BSON("md" << mdObj << "idxIdent" << idxIdentObj << "ns" << nss.ns())),
-        AssertionException,
-        ErrorCodes::BadValue);
+    ASSERT_THROWS_CODE(importCollectionTest(tenantNs,
+                                            BSON("md" << mdObj << "idxIdent" << idxIdentObj << "ns"
+                                                      << tenantNs.toString())),
+                       AssertionException,
+                       ErrorCodes::BadValue);
 
     // Import should success with validate inputs.
-    auto swImportResult = importCollectionTest(
-        nss,
-        BSON("md" << mdObj << "idxIdent" << idxIdentObj << "ns" << nss.ns() << "ident" << ident));
+    auto swImportResult =
+        importCollectionTest(tenantNs,
+                             BSON("md" << mdObj << "idxIdent" << idxIdentObj << "ns"
+                                       << tenantNs.toString() << "ident" << ident));
     ASSERT_OK(swImportResult.getStatus());
     DurableCatalog::ImportResult importResult = std::move(swImportResult.getValue());
 
     // Validate the catalog entry for the imported collection.
     auto entry = getCatalog()->getEntry(importResult.catalogId);
-    ASSERT_EQ(entry.nss, nss);
+    ASSERT_EQ(entry.tenantNs, tenantNs);
     ASSERT_EQ(entry.ident, ident);
     ASSERT_EQ(getCatalog()->getIndexIdent(operationContext(), importResult.catalogId, "_id_"),
               idxIdent);
@@ -579,17 +584,17 @@ TEST_F(DurableCatalogTest, ImportCollection) {
     // match.
     md.options.uuid = importResult.uuid;
     ASSERT_BSONOBJ_EQ(getCatalog()->getCatalogEntry(operationContext(), importResult.catalogId),
-                      BSON("md" << md.toBSON() << "idxIdent" << idxIdentObj << "ns" << nss.ns()
-                                << "ident" << ident));
+                      BSON("md" << md.toBSON() << "idxIdent" << idxIdentObj << "ns"
+                                << tenantNs.toString() << "ident" << ident));
 }
 
 TEST_F(DurableCatalogTest, IdentSuffixUsesRand) {
     const std::string rand = "0000000000000000000";
     getCatalog()->setRand_forTest(rand);
 
-    const NamespaceString nss = NamespaceString("a.b");
+    const TenantNamespace tenantNs = TenantNamespace(boost::none, NamespaceString("a.b"));
 
-    auto uuid = createCollection(nss, CollectionOptions());
+    auto uuid = createCollection(tenantNs, CollectionOptions());
     auto collection = CollectionCatalog::get(operationContext())
                           ->lookupCollectionByUUID(operationContext(), uuid);
     RecordId catalogId = collection->getCatalogId();
@@ -604,9 +609,10 @@ TEST_F(DurableCatalogTest, ImportCollectionRandConflict) {
     {
         // Import a collection with the 'rand' suffix as part of the ident. This will force 'rand'
         // to be changed in the durable catalog internals.
-        const auto nss = NamespaceString("unittest.import");
+        const TenantNamespace tenantNs =
+            TenantNamespace(boost::none, NamespaceString("unittest.import"));
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = nss.ns();
+        md.tenantNs = tenantNs;
 
         CollectionOptions optionsWithUUID;
         optionsWithUUID.uuid = UUID::gen();
@@ -624,9 +630,9 @@ TEST_F(DurableCatalogTest, ImportCollectionRandConflict) {
         auto idxIdentObj = BSON("_id_" << idxIdent);
 
         auto swImportResult =
-            importCollectionTest(nss,
-                                 BSON("md" << mdObj << "idxIdent" << idxIdentObj << "ns" << nss.ns()
-                                           << "ident" << ident));
+            importCollectionTest(tenantNs,
+                                 BSON("md" << mdObj << "idxIdent" << idxIdentObj << "ns"
+                                           << tenantNs.toString() << "ident" << ident));
         ASSERT_OK(swImportResult.getStatus());
     }
 
@@ -634,8 +640,8 @@ TEST_F(DurableCatalogTest, ImportCollectionRandConflict) {
 
     {
         // Check that a newly created collection doesn't use 'rand' as the suffix in the ident.
-        const NamespaceString nss = NamespaceString("a.b");
-        createCollection(nss, CollectionOptions());
+        const TenantNamespace tenantNs = TenantNamespace(boost::none, NamespaceString("a.b"));
+        createCollection(tenantNs, CollectionOptions());
 
         RecordId catalogId = getCollection()->getCatalogId();
         ASSERT(!StringData(getCatalog()->getEntry(catalogId).ident).endsWith(rand));
@@ -649,11 +655,13 @@ TEST_F(DurableCatalogTest, CheckTimeseriesBucketsMayHaveMixedSchemaDataFlagFCVLa
     serverGlobalParams.mutableFeatureCompatibility.setVersion(multiversion::GenericFCV::kLatest);
 
     {
-        const NamespaceString regularNss = NamespaceString("test.regular");
-        createCollection(regularNss, CollectionOptions());
+        const TenantNamespace regularTenantNs =
+            TenantNamespace(boost::none, NamespaceString("test.regular"));
+        createCollection(regularTenantNs, CollectionOptions());
 
-        auto collection = CollectionCatalog::get(operationContext())
-                              ->lookupCollectionByNamespace(operationContext(), regularNss);
+        auto collection =
+            CollectionCatalog::get(operationContext())
+                ->lookupCollectionByNamespace(operationContext(), regularTenantNs.getNss());
         RecordId catalogId = collection->getCatalogId();
         ASSERT(!getCatalog()
                     ->getMetaData(operationContext(), catalogId)
@@ -661,13 +669,15 @@ TEST_F(DurableCatalogTest, CheckTimeseriesBucketsMayHaveMixedSchemaDataFlagFCVLa
     }
 
     {
-        const NamespaceString bucketsNss = NamespaceString("system.buckets.ts");
+        const TenantNamespace bucketsTenantNs =
+            TenantNamespace(boost::none, NamespaceString("system.buckets.ts"));
         CollectionOptions options;
         options.timeseries = TimeseriesOptions(/*timeField=*/"t");
-        createCollection(bucketsNss, options);
+        createCollection(bucketsTenantNs, options);
 
-        auto collection = CollectionCatalog::get(operationContext())
-                              ->lookupCollectionByNamespace(operationContext(), bucketsNss);
+        auto collection =
+            CollectionCatalog::get(operationContext())
+                ->lookupCollectionByNamespace(operationContext(), bucketsTenantNs.getNss());
         RecordId catalogId = collection->getCatalogId();
         ASSERT(getCatalog()
                    ->getMetaData(operationContext(), catalogId)
@@ -678,5 +688,41 @@ TEST_F(DurableCatalogTest, CheckTimeseriesBucketsMayHaveMixedSchemaDataFlagFCVLa
     }
 }
 
+TEST_F(DurableCatalogTest, CreateCollectionCatalogEntryHasCorrectTenantNamespace) {
+    gMultitenancySupport = true;
+
+    auto tenantId = TenantId(OID::gen());
+    const TenantNamespace tenantNs = TenantNamespace(tenantId, NamespaceString("test.regular"));
+    createCollection(tenantNs, CollectionOptions());
+
+    auto collection = CollectionCatalog::get(operationContext())
+                          ->lookupCollectionByNamespace(operationContext(), tenantNs.getNss());
+    RecordId catalogId = collection->getCatalogId();
+    ASSERT_EQ(getCatalog()->getEntry(catalogId).tenantNs, tenantNs);
+}
+
+TEST_F(DurableCatalogTest, ImportCollectionCatalogEntryHasCorrectTenantNamespace) {
+    gMultitenancySupport = true;
+
+    auto tenantId = TenantId(OID::gen());
+    const TenantNamespace tenantNs = TenantNamespace(tenantId, NamespaceString("unittest.import"));
+
+    BSONCollectionCatalogEntry::MetaData md;
+    md.tenantNs = tenantNs;
+    CollectionOptions optionsWithUUID;
+    optionsWithUUID.uuid = UUID::gen();
+    md.options = optionsWithUUID;
+    auto mdObj = md.toBSON();
+    const auto ident = "collection-1-1234567891234567899";
+
+    auto swImportResult = importCollectionTest(
+        tenantNs, BSON("md" << mdObj << "ns" << tenantNs.toString() << "ident" << ident));
+    ASSERT_OK(swImportResult.getStatus());
+
+    auto entry = getCatalog()->getEntry(swImportResult.getValue().catalogId);
+    ASSERT_EQ(entry.tenantNs, tenantNs);
+    ASSERT_EQ(entry.ident, ident);
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
index 1b0fb6db685..3b314acec8e 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/catalog/collection_impl.h"
 #include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/multitenancy.h"
 #include "mongo/db/operation_context_noop.h"
 #include "mongo/db/service_context_test_fixture.h"
 #include "mongo/db/storage/durable_catalog_impl.h"
@@ -84,7 +85,8 @@ protected:
                            const CollectionOptions& options,
                            DurableCatalogImpl* catalog) {
         Lock::DBLock dbLk(opCtx, ns.db(), MODE_IX);
-        auto swEntry = catalog->_addEntry(opCtx, ns, options);
+        TenantNamespace tenantNs(boost::none, ns);
+        auto swEntry = catalog->_addEntry(opCtx, tenantNs, options);
         ASSERT_OK(swEntry.getStatus());
         return swEntry.getValue().catalogId;
     }
@@ -1261,7 +1263,7 @@ TEST_F(DurableCatalogImplTest, Idx1) {
         WriteUnitOfWork uow(opCtx);
 
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = "a.b";
+        md.tenantNs = TenantNamespace(boost::none, NamespaceString("a.b"));
 
         BSONCollectionCatalogEntry::IndexMetaData imd;
         imd.spec = BSON("name"
@@ -1295,7 +1297,7 @@ TEST_F(DurableCatalogImplTest, Idx1) {
         WriteUnitOfWork uow(opCtx);
 
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = "a.b";
+        md.tenantNs = TenantNamespace(boost::none, NamespaceString("a.b"));
         putMetaData(opCtx, catalog.get(), catalogId, md);  // remove index
 
         BSONCollectionCatalogEntry::IndexMetaData imd;
@@ -1349,7 +1351,7 @@ TEST_F(DurableCatalogImplTest, DirectoryPerDb1) {
         WriteUnitOfWork uow(opCtx);
 
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = "a.b";
+        md.tenantNs = TenantNamespace(boost::none, NamespaceString("a.b"));
 
         BSONCollectionCatalogEntry::IndexMetaData imd;
         imd.spec = BSON("name"
@@ -1399,7 +1401,7 @@ TEST_F(DurableCatalogImplTest, Split1) {
         WriteUnitOfWork uow(opCtx);
 
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = "a.b";
+        md.tenantNs = TenantNamespace(boost::none, NamespaceString("a.b"));
 
         BSONCollectionCatalogEntry::IndexMetaData imd;
         imd.spec = BSON("name"
@@ -1449,7 +1451,7 @@ TEST_F(DurableCatalogImplTest, DirectoryPerAndSplit1) {
         WriteUnitOfWork uow(opCtx);
 
         BSONCollectionCatalogEntry::MetaData md;
-        md.ns = "a.b";
+        md.tenantNs = TenantNamespace(boost::none, NamespaceString("a.b"));
 
         BSONCollectionCatalogEntry::IndexMetaData imd;
         imd.spec = BSON("name"
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 922e8d55fe6..44afa63a261 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -175,7 +175,7 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
                                "Historical entry",
                                "catalogId"_attr = entry.catalogId,
                                "ident"_attr = entry.ident,
-                               "namespace"_attr = entry.nss);
+                               logAttrs(entry.tenantNs));
         }
     }
 
@@ -246,14 +246,16 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
             // store, drop it from the catalog and skip initializing it by continuing past the
             // following logic.
             if (orphan) {
-                auto status =
-                    _recoverOrphanedCollection(opCtx, entry.catalogId, entry.nss, collectionIdent);
+                // TODO SERVER-62917 pass the TenantNamespace here so it's accessible when creating
+                // the RecordStore in recoverOrphanedIdent.
+                auto status = _recoverOrphanedCollection(
+                    opCtx, entry.catalogId, entry.tenantNs.getNss(), collectionIdent);
                 if (!status.isOK()) {
                     LOGV2_WARNING(22266,
                                   "Failed to recover orphaned data file for collection "
                                   "'{namespace}': {error}",
                                   "Failed to recover orphaned data file for collection",
-                                  "namespace"_attr = entry.nss,
+                                  "namespace"_attr = entry.tenantNs,
                                   "error"_attr = status);
                     WriteUnitOfWork wuow(opCtx);
                     fassert(50716, _catalog->_removeEntry(opCtx, entry.catalogId));
@@ -261,7 +263,7 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
                     if (_options.forRepair) {
                         StorageRepairObserver::get(getGlobalServiceContext())
                             ->invalidatingModification(str::stream()
-                                                       << "Collection " << entry.nss
+                                                       << "Collection " << entry.tenantNs.toString()
                                                        << " dropped: " << status.reason());
                     }
                     wuow.commit();
@@ -270,7 +272,7 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
             }
         }
 
-        if (!entry.nss.isReplicated() &&
+        if (!entry.tenantNs.getNss().isReplicated() &&
             !std::binary_search(identsKnownToStorageEngine.begin(),
                                 identsKnownToStorageEngine.end(),
                                 entry.ident)) {
@@ -285,7 +287,7 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
             LOGV2_INFO(5555201,
                        "Removed unknown unreplicated collection from the catalog",
                        "catalogId"_attr = entry.catalogId,
-                       logAttrs(entry.nss),
+                       logAttrs(entry.tenantNs),
                        "ident"_attr = entry.ident);
             continue;
         }
@@ -315,10 +317,11 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx, LastShutdownState l
                 });
         }
 
-        _initCollection(opCtx, entry.catalogId, entry.nss, _options.forRepair, minVisibleTs);
+        _initCollection(
+            opCtx, entry.catalogId, entry.tenantNs.getNss(), _options.forRepair, minVisibleTs);
 
-        if (entry.nss.isOrphanCollection()) {
-            LOGV2(22248, "Orphaned collection found", "namespace"_attr = entry.nss);
+        if (entry.tenantNs.getNss().isOrphanCollection()) {
+            LOGV2(22248, "Orphaned collection found", logAttrs(entry.tenantNs));
         }
     }
 
@@ -568,7 +571,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
             if (engineIdents.find(entry.ident) == engineIdents.end()) {
                 return {ErrorCodes::UnrecoverableRollbackError,
                         str::stream() << "Expected collection does not exist. Collection: "
-                                      << entry.nss << " Ident: " << entry.ident};
+                                      << entry.tenantNs.toString() << " Ident: " << entry.ident};
             }
         }
     }
@@ -581,7 +584,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
     for (DurableCatalog::Entry entry : catalogEntries) {
         std::shared_ptr<BSONCollectionCatalogEntry::MetaData> metaData =
             _catalog->getMetaData(opCtx, entry.catalogId);
-        NamespaceString coll(metaData->ns);
+        auto tenantNs = metaData->tenantNs;
 
         // Batch up the indexes to remove them from `metaData` outside of the iterator.
         std::vector<std::string> indexesToDrop;
@@ -609,7 +612,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
                     "the index, or rerunning with the --repair option. See "
                     "http://dochub.mongodb.org/core/repair for more information",
                     "index"_attr = indexName,
-                    "namespace"_attr = coll);
+                    logAttrs(tenantNs));
             }
 
             // Two-phase index drop ensures that the underlying data table for an index in the
@@ -626,7 +629,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
                           lastShutdownState == LastShutdownState::kUnclean,
                       str::stream() << "Failed to find an index data table matching " << indexIdent
                                     << " for durable index catalog entry " << indexMetaData.spec
-                                    << " in collection " << coll);
+                                    << " in collection " << tenantNs.toString());
 
             // Any index build with a UUID is an unfinished two-phase build and must be restarted.
             // There are no special cases to handle on primaries or secondaries. An index build may
@@ -641,7 +644,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
 
                 LOGV2(22253,
                       "Found index from unfinished build",
-                      "namespace"_attr = coll,
+                      logAttrs(tenantNs),
                       "uuid"_attr = *collUUID,
                       "index"_attr = indexName,
                       "buildUUID"_attr = buildUUID);
@@ -665,10 +668,10 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
                 LOGV2(22255,
                       "Expected background index build did not complete, rebuilding in foreground "
                       "- see SERVER-43097",
-                      "namespace"_attr = coll,
+                      logAttrs(tenantNs),
                       "index"_attr = indexName);
                 reconcileResult.indexesToRebuild.push_back(
-                    {entry.catalogId, coll, indexName.toString()});
+                    {entry.catalogId, tenantNs.getNss(), indexName.toString()});
                 continue;
             }
 
@@ -681,7 +684,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
             if (!indexMetaData.ready && !indexMetaData.isBackgroundSecondaryBuild) {
                 LOGV2(22256,
                       "Dropping unfinished index",
-                      "namespace"_attr = coll,
+                      logAttrs(tenantNs),
                       "index"_attr = indexName);
                 // Ensure the `ident` is dropped while we have the `indexIdent` value.
                 fassert(50713, _engine->dropIdent(opCtx->recoveryUnit(), indexIdent));
@@ -692,14 +695,14 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
 
         for (auto&& indexName : indexesToDrop) {
             invariant(metaData->eraseIndex(indexName),
-                      str::stream()
-                          << "Index is missing. Collection: " << coll << " Index: " << indexName);
+                      str::stream() << "Index is missing. Collection: " << tenantNs.toString()
+                                    << " Index: " << indexName);
         }
         if (indexesToDrop.size() > 0) {
             WriteUnitOfWork wuow(opCtx);
             auto collection =
                 CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForMetadataWrite(
-                    opCtx, CollectionCatalog::LifetimeMode::kInplace, entry.nss);
+                    opCtx, CollectionCatalog::LifetimeMode::kInplace, entry.tenantNs.getNss());
             invariant(collection->getCatalogId() == entry.catalogId);
             collection->replaceMetadata(opCtx, std::move(metaData));
             wuow.commit();
diff --git a/src/mongo/db/storage/storage_engine_test_fixture.h b/src/mongo/db/storage/storage_engine_test_fixture.h
index 3abdf6b96d3..ae38723188a 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -67,7 +67,7 @@ public:
 
     StatusWith<DurableCatalog::Entry> createCollection(OperationContext* opCtx,
                                                        NamespaceString ns) {
-        TenantNamespace tenantNs(getActiveTenant(opCtx), ns);
+        TenantNamespace tenantNs(boost::none, ns);
         AutoGetDb db(opCtx, ns.db(), LockMode::MODE_X);
         CollectionOptions options;
         options.uuid = UUID::gen();
@@ -76,7 +76,7 @@ public:
         {
             WriteUnitOfWork wuow(opCtx);
             std::tie(catalogId, rs) = unittest::assertGet(
-                _storageEngine->getCatalog()->createCollection(opCtx, ns, options, true));
+                _storageEngine->getCatalog()->createCollection(opCtx, tenantNs, options, true));
             wuow.commit();
         }
         std::shared_ptr<Collection> coll = std::make_shared<CollectionImpl>(
@@ -139,8 +139,9 @@ public:
     bool collectionExists(OperationContext* opCtx, const NamespaceString& nss) {
         std::vector<DurableCatalog::Entry> allCollections =
             _storageEngine->getCatalog()->getAllCatalogEntries(opCtx);
+        TenantNamespace tenantNs(boost::none, nss);
         return std::count_if(allCollections.begin(), allCollections.end(), [&](auto& entry) {
-            return nss == entry.nss;
+            return tenantNs == entry.tenantNs;
         });
     }
 
diff --git a/src/mongo/db/tenant_namespace.cpp b/src/mongo/db/tenant_namespace.cpp
index e482b954703..1c08a6e2357 100644
--- a/src/mongo/db/tenant_namespace.cpp
+++ b/src/mongo/db/tenant_namespace.cpp
@@ -51,12 +51,24 @@ TenantNamespace TenantNamespace::parseTenantNamespaceFromDisk(StringData ns) {
     }
 
     auto tenantDelim = ns.find('_');
-    if (tenantDelim == std::string::npos)
+    auto collDelim = ns.find('.');
+    // If the first '_' is after the '.' that separates the db and coll names, the '_' is part
+    // of the coll name and is not a db prefix.
+    if (tenantDelim == std::string::npos || collDelim < tenantDelim) {
         return TenantNamespace(boost::none, NamespaceString(ns));
+    }
 
     const TenantId tenantId(OID(ns.substr(0, tenantDelim)));
     auto nss = NamespaceString(ns.substr(tenantDelim + 1, ns.size() - 1 - tenantDelim));
     return TenantNamespace(tenantId, nss);
 }
 
+std::ostream& operator<<(std::ostream& stream, const TenantNamespace& tenantNs) {
+    return stream << tenantNs.toString();
+}
+
+StringBuilder& operator<<(StringBuilder& builder, const TenantNamespace& tenantNs) {
+    return builder << tenantNs.toString();
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/tenant_namespace.h b/src/mongo/db/tenant_namespace.h
index 868792d96f6..1e70d99a55a 100644
--- a/src/mongo/db/tenant_namespace.h
+++ b/src/mongo/db/tenant_namespace.h
@@ -44,6 +44,11 @@ namespace mongo {
 
 class TenantNamespace {
 public:
+    TenantNamespace(const TenantNamespace& tenantNs)
+        : _tenantId(tenantNs.tenantId()),
+          _nss(tenantNs.getNss()),
+          _tenantNsStr(tenantNs.toString()) {}
+
     /**
      * Constructs an empty TenantNamespace.
      */
@@ -126,7 +131,7 @@ public:
     }
 
     friend auto logAttrs(const TenantNamespace& nss) {
-        return "tenantNamespace"_attr = nss;
+        return "namespace"_attr = nss;
     }
 
 private:
@@ -135,4 +140,7 @@ private:
     boost::optional<std::string> _tenantNsStr;  // Only set if _tenantId exists
 };
 
+std::ostream& operator<<(std::ostream& stream, const TenantNamespace& tenantNs);
+StringBuilder& operator<<(StringBuilder& builder, const TenantNamespace& tenantNs);
+
 }  // namespace mongo
diff --git a/src/mongo/dbtests/SConscript b/src/mongo/dbtests/SConscript
index 037a9c9178e..9074965c16f 100644
--- a/src/mongo/dbtests/SConscript
+++ b/src/mongo/dbtests/SConscript
@@ -153,6 +153,7 @@ env.Program(
         "$BUILD_DIR/mongo/db/index/skipped_record_tracker",
         "$BUILD_DIR/mongo/db/logical_time_metadata_hook",
         "$BUILD_DIR/mongo/db/mongohasher",
+        "$BUILD_DIR/mongo/db/multitenancy",
         "$BUILD_DIR/mongo/db/op_observer_impl",
         "$BUILD_DIR/mongo/db/query/collation/collator_interface_mock",
         "$BUILD_DIR/mongo/db/query/command_request_response",
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index 9622e723b80..79390ab1f2a 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -84,6 +84,7 @@
 #include "mongo/db/session_catalog_mongod.h"
 #include "mongo/db/storage/snapshot_manager.h"
 #include "mongo/db/storage/storage_engine_impl.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/db/transaction_participant_gen.h"
 #include "mongo/db/vector_clock_mutable.h"
@@ -3263,7 +3264,8 @@ public:
                 << " incorrectly exists before creation. CreateTs: " << systemViewsCreateTs;
 
             systemViewsMd = getMetaDataAtTime(durableCatalog, catalogId, systemViewsCreateTs);
-            ASSERT_EQ(systemViewsNss.ns(), systemViewsMd->ns);
+            TenantNamespace tenantNs = systemViewsMd->tenantNs;
+            ASSERT_EQ(systemViewsNss.ns(), tenantNs.getNss().ns());
 
             assertDocumentAtTimestamp(autoColl.getCollection(), systemViewsCreateTs, BSONObj());
             assertDocumentAtTimestamp(autoColl.getCollection(),