SERVER-61987 Change DatabaseHolder's map to be keyed by TenantDatabaseName

70 files changed, 479 insertions, 277 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index d4a0114c395..f9454dfe6d5 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -85,6 +85,7 @@ env.Library(
     source=[
         'multitenancy.cpp',
         'tenant_namespace.cpp',
+        'tenant_database_name.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
@@ -711,6 +712,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/idl/server_parameter',
         'catalog/database_holder',
+        'multitenancy',
         'storage/snapshot_helper',
     ],
 )
@@ -770,6 +772,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/catalog/local_oplog_info',
         '$BUILD_DIR/mongo/db/s/sharding_api_d',
+        'multitenancy',
     ]
 )
 
@@ -807,6 +810,7 @@ env.Library(
         'index_builds_coordinator_interface',
         'index_commands_idl',
         'internal_transactions_feature_flag',
+        'multitenancy',
         'not_primary_error_tracker',
         'query_exec',
         'repl/apply_ops_command_info',
@@ -968,6 +972,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/pipeline/change_stream_pre_image_helpers',
         'dbhelpers',
         'internal_transactions_feature_flag',
+        'multitenancy',
         'repl/image_collection_entry',
         'repl/repl_server_parameters',
         'transaction',
@@ -1148,6 +1153,7 @@ env.Library(
         'catalog/database_holder',
         'commands/list_collections_filter',
         'index_builds_coordinator_interface',
+        'multitenancy',
     ],
 )
 
@@ -1254,6 +1260,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/storage_repair_observer',
         '$BUILD_DIR/mongo/db/storage/storage_util',
         'index_builds_coordinator_interface',
+        'multitenancy',
         'rebuild_indexes',
     ],
 )
@@ -1463,6 +1470,7 @@ env.Library(
         'dbdirectclient',
         'dbhelpers',
         'index_builds_coordinator_interface',
+        'multitenancy',
         'rebuild_indexes',
         'repair',
         'server_feature_flags',
@@ -2609,7 +2617,6 @@ if wiredtiger:
             'repl/storage_interface_impl',
             'rw_concern_d',
             's/shard_server_test_fixture',
-            'server_feature_flags',
             'server_options_core',
             'server_options_servers',
             'service_context',
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index ac907b5a8ae..4f065749663 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -274,12 +274,14 @@ env.Library(
         'uncommitted_collections.cpp',
         'uncommitted_multikey.cpp',
     ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/db/multitenancy',
+    ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/profile_filter',
-        '$BUILD_DIR/mongo/db/server_feature_flags',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/storage/bson_collection_catalog_entry',
@@ -320,6 +322,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/rebuild_indexes',
         '$BUILD_DIR/mongo/db/service_context',
         'collection',
@@ -336,6 +339,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         "$BUILD_DIR/mongo/base",
         "$BUILD_DIR/mongo/db/catalog_raii",
+        "$BUILD_DIR/mongo/db/multitenancy",
         "$BUILD_DIR/mongo/db/views/views",
         "$BUILD_DIR/mongo/util/fail_point",
         "collection_catalog",
@@ -415,6 +419,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/db_raii',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/views/views',
         'database_holder',
     ]
@@ -501,6 +506,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
         '$BUILD_DIR/mongo/db/index_commands_idl',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/storage/index_entry_comparison',
@@ -638,7 +644,6 @@ if wiredtiger:
             '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
             '$BUILD_DIR/mongo/db/repl/replmocks',
             '$BUILD_DIR/mongo/db/repl/storage_interface_impl',
-            '$BUILD_DIR/mongo/db/server_feature_flags',
             '$BUILD_DIR/mongo/db/service_context',
             '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
             '$BUILD_DIR/mongo/db/service_context_test_fixture',
diff --git a/src/mongo/db/catalog/catalog_control.cpp b/src/mongo/db/catalog/catalog_control.cpp
index 4e73183ca3a..e9dc0e6691c 100644
--- a/src/mongo/db/catalog/catalog_control.cpp
+++ b/src/mongo/db/catalog/catalog_control.cpp
@@ -43,6 +43,8 @@
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/rebuild_indexes.h"
+#include "mongo/db/tenant_database_name.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -64,7 +66,7 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
     for (auto&& tenantDbName : databasesToOpen) {
         LOGV2_FOR_RECOVERY(
             23992, 1, "openCatalog: dbholder reopening database", "db"_attr = tenantDbName);
-        auto db = databaseHolder->openDb(opCtx, tenantDbName.dbName());
+        auto db = databaseHolder->openDb(opCtx, tenantDbName);
         invariant(db, str::stream() << "failed to reopen database " << tenantDbName.toString());
         for (auto&& collNss : catalog->getAllCollectionNamesFromDb(opCtx, tenantDbName.dbName())) {
             // Note that the collection name already includes the database component.
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index 31c1e969bb9..832b7dcbeb8 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -36,6 +36,7 @@
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/profile_filter.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/tenant_database_name.h"
 #include "mongo/stdx/unordered_map.h"
 #include "mongo/util/uuid.h"
 
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 22f097048b3..657ae0a610c 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -35,7 +35,6 @@
 #include "mongo/db/catalog/collection_catalog_helper.h"
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/concurrency/lock_manager_defs.h"
-#include "mongo/db/multitenancy.h"
 #include "mongo/db/operation_context_noop.h"
 #include "mongo/db/service_context_d_test_fixture.h"
 #include "mongo/unittest/death_test.h"
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 00efbf86ec1..97a437950cc 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -58,7 +58,6 @@
 #include "mongo/db/matcher/doc_validation_error.h"
 #include "mongo/db/matcher/expression_always_boolean.h"
 #include "mongo/db/matcher/expression_parser.h"
-#include "mongo/db/multitenancy.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/ops/update_request.h"
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index 7f4961cc306..f111892b265 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -74,7 +74,7 @@ public:
     }
 
     const TenantNamespace& tenantNs() const {
-        MONGO_UNREACHABLE;
+        return _tenantNs;
     }
 
     Status rename(OperationContext* opCtx, const TenantNamespace& tenantNs, bool stayTemp) final {
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index fc657182db2..d5c4caef4cf 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -55,6 +55,8 @@
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
+#include "mongo/db/tenant_database_name.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/db/timeseries/timeseries_options.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/idl/command_generic_argument.h"
@@ -711,6 +713,7 @@ void createChangeStreamPreImagesCollection(OperationContext* opCtx) {
             status.isOK() || status.code() == ErrorCodes::NamespaceExists);
 }
 
+// TODO SERVER-62880 pass TenantDatabaseName instead of dbName.
 Status createCollectionForApplyOps(OperationContext* opCtx,
                                    const std::string& dbName,
                                    const boost::optional<UUID>& ui,
@@ -723,7 +726,8 @@ Status createCollectionForApplyOps(OperationContext* opCtx,
     auto newCmd = cmdObj;
 
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto* const db = databaseHolder->getDb(opCtx, dbName);
+    const TenantDatabaseName tenantDbName(boost::none, dbName);
+    auto* const db = databaseHolder->getDb(opCtx, tenantDbName);
 
     // If a UUID is given, see if we need to rename a collection out of the way, and whether the
     // collection already exists under a different name. If so, rename it into place. As this is
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index 403b39b3a0a..af85dbd71e8 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -40,6 +40,7 @@
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/repl/optime.h"
+#include "mongo/db/tenant_database_name.h"
 #include "mongo/util/string_map.h"
 
 namespace mongo {
@@ -79,7 +80,7 @@ public:
      */
     virtual Status init(OperationContext* opCtx) = 0;
 
-    virtual const std::string& name() const = 0;
+    virtual const TenantDatabaseName& name() const = 0;
 
     virtual void clearTmpCollections(OperationContext* opCtx) const = 0;
 
diff --git a/src/mongo/db/catalog/database_holder.h b/src/mongo/db/catalog/database_holder.h
index efdaa4b5bcd..484d8dc6cc2 100644
--- a/src/mongo/db/catalog/database_holder.h
+++ b/src/mongo/db/catalog/database_holder.h
@@ -35,6 +35,7 @@
 #include "mongo/base/string_data.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_options.h"
+#include "mongo/db/tenant_database_name.h"
 
 namespace mongo {
 
@@ -63,25 +64,28 @@ public:
      * Retrieves an already opened database or returns nullptr. Must be called with the database
      * locked in at least IS-mode.
      */
-    virtual Database* getDb(OperationContext* opCtx, StringData ns) const = 0;
+    virtual Database* getDb(OperationContext* opCtx,
+                            const TenantDatabaseName& tenantDbName) const = 0;
 
     /**
      * Checks if a database exists without holding a database-level lock. This class' internal mutex
-     * provides concurrency protection around looking up the db name of 'ns'.
+     * provides concurrency protection around looking up the db name of 'tenantDbName'.
      */
-    virtual bool dbExists(OperationContext* opCtx, StringData ns) const = 0;
+    virtual bool dbExists(OperationContext* opCtx,
+                          const TenantDatabaseName& tenantDbName) const = 0;
 
     /**
-     * Fetches the ViewCatalog decorating the Database matching 'dbName', or returns nullptr if the
-     * database does not exist. The returned ViewCatalog is safe to access without a lock because it
-     * is held as a shared_ptr.
+     * Fetches the ViewCatalog decorating the Database matching 'tenantDbName', or returns nullptr
+     * if the database does not exist. The returned ViewCatalog is safe to access without a lock
+     * because it is held as a shared_ptr.
      *
      * The ViewCatalog must be fetched through this interface if the caller holds no database lock
      * to ensure the Database object is safe to access. This class' internal mutex provides
-     * concurrency protection around looking up and accessing the Database object matching 'dbName'.
+     * concurrency protection around looking up and accessing the Database object matching
+     * 'tenantDbName'.
      */
-    virtual std::shared_ptr<const ViewCatalog> getViewCatalog(OperationContext* opCtx,
-                                                              StringData dbName) const = 0;
+    virtual std::shared_ptr<const ViewCatalog> getViewCatalog(
+        OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const = 0;
 
     /**
      * Retrieves a database reference if it is already opened, or opens it if it hasn't been
@@ -91,7 +95,7 @@ public:
      *          existed (false). Can be NULL if this information is not necessary.
      */
     virtual Database* openDb(OperationContext* opCtx,
-                             StringData ns,
+                             const TenantDatabaseName& tenantDbName,
                              bool* justCreated = nullptr) = 0;
 
     /**
@@ -108,7 +112,7 @@ public:
      * Closes the specified database. Must be called with the database locked in X-mode.
      * No background jobs must be in progress on the database when this function is called.
      */
-    virtual void close(OperationContext* opCtx, StringData ns) = 0;
+    virtual void close(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) = 0;
 
     /**
      * Closes all opened databases. Must be called with the global lock acquired in X-mode.
@@ -121,14 +125,15 @@ public:
     /**
      * Returns the set of existing database names that differ only in casing.
      */
-    virtual std::set<std::string> getNamesWithConflictingCasing(StringData name) = 0;
+    virtual std::set<TenantDatabaseName> getNamesWithConflictingCasing(
+        const TenantDatabaseName& tenantDbName) = 0;
 
     /**
      * Returns all the database names (including those which are empty).
      *
      * Unlike CollectionCatalog::getAllDbNames(), this returns databases that are empty.
      */
-    virtual std::vector<std::string> getNames() = 0;
+    virtual std::vector<TenantDatabaseName> getNames() = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index 39e892cbfbc..8115cd59fc6 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -48,34 +48,19 @@
 #include "mongo/logv2/log.h"
 
 namespace mongo {
-namespace {
 
-StringData _todb(StringData ns) {
-    std::size_t i = ns.find('.');
-    if (i == std::string::npos) {
-        uassert(13074, "db name can't be empty", ns.size());
-        return ns;
-    }
-
-    uassert(13075, "db name can't be empty", i > 0);
-
-    const StringData d = ns.substr(0, i);
+Database* DatabaseHolderImpl::getDb(OperationContext* opCtx,
+                                    const TenantDatabaseName& tenantDbName) const {
     uassert(13280,
-            "invalid db name: " + ns.toString(),
-            NamespaceString::validDBName(d, NamespaceString::DollarInDbNameBehavior::Allow));
-
-    return d;
-}
-
-}  // namespace
+            "invalid db name: " + tenantDbName.dbName(),
+            NamespaceString::validDBName(tenantDbName.dbName(),
+                                         NamespaceString::DollarInDbNameBehavior::Allow));
 
-Database* DatabaseHolderImpl::getDb(OperationContext* opCtx, StringData ns) const {
-    const StringData db = _todb(ns);
-    invariant(opCtx->lockState()->isDbLockedForMode(db, MODE_IS) ||
-              (db.compare("local") == 0 && opCtx->lockState()->isLocked()));
+    invariant(opCtx->lockState()->isDbLockedForMode(tenantDbName.dbName(), MODE_IS) ||
+              (tenantDbName.dbName().compare("local") == 0 && opCtx->lockState()->isLocked()));
 
     stdx::lock_guard<SimpleMutex> lk(_m);
-    DBs::const_iterator it = _dbs.find(db);
+    DBs::const_iterator it = _dbs.find(tenantDbName);
     if (it != _dbs.end()) {
         return it->second;
     }
@@ -83,15 +68,20 @@ Database* DatabaseHolderImpl::getDb(OperationContext* opCtx, StringData ns) cons
     return nullptr;
 }
 
-bool DatabaseHolderImpl::dbExists(OperationContext* opCtx, StringData ns) const {
+bool DatabaseHolderImpl::dbExists(OperationContext* opCtx,
+                                  const TenantDatabaseName& tenantDbName) const {
+    uassert(6198702,
+            "invalid db name: " + tenantDbName.dbName(),
+            NamespaceString::validDBName(tenantDbName.dbName(),
+                                         NamespaceString::DollarInDbNameBehavior::Allow));
     stdx::lock_guard<SimpleMutex> lk(_m);
-    return _dbs.find(_todb(ns)) != _dbs.end();
+    return _dbs.find(tenantDbName) != _dbs.end();
 }
 
-std::shared_ptr<const ViewCatalog> DatabaseHolderImpl::getViewCatalog(OperationContext* opCtx,
-                                                                      StringData dbName) const {
+std::shared_ptr<const ViewCatalog> DatabaseHolderImpl::getViewCatalog(
+    OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const {
     stdx::lock_guard<SimpleMutex> lk(_m);
-    DBs::const_iterator it = _dbs.find(dbName);
+    DBs::const_iterator it = _dbs.find(tenantDbName);
     if (it != _dbs.end()) {
         const Database* db = it->second;
         if (db) {
@@ -102,34 +92,42 @@ std::shared_ptr<const ViewCatalog> DatabaseHolderImpl::getViewCatalog(OperationC
     return nullptr;
 }
 
-std::set<std::string> DatabaseHolderImpl::_getNamesWithConflictingCasing_inlock(StringData name) {
-    std::set<std::string> duplicates;
+std::set<TenantDatabaseName> DatabaseHolderImpl::_getNamesWithConflictingCasing_inlock(
+    const TenantDatabaseName& tenantDbName) {
+    std::set<TenantDatabaseName> duplicates;
 
     for (const auto& nameAndPointer : _dbs) {
         // A name that's equal with case-insensitive match must be identical, or it's a duplicate.
-        if (name.equalCaseInsensitive(nameAndPointer.first) && name != nameAndPointer.first)
+        if (tenantDbName.equalCaseInsensitive(nameAndPointer.first) &&
+            tenantDbName != nameAndPointer.first)
             duplicates.insert(nameAndPointer.first);
     }
     return duplicates;
 }
 
-std::set<std::string> DatabaseHolderImpl::getNamesWithConflictingCasing(StringData name) {
+std::set<TenantDatabaseName> DatabaseHolderImpl::getNamesWithConflictingCasing(
+    const TenantDatabaseName& tenantDbName) {
     stdx::lock_guard<SimpleMutex> lk(_m);
-    return _getNamesWithConflictingCasing_inlock(name);
+    return _getNamesWithConflictingCasing_inlock(tenantDbName);
 }
 
-std::vector<std::string> DatabaseHolderImpl::getNames() {
+std::vector<TenantDatabaseName> DatabaseHolderImpl::getNames() {
     stdx::lock_guard<SimpleMutex> lk(_m);
-    std::vector<std::string> names;
+    std::vector<TenantDatabaseName> tenantDbNames;
     for (const auto& nameAndPointer : _dbs) {
-        names.push_back(nameAndPointer.first);
+        tenantDbNames.push_back(nameAndPointer.first);
     }
-    return names;
+    return tenantDbNames;
 }
 
-Database* DatabaseHolderImpl::openDb(OperationContext* opCtx, StringData ns, bool* justCreated) {
-    const StringData dbname = _todb(ns);
-    invariant(opCtx->lockState()->isDbLockedForMode(dbname, MODE_IX));
+Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
+                                     const TenantDatabaseName& tenantDbName,
+                                     bool* justCreated) {
+    uassert(6198701,
+            "invalid db name: " + tenantDbName.dbName(),
+            NamespaceString::validDBName(tenantDbName.dbName(),
+                                         NamespaceString::DollarInDbNameBehavior::Allow));
+    invariant(opCtx->lockState()->isDbLockedForMode(tenantDbName.dbName(), MODE_IX));
 
     if (justCreated)
         *justCreated = false;  // Until proven otherwise.
@@ -138,15 +136,15 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx, StringData ns, boo
 
     // The following will insert a nullptr for dbname, which will treated the same as a non-
     // existant database by the get method, yet still counts in getNamesWithConflictingCasing.
-    if (auto db = _dbs[dbname])
+    if (auto db = _dbs[tenantDbName])
         return db;
 
     std::unique_ptr<DatabaseImpl> newDb;
     // We've inserted a nullptr entry for dbname: make sure to remove it on unsuccessful exit.
-    ScopeGuard removeDbGuard([this, &lk, &newDb, opCtx, dbname] {
+    ScopeGuard removeDbGuard([this, &lk, &newDb, opCtx, tenantDbName] {
         if (!lk.owns_lock())
             lk.lock();
-        auto it = _dbs.find(dbname);
+        auto it = _dbs.find(tenantDbName);
         // If someone else hasn't either already removed it or already set it successfully, remove.
         if (it != _dbs.end() && !it->second) {
             _dbs.erase(it);
@@ -161,24 +159,24 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx, StringData ns, boo
     });
 
     // Check casing in lock to avoid transient duplicates.
-    auto duplicates = _getNamesWithConflictingCasing_inlock(dbname);
+    auto duplicates = _getNamesWithConflictingCasing_inlock(tenantDbName);
     uassert(ErrorCodes::DatabaseDifferCase,
             str::stream() << "db already exists with different case already have: ["
-                          << *duplicates.cbegin() << "] trying to create [" << dbname.toString()
-                          << "]",
+                          << (*duplicates.cbegin()) << "] trying to create ["
+                          << tenantDbName.toString() << "]",
             duplicates.empty());
 
     // Do the catalog lookup and database creation outside of the scoped lock, because these may
     // block.
     lk.unlock();
 
-    if (CollectionCatalog::get(opCtx)->getAllCollectionUUIDsFromDb(dbname).empty()) {
-        audit::logCreateDatabase(opCtx->getClient(), dbname);
+    if (CollectionCatalog::get(opCtx)->getAllCollectionUUIDsFromDb(tenantDbName.dbName()).empty()) {
+        audit::logCreateDatabase(opCtx->getClient(), tenantDbName.dbName());
         if (justCreated)
             *justCreated = true;
     }
 
-    newDb = std::make_unique<DatabaseImpl>(dbname);
+    newDb = std::make_unique<DatabaseImpl>(tenantDbName);
     Status status = newDb->init(opCtx);
     while (!status.isOK()) {
         // If we get here, then initializing the database failed because another concurrent writer
@@ -186,7 +184,7 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx, StringData ns, boo
         // them to finish.
         lk.lock();
 
-        auto it = _dbs.find(dbname);
+        auto it = _dbs.find(tenantDbName);
         if (it != _dbs.end() && it->second) {
             // Creating databases only requires a DB lock in MODE_IX. Thus databases can be created
             // concurrently. If this thread "lost the race", return the database object that was
@@ -199,7 +197,7 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx, StringData ns, boo
         // in such a way that we can easily express it as a predicate for that function.
         _c.wait_for(lk, stdx::chrono::milliseconds(1));
 
-        it = _dbs.find(dbname);
+        it = _dbs.find(tenantDbName);
         if (it != _dbs.end() && it->second) {
             // As above, another writer finished successfully, return the persisted object.
             removeDbGuard.dismiss();
@@ -220,10 +218,10 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx, StringData ns, boo
     removeDbGuard.dismiss();
     lk.lock();
 
-    invariant(!_dbs[dbname]);
+    invariant(!_dbs[tenantDbName]);
     auto* db = newDb.release();
-    _dbs[dbname] = db;
-    invariant(_getNamesWithConflictingCasing_inlock(dbname.toString()).empty());
+    _dbs[tenantDbName] = db;
+    invariant(_getNamesWithConflictingCasing_inlock(tenantDbName).empty());
     _c.notify_all();
 
     return db;
@@ -237,10 +235,11 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
 
     LOGV2_DEBUG(20310, 1, "dropDatabase {name}", "name"_attr = name);
 
-    invariant(opCtx->lockState()->isDbLockedForMode(name, MODE_X));
+    invariant(opCtx->lockState()->isDbLockedForMode(name.dbName(), MODE_X));
 
     auto catalog = CollectionCatalog::get(opCtx);
-    for (auto collIt = catalog->begin(opCtx, name); collIt != catalog->end(opCtx); ++collIt) {
+    for (auto collIt = catalog->begin(opCtx, name.dbName()); collIt != catalog->end(opCtx);
+         ++collIt) {
         auto coll = *collIt;
         if (!coll) {
             break;
@@ -252,11 +251,12 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
                   str::stream() << "An index is building on collection '" << coll->ns() << "'.");
     }
 
-    audit::logDropDatabase(opCtx->getClient(), name);
+    audit::logDropDatabase(opCtx->getClient(), name.dbName());
 
     auto const serviceContext = opCtx->getServiceContext();
 
-    for (auto collIt = catalog->begin(opCtx, name); collIt != catalog->end(opCtx); ++collIt) {
+    for (auto collIt = catalog->begin(opCtx, name.dbName()); collIt != catalog->end(opCtx);
+         ++collIt) {
         auto coll = *collIt;
         if (!coll) {
             break;
@@ -280,23 +280,28 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
     }
 
     // Clean up the in-memory database state.
-    CollectionCatalog::write(
-        opCtx, [&](CollectionCatalog& catalog) { catalog.clearDatabaseProfileSettings(name); });
+    CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
+        catalog.clearDatabaseProfileSettings(name.dbName());
+    });
     close(opCtx, name);
 
     auto const storageEngine = serviceContext->getStorageEngine();
-    writeConflictRetry(opCtx, "dropDatabase", name, [&] {
-        storageEngine->dropDatabase(opCtx, name).transitional_ignore();
+    writeConflictRetry(opCtx, "dropDatabase", name.dbName(), [&] {
+        // TODO SERVER-63187 Call dropDatabase with name which is TenantDatabaseName.
+        storageEngine->dropDatabase(opCtx, name.fullName()).transitional_ignore();
     });
 }
 
-void DatabaseHolderImpl::close(OperationContext* opCtx, StringData ns) {
-    const StringData dbName = _todb(ns);
-    invariant(opCtx->lockState()->isDbLockedForMode(dbName, MODE_X));
+void DatabaseHolderImpl::close(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) {
+    uassert(6198700,
+            "invalid db name: " + tenantDbName.dbName(),
+            NamespaceString::validDBName(tenantDbName.dbName(),
+                                         NamespaceString::DollarInDbNameBehavior::Allow));
+    invariant(opCtx->lockState()->isDbLockedForMode(tenantDbName.dbName(), MODE_X));
 
     stdx::unique_lock<SimpleMutex> lk(_m);
 
-    DBs::const_iterator it = _dbs.find(dbName);
+    DBs::const_iterator it = _dbs.find(tenantDbName);
     if (it == _dbs.end()) {
         return;
     }
@@ -309,15 +314,15 @@ void DatabaseHolderImpl::close(OperationContext* opCtx, StringData ns) {
 
     // It's possible another thread altered the record before we reacquired the lock, so make sure
     // we still have the same database.
-    it = _dbs.find(dbName);
+    it = _dbs.find(tenantDbName);
     if (it == _dbs.end() || db != it->second) {
         return;
     }
 
-    LOGV2_DEBUG(20311, 2, "DatabaseHolder::close", "db"_attr = dbName);
+    LOGV2_DEBUG(20311, 2, "DatabaseHolder::close", "db"_attr = tenantDbName);
 
     CollectionCatalog::write(opCtx, [&](CollectionCatalog& catalog) {
-        catalog.onCloseDatabase(opCtx, dbName.toString());
+        catalog.onCloseDatabase(opCtx, tenantDbName.dbName());
     });
 
     delete db;
@@ -326,20 +331,21 @@ void DatabaseHolderImpl::close(OperationContext* opCtx, StringData ns) {
     _dbs.erase(it);
 
     auto* const storageEngine = opCtx->getServiceContext()->getStorageEngine();
-    storageEngine->closeDatabase(opCtx, dbName.toString()).transitional_ignore();
+    // TODO SERVER-63187 Call dropDatabase with name which is TenantDatabaseName.
+    storageEngine->closeDatabase(opCtx, tenantDbName.fullName()).transitional_ignore();
 }
 
 void DatabaseHolderImpl::closeAll(OperationContext* opCtx) {
     invariant(opCtx->lockState()->isW());
 
     while (true) {
-        std::vector<std::string> dbs;
+        std::vector<TenantDatabaseName> dbs;
         {
             stdx::lock_guard<SimpleMutex> lk(_m);
             for (DBs::const_iterator i = _dbs.begin(); i != _dbs.end(); ++i) {
                 // It is the caller's responsibility to ensure that no index builds are active in
                 // the database.
-                IndexBuildsCoordinator::get(opCtx)->assertNoBgOpInProgForDb(i->first);
+                IndexBuildsCoordinator::get(opCtx)->assertNoBgOpInProgForDb(i->first.dbName());
                 dbs.push_back(i->first);
             }
         }
diff --git a/src/mongo/db/catalog/database_holder_impl.h b/src/mongo/db/catalog/database_holder_impl.h
index 6837dcfc05b..7e5c49f9c56 100644
--- a/src/mongo/db/catalog/database_holder_impl.h
+++ b/src/mongo/db/catalog/database_holder_impl.h
@@ -31,6 +31,7 @@
 
 #include "mongo/db/catalog/database_holder.h"
 
+#include "mongo/db/tenant_database_name.h"
 #include "mongo/stdx/condition_variable.h"
 #include "mongo/util/concurrency/mutex.h"
 #include "mongo/util/string_map.h"
@@ -41,29 +42,33 @@ class DatabaseHolderImpl : public DatabaseHolder {
 public:
     DatabaseHolderImpl() = default;
 
-    Database* getDb(OperationContext* opCtx, StringData ns) const override;
+    Database* getDb(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const override;
 
-    bool dbExists(OperationContext* opCtx, StringData ns) const override;
+    bool dbExists(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const override;
 
-    std::shared_ptr<const ViewCatalog> getViewCatalog(OperationContext* opCtx,
-                                                      StringData dbName) const override;
+    std::shared_ptr<const ViewCatalog> getViewCatalog(
+        OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const override;
 
-    Database* openDb(OperationContext* opCtx, StringData ns, bool* justCreated = nullptr) override;
+    Database* openDb(OperationContext* opCtx,
+                     const TenantDatabaseName& tenantDbName,
+                     bool* justCreated = nullptr) override;
 
     void dropDb(OperationContext* opCtx, Database* db) override;
 
-    void close(OperationContext* opCtx, StringData ns) override;
+    void close(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) override;
 
     void closeAll(OperationContext* opCtx) override;
 
-    std::set<std::string> getNamesWithConflictingCasing(StringData name) override;
+    std::set<TenantDatabaseName> getNamesWithConflictingCasing(
+        const TenantDatabaseName& tenantDbName) override;
 
-    std::vector<std::string> getNames() override;
+    std::vector<TenantDatabaseName> getNames() override;
 
 private:
-    std::set<std::string> _getNamesWithConflictingCasing_inlock(StringData name);
+    std::set<TenantDatabaseName> _getNamesWithConflictingCasing_inlock(
+        const TenantDatabaseName& tdb);
 
-    typedef StringMap<Database*> DBs;
+    typedef stdx::unordered_map<TenantDatabaseName, Database*> DBs;
     mutable SimpleMutex _m;
     mutable stdx::condition_variable _c;
     DBs _dbs;
diff --git a/src/mongo/db/catalog/database_holder_mock.h b/src/mongo/db/catalog/database_holder_mock.h
index 9c5e4e1d25b..8086e0d9164 100644
--- a/src/mongo/db/catalog/database_holder_mock.h
+++ b/src/mongo/db/catalog/database_holder_mock.h
@@ -37,34 +37,38 @@ class DatabaseHolderMock : public DatabaseHolder {
 public:
     DatabaseHolderMock() = default;
 
-    Database* getDb(OperationContext* opCtx, StringData ns) const override {
+    Database* getDb(OperationContext* opCtx,
+                    const TenantDatabaseName& tenantDbName) const override {
         return nullptr;
     }
 
-    bool dbExists(OperationContext* opCtx, StringData dbName) const override {
+    bool dbExists(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) const override {
         return false;
     }
 
-    std::shared_ptr<const ViewCatalog> getViewCatalog(OperationContext* const opCtx,
-                                                      StringData dbName) const override {
+    std::shared_ptr<const ViewCatalog> getViewCatalog(
+        OperationContext* const opCtx, const TenantDatabaseName& tenantDbName) const override {
         return nullptr;
     }
 
-    Database* openDb(OperationContext* opCtx, StringData ns, bool* justCreated = nullptr) override {
+    Database* openDb(OperationContext* opCtx,
+                     const TenantDatabaseName& tenantDbName,
+                     bool* justCreated = nullptr) override {
         return nullptr;
     }
 
     void dropDb(OperationContext* opCtx, Database* db) override {}
 
-    void close(OperationContext* opCtx, StringData ns) override {}
+    void close(OperationContext* opCtx, const TenantDatabaseName& tenantDbName) override {}
 
     void closeAll(OperationContext* opCtx) override {}
 
-    std::set<std::string> getNamesWithConflictingCasing(StringData name) override {
-        return std::set<std::string>();
+    std::set<TenantDatabaseName> getNamesWithConflictingCasing(
+        const TenantDatabaseName& tenantDbName) override {
+        return std::set<TenantDatabaseName>();
     }
 
-    std::vector<std::string> getNames() override {
+    std::vector<TenantDatabaseName> getNames() override {
         return {};
     }
 };
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 75b48c8bcf0..922f7068f72 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -58,7 +58,6 @@
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/introspect.h"
-#include "mongo/db/multitenancy.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/repl/drop_pending_collection_reaper.h"
@@ -152,12 +151,12 @@ Status DatabaseImpl::validateDBName(StringData dbname) {
     return Status::OK();
 }
 
-DatabaseImpl::DatabaseImpl(const StringData name)
-    : _name(name.toString()),
-      _viewsName(_name + "." + DurableViewCatalog::viewsCollectionName().toString()) {}
+DatabaseImpl::DatabaseImpl(const TenantDatabaseName& tenantDbName)
+    : _name(tenantDbName),
+      _viewsName(_name.dbName() + "." + DurableViewCatalog::viewsCollectionName().toString()) {}
 
 Status DatabaseImpl::init(OperationContext* const opCtx) {
-    Status status = validateDBName(_name);
+    Status status = validateDBName(_name.dbName());
 
     if (!status.isOK()) {
         LOGV2_WARNING(20325,
@@ -168,13 +167,13 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
     }
 
     auto durableViewCatalog = std::make_unique<DurableViewCatalogImpl>(this);
-    status = ViewCatalog::registerDatabase(opCtx, _name, std::move(durableViewCatalog));
+    status = ViewCatalog::registerDatabase(opCtx, _name.dbName(), std::move(durableViewCatalog));
     if (!status.isOK()) {
         return status;
     }
 
     auto catalog = CollectionCatalog::get(opCtx);
-    for (const auto& uuid : catalog->getAllCollectionUUIDsFromDb(_name)) {
+    for (const auto& uuid : catalog->getAllCollectionUUIDsFromDb(_name.dbName())) {
         CollectionWriter collection(
             opCtx,
             uuid,
@@ -199,9 +198,11 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
         // Realistically no one else can be accessing the collection, and there's no chance of this
         // blocking.
         Lock::CollectionLock systemViewsLock(
-            opCtx, NamespaceString(_name, NamespaceString::kSystemDotViewsCollectionName), MODE_IS);
-        Status reloadStatus =
-            ViewCatalog::reload(opCtx, _name, ViewCatalogLookupBehavior::kValidateDurableViews);
+            opCtx,
+            NamespaceString(_name.dbName(), NamespaceString::kSystemDotViewsCollectionName),
+            MODE_IS);
+        Status reloadStatus = ViewCatalog::reload(
+            opCtx, _name.dbName(), ViewCatalogLookupBehavior::kValidateDurableViews);
         if (!reloadStatus.isOK()) {
             LOGV2_WARNING_OPTIONS(20326,
                                   {logv2::LogTag::kStartupWarnings},
@@ -218,7 +219,7 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
 
         try {
             auto viewCatalog = ViewCatalog::get(opCtx);
-            viewCatalog->iterate(_name, [&](const ViewDefinition& view) {
+            viewCatalog->iterate(_name.dbName(), [&](const ViewDefinition& view) {
                 auto swResolvedView = viewCatalog->resolveView(opCtx, view.name(), boost::none);
                 if (!swResolvedView.isOK()) {
                     LOGV2_WARNING(6260802,
@@ -270,7 +271,7 @@ Status DatabaseImpl::init(OperationContext* const opCtx) {
 }
 
 void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
-    invariant(opCtx->lockState()->isDbLockedForMode(name(), MODE_IX));
+    invariant(opCtx->lockState()->isDbLockedForMode(name().dbName(), MODE_IX));
 
     CollectionCatalog::CollectionInfoFn callback = [&](const CollectionPtr& collection) {
         try {
@@ -299,17 +300,17 @@ void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
         return collection->getCollectionOptions().temp;
     };
 
-    catalog::forEachCollectionFromDb(opCtx, name(), MODE_X, callback, predicate);
+    catalog::forEachCollectionFromDb(opCtx, name().dbName(), MODE_X, callback, predicate);
 }
 
 void DatabaseImpl::setDropPending(OperationContext* opCtx, bool dropPending) {
     auto mode = dropPending ? MODE_X : MODE_IX;
-    invariant(opCtx->lockState()->isDbLockedForMode(name(), mode));
+    invariant(opCtx->lockState()->isDbLockedForMode(name().dbName(), mode));
     _dropPending.store(dropPending);
 }
 
 bool DatabaseImpl::isDropPending(OperationContext* opCtx) const {
-    invariant(opCtx->lockState()->isDbLockedForMode(name(), MODE_IS));
+    invariant(opCtx->lockState()->isDbLockedForMode(name().dbName(), MODE_IS));
     return _dropPending.load();
 }
 
@@ -328,10 +329,10 @@ void DatabaseImpl::getStats(OperationContext* opCtx,
     long long indexSize = 0;
     long long indexFreeStorageSize = 0;
 
-    invariant(opCtx->lockState()->isDbLockedForMode(name(), MODE_IS));
+    invariant(opCtx->lockState()->isDbLockedForMode(name().dbName(), MODE_IS));
 
     catalog::forEachCollectionFromDb(
-        opCtx, name(), MODE_IS, [&](const CollectionPtr& collection) -> bool {
+        opCtx, name().dbName(), MODE_IS, [&](const CollectionPtr& collection) -> bool {
             nCollections += 1;
             objects += collection->numRecords(opCtx);
             size += collection->dataSize(opCtx);
@@ -351,7 +352,7 @@ void DatabaseImpl::getStats(OperationContext* opCtx,
         });
 
 
-    ViewCatalog::get(opCtx)->iterate(name(), [&](const ViewDefinition& view) {
+    ViewCatalog::get(opCtx)->iterate(name().dbName(), [&](const ViewDefinition& view) {
         nViews += 1;
         return true;
     });
@@ -378,8 +379,10 @@ void DatabaseImpl::getStats(OperationContext* opCtx,
     output->appendNumber("scaleFactor", scale);
 
     if (!opCtx->getServiceContext()->getStorageEngine()->isEphemeral()) {
+        // It does not matter whether _name.dbName() or _name.fullName() is passed in here since
+        // directoryPerDB isn't supported in Serverless. We choose _name.dbName().
         boost::filesystem::path dbpath(
-            opCtx->getServiceContext()->getStorageEngine()->getFilesystemPathForDb(_name));
+            opCtx->getServiceContext()->getStorageEngine()->getFilesystemPathForDb(_name.dbName()));
         boost::system::error_code ec;
         boost::filesystem::space_info spaceInfo = boost::filesystem::space(dbpath, ec);
         if (!ec) {
@@ -398,7 +401,7 @@ void DatabaseImpl::getStats(OperationContext* opCtx,
 }
 
 Status DatabaseImpl::dropView(OperationContext* opCtx, NamespaceString viewName) const {
-    dassert(opCtx->lockState()->isDbLockedForMode(name(), MODE_IX));
+    dassert(opCtx->lockState()->isDbLockedForMode(name().dbName(), MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(NamespaceString(_viewsName), MODE_X));
 
@@ -418,18 +421,17 @@ Status DatabaseImpl::dropCollection(OperationContext* opCtx,
         return Status::OK();
     }
 
-    invariant(nss.db() == _name);
+    invariant(nss.db() == _name.dbName());
 
     if (nss.isSystem()) {
         if (nss.isSystemDotProfile()) {
-            if (CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(_name) != 0)
+            if (CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(_name.dbName()) != 0)
                 return Status(ErrorCodes::IllegalOperation,
                               "turn off profiling before dropping system.profile collection");
         } else if (nss.isSystemDotViews()) {
             if (!MONGO_unlikely(allowSystemViewsDrop.shouldFail())) {
-                const auto viewCatalog =
-                    DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, nss.db());
-                const auto viewStats = viewCatalog->getStats(nss.db());
+                const auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, _name);
+                const auto viewStats = viewCatalog->getStats(_name.dbName());
                 uassert(ErrorCodes::CommandFailed,
                         str::stream() << "cannot drop collection " << nss
                                       << " when time-series collections are present.",
@@ -600,7 +602,7 @@ Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
 void DatabaseImpl::_dropCollectionIndexes(OperationContext* opCtx,
                                           const NamespaceString& nss,
                                           Collection* collection) const {
-    invariant(_name == nss.db());
+    invariant(_name.dbName() == nss.db());
     LOGV2_DEBUG(
         20316, 1, "dropCollection: {namespace} - dropAllIndexes start", "namespace"_attr = nss);
     collection->getIndexCatalog()->dropAllIndexes(opCtx, collection, true);
@@ -640,8 +642,8 @@ Status DatabaseImpl::renameCollection(OperationContext* opCtx,
     invariant(opCtx->lockState()->isCollectionLockedForMode(fromNss, MODE_X));
     invariant(opCtx->lockState()->isCollectionLockedForMode(toNss, MODE_X));
 
-    invariant(fromNss.db() == _name);
-    invariant(toNss.db() == _name);
+    invariant(fromNss.db() == _name.dbName());
+    invariant(toNss.db() == _name.dbName());
     if (CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, toNss)) {
         return Status(ErrorCodes::NamespaceExists,
                       str::stream() << "Cannot rename '" << fromNss << "' to '" << toNss
@@ -668,7 +670,7 @@ Status DatabaseImpl::renameCollection(OperationContext* opCtx,
     // because the CollectionCatalog manages the necessary isolation for this Collection until the
     // WUOW commits.
     auto writableCollection = collToRename.getWritableCollection();
-    TenantNamespace toTenantNs(getActiveTenant(opCtx), toNss);
+    TenantNamespace toTenantNs(boost::none, toNss);
     Status status = writableCollection->rename(opCtx, toTenantNs, stayTemp);
     if (!status.isOK())
         return status;
@@ -715,7 +717,7 @@ void DatabaseImpl::_checkCanCreateCollection(OperationContext* opCtx,
 Status DatabaseImpl::createView(OperationContext* opCtx,
                                 const NamespaceString& viewName,
                                 const CollectionOptions& options) const {
-    dassert(opCtx->lockState()->isDbLockedForMode(name(), MODE_IX));
+    dassert(opCtx->lockState()->isDbLockedForMode(name().dbName(), MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(NamespaceString(_viewsName), MODE_X));
 
@@ -806,7 +808,7 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
 
     // Create Collection object
     auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
-    TenantNamespace tenantNs(getActiveTenant(opCtx), nss);
+    TenantNamespace tenantNs(boost::none, nss);
     std::pair<RecordId, std::unique_ptr<RecordStore>> catalogIdRecordStorePair =
         uassertStatusOK(storageEngine->getCatalog()->createCollection(
             opCtx, tenantNs, optionsWithUUID, true /*allocateDefaultSpace*/));
@@ -864,7 +866,7 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
 
 StatusWith<NamespaceString> DatabaseImpl::makeUniqueCollectionNamespace(
     OperationContext* opCtx, StringData collectionNameModel) const {
-    invariant(opCtx->lockState()->isDbLockedForMode(name(), MODE_IX));
+    invariant(opCtx->lockState()->isDbLockedForMode(name().dbName(), MODE_IX));
 
     // There must be at least one percent sign in the collection name model.
     auto numPercentSign = std::count(collectionNameModel.begin(), collectionNameModel.end(), '%');
@@ -900,7 +902,7 @@ StatusWith<NamespaceString> DatabaseImpl::makeUniqueCollectionNamespace(
                        collectionName.begin(),
                        replacePercentSign);
 
-        NamespaceString nss(_name, collectionName);
+        NamespaceString nss(_name.dbName(), collectionName);
         if (!CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss)) {
             return nss;
         }
@@ -914,7 +916,7 @@ StatusWith<NamespaceString> DatabaseImpl::makeUniqueCollectionNamespace(
 }
 
 void DatabaseImpl::checkForIdIndexesAndDropPendingCollections(OperationContext* opCtx) const {
-    if (name() == "local") {
+    if (name().dbName() == "local") {
         // Collections in the local database are not replicated, so we do not need an _id index on
         // any collection. For the same reason, it is not possible for the local database to contain
         // any drop-pending collections (drops are effective immediately).
@@ -922,7 +924,7 @@ void DatabaseImpl::checkForIdIndexesAndDropPendingCollections(OperationContext*
     }
 
     auto catalog = CollectionCatalog::get(opCtx);
-    for (const auto& nss : catalog->getAllCollectionNamesFromDb(opCtx, _name)) {
+    for (const auto& nss : catalog->getAllCollectionNamesFromDb(opCtx, _name.dbName())) {
         if (nss.isDropPendingNamespace()) {
             auto dropOpTime = fassert(40459, nss.getDropPendingNamespaceOpTime());
             LOGV2(20321,
diff --git a/src/mongo/db/catalog/database_impl.h b/src/mongo/db/catalog/database_impl.h
index db5c793c980..b4a5f41afec 100644
--- a/src/mongo/db/catalog/database_impl.h
+++ b/src/mongo/db/catalog/database_impl.h
@@ -30,16 +30,17 @@
 #pragma once
 
 #include "mongo/db/catalog/database.h"
+#include "mongo/db/tenant_database_name.h"
 
 namespace mongo {
 
 class DatabaseImpl final : public Database {
 public:
-    explicit DatabaseImpl(StringData name);
+    explicit DatabaseImpl(const TenantDatabaseName& tenantDbName);
 
     Status init(OperationContext*) final;
 
-    const std::string& name() const final {
+    const TenantDatabaseName& name() const final {
         return _name;
     }
 
@@ -142,7 +143,7 @@ private:
                                 const NamespaceString& nss,
                                 Collection* collection) const;
 
-    const std::string _name;  // "dbname"
+    const TenantDatabaseName _name;  // "dbname"
 
     const NamespaceString _viewsName;  // "dbname.system.views"
 
diff --git a/src/mongo/db/catalog/drop_database.cpp b/src/mongo/db/catalog/drop_database.cpp
index 153f986a1d1..5131dd52b3c 100644
--- a/src/mongo/db/catalog/drop_database.cpp
+++ b/src/mongo/db/catalog/drop_database.cpp
@@ -222,7 +222,8 @@ 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()); collIt != catalog->end(opCtx);
+        for (auto collIt = catalog->begin(opCtx, db->name().dbName());
+             collIt != catalog->end(opCtx);
              ++collIt) {
             auto collection = *collIt;
             if (!collection) {
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index 2d21cd822f9..1827c7aa084 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -99,7 +99,10 @@ Status checkSourceAndTargetNamespaces(OperationContext* opCtx,
         return {ErrorCodes::IllegalOperation,
                 "Cannot rename collections between a replicated and an unreplicated database"};
 
-    auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, source.db());
+    // TODO SERVER-63105 Make renameCollection accept TenantNamespace and create tenantDbName with
+    // the right tenant instead boost::none.
+    const TenantDatabaseName tenantDbName(boost::none, source.db());
+    auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
     if (!db || db->isDropPending(opCtx))
         return Status(ErrorCodes::NamespaceNotFound,
                       str::stream()
@@ -141,7 +144,7 @@ Status renameTargetCollectionToTmp(OperationContext* opCtx,
 
     // The generated unique collection name is only guaranteed to exist if the database is
     // exclusively locked.
-    invariant(opCtx->lockState()->isDbLockedForMode(targetDB->name(), LockMode::MODE_X));
+    invariant(opCtx->lockState()->isDbLockedForMode(targetDB->name().dbName(), LockMode::MODE_X));
     auto tmpNameResult = targetDB->makeUniqueCollectionNamespace(opCtx, "tmp%%%%%.rename");
     if (!tmpNameResult.isOK()) {
         return tmpNameResult.getStatus().withContext(
@@ -314,7 +317,8 @@ Status renameCollectionWithinDB(OperationContext* opCtx,
     if (!status.isOK())
         return status;
 
-    auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, source.db());
+    const TenantDatabaseName tenantDbName(boost::none, source.db());
+    auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
     auto catalog = CollectionCatalog::get(opCtx);
     const auto sourceColl = catalog->lookupCollectionByNamespace(opCtx, source);
     const auto targetColl = catalog->lookupCollectionByNamespace(opCtx, target);
@@ -359,7 +363,8 @@ Status renameCollectionWithinDBForApplyOps(OperationContext* opCtx,
     if (!status.isOK())
         return status;
 
-    auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, source.db());
+    const TenantDatabaseName tenantDbName(boost::none, source.db());
+    auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
     const auto sourceColl =
         CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, source);
 
@@ -490,7 +495,8 @@ Status renameBetweenDBs(OperationContext* opCtx,
 
     DisableDocumentValidation validationDisabler(opCtx);
 
-    auto sourceDB = DatabaseHolder::get(opCtx)->getDb(opCtx, source.db());
+    const TenantDatabaseName sourceTenantDbName(boost::none, source.db());
+    auto sourceDB = DatabaseHolder::get(opCtx)->getDb(opCtx, sourceTenantDbName);
     if (!sourceDB)
         return Status(ErrorCodes::NamespaceNotFound, "source namespace does not exist");
 
@@ -517,7 +523,8 @@ Status renameBetweenDBs(OperationContext* opCtx,
 
     IndexBuildsCoordinator::get(opCtx)->assertNoIndexBuildInProgForCollection(sourceColl->uuid());
 
-    auto targetDB = DatabaseHolder::get(opCtx)->getDb(opCtx, target.db());
+    const TenantDatabaseName targetTenantDbName(boost::none, target.db());
+    auto targetDB = DatabaseHolder::get(opCtx)->getDb(opCtx, targetTenantDbName);
 
     // Check if the target namespace exists and if dropTarget is true.
     // Return a non-OK status if target exists and dropTarget is not true or if the collection
@@ -542,12 +549,13 @@ Status renameBetweenDBs(OperationContext* opCtx,
     // Create a temporary collection in the target database. It will be removed if we fail to
     // copy the collection, or on restart, so there is no need to replicate these writes.
     if (!targetDB) {
-        targetDB = DatabaseHolder::get(opCtx)->openDb(opCtx, target.db());
+        const TenantDatabaseName tenantDbName(boost::none, target.db());
+        targetDB = DatabaseHolder::get(opCtx)->openDb(opCtx, tenantDbName);
     }
 
     // The generated unique collection name is only guaranteed to exist if the database is
     // exclusively locked.
-    invariant(opCtx->lockState()->isDbLockedForMode(targetDB->name(), LockMode::MODE_X));
+    invariant(opCtx->lockState()->isDbLockedForMode(targetDB->name().dbName(), LockMode::MODE_X));
 
     // Note that this temporary collection name is used by MongoMirror and thus must not be changed
     // without consultation.
diff --git a/src/mongo/db/catalog/rename_collection_test.cpp b/src/mongo/db/catalog/rename_collection_test.cpp
index e51ad09c0d1..9730c23d46c 100644
--- a/src/mongo/db/catalog/rename_collection_test.cpp
+++ b/src/mongo/db/catalog/rename_collection_test.cpp
@@ -505,7 +505,7 @@ void _insertDocument(OperationContext* opCtx, const NamespaceString& nss, const
 CollectionPtr _getCollection_inlock(OperationContext* opCtx, const NamespaceString& nss) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_IS));
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto* db = databaseHolder->getDb(opCtx, nss.db());
+    auto* db = databaseHolder->getDb(opCtx, TenantDatabaseName(boost::none, nss.db()));
     if (!db) {
         return nullptr;
     }
diff --git a/src/mongo/db/catalog/validate_state.cpp b/src/mongo/db/catalog/validate_state.cpp
index d3a7784d024..4cf13eb010b 100644
--- a/src/mongo/db/catalog/validate_state.cpp
+++ b/src/mongo/db/catalog/validate_state.cpp
@@ -298,7 +298,9 @@ void ValidateState::_relockDatabaseAndCollection(OperationContext* opCtx) {
         << " while validating collection: " << _nss << " (" << *_uuid << ")";
 
     _databaseLock.emplace(opCtx, _nss.db(), MODE_IS);
-    _database = DatabaseHolder::get(opCtx)->getDb(opCtx, _nss.db());
+    // TODO SERVER-63106 Have the ValidateState implementation use TenantNamespace
+    const TenantDatabaseName tenantDbName(boost::none, _nss.db());
+    _database = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
     uassert(ErrorCodes::Interrupted, dbErrMsg, _database);
     uassert(ErrorCodes::Interrupted, dbErrMsg, !_database->isDropPending(opCtx));
 
diff --git a/src/mongo/db/catalog_raii.cpp b/src/mongo/db/catalog_raii.cpp
index cc986ecadd0..8645ea5fc40 100644
--- a/src/mongo/db/catalog_raii.cpp
+++ b/src/mongo/db/catalog_raii.cpp
@@ -50,7 +50,8 @@ MONGO_FAIL_POINT_DEFINE(setAutoGetCollectionWait);
  * Returns true if 'nss' is a view. False if the namespace or view doesn't exist.
  */
 bool isSecondaryNssAView(OperationContext* opCtx, const NamespaceString& nss) {
-    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, nss.db());
+    TenantDatabaseName tenantDbName(boost::none, nss.db());
+    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
     return viewCatalog && viewCatalog->lookup(opCtx, nss);
 }
 
@@ -167,14 +168,16 @@ void acquireCollectionLocksInResourceIdOrder(
 
 }  // namespace
 
+// TODO SERVER-62918 Pass TenantDatabaseName instead of string for dbName.
 AutoGetDb::AutoGetDb(OperationContext* opCtx,
                      StringData dbName,
                      LockMode mode,
                      Date_t deadline,
                      const std::set<StringData>& secondaryDbNames)
     : _dbName(dbName), _dbLock(opCtx, dbName, mode, deadline), _db([&] {
+          const TenantDatabaseName tenantDbName(boost::none, dbName);
           auto databaseHolder = DatabaseHolder::get(opCtx);
-          return databaseHolder->getDb(opCtx, dbName);
+          return databaseHolder->getDb(opCtx, tenantDbName);
       }()) {
     // Locking multiple databases is only supported in intent read mode (MODE_IS).
     invariant(secondaryDbNames.empty() || mode == MODE_IS);
@@ -206,7 +209,8 @@ Database* AutoGetDb::ensureDbExists(OperationContext* opCtx) {
     }
 
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    _db = databaseHolder->openDb(opCtx, _dbName, nullptr);
+    const TenantDatabaseName tenantDbName(boost::none, _dbName);
+    _db = databaseHolder->openDb(opCtx, tenantDbName, nullptr);
 
     auto dss = DatabaseShardingState::get(opCtx, _dbName);
     auto dssLock = DatabaseShardingState::DSSLock::lockShared(opCtx, dss);
@@ -275,12 +279,14 @@ AutoGetCollection::AutoGetCollection(
         auto secondaryResolvedNss =
             catalog->resolveNamespaceStringOrUUID(opCtx, secondaryNssOrUUID);
         auto secondaryColl = catalog->lookupCollectionByNamespace(opCtx, secondaryResolvedNss);
+        // TODO SERVER-62926 Change collection lock RAII types to use TenantNamespace
+        const TenantDatabaseName secondaryTenantDbName(boost::none, secondaryNssOrUUID.db());
         verifyDbAndCollection(opCtx,
                               MODE_IS,
                               secondaryNssOrUUID,
                               secondaryResolvedNss,
                               secondaryColl,
-                              databaseHolder->getDb(opCtx, secondaryNssOrUUID.db()));
+                              databaseHolder->getDb(opCtx, secondaryTenantDbName));
 
         // Flag if a secondary namespace is a view.
         if (!_secondaryNssIsView && isSecondaryNssAView(opCtx, secondaryResolvedNss)) {
@@ -401,7 +407,8 @@ AutoGetCollectionLockFree::AutoGetCollectionLockFree(OperationContext* opCtx,
     }
 
     // Returns nullptr for 'viewCatalog' if db does not exist.
-    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, _resolvedNss.db());
+    const TenantDatabaseName tenantDbName(boost::none, _resolvedNss.db());
+    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
     if (!viewCatalog) {
         return;
     }
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index 9f3aa6fd8aa..7d289892fa5 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -102,7 +102,9 @@ struct Cloner::Fun {
 
         // Make sure database still exists after we resume from the temp release
         auto databaseHolder = DatabaseHolder::get(opCtx);
-        auto db = databaseHolder->openDb(opCtx, _dbName);
+        // TODO SERVER-63111 use TenantDatabase in the Cloner.
+        const TenantDatabaseName tenantDbName(boost::none, _dbName);
+        auto db = databaseHolder->openDb(opCtx, tenantDbName);
         auto catalog = CollectionCatalog::get(opCtx);
         auto collection = catalog->lookupCollectionByNamespace(opCtx, nss);
         if (!collection) {
@@ -149,7 +151,7 @@ struct Cloner::Fun {
                         repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesFor(opCtx, nss));
                 }
 
-                db = databaseHolder->getDb(opCtx, _dbName);
+                db = databaseHolder->getDb(opCtx, tenantDbName);
                 uassert(28593,
                         str::stream() << "Database " << _dbName << " dropped while cloning",
                         db != nullptr);
@@ -338,7 +340,8 @@ Status Cloner::_createCollectionsForDb(
     const std::vector<CreateCollectionParams>& createCollectionParams,
     const std::string& dbName) {
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->openDb(opCtx, dbName);
+    const TenantDatabaseName tenantDbName(boost::none, dbName);
+    auto db = databaseHolder->openDb(opCtx, tenantDbName);
     invariant(opCtx->lockState()->isDbLockedForMode(dbName, MODE_X));
 
     auto catalog = CollectionCatalog::get(opCtx);
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index 99f44396f33..58f78dca5f0 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -310,8 +310,9 @@ CreateIndexesReply runCreateIndexesOnNewCollection(
     bool createCollImplicitly) {
     WriteUnitOfWork wunit(opCtx);
 
+    const TenantDatabaseName tenantDbName(boost::none, ns.db());
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, ns.db());
+    auto db = databaseHolder->getDb(opCtx, tenantDbName);
     uassert(ErrorCodes::CommandNotSupportedOnView,
             "Cannot create indexes on a view",
             !db || !ViewCatalog::get(opCtx)->lookup(opCtx, ns));
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index 4bcc4350455..41969950a43 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -178,7 +178,9 @@ protected:
                 // When setting the profiling level, create the database if it didn't already exist.
                 // When just reading the profiling level, we do not create the database.
                 auto databaseHolder = DatabaseHolder::get(opCtx);
-                db = databaseHolder->openDb(opCtx, dbName);
+                // TODO SERVER-63109 Make _setProfileSettings pass TenantDatabaseName.
+                const TenantDatabaseName tenantDbName(boost::none, dbName);
+                db = databaseHolder->openDb(opCtx, tenantDbName);
             }
 
             auto newSettings = oldSettings;
diff --git a/src/mongo/db/commands/dbhash.cpp b/src/mongo/db/commands/dbhash.cpp
index a17c4bc77c0..df4f5f5b4dd 100644
--- a/src/mongo/db/commands/dbhash.cpp
+++ b/src/mongo/db/commands/dbhash.cpp
@@ -341,7 +341,7 @@ private:
                                   << minSnapshot->toString(),
                     !minSnapshot || *mySnapshot >= *minSnapshot);
         } else {
-            invariant(opCtx->lockState()->isDbLockedForMode(db->name(), MODE_S));
+            invariant(opCtx->lockState()->isDbLockedForMode(db->name().dbName(), MODE_S));
         }
 
         auto desc = collection->getIndexCatalog()->findIdIndex(opCtx);
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index eaa5ea0b7a6..89d6c775d05 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -330,7 +330,8 @@ public:
                 // Acquire only the global lock and set up a consistent in-memory catalog and
                 // storage snapshot.
                 AutoGetDbForReadMaybeLockFree lockFreeReadBlock(opCtx, dbName);
-                auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, dbName);
+                const TenantDatabaseName tenantDbName(boost::none, dbName);
+                auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
 
                 CurOpFailpointHelpers::waitWhileFailPointEnabled(&hangBeforeListCollections,
                                                                  opCtx,
diff --git a/src/mongo/db/commands/list_databases.cpp b/src/mongo/db/commands/list_databases.cpp
index 6111555e8ce..ccd34311ee1 100644
--- a/src/mongo/db/commands/list_databases.cpp
+++ b/src/mongo/db/commands/list_databases.cpp
@@ -43,6 +43,8 @@
 #include "mongo/db/operation_context.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/storage_engine.h"
+#include "mongo/db/tenant_database_name.h"
+#include "mongo/db/tenant_namespace.h"
 
 namespace mongo {
 
@@ -155,7 +157,7 @@ public:
 
                     AutoGetDbForReadMaybeLockFree lockFreeReadBlock(opCtx, tenantDbName.dbName());
                     // The database could have been dropped since we called 'listDatabases()' above.
-                    if (!DatabaseHolder::get(opCtx)->dbExists(opCtx, tenantDbName.dbName())) {
+                    if (!DatabaseHolder::get(opCtx)->dbExists(opCtx, tenantDbName)) {
                         continue;
                     }
 
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 2df3319d322..ee20bedd196 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -296,8 +296,8 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
     // 'resolvedNamespaces' from changing relative to those in the acquired ViewCatalog. The
     // resolution of the view definitions below might lead into an endless cycle if any are allowed
     // to change.
-    auto viewCatalog =
-        DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, request.getNamespace().db());
+    const TenantDatabaseName tenantDbName(boost::none, request.getNamespace().db());
+    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
 
     std::deque<NamespaceString> involvedNamespacesQueue(pipelineInvolvedNamespaces.begin(),
                                                         pipelineInvolvedNamespaces.end());
@@ -348,8 +348,9 @@ StatusWith<StringMap<ExpressionContext::ResolvedNamespace>> resolveInvolvedNames
                 // require a lookup stage involving a view on the 'local' database.
                 // If the involved namespace is 'local.system.tenantMigration.oplogView', resolve
                 // its view definition.
+                const TenantDatabaseName involvedTenantDbName(boost::none, involvedNs.db());
                 auto involvedDbViewCatalog =
-                    DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, involvedNs.db());
+                    DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, involvedTenantDbName);
 
                 // It is safe to assume that the ViewCatalog for the `local` database always
                 // exists because replica sets forbid dropping the oplog and the `local` database.
@@ -408,7 +409,8 @@ Status collatorCompatibleWithPipeline(OperationContext* opCtx,
                                       StringData dbName,
                                       const CollatorInterface* collator,
                                       const LiteParsedPipeline& liteParsedPipeline) {
-    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, dbName);
+    const TenantDatabaseName tenantDbName(boost::none, dbName);
+    auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
     if (!viewCatalog) {
         return Status::OK();
     }
@@ -714,8 +716,10 @@ Status runAggregate(OperationContext* opCtx,
 
             // Raise an error if 'origNss' is a view. We do not need to check this if we are opening
             // a stream on an entire db or across the cluster.
+            const TenantDatabaseName origTenantDbName(boost::none, origNss.db());
             if (!origNss.isCollectionlessAggregateNS()) {
-                auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, origNss.db());
+                auto viewCatalog =
+                    DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, origTenantDbName);
                 if (viewCatalog) {
                     auto view = viewCatalog->lookup(opCtx, origNss);
                     uassert(ErrorCodes::CommandNotSupportedOnView,
@@ -814,7 +818,8 @@ Status runAggregate(OperationContext* opCtx,
             // Check that the database/view catalog still exist, in case this is a lock-free
             // operation. It's possible for a view to disappear after we release locks below, so
             // it's safe to quit early if the view disappears while running lock-free.
-            auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, nss.db());
+            const TenantDatabaseName tenantDbName(boost::none, nss.db());
+            auto viewCatalog = DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName);
             uassert(ErrorCodes::NamespaceNotFound,
                     str::stream() << "Namespace '" << nss << "' no longer exists",
                     viewCatalog);
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 fab910caad1..4de98443c06 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -489,7 +489,8 @@ private:
         // (Generic FCV reference): TODO SERVER-60912: When kLastLTS is 6.0, remove this FCV-gated
         // upgrade code.
         if (requestedVersion == multiversion::GenericFCV::kLatest) {
-            for (const auto& dbName : DatabaseHolder::get(opCtx)->getNames()) {
+            for (const auto& tenantDbName : DatabaseHolder::get(opCtx)->getNames()) {
+                const auto& dbName = tenantDbName.dbName();
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
                     opCtx,
@@ -603,7 +604,8 @@ private:
         // (Generic FCV reference): TODO SERVER-60912: When kLastLTS is 6.0, remove this FCV-gated
         // downgrade code.
         if (requestedVersion == multiversion::GenericFCV::kLastLTS) {
-            for (const auto& dbName : DatabaseHolder::get(opCtx)->getNames()) {
+            for (const auto& tenantDbName : DatabaseHolder::get(opCtx)->getNames()) {
+                const auto& dbName = tenantDbName.dbName();
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
                     opCtx,
@@ -662,7 +664,6 @@ private:
                             // downgrade process cannot be aborted at this point.
                             return true;
                         }
-
                         NamespaceStringOrUUID nsOrUUID(dbName, collection->uuid());
                         CollMod collModCmd(collection->ns());
                         BSONObjBuilder unusedBuilder;
@@ -690,7 +691,8 @@ private:
 
         if (serverGlobalParams.featureCompatibility
                 .isFCVDowngradingOrAlreadyDowngradedFromLatest()) {
-            for (const auto& dbName : DatabaseHolder::get(opCtx)->getNames()) {
+            for (const auto& tenantDbName : DatabaseHolder::get(opCtx)->getNames()) {
+                const auto& dbName = tenantDbName.dbName();
                 Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
                 catalog::forEachCollectionFromDb(
                     opCtx,
@@ -718,10 +720,10 @@ private:
 
             // TODO SERVER-63171: Only check on last-lts when FCV 5.3 becomes last-continuous.
             // TODO SERVER-63172: Remove once FCV 6.0 becomes last-lts.
-            for (const auto& dbName : DatabaseHolder::get(opCtx)->getNames()) {
-                Lock::DBLock dbLock(opCtx, dbName, MODE_IX);
+            for (const auto& tenantDbName : DatabaseHolder::get(opCtx)->getNames()) {
+                Lock::DBLock dbLock(opCtx, tenantDbName.dbName(), MODE_IX);
                 catalog::forEachCollectionFromDb(
-                    opCtx, dbName, MODE_X, [&](const CollectionPtr& collection) {
+                    opCtx, tenantDbName.dbName(), MODE_X, [&](const CollectionPtr& collection) {
                         auto indexCatalog = collection->getIndexCatalog();
                         auto indexIt = indexCatalog->getIndexIterator(
                             opCtx, true /* includeUnfinishedIndexes */);
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index 13b832c3257..fe481cb4880 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -143,7 +143,7 @@ public:
                 // If there is no collection name present in the input, run validation against all
                 // the collections.
                 if (auto viewCatalog =
-                        DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName.dbName())) {
+                        DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName)) {
                     viewCatalog->iterate(tenantDbName.dbName(),
                                          [this, opCtx](const ViewDefinition& view) {
                                              return _validateView(opCtx, view);
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index aaab7666d87..0fe457bd152 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -709,10 +709,13 @@ AutoGetCollectionMultiForReadCommandLockFree::AutoGetCollectionMultiForReadComma
 }
 
 OldClientContext::OldClientContext(OperationContext* opCtx, const std::string& ns, bool doVersion)
-    : _opCtx(opCtx), _db(DatabaseHolder::get(opCtx)->getDb(opCtx, ns)) {
+    : _opCtx(opCtx) {
+    const auto dbName = nsToDatabaseSubstring(ns);
+    const TenantDatabaseName tenantDbName(boost::none, dbName);
+    _db = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
+
     if (!_db) {
-        const auto dbName = nsToDatabaseSubstring(ns);
-        _db = DatabaseHolder::get(opCtx)->openDb(_opCtx, dbName, &_justCreated);
+        _db = DatabaseHolder::get(opCtx)->openDb(_opCtx, tenantDbName, &_justCreated);
         invariant(_db);
     }
 
@@ -732,8 +735,8 @@ OldClientContext::OldClientContext(OperationContext* opCtx, const std::string& n
     }
 
     stdx::lock_guard<Client> lk(*_opCtx->getClient());
-    currentOp->enter_inlock(ns.c_str(),
-                            CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(_db->name()));
+    currentOp->enter_inlock(
+        ns.c_str(), CollectionCatalog::get(opCtx)->getDatabaseProfileLevel(_db->name().dbName()));
 }
 
 AutoGetCollectionForReadCommandMaybeLockFree::AutoGetCollectionForReadCommandMaybeLockFree(
diff --git a/src/mongo/db/introspect.cpp b/src/mongo/db/introspect.cpp
index 459ddc315b4..6556f8979ad 100644
--- a/src/mongo/db/introspect.cpp
+++ b/src/mongo/db/introspect.cpp
@@ -161,10 +161,10 @@ void profile(OperationContext* opCtx, NetworkOp op) {
 
 
 Status createProfileCollection(OperationContext* opCtx, Database* db) {
-    invariant(opCtx->lockState()->isDbLockedForMode(db->name(), MODE_IX));
+    invariant(opCtx->lockState()->isDbLockedForMode(db->name().dbName(), MODE_IX));
     invariant(!opCtx->shouldParticipateInFlowControl());
 
-    const auto dbProfilingNS = NamespaceString(db->name(), "system.profile");
+    const auto dbProfilingNS = NamespaceString(db->name().dbName(), "system.profile");
 
     // Checking the collection exists must also be done in the WCE retry loop. Only retrying
     // collection creation would endlessly throw errors because the collection exists: must check
diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp
index a47a6055603..08cf284929c 100644
--- a/src/mongo/db/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer_impl.cpp
@@ -1010,8 +1010,9 @@ void OpObserverImpl::onCollMod(OperationContext* opCtx,
     // Make sure the UUID values in the Collection metadata, the Collection object, and the UUID
     // catalog are all present and equal.
     invariant(opCtx->lockState()->isCollectionLockedForMode(nss, MODE_X));
+    const TenantDatabaseName tenantDbName(boost::none, nss.db());
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, nss.db());
+    auto db = databaseHolder->getDb(opCtx, tenantDbName);
     // Some unit tests call the op observer on an unregistered Database.
     if (!db) {
         return;
diff --git a/src/mongo/db/pipeline/process_interface/SConscript b/src/mongo/db/pipeline/process_interface/SConscript
index 930ff5090f3..87c6fe453dd 100644
--- a/src/mongo/db/pipeline/process_interface/SConscript
+++ b/src/mongo/db/pipeline/process_interface/SConscript
@@ -49,6 +49,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/collection_index_usage_tracker',
         '$BUILD_DIR/mongo/db/concurrency/flow_control_ticketholder',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_mongod',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/repl/primary_only_service',
         '$BUILD_DIR/mongo/db/session_catalog',
         '$BUILD_DIR/mongo/db/stats/fill_locker_info',
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
index 3e74385e31a..2cf8fa19807 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
@@ -450,13 +450,15 @@ bool CommonMongodProcessInterface::fieldsHaveSupportingUniqueIndex(
     const NamespaceString& nss,
     const std::set<FieldPath>& fieldPaths) const {
     auto* opCtx = expCtx->opCtx;
+
     // We purposefully avoid a helper like AutoGetCollection here because we don't want to check the
     // db version or do anything else. We simply want to protect against concurrent modifications to
     // the catalog.
     Lock::DBLock dbLock(opCtx, nss.db(), MODE_IS);
     Lock::CollectionLock collLock(opCtx, nss, MODE_IS);
+    const TenantDatabaseName tenantDbName(boost::none, nss.db());
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, nss.db());
+    auto db = databaseHolder->getDb(opCtx, tenantDbName);
     auto collection =
         db ? CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss) : nullptr;
     if (!collection) {
diff --git a/src/mongo/db/repair.cpp b/src/mongo/db/repair.cpp
index 31bbc302e9d..7332440372b 100644
--- a/src/mongo/db/repair.cpp
+++ b/src/mongo/db/repair.cpp
@@ -59,6 +59,7 @@
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_repair_observer.h"
 #include "mongo/db/storage/storage_util.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/db/vector_clock.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/scopeguard.h"
@@ -128,31 +129,33 @@ Status repairCollections(OperationContext* opCtx,
 }  // namespace
 
 namespace repair {
-Status repairDatabase(OperationContext* opCtx, StorageEngine* engine, const std::string& dbName) {
+Status repairDatabase(OperationContext* opCtx,
+                      StorageEngine* engine,
+                      const TenantDatabaseName& tenantDbName) {
     DisableDocumentValidation validationDisabler(opCtx);
 
     // We must hold some form of lock here
     invariant(opCtx->lockState()->isW());
-    invariant(dbName.find('.') == std::string::npos);
+    invariant(tenantDbName.dbName().find('.') == std::string::npos);
 
-    LOGV2(21029, "repairDatabase", "db"_attr = dbName);
+    LOGV2(21029, "repairDatabase", "db"_attr = tenantDbName);
 
 
     opCtx->checkForInterrupt();
 
     // Close the db and invalidate all current users and caches.
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    databaseHolder->close(opCtx, dbName);
+    databaseHolder->close(opCtx, tenantDbName);
 
     // Reopening db is necessary for repairCollections.
-    databaseHolder->openDb(opCtx, dbName);
+    databaseHolder->openDb(opCtx, tenantDbName);
 
-    auto status = repairCollections(opCtx, engine, dbName);
+    auto status = repairCollections(opCtx, engine, tenantDbName.dbName());
     if (!status.isOK()) {
         LOGV2_FATAL_CONTINUE(21030,
                              "Failed to repair database {dbName}: {status_reason}",
                              "Failed to repair database",
-                             "db"_attr = dbName,
+                             "db"_attr = tenantDbName,
                              "error"_attr = status);
     }
 
diff --git a/src/mongo/db/repair.h b/src/mongo/db/repair.h
index 719b11c57a3..391c391f91d 100644
--- a/src/mongo/db/repair.h
+++ b/src/mongo/db/repair.h
@@ -34,6 +34,7 @@
 
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/record_id.h"
+#include "mongo/db/tenant_database_name.h"
 
 namespace mongo {
 class StorageEngine;
@@ -50,7 +51,9 @@ namespace repair {
  *
  * It is expected that the local database will be repaired first when running in repair mode.
  */
-Status repairDatabase(OperationContext* opCtx, StorageEngine* engine, const std::string& dbName);
+Status repairDatabase(OperationContext* opCtx,
+                      StorageEngine* engine,
+                      const TenantDatabaseName& tenantDbName);
 
 /**
  * Repairs a collection using a storage engine-specific, best-effort process.
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 0df82726b10..20c70ef15ec 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -61,6 +61,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/dbdirectclient',
         '$BUILD_DIR/mongo/db/dbhelpers',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/op_observer',
         '$BUILD_DIR/mongo/db/op_observer_util',
         '$BUILD_DIR/mongo/db/pipeline/change_stream_pre_image_helpers',
@@ -245,6 +246,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/database_holder',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/record_id_helpers',
         '$BUILD_DIR/mongo/db/storage/oplog_cap_maintainer_thread',
         '$BUILD_DIR/mongo/db/storage/record_store_base',
@@ -435,6 +437,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/index_build_oplog_entry',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/s/sharding_runtime_d',
         '$BUILD_DIR/mongo/db/storage/journal_flusher',
     ],
@@ -472,6 +475,7 @@ env.Library(
         'rollback_test_fixture.cpp',
     ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/op_observer',
         '$BUILD_DIR/mongo/db/query_exec',
         '$BUILD_DIR/mongo/db/read_write_concern_defaults_mock',
@@ -506,6 +510,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/import_collection_oplog_entry',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/s/sharding_runtime_d',
         '$BUILD_DIR/mongo/idl/server_parameter',
         'drop_pending_collection_reaper',
@@ -1540,6 +1545,7 @@ if wiredtiger:
             '$BUILD_DIR/mongo/db/catalog/database_holder',
             '$BUILD_DIR/mongo/db/catalog/document_validation',
             '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
+            '$BUILD_DIR/mongo/db/multitenancy',
             '$BUILD_DIR/mongo/db/service_context_d_test_fixture',
             '$BUILD_DIR/mongo/db/storage/wiredtiger/storage_wiredtiger',
             'drop_pending_collection_reaper',
diff --git a/src/mongo/db/repl/apply_ops.cpp b/src/mongo/db/repl/apply_ops.cpp
index 8cd9fb586b3..2e4ba55139d 100644
--- a/src/mongo/db/repl/apply_ops.cpp
+++ b/src/mongo/db/repl/apply_ops.cpp
@@ -53,6 +53,7 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/session_catalog_mongod.h"
+#include "mongo/db/tenant_database_name.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/logv2/log.h"
 #include "mongo/rpc/get_status_from_command_result.h"
@@ -106,8 +107,10 @@ Status _applyOps(OperationContext* opCtx,
             // ApplyOps does not have the global writer lock when applying transaction
             // operations, so we need to acquire the DB and Collection locks.
             Lock::DBLock dbLock(opCtx, nss.db(), MODE_IX);
+            // TODO SERVER-62880 Parse the tenant id from the TenantNamespace.
+            const TenantDatabaseName tenantDbName(boost::none, nss.db());
             auto databaseHolder = DatabaseHolder::get(opCtx);
-            auto db = databaseHolder->getDb(opCtx, nss.ns());
+            auto db = databaseHolder->getDb(opCtx, tenantDbName);
             if (!db) {
                 // Retry in non-atomic mode, since MMAP cannot implicitly create a new database
                 // within an active WriteUnitOfWork.
@@ -296,8 +299,9 @@ Status _checkPrecondition(OperationContext* opCtx,
         BSONObj realres = cursor->more() ? cursor->nextSafe() : BSONObj{};
 
         // Get collection default collation.
+        const TenantDatabaseName tenantDbName(boost::none, nss.db());
         auto databaseHolder = DatabaseHolder::get(opCtx);
-        auto database = databaseHolder->getDb(opCtx, nss.db());
+        auto database = databaseHolder->getDb(opCtx, tenantDbName);
         if (!database) {
             return {ErrorCodes::NamespaceNotFound, "database in ns does not exist: " + nss.ns()};
         }
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 61049e2df5d..232bf268931 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -199,8 +199,9 @@ void createIndexForApplyOps(OperationContext* opCtx,
     invariant(opCtx->lockState()->isCollectionLockedForMode(indexNss, MODE_X));
 
     // Check if collection exists.
+    const TenantDatabaseName tenantDbName(boost::none, indexNss.db());
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, indexNss.ns());
+    auto db = databaseHolder->getDb(opCtx, tenantDbName);
     auto indexCollection =
         db ? CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, indexNss) : nullptr;
     uassert(ErrorCodes::NamespaceNotFound,
@@ -1827,8 +1828,9 @@ Status applyCommand_inlock(OperationContext* opCtx,
         // Command application doesn't always acquire the global writer lock for transaction
         // commands, so we acquire its own locks here.
         Lock::DBLock lock(opCtx, nss.db(), MODE_IS);
+        const TenantDatabaseName tenantDbName(boost::none, nss.db());
         auto databaseHolder = DatabaseHolder::get(opCtx);
-        auto db = databaseHolder->getDb(opCtx, nss.ns());
+        auto db = databaseHolder->getDb(opCtx, tenantDbName);
         if (db && !CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss) &&
             ViewCatalog::get(opCtx)->lookup(opCtx, nss)) {
             return {ErrorCodes::CommandNotSupportedOnView,
diff --git a/src/mongo/db/repl/oplog_applier_impl_test.cpp b/src/mongo/db/repl/oplog_applier_impl_test.cpp
index 283482ac5c3..5256b85ff9b 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test.cpp
@@ -2111,7 +2111,8 @@ TEST_F(OplogApplierImplTest, ApplyGroupIgnoresUpdateOperationIfDocumentIsMissing
         Lock::GlobalWrite globalLock(_opCtx.get());
         bool justCreated = false;
         auto databaseHolder = DatabaseHolder::get(_opCtx.get());
-        auto db = databaseHolder->openDb(_opCtx.get(), nss.db(), &justCreated);
+        const TenantDatabaseName tenantDbName(boost::none, nss.db());
+        auto db = databaseHolder->openDb(_opCtx.get(), tenantDbName, &justCreated);
         ASSERT_TRUE(db);
         ASSERT_TRUE(justCreated);
     }
diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
index 4900d1905c0..2c10476876f 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
@@ -472,7 +472,8 @@ void createDatabase(OperationContext* opCtx, StringData dbName) {
     Lock::GlobalWrite globalLock(opCtx);
     bool justCreated;
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->openDb(opCtx, dbName, &justCreated);
+    const TenantDatabaseName tenantDbName(boost::none, dbName);
+    auto db = databaseHolder->openDb(opCtx, tenantDbName, &justCreated);
     ASSERT_TRUE(db);
     ASSERT_TRUE(justCreated);
 }
diff --git a/src/mongo/db/repl/rollback_impl.cpp b/src/mongo/db/repl/rollback_impl.cpp
index cc543b5b5ce..b80cc9c0cf9 100644
--- a/src/mongo/db/repl/rollback_impl.cpp
+++ b/src/mongo/db/repl/rollback_impl.cpp
@@ -64,6 +64,8 @@
 #include "mongo/db/session_catalog_mongod.h"
 #include "mongo/db/session_txn_record_gen.h"
 #include "mongo/db/storage/remove_saver.h"
+#include "mongo/db/tenant_database_name.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/db/transaction_history_iterator.h"
 #include "mongo/logv2/log.h"
 #include "mongo/s/catalog/type_config_version.h"
@@ -1365,7 +1367,7 @@ void RollbackImpl::_resetDropPendingState(OperationContext* opCtx) {
     auto databaseHolder = DatabaseHolder::get(opCtx);
     for (const auto& tenantDbName : tenantDbNames) {
         Lock::DBLock dbLock(opCtx, tenantDbName.dbName(), MODE_X);
-        auto db = databaseHolder->openDb(opCtx, tenantDbName.dbName());
+        auto db = databaseHolder->openDb(opCtx, tenantDbName);
         db->checkForIdIndexesAndDropPendingCollections(opCtx);
     }
 }
diff --git a/src/mongo/db/repl/rollback_test_fixture.cpp b/src/mongo/db/repl/rollback_test_fixture.cpp
index dd137984a26..b4a064a7bce 100644
--- a/src/mongo/db/repl/rollback_test_fixture.cpp
+++ b/src/mongo/db/repl/rollback_test_fixture.cpp
@@ -213,7 +213,8 @@ Collection* RollbackTest::_createCollection(OperationContext* opCtx,
     Lock::DBLock dbLock(opCtx, nss.db(), MODE_X);
     mongo::WriteUnitOfWork wuow(opCtx);
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->openDb(opCtx, nss.db());
+    const TenantDatabaseName tenantDbName(boost::none, nss.db());
+    auto db = databaseHolder->openDb(opCtx, tenantDbName);
     ASSERT_TRUE(db);
     db->dropCollection(opCtx, nss).transitional_ignore();
     auto coll = db->createCollection(opCtx, nss, options);
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 92496e9ef46..3fc5bd47d41 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -1087,7 +1087,7 @@ void renameOutOfTheWay(OperationContext* opCtx, RenameCollectionInfo info, Datab
 
     // The generated unique collection name is only guaranteed to exist if the database is
     // exclusively locked.
-    invariant(opCtx->lockState()->isDbLockedForMode(db->name(), LockMode::MODE_X));
+    invariant(opCtx->lockState()->isDbLockedForMode(db->name().dbName(), LockMode::MODE_X));
     // Creates the oplog entry to temporarily rename the collection that is
     // preventing the renameCollection command from rolling back to a unique
     // namespace.
@@ -1139,6 +1139,7 @@ void renameOutOfTheWay(OperationContext* opCtx, RenameCollectionInfo info, Datab
 void rollbackRenameCollection(OperationContext* opCtx, UUID uuid, RenameCollectionInfo info) {
 
     auto dbName = info.renameFrom.db();
+    const TenantDatabaseName tenantDbName(boost::none, dbName);
 
     LOGV2(21679,
           "Attempting to rename collection with UUID: {uuid}, from: {renameFrom}, to: "
@@ -1149,7 +1150,7 @@ void rollbackRenameCollection(OperationContext* opCtx, UUID uuid, RenameCollecti
           "renameTo"_attr = info.renameTo);
     Lock::DBLock dbLock(opCtx, dbName, MODE_X);
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->openDb(opCtx, dbName);
+    auto db = databaseHolder->openDb(opCtx, tenantDbName);
     invariant(db);
 
     auto status = renameCollectionForRollback(opCtx, info.renameTo, uuid);
@@ -1576,10 +1577,11 @@ void rollback_internal::syncFixUp(OperationContext* opCtx,
                   "namespace"_attr = *nss,
                   "uuid"_attr = uuid);
 
+            const TenantDatabaseName tenantDbName(boost::none, nss->db());
             Lock::DBLock dbLock(opCtx, nss->db(), MODE_X);
 
             auto databaseHolder = DatabaseHolder::get(opCtx);
-            auto db = databaseHolder->openDb(opCtx, nss->db().toString());
+            auto db = databaseHolder->openDb(opCtx, tenantDbName);
             invariant(db);
 
             CollectionWriter collection(opCtx, uuid);
diff --git a/src/mongo/db/repl/rs_rollback_test.cpp b/src/mongo/db/repl/rs_rollback_test.cpp
index d95aae3f5a8..cdeff56a1d5 100644
--- a/src/mongo/db/repl/rs_rollback_test.cpp
+++ b/src/mongo/db/repl/rs_rollback_test.cpp
@@ -386,7 +386,7 @@ int _testRollbackDelete(OperationContext* opCtx,
     Lock::DBLock dbLock(opCtx, "test", MODE_S);
     Lock::CollectionLock collLock(opCtx, NamespaceString("test.t"), MODE_S);
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, "test");
+    auto db = databaseHolder->getDb(opCtx, TenantDatabaseName(boost::none, "test"));
     ASSERT_TRUE(db);
     auto collection = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(
         opCtx, NamespaceString("test.t"));
@@ -411,7 +411,7 @@ TEST_F(RSRollbackTest, RollbackDeleteDocCmdCollectionAtSourceDropped) {
     {
         Lock::DBLock dbLock(_opCtx.get(), nss.db(), MODE_X);
         auto databaseHolder = DatabaseHolder::get(_opCtx.get());
-        auto db = databaseHolder->openDb(_opCtx.get(), nss.db());
+        auto db = databaseHolder->openDb(_opCtx.get(), TenantDatabaseName(boost::none, nss.db()));
         ASSERT_TRUE(db);
     }
     ASSERT_EQUALS(-1,
@@ -1961,7 +1961,7 @@ TEST_F(RSRollbackTest, RollbackCreateCollectionCommand) {
     {
         Lock::DBLock dbLock(_opCtx.get(), "test", MODE_S);
         auto databaseHolder = DatabaseHolder::get(_opCtx.get());
-        auto db = databaseHolder->getDb(_opCtx.get(), "test");
+        auto db = databaseHolder->getDb(_opCtx.get(), TenantDatabaseName(boost::none, "test"));
         ASSERT_TRUE(db);
         ASSERT_FALSE(CollectionCatalog::get(_opCtx.get())
                          ->lookupCollectionByNamespace(_opCtx.get(), NamespaceString("test.t")));
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 716743743a3..8e27ac5a4ca 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -80,6 +80,8 @@
 #include "mongo/db/storage/control/journal_flusher.h"
 #include "mongo/db/storage/control/storage_control.h"
 #include "mongo/db/storage/oplog_cap_maintainer_thread.h"
+#include "mongo/db/tenant_database_name.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/background.h"
@@ -429,7 +431,7 @@ Status StorageInterfaceImpl::dropReplicatedDatabases(OperationContext* opCtx) {
             continue;
         }
         writeConflictRetry(opCtx, "dropReplicatedDatabases", tenantDbName.dbName(), [&] {
-            if (auto db = databaseHolder->getDb(opCtx, tenantDbName.dbName())) {
+            if (auto db = databaseHolder->getDb(opCtx, tenantDbName)) {
                 databaseHolder->dropDb(opCtx, db);
             } else {
                 // This is needed since dropDatabase can't be rolled back.
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index 3afeea37291..8d128178c2c 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -377,6 +377,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/db/commands/txn_cmd_request',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/repl/primary_only_service',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/repl/replica_set_messages',
diff --git a/src/mongo/db/s/rename_collection_coordinator.cpp b/src/mongo/db/s/rename_collection_coordinator.cpp
index 655faf971f4..d5c76a048b3 100644
--- a/src/mongo/db/s/rename_collection_coordinator.cpp
+++ b/src/mongo/db/s/rename_collection_coordinator.cpp
@@ -194,7 +194,8 @@ ExecutorFuture<void> RenameCollectionCoordinator::_runImpl(
                     // Make sure the target namespace is not a view
                     {
                         Lock::DBLock dbLock(opCtx, toNss.db(), MODE_IS);
-                        const auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, toNss.db());
+                        const TenantDatabaseName tenantDbName(boost::none, toNss.db());
+                        const auto db = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
                         if (db) {
                             uassert(ErrorCodes::CommandNotSupportedOnView,
                                     str::stream() << "Can't rename to target collection `" << toNss
diff --git a/src/mongo/db/startup_recovery.cpp b/src/mongo/db/startup_recovery.cpp
index 0c2aed85986..e9daca7e46d 100644
--- a/src/mongo/db/startup_recovery.cpp
+++ b/src/mongo/db/startup_recovery.cpp
@@ -54,6 +54,7 @@
 #include "mongo/db/repl_set_member_in_standalone_mode.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/storage/storage_repair_observer.h"
+#include "mongo/db/tenant_namespace.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/fail_point.h"
@@ -84,11 +85,12 @@ Status restoreMissingFeatureCompatibilityVersionDocument(OperationContext* opCtx
     // If the admin database, which contains the server configuration collection with the
     // featureCompatibilityVersion document, does not exist, create it.
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, fcvNss.db());
+    const TenantDatabaseName fcvTenantDbName(boost::none, fcvNss.db());
+    auto db = databaseHolder->getDb(opCtx, fcvTenantDbName);
     if (!db) {
         LOGV2(20998, "Re-creating admin database that was dropped.");
     }
-    db = databaseHolder->openDb(opCtx, fcvNss.db());
+    db = databaseHolder->openDb(opCtx, fcvTenantDbName);
     invariant(db);
 
     // If the server configuration collection, which contains the FCV document, does not exist, then
@@ -212,7 +214,8 @@ Status ensureCollectionProperties(OperationContext* opCtx,
                                   Database* db,
                                   EnsureIndexPolicy ensureIndexPolicy) {
     auto catalog = CollectionCatalog::get(opCtx);
-    for (auto collIt = catalog->begin(opCtx, db->name()); collIt != catalog->end(opCtx); ++collIt) {
+    for (auto collIt = catalog->begin(opCtx, db->name().dbName()); collIt != catalog->end(opCtx);
+         ++collIt) {
         auto coll = collIt.getWritableCollection(opCtx, CollectionCatalog::LifetimeMode::kInplace);
         if (!coll) {
             break;
@@ -260,7 +263,7 @@ void openDatabases(OperationContext* opCtx, const StorageEngine* storageEngine,
     auto tenantDbNames = storageEngine->listDatabases();
     for (const auto& tenantDbName : tenantDbNames) {
         LOGV2_DEBUG(21010, 1, "    Opening database: {dbName}", "dbName"_attr = tenantDbName);
-        auto db = databaseHolder->openDb(opCtx, tenantDbName.dbName());
+        auto db = databaseHolder->openDb(opCtx, tenantDbName);
         invariant(db);
 
         onDatabase(db);
@@ -277,7 +280,8 @@ bool hasReplSetConfigDoc(OperationContext* opCtx) {
     // 'kSystemReplSetNamespace' collection have been populated if the collection exists. If the
     // "local" database doesn't exist at this point yet, then it will be created.
     const auto nss = NamespaceString::kSystemReplSetNamespace;
-    databaseHolder->openDb(opCtx, nss.db());
+    const TenantDatabaseName tenantDbName(boost::none, nss.db());
+    databaseHolder->openDb(opCtx, tenantDbName);
     BSONObj config;
     return Helpers::getSingleton(opCtx, nss.ns().c_str(), config);
 }
@@ -470,7 +474,9 @@ void startupRepair(OperationContext* opCtx, StorageEngine* storageEngine) {
     if (auto fcvColl = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(
             opCtx, NamespaceString::kServerConfigurationNamespace)) {
         auto databaseHolder = DatabaseHolder::get(opCtx);
-        databaseHolder->openDb(opCtx, fcvColl->ns().db());
+
+        const TenantDatabaseName fcvTenantDbName(boost::none, fcvColl->ns().db());
+        databaseHolder->openDb(opCtx, fcvTenantDbName);
         fassertNoTrace(4805000,
                        repair::repairCollection(
                            opCtx, storageEngine, NamespaceString::kServerConfigurationNamespace));
@@ -487,7 +493,7 @@ void startupRepair(OperationContext* opCtx, StorageEngine* storageEngine) {
                             tenantDbNames.end(),
                             TenantDatabaseName(boost::none, NamespaceString::kLocalDb));
         it != tenantDbNames.end()) {
-        fassertNoTrace(4805001, repair::repairDatabase(opCtx, storageEngine, it->dbName()));
+        fassertNoTrace(4805001, repair::repairDatabase(opCtx, storageEngine, *it));
 
         // This must be set before rebuilding index builds on replicated collections.
         setReplSetMemberInStandaloneMode(opCtx, StartupRecoveryMode::kAuto);
@@ -496,7 +502,7 @@ void startupRepair(OperationContext* opCtx, StorageEngine* storageEngine) {
 
     // Repair the remaining databases.
     for (const auto& tenantDbName : tenantDbNames) {
-        fassertNoTrace(18506, repair::repairDatabase(opCtx, storageEngine, tenantDbName.dbName()));
+        fassertNoTrace(18506, repair::repairDatabase(opCtx, storageEngine, tenantDbName));
     }
 
     openDatabases(opCtx, storageEngine, [&](auto db) {
@@ -575,7 +581,7 @@ void startupRecovery(OperationContext* opCtx,
         !(hasReplSetConfigDoc(opCtx) || replSettings.usingReplSets());
 
     openDatabases(opCtx, storageEngine, [&](auto db) {
-        auto dbName = db->name();
+        auto dbName = db->name().dbName();
 
         // Ensures all collections meet requirements such as having _id indexes, and corrects them
         // if needed.
@@ -586,7 +592,7 @@ void startupRecovery(OperationContext* opCtx,
             db->checkForIdIndexesAndDropPendingCollections(opCtx);
             // Ensure oplog is capped (mongodb does not guarantee order of inserts on noncapped
             // collections)
-            if (db->name() == NamespaceString::kLocalDb) {
+            if (dbName == NamespaceString::kLocalDb) {
                 assertCappedOplog(opCtx, db);
             }
         }
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index 3fd2766f798..8594880f8e2 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -40,7 +40,6 @@
 #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"
diff --git a/src/mongo/db/storage/kv/durable_catalog_test.cpp b/src/mongo/db/storage/kv/durable_catalog_test.cpp
index d3d6d8a8de6..6c6190ce5be 100644
--- a/src/mongo/db/storage/kv/durable_catalog_test.cpp
+++ b/src/mongo/db/storage/kv/durable_catalog_test.cpp
@@ -40,7 +40,6 @@
 #include "mongo/db/index/index_descriptor.h"
 #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"
diff --git a/src/mongo/db/storage/wiredtiger/SConscript b/src/mongo/db/storage/wiredtiger/SConscript
index d93b4bb07fc..15693d5bf7b 100644
--- a/src/mongo/db/storage/wiredtiger/SConscript
+++ b/src/mongo/db/storage/wiredtiger/SConscript
@@ -56,6 +56,7 @@ wtEnv.Library(
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/global_settings',
         '$BUILD_DIR/mongo/db/index/index_descriptor',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/prepare_conflict_tracker',
         '$BUILD_DIR/mongo/db/record_id_helpers',
diff --git a/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp b/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp
index 58788e9439a..08011cb3a5a 100644
--- a/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp
+++ b/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp
@@ -35,6 +35,7 @@
 #include "mongo/db/commands/server_status.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/tenant_database_name.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
@@ -68,8 +69,8 @@ public:
         AutoGetOplog oplogRead(opCtx, OplogAccessMode::kRead);
         const auto& oplog = oplogRead.getCollection();
         if (oplog) {
-            const auto localDb =
-                DatabaseHolder::get(opCtx)->getDb(opCtx, NamespaceString::kLocalDb);
+            const TenantDatabaseName tenantDbName(boost::none, NamespaceString::kLocalDb);
+            const auto localDb = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
             invariant(localDb);
             AutoStatsTracker statsTracker(
                 opCtx,
diff --git a/src/mongo/db/tenant_database_name.cpp b/src/mongo/db/tenant_database_name.cpp
new file mode 100644
index 00000000000..457763d206b
--- /dev/null
+++ b/src/mongo/db/tenant_database_name.cpp
@@ -0,0 +1,58 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/tenant_database_name.h"
+
+#include "mongo/db/multitenancy_gen.h"
+#include "mongo/db/server_feature_flags_gen.h"
+
+namespace mongo {
+
+TenantDatabaseName::TenantDatabaseName(boost::optional<TenantId> tenantId, StringData dbName) {
+    // TODO SERVER-62114 Check instead if gMultitenancySupport is enabled.
+    if (gFeatureFlagRequireTenantID.isEnabledAndIgnoreFCV())
+        invariant(tenantId);
+
+    _tenantId = tenantId;
+    _dbName = dbName.toString();
+
+    _tenantDbName =
+        _tenantId ? boost::make_optional(_tenantId->toString() + "_" + _dbName) : boost::none;
+}
+
+TenantDatabaseName TenantDatabaseName::createSystemTenantDbName(StringData dbName) {
+    // TODO SERVER-62114 Check instead if gMultitenancySupport is enabled.
+    if (gFeatureFlagRequireTenantID.isEnabledAndIgnoreFCV()) {
+        return TenantDatabaseName(TenantId::kSystemTenantId, dbName);
+    }
+
+    return TenantDatabaseName(boost::none, dbName);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/tenant_database_name.h b/src/mongo/db/tenant_database_name.h
index 615e832c735..ade9538728a 100644
--- a/src/mongo/db/tenant_database_name.h
+++ b/src/mongo/db/tenant_database_name.h
@@ -29,14 +29,21 @@
 
 #pragma once
 #include <algorithm>
+#include <boost/algorithm/string/predicate.hpp>
 #include <boost/optional.hpp>
 #include <string>
 
 #include "mongo/base/string_data.h"
-#include "mongo/db/server_feature_flags_gen.h"
 #include "mongo/db/tenant_id.h"
+#include "mongo/logv2/log_attr.h"
 
 namespace mongo {
+
+/**
+ * A TenantDatabaseName is a unique name for database.
+ * It holds a database name and tenant id, if one exists. In a serverless environment, a tenant id
+ * is expected to exist so that a database can be uniquely identified.
+ */
 class TenantDatabaseName {
 public:
     /**
@@ -46,17 +53,9 @@ public:
      *
      * If featureFlagRequireTenantID is set, tenantId is required.
      */
-    TenantDatabaseName(boost::optional<TenantId> tenantId, StringData dbName) {
-        // TODO SERVER-62114 Check instead if gMultitenancySupport is enabled.
-        if (gFeatureFlagRequireTenantID.isEnabledAndIgnoreFCV())
-            invariant(tenantId);
+    TenantDatabaseName(boost::optional<TenantId> tenantId, StringData dbName);
 
-        _tenantId = tenantId;
-        _dbName = dbName.toString();
-
-        _tenantDbName =
-            _tenantId ? boost::make_optional(_tenantId->toString() + "_" + _dbName) : boost::none;
-    }
+    static TenantDatabaseName createSystemTenantDbName(StringData dbName);
 
     const boost::optional<TenantId> tenantId() const {
         return _tenantId;
@@ -78,6 +77,10 @@ public:
         return fullName();
     }
 
+    bool equalCaseInsensitive(const TenantDatabaseName& other) const {
+        return boost::iequals(fullName(), other.fullName());
+    }
+
     /**
      * Returns -1, 0, or 1 if 'this' is less, equal, or greater than 'other' in
      * lexicographical order.
@@ -101,6 +104,14 @@ private:
     boost::optional<std::string> _tenantDbName;
 };
 
+inline std::ostream& operator<<(std::ostream& stream, const TenantDatabaseName& tdb) {
+    return stream << tdb.fullName();
+}
+
+inline StringBuilder& operator<<(StringBuilder& builder, const TenantDatabaseName& tdb) {
+    return builder << tdb.fullName();
+}
+
 inline bool operator==(const TenantDatabaseName& lhs, const TenantDatabaseName& rhs) {
     return lhs.compare(rhs) == 0;
 }
diff --git a/src/mongo/db/tenant_database_name_test.cpp b/src/mongo/db/tenant_database_name_test.cpp
index 4c91d9da8c1..75d9550afef 100644
--- a/src/mongo/db/tenant_database_name_test.cpp
+++ b/src/mongo/db/tenant_database_name_test.cpp
@@ -27,7 +27,6 @@
  *    it in the license file.
  */
 
-#include "mongo/db/multitenancy.h"
 #include "mongo/db/server_feature_flags_gen.h"
 #include "mongo/db/tenant_database_name.h"
 #include "mongo/idl/server_parameter_test_util.h"
@@ -39,11 +38,12 @@ namespace {
 
 TEST(TenantDatabaseNameTest, MultitenancySupportDisabled) {
     TenantDatabaseName tdnWithoutTenant1(boost::none, "a");
+
     ASSERT(!tdnWithoutTenant1.tenantId());
     ASSERT_EQUALS(std::string("a"), tdnWithoutTenant1.dbName());
     ASSERT_EQUALS(std::string("a"), tdnWithoutTenant1.fullName());
 
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     TenantDatabaseName tdnWithTenant(tenantId, "a");
     ASSERT(tdnWithTenant.tenantId());
     ASSERT_EQUALS(tenantId, *tdnWithTenant.tenantId());
@@ -60,7 +60,7 @@ TEST(TenantDatabaseNameTest, MultitenancySupportEnabledTenantIDNotRequired) {
     ASSERT_EQUALS(std::string("a"), tdnWithoutTenant.dbName());
     ASSERT_EQUALS(std::string("a"), tdnWithoutTenant.fullName());
 
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     TenantDatabaseName tdnWithTenant(tenantId, "a");
     ASSERT(tdnWithTenant.tenantId());
     ASSERT_EQUALS(tenantId, *tdnWithTenant.tenantId());
@@ -81,7 +81,7 @@ TEST(TenantDatabaseNameTest, TenantIDRequiredBasic) {
     // TODO SERVER-62114 Remove enabling this feature flag.
     RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
 
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     TenantDatabaseName tdn(tenantId, "a");
     ASSERT(tdn.tenantId());
     ASSERT_EQUALS(tenantId, *tdn.tenantId());
@@ -90,7 +90,7 @@ TEST(TenantDatabaseNameTest, TenantIDRequiredBasic) {
 }
 
 TEST(TenantDatabaseNameTest, VerifyEqualsOperator) {
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     TenantDatabaseName tdn(tenantId, "a");
     ASSERT_TRUE(TenantDatabaseName(tenantId, "a") == tdn);
     ASSERT_TRUE(TenantDatabaseName(tenantId, "b") != tdn);
@@ -101,8 +101,8 @@ TEST(TenantDatabaseNameTest, VerifyEqualsOperator) {
 }
 
 TEST(TenantDatabaseNameTest, VerifyHashFunction) {
-    TenantId tenantId1 = TenantId(OID::gen());
-    TenantId tenantId2 = TenantId(OID::gen());
+    TenantId tenantId1(OID::gen());
+    TenantId tenantId2(OID::gen());
     TenantDatabaseName tdn1 = TenantDatabaseName(tenantId1, "a");
     TenantDatabaseName tdn2 = TenantDatabaseName(tenantId2, "a");
     TenantDatabaseName tdn3 = TenantDatabaseName(boost::none, "a");
diff --git a/src/mongo/db/tenant_namespace.cpp b/src/mongo/db/tenant_namespace.cpp
index 1c08a6e2357..d1742e639d8 100644
--- a/src/mongo/db/tenant_namespace.cpp
+++ b/src/mongo/db/tenant_namespace.cpp
@@ -45,6 +45,10 @@ TenantNamespace::TenantNamespace(boost::optional<mongo::TenantId> tenantId, Name
                              : boost::none;
 }
 
+TenantDatabaseName TenantNamespace::createTenantDatabaseName() const {
+    return TenantDatabaseName(_tenantId, _nss.db());
+}
+
 TenantNamespace TenantNamespace::parseTenantNamespaceFromDisk(StringData ns) {
     if (!gMultitenancySupport) {
         return TenantNamespace(boost::none, NamespaceString(ns));
diff --git a/src/mongo/db/tenant_namespace.h b/src/mongo/db/tenant_namespace.h
index 1e70d99a55a..1aebcb5a43c 100644
--- a/src/mongo/db/tenant_namespace.h
+++ b/src/mongo/db/tenant_namespace.h
@@ -37,6 +37,7 @@
 #include "mongo/base/string_data.h"
 #include "mongo/bson/util/builder.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/tenant_database_name.h"
 #include "mongo/db/tenant_id.h"
 #include "mongo/logv2/log_attr.h"
 
@@ -62,6 +63,11 @@ public:
     TenantNamespace(boost::optional<mongo::TenantId> tenantId, NamespaceString nss);
 
     /**
+     * Create a TenantDatabaseName from the TenantNamespace.
+     */
+    TenantDatabaseName createTenantDatabaseName() const;
+
+    /**
      * Constructs a TenantNamespace from the string "ns". When the server parameter
      * "multitenancySupport” is enabled, the tenantId will be parsed separately from the database
      * name. If it is disabled, the tenantId will be parsed as a prefix of the database name, and
diff --git a/src/mongo/db/tenant_namespace_test.cpp b/src/mongo/db/tenant_namespace_test.cpp
index 3c1c4684ecb..8cf26821daf 100644
--- a/src/mongo/db/tenant_namespace_test.cpp
+++ b/src/mongo/db/tenant_namespace_test.cpp
@@ -27,8 +27,6 @@
  *    it in the license file.
  */
 
-#include "mongo/platform/basic.h"
-
 #include "mongo/db/multitenancy_gen.h"
 #include "mongo/db/server_feature_flags_gen.h"
 #include "mongo/db/tenant_namespace.h"
@@ -55,7 +53,7 @@ TEST(TenantNamespaceTest, TenantNamespaceParseFromDiskMultitenancySupportDisable
     ASSERT_EQUALS(std::string("a"), tenantNs.db());
     ASSERT_EQUALS(std::string("b"), tenantNs.coll());
 
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     std::string ns = tenantId.toString() + "_a.b";
     TenantNamespace tenantNs2 = TenantNamespace::parseTenantNamespaceFromDisk(ns);
     ASSERT(!tenantNs2.tenantId());
@@ -76,7 +74,7 @@ TEST(TenantNamespaceTest, TenantNamespaceMultitenancySupportEnabledFeatureFlagDi
 
     // If the feature flag is disabled but a tenantId is given, the tenantId should be parsed
     // separately from the db name.
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     TenantNamespace tenantNs2(tenantId, NamespaceString("a.b"));
     ASSERT(tenantNs2.tenantId());
     ASSERT_EQUALS(tenantId, *tenantNs2.tenantId());
@@ -112,7 +110,7 @@ TEST(TenantNamespaceTest, TenantNamespaceMultitenancySupportEnabledBasic) {
     // TODO SERVER-62114 Remove enabling this feature flag.
     RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
 
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     TenantNamespace tenantNs(tenantId, NamespaceString("a.b"));
     ASSERT(tenantNs.tenantId());
     ASSERT_EQUALS(tenantId, *tenantNs.tenantId());
@@ -126,7 +124,7 @@ TEST(TenantNamespaceTest, TenantNamespaceParseFromDiskMultitenancySupportEnabled
     // TODO SERVER-62114 Remove enabling this feature flag.
     RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
 
-    TenantId tenantId = TenantId(OID::gen());
+    TenantId tenantId(OID::gen());
     std::string tenantNsStr = str::stream() << tenantId << "_a.b";
 
     TenantNamespace tenantNs = TenantNamespace::parseTenantNamespaceFromDisk(tenantNsStr);
diff --git a/src/mongo/db/transaction_history_iterator.cpp b/src/mongo/db/transaction_history_iterator.cpp
index b01259efd5e..5de4a6c90ae 100644
--- a/src/mongo/db/transaction_history_iterator.cpp
+++ b/src/mongo/db/transaction_history_iterator.cpp
@@ -76,7 +76,8 @@ BSONObj findOneOplogEntry(OperationContext* opCtx,
     std::unique_ptr<CanonicalQuery> cq = std::move(statusWithCQ.getValue());
 
     AutoGetOplog oplogRead(opCtx, OplogAccessMode::kRead);
-    const auto localDb = DatabaseHolder::get(opCtx)->getDb(opCtx, NamespaceString::kLocalDb);
+    const TenantDatabaseName tenantDbName(boost::none, NamespaceString::kLocalDb);
+    const auto localDb = DatabaseHolder::get(opCtx)->getDb(opCtx, tenantDbName);
     invariant(localDb);
     AutoStatsTracker statsTracker(
         opCtx,
diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp
index 5f0a6adf34d..d864b9e8a1c 100644
--- a/src/mongo/db/transaction_participant.cpp
+++ b/src/mongo/db/transaction_participant.cpp
@@ -540,8 +540,9 @@ TransactionParticipant::getOldestActiveTimestamp(Timestamp stableTimestamp) {
         Lock::DBLock dbLock(opCtx.get(), nss.db(), MODE_IS, deadline);
         Lock::CollectionLock collLock(opCtx.get(), nss, MODE_IS, deadline);
 
+        const TenantDatabaseName tenantDbName(boost::none, nss.db());
         auto databaseHolder = DatabaseHolder::get(opCtx.get());
-        auto db = databaseHolder->getDb(opCtx.get(), nss.db());
+        auto db = databaseHolder->getDb(opCtx.get(), tenantDbName);
         if (!db) {
             // There is no config database, so there cannot be any active transactions.
             return boost::none;
diff --git a/src/mongo/db/ttl.cpp b/src/mongo/db/ttl.cpp
index adb4fb07221..78c77202d75 100644
--- a/src/mongo/db/ttl.cpp
+++ b/src/mongo/db/ttl.cpp
@@ -271,14 +271,14 @@ private:
         if (coll.getDb() &&
             nullptr !=
                 (mtab = TenantMigrationAccessBlockerRegistry::get(opCtx->getServiceContext())
-                            .getTenantMigrationAccessBlockerForDbName(coll.getDb()->name(),
-                                                                      MtabType::kRecipient)) &&
+                            .getTenantMigrationAccessBlockerForDbName(
+                                coll.getDb()->name().fullName(), MtabType::kRecipient)) &&
             mtab->checkIfShouldBlockTTL()) {
             LOGV2_DEBUG(53768,
                         1,
                         "Postpone TTL of DB because of active tenant migration",
                         "tenantMigrationAccessBlocker"_attr = mtab->getDebugInfo().jsonString(),
-                        "database"_attr = coll.getDb()->name());
+                        "database"_attr = coll.getDb()->name().dbName());
             return;
         }
 
diff --git a/src/mongo/db/views/SConscript b/src/mongo/db/views/SConscript
index 613f3273116..1a5db103452 100644
--- a/src/mongo/db/views/SConscript
+++ b/src/mongo/db/views/SConscript
@@ -15,6 +15,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog/database_holder',
+        '$BUILD_DIR/mongo/db/multitenancy',
     ],
 )
 
diff --git a/src/mongo/db/views/durable_view_catalog.cpp b/src/mongo/db/views/durable_view_catalog.cpp
index 0aff1e47984..1cf16491115 100644
--- a/src/mongo/db/views/durable_view_catalog.cpp
+++ b/src/mongo/db/views/durable_view_catalog.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/record_data.h"
+#include "mongo/db/tenant_database_name.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 #include "mongo/stdx/unordered_set.h"
@@ -78,8 +79,9 @@ void DurableViewCatalog::onSystemViewsCollectionDrop(OperationContext* opCtx,
         NamespaceString(name.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
     dassert(name.coll() == NamespaceString::kSystemDotViewsCollectionName);
 
+    const TenantDatabaseName tenantDbName(boost::none, name.db());
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, name.db());
+    auto db = databaseHolder->getDb(opCtx, tenantDbName);
     if (db) {
         // If the 'system.views' collection is dropped, we need to clear the in-memory state of the
         // view catalog.
@@ -90,7 +92,7 @@ void DurableViewCatalog::onSystemViewsCollectionDrop(OperationContext* opCtx,
 // DurableViewCatalogImpl
 
 const std::string& DurableViewCatalogImpl::getName() const {
-    return _db->name();
+    return _db->name().dbName();
 }
 
 const bool DurableViewCatalogImpl::belongsTo(const Database* db) const {
@@ -156,7 +158,7 @@ BSONObj DurableViewCatalogImpl::_validateViewDefinition(OperationContext* opCtx,
     // be valid. If not valid then the NamespaceString constructor will uassert.
     if (viewNameIsValid) {
         NamespaceString viewNss(viewName);
-        valid &= viewNss.isValid() && viewNss.db() == _db->name();
+        valid &= viewNss.isValid() && viewNss.db() == _db->name().dbName();
     }
 
     valid &= NamespaceString::validCollectionName(viewDefinition["viewOn"].str());
@@ -184,7 +186,7 @@ BSONObj DurableViewCatalogImpl::_validateViewDefinition(OperationContext* opCtx,
 void DurableViewCatalogImpl::upsert(OperationContext* opCtx,
                                     const NamespaceString& name,
                                     const BSONObj& view) {
-    dassert(opCtx->lockState()->isDbLockedForMode(_db->name(), MODE_IX));
+    dassert(opCtx->lockState()->isDbLockedForMode(_db->name().dbName(), MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(name, MODE_IX));
 
     NamespaceString systemViewsNs(_db->getSystemViewsName());
@@ -218,7 +220,7 @@ void DurableViewCatalogImpl::upsert(OperationContext* opCtx,
 }
 
 void DurableViewCatalogImpl::remove(OperationContext* opCtx, const NamespaceString& name) {
-    dassert(opCtx->lockState()->isDbLockedForMode(_db->name(), MODE_IX));
+    dassert(opCtx->lockState()->isDbLockedForMode(_db->name().dbName(), MODE_IX));
     dassert(opCtx->lockState()->isCollectionLockedForMode(name, MODE_IX));
 
     CollectionPtr systemViews = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(
diff --git a/src/mongo/db/views/view_catalog.cpp b/src/mongo/db/views/view_catalog.cpp
index 4b2b4a3ea83..1e49e7abd92 100644
--- a/src/mongo/db/views/view_catalog.cpp
+++ b/src/mongo/db/views/view_catalog.cpp
@@ -192,7 +192,7 @@ Status ViewCatalog::registerDatabase(OperationContext* opCtx,
 
 void ViewCatalog::unregisterDatabase(OperationContext* opCtx, Database* db) {
     auto catalog = getViewCatalog(opCtx->getServiceContext()).writer();
-    auto it = catalog.writable()->_viewsForDatabase.find(db->name());
+    auto it = catalog.writable()->_viewsForDatabase.find(db->name().dbName());
     if (it != catalog.writable()->_viewsForDatabase.end() && it->second.durable->belongsTo(db)) {
         catalog.writable()->_viewsForDatabase.erase(it);
         catalog.commit();
diff --git a/src/mongo/dbtests/catalogtests.cpp b/src/mongo/dbtests/catalogtests.cpp
index 5df1d340b30..445a91b8502 100644
--- a/src/mongo/dbtests/catalogtests.cpp
+++ b/src/mongo/dbtests/catalogtests.cpp
@@ -61,14 +61,14 @@ public:
         auto op1 = client1->makeOperationContext();
         auto op2 = client2->makeOperationContext();
 
-
         Lock::DBLock dbLk1(op1.get(), competingNss.db(), LockMode::MODE_IX);
         Lock::CollectionLock collLk1(op1.get(), competingNss, LockMode::MODE_IX);
         Lock::DBLock dbLk2(op2.get(), competingNss.db(), LockMode::MODE_IX);
         Lock::CollectionLock collLk2(op2.get(), competingNss, LockMode::MODE_IX);
 
+        const TenantDatabaseName competingTenantDbName(boost::none, competingNss.db());
         Database* db =
-            DatabaseHolder::get(op1.get())->openDb(op1.get(), competingNss.db(), nullptr);
+            DatabaseHolder::get(op1.get())->openDb(op1.get(), competingTenantDbName, nullptr);
 
         {
             WriteUnitOfWork wuow1(op1.get());
diff --git a/src/mongo/dbtests/dbhelper_tests.cpp b/src/mongo/dbtests/dbhelper_tests.cpp
index 04dee799791..5dbc2f05ca4 100644
--- a/src/mongo/dbtests/dbhelper_tests.cpp
+++ b/src/mongo/dbtests/dbhelper_tests.cpp
@@ -137,7 +137,9 @@ public:
         Lock::DBLock dbLk2(opCtx2.get(), nss.db(), LockMode::MODE_IX);
         Lock::CollectionLock collLk2(opCtx2.get(), nss, LockMode::MODE_IX);
 
-        Database* db = DatabaseHolder::get(opCtx1.get())->openDb(opCtx1.get(), nss.db(), nullptr);
+        const TenantDatabaseName tenantDbName(boost::none, nss.db());
+        Database* db =
+            DatabaseHolder::get(opCtx1.get())->openDb(opCtx1.get(), tenantDbName, nullptr);
 
         // Create the collection and insert one doc
         BSONObj doc = BSON("_id" << 1 << "x" << 2);
diff --git a/src/mongo/dbtests/query_stage_cached_plan.cpp b/src/mongo/dbtests/query_stage_cached_plan.cpp
index d3cf8bd8a63..bbea1b278ac 100644
--- a/src/mongo/dbtests/query_stage_cached_plan.cpp
+++ b/src/mongo/dbtests/query_stage_cached_plan.cpp
@@ -101,8 +101,9 @@ public:
 
     void dropCollection() {
         Lock::DBLock dbLock(&_opCtx, nss.db(), MODE_X);
+        const TenantDatabaseName tenantDbName(boost::none, nss.db());
         auto databaseHolder = DatabaseHolder::get(&_opCtx);
-        auto database = databaseHolder->getDb(&_opCtx, nss.db());
+        auto database = databaseHolder->getDb(&_opCtx, tenantDbName);
         if (!database) {
             return;
         }
diff --git a/src/mongo/dbtests/querytests.cpp b/src/mongo/dbtests/querytests.cpp
index 06757ea35ee..3310f766ff4 100644
--- a/src/mongo/dbtests/querytests.cpp
+++ b/src/mongo/dbtests/querytests.cpp
@@ -1169,7 +1169,7 @@ public:
         Lock::GlobalWrite lk(&_opCtx);
         OldClientContext ctx(&_opCtx, "unittests.DirectLocking");
         _client.remove("a.b", BSONObj());
-        ASSERT_EQUALS("unittests", ctx.db()->name());
+        ASSERT_EQUALS("unittests", ctx.db()->name().dbName());
     }
     const char* ns;
 };
diff --git a/src/mongo/dbtests/rollbacktests.cpp b/src/mongo/dbtests/rollbacktests.cpp
index 95310d74bc1..10586a30ee7 100644
--- a/src/mongo/dbtests/rollbacktests.cpp
+++ b/src/mongo/dbtests/rollbacktests.cpp
@@ -55,7 +55,7 @@ const auto kIndexVersion = IndexDescriptor::IndexVersion::kV2;
 void dropDatabase(OperationContext* opCtx, const NamespaceString& nss) {
     Lock::GlobalWrite globalWriteLock(opCtx);
     auto databaseHolder = DatabaseHolder::get(opCtx);
-    auto db = databaseHolder->getDb(opCtx, nss.db());
+    auto db = databaseHolder->getDb(opCtx, TenantDatabaseName(boost::none, nss.db()));
 
     if (db) {
         databaseHolder->dropDb(opCtx, db);