SERVER-63131 Change CollectionCatalog::_orderedCollections to be keyed by std::pair<TenantDatabaseName, UUID>

25 files changed, 175 insertions, 149 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 0a923a52edb..0a7df2c5649 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -88,10 +88,10 @@ env.Library(
         'tenant_database_name.cpp',
     ],
     LIBDEPS=[
-        '$BUILD_DIR/mongo/base',
         'multitenancy_params',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/auth/auth',
         '$BUILD_DIR/mongo/db/auth/security_token',
         '$BUILD_DIR/mongo/idl/feature_flag',
@@ -284,10 +284,8 @@ env.Library(
         'multitenancy.idl',
         'tenant_id.cpp',
     ],
-    LIBDEPS=[
-        '$BUILD_DIR/mongo/base',
-    ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/idl/server_parameter',
     ],
 )
diff --git a/src/mongo/db/catalog/catalog_control.cpp b/src/mongo/db/catalog/catalog_control.cpp
index e9dc0e6691c..f52b4cd0267 100644
--- a/src/mongo/db/catalog/catalog_control.cpp
+++ b/src/mongo/db/catalog/catalog_control.cpp
@@ -68,7 +68,7 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
             23992, 1, "openCatalog: dbholder reopening database", "db"_attr = tenantDbName);
         auto db = databaseHolder->openDb(opCtx, tenantDbName);
         invariant(db, str::stream() << "failed to reopen database " << tenantDbName.toString());
-        for (auto&& collNss : catalog->getAllCollectionNamesFromDb(opCtx, tenantDbName.dbName())) {
+        for (auto&& collNss : catalog->getAllCollectionNamesFromDb(opCtx, tenantDbName)) {
             // Note that the collection name already includes the database component.
             auto collection = catalog->lookupCollectionByNamespaceForMetadataWrite(
                 opCtx, CollectionCatalog::LifetimeMode::kInplace, collNss);
@@ -126,8 +126,7 @@ MinVisibleTimestampMap closeCatalog(OperationContext* opCtx) {
     auto databaseHolder = DatabaseHolder::get(opCtx);
     auto catalog = CollectionCatalog::get(opCtx);
     for (auto&& tenantDbName : allDbs) {
-        for (auto collIt = catalog->begin(opCtx, tenantDbName.dbName());
-             collIt != catalog->end(opCtx);
+        for (auto collIt = catalog->begin(opCtx, tenantDbName); collIt != catalog->end(opCtx);
              ++collIt) {
             auto coll = *collIt;
             if (!coll) {
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index eed78a60f3d..5019ef38dd4 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -261,8 +261,9 @@ public:
                         [collection = std::move(entry.collection)](CollectionCatalog& catalog) {
                             catalog._collections[collection->ns()] = collection;
                             catalog._catalog[collection->uuid()] = collection;
-                            auto dbIdPair = std::make_pair(collection->ns().db().toString(),
-                                                           collection->uuid());
+                            auto dbIdPair =
+                                std::make_pair(collection->tenantNs().createTenantDatabaseName(),
+                                               collection->uuid());
                             catalog._orderedCollections[dbIdPair] = collection;
                         });
                     break;
@@ -317,12 +318,12 @@ private:
 };
 
 CollectionCatalog::iterator::iterator(OperationContext* opCtx,
-                                      StringData dbName,
+                                      const TenantDatabaseName& tenantDbName,
                                       const CollectionCatalog& catalog)
-    : _opCtx(opCtx), _dbName(dbName), _catalog(&catalog) {
+    : _opCtx(opCtx), _tenantDbName(tenantDbName), _catalog(&catalog) {
     auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
 
-    _mapIter = _catalog->_orderedCollections.lower_bound(std::make_pair(_dbName, minUuid));
+    _mapIter = _catalog->_orderedCollections.lower_bound(std::make_pair(_tenantDbName, minUuid));
 
     // Start with the first collection that is visible outside of its transaction.
     while (!_exhausted() && !_mapIter->second->isCommitted()) {
@@ -334,10 +335,10 @@ CollectionCatalog::iterator::iterator(OperationContext* opCtx,
     }
 }
 
-CollectionCatalog::iterator::iterator(
-    OperationContext* opCtx,
-    std::map<std::pair<std::string, UUID>, std::shared_ptr<Collection>>::const_iterator mapIter,
-    const CollectionCatalog& catalog)
+CollectionCatalog::iterator::iterator(OperationContext* opCtx,
+                                      std::map<std::pair<TenantDatabaseName, UUID>,
+                                               std::shared_ptr<Collection>>::const_iterator mapIter,
+                                      const CollectionCatalog& catalog)
     : _opCtx(opCtx), _mapIter(mapIter), _catalog(&catalog) {}
 
 CollectionCatalog::iterator::value_type CollectionCatalog::iterator::operator*() {
@@ -398,7 +399,8 @@ bool CollectionCatalog::iterator::operator!=(const iterator& other) const {
 }
 
 bool CollectionCatalog::iterator::_exhausted() {
-    return _mapIter == _catalog->_orderedCollections.end() || _mapIter->first.first != _dbName;
+    return _mapIter == _catalog->_orderedCollections.end() ||
+        _mapIter->first.first != _tenantDbName;
 }
 
 std::shared_ptr<const CollectionCatalog> CollectionCatalog::get(ServiceContext* svcCtx) {
@@ -864,12 +866,13 @@ bool CollectionCatalog::checkIfCollectionSatisfiable(UUID uuid, CollectionInfoFn
     return predicate(collection.get());
 }
 
-std::vector<UUID> CollectionCatalog::getAllCollectionUUIDsFromDb(StringData dbName) const {
+std::vector<UUID> CollectionCatalog::getAllCollectionUUIDsFromDb(
+    const TenantDatabaseName& tenantDbName) const {
     auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
-    auto it = _orderedCollections.lower_bound(std::make_pair(dbName.toString(), minUuid));
+    auto it = _orderedCollections.lower_bound(std::make_pair(tenantDbName, minUuid));
 
     std::vector<UUID> ret;
-    while (it != _orderedCollections.end() && it->first.first == dbName) {
+    while (it != _orderedCollections.end() && it->first.first == tenantDbName) {
         if (it->second->isCommitted()) {
             ret.push_back(it->first.second);
         }
@@ -879,14 +882,14 @@ std::vector<UUID> CollectionCatalog::getAllCollectionUUIDsFromDb(StringData dbNa
 }
 
 std::vector<NamespaceString> CollectionCatalog::getAllCollectionNamesFromDb(
-    OperationContext* opCtx, StringData dbName) const {
-    invariant(opCtx->lockState()->isDbLockedForMode(dbName, MODE_S));
+    OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const {
+    invariant(opCtx->lockState()->isDbLockedForMode(tenantDbName.dbName(), MODE_S));
 
     auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
 
     std::vector<NamespaceString> ret;
-    for (auto it = _orderedCollections.lower_bound(std::make_pair(dbName.toString(), minUuid));
-         it != _orderedCollections.end() && it->first.first == dbName;
+    for (auto it = _orderedCollections.lower_bound(std::make_pair(tenantDbName, minUuid));
+         it != _orderedCollections.end() && it->first.first == tenantDbName;
          ++it) {
         if (it->second->isCommitted()) {
             ret.push_back(it->second->ns());
@@ -898,21 +901,20 @@ std::vector<NamespaceString> CollectionCatalog::getAllCollectionNamesFromDb(
 std::vector<TenantDatabaseName> CollectionCatalog::getAllDbNames() const {
     std::vector<TenantDatabaseName> ret;
     auto maxUuid = UUID::parse("FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF").getValue();
-    auto iter = _orderedCollections.upper_bound(std::make_pair("", maxUuid));
+    auto iter = _orderedCollections.upper_bound(std::make_pair(TenantDatabaseName(), maxUuid));
     while (iter != _orderedCollections.end()) {
-        auto dbName = iter->first.first;
+        auto tenantDbName = iter->first.first;
         if (iter->second->isCommitted()) {
-            // TODO SERVER-61988: Once iter->first.first has TenantDatabaseName object, we need not
-            // create the TenantDatabaseName here.
-            ret.push_back(TenantDatabaseName(boost::none, dbName));
+            ret.push_back(tenantDbName);
         } else {
-            // If the first collection found for `dbName` is not yet committed, increment the
-            // iterator to find the next visible collection (possibly under a different `dbName`).
+            // If the first collection found for `tenantDbName` is not yet committed, increment the
+            // iterator to find the next visible collection (possibly under a different
+            // `tenantDbName`).
             iter++;
             continue;
         }
-        // Move on to the next database after `dbName`.
-        iter = _orderedCollections.upper_bound(std::make_pair(dbName, maxUuid));
+        // Move on to the next database after `tenantDbName`.
+        iter = _orderedCollections.upper_bound(std::make_pair(tenantDbName, maxUuid));
     }
     return ret;
 }
@@ -982,8 +984,7 @@ void CollectionCatalog::registerCollection(OperationContext* opCtx,
                 logAttrs(tenantNs),
                 "uuid"_attr = uuid);
 
-    auto dbName = tenantDbName.dbName();
-    auto dbIdPair = std::make_pair(dbName, uuid);
+    auto dbIdPair = std::make_pair(tenantDbName, uuid);
 
     // Make sure no entry related to this uuid.
     invariant(_catalog.find(uuid) == _catalog.end());
@@ -1004,8 +1005,9 @@ void CollectionCatalog::registerCollection(OperationContext* opCtx,
 
     invariant(static_cast<size_t>(_stats.internal + _stats.userCollections) == _collections.size());
 
-    auto dbRid = ResourceId(RESOURCE_DATABASE, dbName);
-    addResource(dbRid, dbName);
+    // TODO SERVER-62918 create ResourceId for db with TenantDatabaseName.
+    auto dbRid = ResourceId(RESOURCE_DATABASE, tenantDbName.dbName());
+    addResource(dbRid, tenantDbName.dbName());
 
     auto collRid = ResourceId(RESOURCE_COLLECTION, tenantNs.getNss().ns());
     addResource(collRid, tenantNs.getNss().ns());
@@ -1017,8 +1019,8 @@ std::shared_ptr<Collection> CollectionCatalog::deregisterCollection(OperationCon
 
     auto coll = std::move(_catalog[uuid]);
     auto ns = coll->ns();
-    auto dbName = ns.db().toString();
-    auto dbIdPair = std::make_pair(dbName, uuid);
+    auto tenantDbName = coll->tenantNs().createTenantDatabaseName();
+    auto dbIdPair = std::make_pair(tenantDbName, uuid);
 
     LOGV2_DEBUG(20281, 1, "Deregistering collection", logAttrs(ns), "uuid"_attr = uuid);
 
@@ -1054,8 +1056,6 @@ void CollectionCatalog::deregisterAllCollectionsAndViews() {
     for (auto& entry : _catalog) {
         auto uuid = entry.first;
         auto ns = entry.second->ns();
-        auto dbName = ns.db().toString();
-        auto dbIdPair = std::make_pair(dbName, uuid);
 
         LOGV2_DEBUG(20283, 1, "Deregistering collection", logAttrs(ns), "uuid"_attr = uuid);
 
@@ -1103,8 +1103,9 @@ void CollectionCatalog::replaceViewsForDatabase(const TenantDatabaseName& tenant
     }
 }
 
-CollectionCatalog::iterator CollectionCatalog::begin(OperationContext* opCtx, StringData db) const {
-    return iterator(opCtx, db, *this);
+CollectionCatalog::iterator CollectionCatalog::begin(OperationContext* opCtx,
+                                                     const TenantDatabaseName& tenantDbName) const {
+    return iterator(opCtx, tenantDbName, *this);
 }
 
 CollectionCatalog::iterator CollectionCatalog::end(OperationContext* opCtx) const {
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index 19214795bd3..c6198e39290 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -69,10 +69,12 @@ public:
     public:
         using value_type = CollectionPtr;
 
-        iterator(OperationContext* opCtx, StringData dbName, const CollectionCatalog& catalog);
         iterator(OperationContext* opCtx,
-                 std::map<std::pair<std::string, UUID>, std::shared_ptr<Collection>>::const_iterator
-                     mapIter,
+                 const TenantDatabaseName& tenantDbName,
+                 const CollectionCatalog& catalog);
+        iterator(OperationContext* opCtx,
+                 std::map<std::pair<TenantDatabaseName, UUID>,
+                          std::shared_ptr<Collection>>::const_iterator mapIter,
                  const CollectionCatalog& catalog);
         value_type operator*();
         iterator operator++();
@@ -92,9 +94,9 @@ public:
         bool _exhausted();
 
         OperationContext* _opCtx;
-        std::string _dbName;
+        TenantDatabaseName _tenantDbName;
         boost::optional<UUID> _uuid;
-        std::map<std::pair<std::string, UUID>, std::shared_ptr<Collection>>::const_iterator
+        std::map<std::pair<TenantDatabaseName, UUID>, std::shared_ptr<Collection>>::const_iterator
             _mapIter;
         const CollectionCatalog* _catalog;
     };
@@ -285,25 +287,25 @@ public:
     bool checkIfCollectionSatisfiable(UUID uuid, CollectionInfoFn predicate) const;
 
     /**
-     * This function gets the UUIDs of all collections from `dbName`.
+     * This function gets the UUIDs of all collections from `tenantDbName`.
      *
      * If the caller does not take a strong database lock, some of UUIDs might no longer exist (due
      * to collection drop) after this function returns.
      *
-     * Returns empty vector if the 'dbName' is not known.
+     * Returns empty vector if the 'tenantDbName' is not known.
      */
-    std::vector<UUID> getAllCollectionUUIDsFromDb(StringData dbName) const;
+    std::vector<UUID> getAllCollectionUUIDsFromDb(const TenantDatabaseName& tenantDbName) const;
 
     /**
-     * This function gets the ns of all collections from `dbName`. The result is not sorted.
+     * This function gets the ns of all collections from `tenantDbName`. The result is not sorted.
      *
      * Caller must take a strong database lock; otherwise, collections returned could be dropped or
      * renamed.
      *
-     * Returns empty vector if the 'dbName' is not known.
+     * Returns empty vector if the 'tenantDbName' is not known.
      */
-    std::vector<NamespaceString> getAllCollectionNamesFromDb(OperationContext* opCtx,
-                                                             StringData dbName) const;
+    std::vector<NamespaceString> getAllCollectionNamesFromDb(
+        OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const;
 
     /**
      * This functions gets all the database names. The result is sorted in alphabetical ascending
@@ -394,7 +396,7 @@ public:
      */
     uint64_t getEpoch() const;
 
-    iterator begin(OperationContext* opCtx, StringData db) const;
+    iterator begin(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const;
     iterator end(OperationContext* opCtx) const;
 
     /**
@@ -428,13 +430,13 @@ private:
 
     using CollectionCatalogMap = stdx::unordered_map<UUID, std::shared_ptr<Collection>, UUID::Hash>;
     using OrderedCollectionMap =
-        std::map<std::pair<std::string, UUID>, std::shared_ptr<Collection>>;
+        std::map<std::pair<TenantDatabaseName, UUID>, std::shared_ptr<Collection>>;
     using NamespaceCollectionMap =
         stdx::unordered_map<NamespaceString, std::shared_ptr<Collection>>;
     using DatabaseProfileSettingsMap = StringMap<ProfileSettings>;
 
     CollectionCatalogMap _catalog;
-    OrderedCollectionMap _orderedCollections;  // Ordered by <dbName, collUUID> pair
+    OrderedCollectionMap _orderedCollections;  // Ordered by <tenantDbName, collUUID> pair
     NamespaceCollectionMap _collections;
 
     // Map of database names to a set of their views. Only databases with views are present.
diff --git a/src/mongo/db/catalog/collection_catalog_helper.cpp b/src/mongo/db/catalog/collection_catalog_helper.cpp
index 60e0d125973..50c78212e58 100644
--- a/src/mongo/db/catalog/collection_catalog_helper.cpp
+++ b/src/mongo/db/catalog/collection_catalog_helper.cpp
@@ -39,13 +39,13 @@ MONGO_FAIL_POINT_DEFINE(hangBeforeGettingNextCollection);
 namespace catalog {
 
 void forEachCollectionFromDb(OperationContext* opCtx,
-                             StringData dbName,
+                             const TenantDatabaseName& tenantDbName,
                              LockMode collLockMode,
                              CollectionCatalog::CollectionInfoFn callback,
                              CollectionCatalog::CollectionInfoFn predicate) {
 
     auto catalogForIteration = CollectionCatalog::get(opCtx);
-    for (auto collectionIt = catalogForIteration->begin(opCtx, dbName);
+    for (auto collectionIt = catalogForIteration->begin(opCtx, tenantDbName);
          collectionIt != catalogForIteration->end(opCtx);
          ++collectionIt) {
         auto uuid = collectionIt.uuid().get();
diff --git a/src/mongo/db/catalog/collection_catalog_helper.h b/src/mongo/db/catalog/collection_catalog_helper.h
index aa5756e7baf..f45b0f54a45 100644
--- a/src/mongo/db/catalog/collection_catalog_helper.h
+++ b/src/mongo/db/catalog/collection_catalog_helper.h
@@ -53,7 +53,7 @@ namespace catalog {
  * Iterating through the remaining collections stops when the callback returns false.
  */
 void forEachCollectionFromDb(OperationContext* opCtx,
-                             StringData dbName,
+                             const TenantDatabaseName& tenantDbName,
                              LockMode collLockMode,
                              CollectionCatalog::CollectionInfoFn callback,
                              CollectionCatalog::CollectionInfoFn predicate = nullptr);
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 657ae0a610c..db498e2aa6d 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -127,12 +127,13 @@ public:
         return it->second.end();
     }
 
-    void checkCollections(std::string dbName) {
+    void checkCollections(const TenantDatabaseName& tenantDbName) {
         unsigned long counter = 0;
 
-        for (auto [orderedIt, catalogIt] =
-                 std::tuple{collsIterator(dbName), catalog.begin(&opCtx, dbName)};
-             catalogIt != catalog.end(&opCtx) && orderedIt != collsIteratorEnd(dbName);
+        for (auto [orderedIt, catalogIt] = std::tuple{collsIterator(tenantDbName.dbName()),
+                                                      catalog.begin(&opCtx, tenantDbName)};
+             catalogIt != catalog.end(&opCtx) &&
+             orderedIt != collsIteratorEnd(tenantDbName.dbName());
              ++catalogIt, ++orderedIt) {
 
             auto catalogColl = *catalogIt;
@@ -142,7 +143,7 @@ public:
             ++counter;
         }
 
-        ASSERT_EQUALS(counter, dbMap[dbName].size());
+        ASSERT_EQUALS(counter, dbMap[tenantDbName.dbName()].size());
     }
 
     void dropColl(const std::string dbName, UUID uuid) {
@@ -286,7 +287,9 @@ public:
         }
 
         int numEntries = 0;
-        for (auto it = catalog.begin(&opCtx, "resourceDb"); it != catalog.end(&opCtx); it++) {
+        for (auto it = catalog.begin(&opCtx, TenantDatabaseName(boost::none, "resourceDb"));
+             it != catalog.end(&opCtx);
+             it++) {
             auto coll = *it;
             std::string collName = coll->ns().ns();
             ResourceId rid(RESOURCE_COLLECTION, collName);
@@ -299,7 +302,9 @@ public:
 
     void tearDown() {
         std::vector<UUID> collectionsToDeregister;
-        for (auto it = catalog.begin(&opCtx, "resourceDb"); it != catalog.end(&opCtx); ++it) {
+        for (auto it = catalog.begin(&opCtx, TenantDatabaseName(boost::none, "resourceDb"));
+             it != catalog.end(&opCtx);
+             ++it) {
             auto coll = *it;
             auto uuid = coll->uuid();
             if (!coll) {
@@ -314,7 +319,9 @@ public:
         }
 
         int numEntries = 0;
-        for (auto it = catalog.begin(&opCtx, "resourceDb"); it != catalog.end(&opCtx); it++) {
+        for (auto it = catalog.begin(&opCtx, TenantDatabaseName(boost::none, "resourceDb"));
+             it != catalog.end(&opCtx);
+             it++) {
             numEntries++;
         }
         ASSERT_EQ(0, numEntries);
@@ -380,18 +387,18 @@ TEST_F(CollectionCatalogResourceTest, RemoveCollection) {
 // Create an iterator over the CollectionCatalog and assert that all collections are present.
 // Iteration ends when the end of the catalog is reached.
 TEST_F(CollectionCatalogIterationTest, EndAtEndOfCatalog) {
-    checkCollections("foo");
+    checkCollections(TenantDatabaseName(boost::none, "foo"));
 }
 
 // Create an iterator over the CollectionCatalog and test that all collections are present.
 // Iteration ends
 // when the end of a database-specific section of the catalog is reached.
 TEST_F(CollectionCatalogIterationTest, EndAtEndOfSection) {
-    checkCollections("bar");
+    checkCollections(TenantDatabaseName(boost::none, "bar"));
 }
 
 TEST_F(CollectionCatalogIterationTest, GetUUIDWontRepositionEvenIfEntryIsDropped) {
-    auto it = catalog.begin(&opCtx, "bar");
+    auto it = catalog.begin(&opCtx, TenantDatabaseName(boost::none, "bar"));
     auto collsIt = collsIterator("bar");
     auto uuid = collsIt->first;
     catalog.deregisterCollection(&opCtx, uuid);
@@ -569,7 +576,8 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
     std::vector<NamespaceString> dCollList = {d1Coll, d2Coll, d3Coll};
 
     Lock::DBLock dbLock(opCtx.get(), "dbD", MODE_S);
-    auto res = catalog.getAllCollectionNamesFromDb(opCtx.get(), "dbD");
+    auto res =
+        catalog.getAllCollectionNamesFromDb(opCtx.get(), TenantDatabaseName(boost::none, "dbD"));
     std::sort(res.begin(), res.end());
     ASSERT(res == dCollList);
 
@@ -628,7 +636,8 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommitt
     invisibleCollA->setCommitted(false);
 
     Lock::DBLock dbLock(opCtx.get(), "dbA", MODE_S);
-    auto res = catalog.getAllCollectionNamesFromDb(opCtx.get(), "dbA");
+    auto res =
+        catalog.getAllCollectionNamesFromDb(opCtx.get(), TenantDatabaseName(boost::none, "dbA"));
     ASSERT(res.empty());
 
     std::vector<TenantDatabaseName> tenantDbNames = {TenantDatabaseName(boost::none, "dbB"),
@@ -650,7 +659,8 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommitt
         invisibleCollD->setCommitted(false);
 
         Lock::DBLock dbLock(opCtx.get(), "dbD", MODE_S);
-        res = catalog.getAllCollectionNamesFromDb(opCtx.get(), "dbD");
+        res = catalog.getAllCollectionNamesFromDb(opCtx.get(),
+                                                  TenantDatabaseName(boost::none, "dbD"));
         std::sort(res.begin(), res.end());
         ASSERT(res == dCollList);
 
@@ -699,11 +709,14 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDb) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db", MODE_IX);
         int numCollectionsTraversed = 0;
-        catalog::forEachCollectionFromDb(opCtx, "db", MODE_X, [&](const CollectionPtr& collection) {
-            ASSERT_TRUE(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
-            numCollectionsTraversed++;
-            return true;
-        });
+        const TenantDatabaseName tenantDbName(boost::none, "db");
+        catalog::forEachCollectionFromDb(
+            opCtx, tenantDbName, MODE_X, [&](const CollectionPtr& collection) {
+                ASSERT_TRUE(
+                    opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
+                numCollectionsTraversed++;
+                return true;
+            });
 
         ASSERT_EQUALS(numCollectionsTraversed, 3);
     }
@@ -711,8 +724,9 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDb) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db2", MODE_IX);
         int numCollectionsTraversed = 0;
+        const TenantDatabaseName tenantDbName(boost::none, "db2");
         catalog::forEachCollectionFromDb(
-            opCtx, "db2", MODE_IS, [&](const CollectionPtr& collection) {
+            opCtx, tenantDbName, MODE_IS, [&](const CollectionPtr& collection) {
                 ASSERT_TRUE(
                     opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_IS));
                 numCollectionsTraversed++;
@@ -725,8 +739,9 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDb) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db3", MODE_IX);
         int numCollectionsTraversed = 0;
+        const TenantDatabaseName tenantDbName(boost::none, "db3");
         catalog::forEachCollectionFromDb(
-            opCtx, "db3", MODE_S, [&](const CollectionPtr& collection) {
+            opCtx, tenantDbName, MODE_S, [&](const CollectionPtr& collection) {
                 numCollectionsTraversed++;
                 return true;
             });
@@ -742,9 +757,10 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db", MODE_IX);
         int numCollectionsTraversed = 0;
+        const TenantDatabaseName tenantDbName(boost::none, "db");
         catalog::forEachCollectionFromDb(
             opCtx,
-            "db",
+            tenantDbName,
             MODE_X,
             [&](const CollectionPtr& collection) {
                 ASSERT_TRUE(
@@ -764,9 +780,10 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
     {
         auto dbLock = std::make_unique<Lock::DBLock>(opCtx, "db", MODE_IX);
         int numCollectionsTraversed = 0;
+        const TenantDatabaseName tenantDbName(boost::none, "db");
         catalog::forEachCollectionFromDb(
             opCtx,
-            "db",
+            tenantDbName,
             MODE_IX,
             [&](const CollectionPtr& collection) {
                 ASSERT_TRUE(
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index 8115cd59fc6..bdca7f4e0bb 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -170,7 +170,7 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
     // block.
     lk.unlock();
 
-    if (CollectionCatalog::get(opCtx)->getAllCollectionUUIDsFromDb(tenantDbName.dbName()).empty()) {
+    if (CollectionCatalog::get(opCtx)->getAllCollectionUUIDsFromDb(tenantDbName).empty()) {
         audit::logCreateDatabase(opCtx->getClient(), tenantDbName.dbName());
         if (justCreated)
             *justCreated = true;
@@ -238,8 +238,7 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
     invariant(opCtx->lockState()->isDbLockedForMode(name.dbName(), MODE_X));
 
     auto catalog = CollectionCatalog::get(opCtx);
-    for (auto collIt = catalog->begin(opCtx, name.dbName()); collIt != catalog->end(opCtx);
-         ++collIt) {
+    for (auto collIt = catalog->begin(opCtx, name); collIt != catalog->end(opCtx); ++collIt) {
         auto coll = *collIt;
         if (!coll) {
             break;
@@ -255,8 +254,7 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
 
     auto const serviceContext = opCtx->getServiceContext();
 
-    for (auto collIt = catalog->begin(opCtx, name.dbName()); collIt != catalog->end(opCtx);
-         ++collIt) {
+    for (auto collIt = catalog->begin(opCtx, name); collIt != catalog->end(opCtx); ++collIt) {
         auto coll = *collIt;
         if (!coll) {
             break;
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 922f7068f72..4df95f05e84 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -173,7 +173,7 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
     }
 
     auto catalog = CollectionCatalog::get(opCtx);
-    for (const auto& uuid : catalog->getAllCollectionUUIDsFromDb(_name.dbName())) {
+    for (const auto& uuid : catalog->getAllCollectionUUIDsFromDb(_name)) {
         CollectionWriter collection(
             opCtx,
             uuid,
@@ -300,7 +300,7 @@ void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
         return collection->getCollectionOptions().temp;
     };
 
-    catalog::forEachCollectionFromDb(opCtx, name().dbName(), MODE_X, callback, predicate);
+    catalog::forEachCollectionFromDb(opCtx, name(), MODE_X, callback, predicate);
 }
 
 void DatabaseImpl::setDropPending(OperationContext* opCtx, bool dropPending) {
@@ -332,7 +332,7 @@ void DatabaseImpl::getStats(OperationContext* opCtx,
     invariant(opCtx->lockState()->isDbLockedForMode(name().dbName(), MODE_IS));
 
     catalog::forEachCollectionFromDb(
-        opCtx, name().dbName(), MODE_IS, [&](const CollectionPtr& collection) -> bool {
+        opCtx, name(), MODE_IS, [&](const CollectionPtr& collection) -> bool {
             nCollections += 1;
             objects += collection->numRecords(opCtx);
             size += collection->dataSize(opCtx);
@@ -924,7 +924,7 @@ void DatabaseImpl::checkForIdIndexesAndDropPendingCollections(OperationContext*
     }
 
     auto catalog = CollectionCatalog::get(opCtx);
-    for (const auto& nss : catalog->getAllCollectionNamesFromDb(opCtx, _name.dbName())) {
+    for (const auto& nss : catalog->getAllCollectionNamesFromDb(opCtx, _name)) {
         if (nss.isDropPendingNamespace()) {
             auto dropOpTime = fassert(40459, nss.getDropPendingNamespaceOpTime());
             LOGV2(20321,
diff --git a/src/mongo/db/catalog/drop_database.cpp b/src/mongo/db/catalog/drop_database.cpp
index 5131dd52b3c..153f986a1d1 100644
--- a/src/mongo/db/catalog/drop_database.cpp
+++ b/src/mongo/db/catalog/drop_database.cpp
@@ -222,8 +222,7 @@ Status _dropDatabase(OperationContext* opCtx, const std::string& dbName, bool ab
 
         std::vector<NamespaceString> collectionsToDrop;
         auto catalog = CollectionCatalog::get(opCtx);
-        for (auto collIt = catalog->begin(opCtx, db->name().dbName());
-             collIt != catalog->end(opCtx);
+        for (auto collIt = catalog->begin(opCtx, db->name()); collIt != catalog->end(opCtx);
              ++collIt) {
             auto collection = *collIt;
             if (!collection) {
diff --git a/src/mongo/db/commands/dbcheck.cpp b/src/mongo/db/commands/dbcheck.cpp
index a1e44d16ff4..291b79740ac 100644
--- a/src/mongo/db/commands/dbcheck.cpp
+++ b/src/mongo/db/commands/dbcheck.cpp
@@ -222,7 +222,9 @@ std::unique_ptr<DbCheckRun> fullDatabaseRun(OperationContext* opCtx,
         result->push_back(info);
         return true;
     };
-    mongo::catalog::forEachCollectionFromDb(opCtx, dbName, MODE_IS, perCollectionWork);
+    // TODO SERVER-63353: Change dbcheck command to use TenantDatabaseName
+    mongo::catalog::forEachCollectionFromDb(
+        opCtx, TenantDatabaseName(boost::none, dbName), MODE_IS, perCollectionWork);
 
     return result;
 }
diff --git a/src/mongo/db/commands/dbhash.cpp b/src/mongo/db/commands/dbhash.cpp
index df4f5f5b4dd..2096551bce8 100644
--- a/src/mongo/db/commands/dbhash.cpp
+++ b/src/mongo/db/commands/dbhash.cpp
@@ -232,8 +232,9 @@ public:
         std::set<std::string> cappedCollectionSet;
 
         bool noError = true;
+        const TenantDatabaseName tenantDbName(boost::none, dbname);
         catalog::forEachCollectionFromDb(
-            opCtx, dbname, MODE_IS, [&](const CollectionPtr& collection) {
+            opCtx, tenantDbName, MODE_IS, [&](const CollectionPtr& collection) {
                 auto collNss = collection->ns();
 
                 if (collNss.size() - 1 <= dbname.size()) {
diff --git a/src/mongo/db/commands/feature_compatibility_version.cpp b/src/mongo/db/commands/feature_compatibility_version.cpp
index 09751b99db6..5adcdffef21 100644
--- a/src/mongo/db/commands/feature_compatibility_version.cpp
+++ b/src/mongo/db/commands/feature_compatibility_version.cpp
@@ -383,7 +383,7 @@ bool FeatureCompatibilityVersion::hasNoReplicatedCollections(OperationContext* o
     auto catalog = CollectionCatalog::get(opCtx);
     for (auto&& tenantDbName : tenantDbNames) {
         Lock::DBLock dbLock(opCtx, tenantDbName.dbName(), MODE_S);
-        for (auto&& collNss : catalog->getAllCollectionNamesFromDb(opCtx, tenantDbName.dbName())) {
+        for (auto&& collNss : catalog->getAllCollectionNamesFromDb(opCtx, tenantDbName)) {
             if (collNss.isReplicated()) {
                 return false;
             }
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index 89d6c775d05..741e19ba454 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/exec/working_set.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/list_collections_gen.h"
+#include "mongo/db/multitenancy.h"
 #include "mongo/db/query/cursor_request.h"
 #include "mongo/db/query/cursor_response.h"
 #include "mongo/db/query/find_common.h"
@@ -307,6 +308,7 @@ public:
 
             const auto listCollRequest = request();
             const auto dbName = listCollRequest.getDbName();
+            const TenantDatabaseName tenantDbName(getActiveTenant(opCtx), dbName);
             const bool nameOnly = listCollRequest.getNameOnly();
             const bool authorizedCollections = listCollRequest.getAuthorizedCollections();
 
@@ -330,7 +332,6 @@ public:
                 // Acquire only the global lock and set up a consistent in-memory catalog and
                 // storage snapshot.
                 AutoGetDbForReadMaybeLockFree lockFreeReadBlock(opCtx, dbName);
-                const TenantDatabaseName tenantDbName(boost::none, dbName);
                 auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
 
                 CurOpFailpointHelpers::waitWhileFailPointEnabled(&hangBeforeListCollections,
@@ -435,14 +436,14 @@ public:
                         // needing to yield as we don't take any locks.
                         if (opCtx->isLockFreeReadsOp()) {
                             auto collectionCatalog = CollectionCatalog::get(opCtx);
-                            for (auto it = collectionCatalog->begin(opCtx, dbName);
+                            for (auto it = collectionCatalog->begin(opCtx, tenantDbName);
                                  it != collectionCatalog->end(opCtx);
                                  ++it) {
                                 perCollectionWork(*it);
                             }
                         } else {
                             mongo::catalog::forEachCollectionFromDb(
-                                opCtx, dbName, MODE_IS, perCollectionWork);
+                                opCtx, tenantDbName, MODE_IS, perCollectionWork);
                         }
                     }
 
diff --git a/src/mongo/db/commands/list_databases.cpp b/src/mongo/db/commands/list_databases.cpp
index ccd34311ee1..d0c708cd145 100644
--- a/src/mongo/db/commands/list_databases.cpp
+++ b/src/mongo/db/commands/list_databases.cpp
@@ -166,7 +166,7 @@ public:
                     });
                     item.setSizeOnDisk(size);
                     item.setEmpty(CollectionCatalog::get(opCtx)
-                                      ->getAllCollectionUUIDsFromDb(tenantDbName.dbName())
+                                      ->getAllCollectionUUIDsFromDb(tenantDbName)
                                       .empty());
                 }
                 if (!filter || filter->matchesBSON(item.toBSON())) {
diff --git a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
index b61878d5233..75a38e8a472 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -494,7 +494,7 @@ private:
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
                     opCtx,
-                    dbName,
+                    tenantDbName,
                     MODE_X,
                     [&](const CollectionPtr& collection) {
                         if (collection->getTimeseriesBucketsMayHaveMixedSchemaData()) {
@@ -609,7 +609,7 @@ private:
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
                     opCtx,
-                    dbName,
+                    tenantDbName,
                     MODE_X,
                     [&](const CollectionPtr& collection) {
                         invariant(collection->getTimeseriesOptions());
@@ -696,7 +696,7 @@ private:
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
                     opCtx,
-                    dbName,
+                    tenantDbName,
                     MODE_X,
                     [&](const CollectionPtr& collection) {
                         // Fail to downgrade if there exists a collection with
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index fe481cb4880..6af54a47d11 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -150,7 +150,7 @@ public:
                                          });
                 }
 
-                for (auto collIt = collectionCatalog->begin(opCtx, tenantDbName.dbName());
+                for (auto collIt = collectionCatalog->begin(opCtx, tenantDbName);
                      collIt != collectionCatalog->end(opCtx);
                      ++collIt) {
                     if (!_validateNamespace(
diff --git a/src/mongo/db/repair.cpp b/src/mongo/db/repair.cpp
index 7332440372b..c889bb7cbc0 100644
--- a/src/mongo/db/repair.cpp
+++ b/src/mongo/db/repair.cpp
@@ -115,8 +115,8 @@ Status dropUnfinishedIndexes(OperationContext* opCtx, Collection* collection) {
 
 Status repairCollections(OperationContext* opCtx,
                          StorageEngine* engine,
-                         const std::string& dbName) {
-    auto colls = CollectionCatalog::get(opCtx)->getAllCollectionNamesFromDb(opCtx, dbName);
+                         const TenantDatabaseName& tenantDbName) {
+    auto colls = CollectionCatalog::get(opCtx)->getAllCollectionNamesFromDb(opCtx, tenantDbName);
 
     for (const auto& nss : colls) {
         auto status = repair::repairCollection(opCtx, engine, nss);
@@ -150,7 +150,7 @@ Status repairDatabase(OperationContext* opCtx,
     // Reopening db is necessary for repairCollections.
     databaseHolder->openDb(opCtx, tenantDbName);
 
-    auto status = repairCollections(opCtx, engine, tenantDbName.dbName());
+    auto status = repairCollections(opCtx, engine, tenantDbName);
     if (!status.isOK()) {
         LOGV2_FATAL_CONTINUE(21030,
                              "Failed to repair database {dbName}: {status_reason}",
diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp
index 08a0d446c85..0043ff54e3e 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.cpp
+++ b/src/mongo/db/repl/idempotency_test_fixture.cpp
@@ -361,8 +361,7 @@ std::vector<CollectionState> IdempotencyTest::validateAllCollections() {
             std::vector<NamespaceString> collectionNames;
             {
                 Lock::DBLock lk(_opCtx.get(), tenantDbName.dbName(), MODE_S);
-                collectionNames =
-                    catalog->getAllCollectionNamesFromDb(_opCtx.get(), tenantDbName.dbName());
+                collectionNames = catalog->getAllCollectionNamesFromDb(_opCtx.get(), tenantDbName);
             }
             for (const auto& nss : collectionNames) {
                 collStates.push_back(validate(nss));
diff --git a/src/mongo/db/s/migration_util.cpp b/src/mongo/db/s/migration_util.cpp
index 63a6806b9fe..219236c5e7a 100644
--- a/src/mongo/db/s/migration_util.cpp
+++ b/src/mongo/db/s/migration_util.cpp
@@ -620,8 +620,7 @@ void submitOrphanRangesForCleanup(OperationContext* opCtx) {
         if (tenantDbName.dbName() == NamespaceString::kLocalDb)
             continue;
 
-        for (auto collIt = catalog->begin(opCtx, tenantDbName.dbName());
-             collIt != catalog->end(opCtx);
+        for (auto collIt = catalog->begin(opCtx, tenantDbName); collIt != catalog->end(opCtx);
              ++collIt) {
             auto uuid = collIt.uuid().get();
             auto nss = catalog->lookupNSSByUUID(opCtx, uuid).get();
diff --git a/src/mongo/db/startup_recovery.cpp b/src/mongo/db/startup_recovery.cpp
index e9daca7e46d..b647e5e526d 100644
--- a/src/mongo/db/startup_recovery.cpp
+++ b/src/mongo/db/startup_recovery.cpp
@@ -214,8 +214,7 @@ Status ensureCollectionProperties(OperationContext* opCtx,
                                   Database* db,
                                   EnsureIndexPolicy ensureIndexPolicy) {
     auto catalog = CollectionCatalog::get(opCtx);
-    for (auto collIt = catalog->begin(opCtx, db->name().dbName()); collIt != catalog->end(opCtx);
-         ++collIt) {
+    for (auto collIt = catalog->begin(opCtx, db->name()); collIt != catalog->end(opCtx); ++collIt) {
         auto coll = collIt.getWritableCollection(opCtx, CollectionCatalog::LifetimeMode::kInplace);
         if (!coll) {
             break;
diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index 39d07b688ed..b9fd8b0ed9b 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -527,6 +527,7 @@ env.Library(
         'storage_util.cpp',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/multitenancy',
         'durable_catalog_impl',
         'kv/kv_drop_pending_ident_reaper',
     ],
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 0d1121215a5..d041cae643a 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -839,7 +839,7 @@ Status StorageEngineImpl::dropDatabase(OperationContext* opCtx, StringData db) {
         }
     }
 
-    std::vector<UUID> toDrop = catalog->getAllCollectionUUIDsFromDb(db);
+    std::vector<UUID> toDrop = catalog->getAllCollectionUUIDsFromDb(tenantDbName);
 
     // Do not timestamp any of the following writes. This will remove entries from the catalog as
     // well as drop any underlying tables. It's not expected for dropping tables to be reversible
@@ -1313,14 +1313,17 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, StringData d
         return true;
     };
 
+    // TODO SERVER-63187: Change StorageEngine APIs to accept TenantDatabaseName.
+    const TenantDatabaseName tenantDbName(boost::none, dbName);
     if (opCtx->isLockFreeReadsOp()) {
         auto collectionCatalog = CollectionCatalog::get(opCtx);
-        for (auto it = collectionCatalog->begin(opCtx, dbName); it != collectionCatalog->end(opCtx);
+        for (auto it = collectionCatalog->begin(opCtx, tenantDbName);
+             it != collectionCatalog->end(opCtx);
              ++it) {
             perCollectionWork(*it);
         }
     } else {
-        catalog::forEachCollectionFromDb(opCtx, dbName, MODE_IS, perCollectionWork);
+        catalog::forEachCollectionFromDb(opCtx, tenantDbName, MODE_IS, perCollectionWork);
     };
 
     return size;
diff --git a/src/mongo/db/storage/storage_util.cpp b/src/mongo/db/storage/storage_util.cpp
index 81380e9eae2..d3d86a47b2f 100644
--- a/src/mongo/db/storage/storage_util.cpp
+++ b/src/mongo/db/storage/storage_util.cpp
@@ -47,38 +47,40 @@
 namespace mongo {
 namespace catalog {
 namespace {
-auto removeEmptyDirectory =
-    [](ServiceContext* svcCtx, StorageEngine* storageEngine, const NamespaceString& ns) {
-        // Nothing to do if not using directoryperdb or there are still collections in the database.
-        // If we don't support supportsPendingDrops then this is executing before the collection is
-        // removed from the catalog. In that case, just blindly attempt to delete the directory, it
-        // will only succeed if it is empty which is the behavior we want.
-        auto collectionCatalog = CollectionCatalog::get(svcCtx);
-        if (!storageEngine->isUsingDirectoryPerDb() ||
-            (storageEngine->supportsPendingDrops() &&
-             collectionCatalog->begin(nullptr, ns.db()) != collectionCatalog->end(nullptr))) {
-            return;
-        }
+auto removeEmptyDirectory = [](ServiceContext* svcCtx,
+                               StorageEngine* storageEngine,
+                               const NamespaceString& ns) {
+    // Nothing to do if not using directoryperdb or there are still collections in the database.
+    // If we don't support supportsPendingDrops then this is executing before the collection is
+    // removed from the catalog. In that case, just blindly attempt to delete the directory, it
+    // will only succeed if it is empty which is the behavior we want.
+    auto collectionCatalog = CollectionCatalog::get(svcCtx);
+    const TenantDatabaseName tenantDbName(boost::none, ns.db());
+    if (!storageEngine->isUsingDirectoryPerDb() ||
+        (storageEngine->supportsPendingDrops() &&
+         collectionCatalog->begin(nullptr, tenantDbName) != collectionCatalog->end(nullptr))) {
+        return;
+    }
 
-        boost::system::error_code ec;
-        boost::filesystem::remove(storageEngine->getFilesystemPathForDb(ns.db().toString()), ec);
-
-        if (!ec) {
-            LOGV2(4888200, "Removed empty database directory", "db"_attr = ns.db());
-        } else if (collectionCatalog->begin(nullptr, ns.db()) == collectionCatalog->end(nullptr)) {
-            // It is possible for a new collection to be created in the database between when we
-            // check whether the database is empty and actually attempting to remove the directory.
-            // In this case, don't log that the removal failed because it is expected. However,
-            // since we attempt to remove the directory for both the collection and index ident
-            // drops, once the database is empty it will be still logged until the final of these
-            // ident drops occurs.
-            LOGV2_DEBUG(4888201,
-                        1,
-                        "Failed to remove database directory",
-                        "db"_attr = ns.db(),
-                        "error"_attr = ec.message());
-        }
-    };
+    boost::system::error_code ec;
+    boost::filesystem::remove(storageEngine->getFilesystemPathForDb(tenantDbName.dbName()), ec);
+
+    if (!ec) {
+        LOGV2(4888200, "Removed empty database directory", "db"_attr = tenantDbName.dbName());
+    } else if (collectionCatalog->begin(nullptr, tenantDbName) == collectionCatalog->end(nullptr)) {
+        // It is possible for a new collection to be created in the database between when we
+        // check whether the database is empty and actually attempting to remove the directory.
+        // In this case, don't log that the removal failed because it is expected. However,
+        // since we attempt to remove the directory for both the collection and index ident
+        // drops, once the database is empty it will be still logged until the final of these
+        // ident drops occurs.
+        LOGV2_DEBUG(4888201,
+                    1,
+                    "Failed to remove database directory",
+                    "db"_attr = tenantDbName.dbName(),
+                    "error"_attr = ec.message());
+    }
+};
 }  // namespace
 
 void removeIndex(OperationContext* opCtx,
diff --git a/src/mongo/db/tenant_database_name.h b/src/mongo/db/tenant_database_name.h
index ade9538728a..bf013d7fd4a 100644
--- a/src/mongo/db/tenant_database_name.h
+++ b/src/mongo/db/tenant_database_name.h
@@ -47,6 +47,11 @@ namespace mongo {
 class TenantDatabaseName {
 public:
     /**
+     * Constructs an empty TenantDatabaseName.
+     */
+    TenantDatabaseName() : _tenantId(boost::none), _dbName(""), _tenantDbName(boost::none){};
+
+    /**
      * Constructs a TenantDatabaseName from the given tenantId and database name.
      * "dbName" is expected only consist of a db name. It is the caller's responsibility to ensure
      * the dbName is a valid db name.