SERVER-74491 Make DatabaseName::toString private

35 files changed, 181 insertions, 95 deletions

diff --git a/src/mongo/db/auth/sasl_commands.cpp b/src/mongo/db/auth/sasl_commands.cpp
index ff160e753e6..a32e4ccee36 100644
--- a/src/mongo/db/auth/sasl_commands.cpp
+++ b/src/mongo/db/auth/sasl_commands.cpp
@@ -207,9 +207,10 @@ void warnIfCompressed(OperationContext* opCtx) {
 SaslReply doSaslStart(OperationContext* opCtx,
                       AuthenticationSession* session,
                       const SaslStartCommand& request) {
-    auto mechanism = uassertStatusOK(
-        SASLServerMechanismRegistry::get(opCtx->getServiceContext())
-            .getServerMechanism(request.getMechanism(), request.getDbName().toString()));
+    auto mechanism =
+        uassertStatusOK(SASLServerMechanismRegistry::get(opCtx->getServiceContext())
+                            .getServerMechanism(request.getMechanism(),
+                                                DatabaseNameUtil::serialize(request.getDbName())));
 
     uassert(ErrorCodes::BadValue,
             "Plaintext mechanisms may not be used with speculativeSaslStart",
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index b2210fec529..8678535c174 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -153,11 +153,10 @@ public:
 
     void checkCollections(const DatabaseName& dbName) {
         unsigned long counter = 0;
-
+        const auto dbNameStr = dbName.toString_forTest();
         for (auto [orderedIt, catalogIt] =
-                 std::tuple{collsIterator(dbName.toString()), catalog.begin(opCtx.get(), dbName)};
-             catalogIt != catalog.end(opCtx.get()) &&
-             orderedIt != collsIteratorEnd(dbName.toString());
+                 std::tuple{collsIterator(dbNameStr), catalog.begin(opCtx.get(), dbName)};
+             catalogIt != catalog.end(opCtx.get()) && orderedIt != collsIteratorEnd(dbNameStr);
              ++catalogIt, ++orderedIt) {
 
             auto catalogColl = *catalogIt;
@@ -167,7 +166,7 @@ public:
             ++counter;
         }
 
-        ASSERT_EQUALS(counter, dbMap[dbName.toString()].size());
+        ASSERT_EQUALS(counter, dbMap[dbNameStr].size());
     }
 
     void dropColl(const std::string dbName, UUID uuid) {
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index b57bff3323d..964dc468284 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -143,7 +143,7 @@ Database* DatabaseHolderImpl::openDb(OperationContext* opCtx,
     lk.unlock();
 
     if (CollectionCatalog::get(opCtx)->getAllCollectionUUIDsFromDb(dbName).empty()) {
-        audit::logCreateDatabase(opCtx->getClient(), dbName.toString());
+        audit::logCreateDatabase(opCtx->getClient(), DatabaseNameUtil::serialize(dbName));
         if (justCreated)
             *justCreated = true;
     }
@@ -192,7 +192,7 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
                                 << coll->ns().toStringForErrorMsg() << "'.");
     }
 
-    audit::logDropDatabase(opCtx->getClient(), name.toString());
+    audit::logDropDatabase(opCtx->getClient(), DatabaseNameUtil::serialize(name));
 
     auto const serviceContext = opCtx->getServiceContext();
 
@@ -232,7 +232,7 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
         });
 
     auto const storageEngine = serviceContext->getStorageEngine();
-    writeConflictRetry(opCtx, "dropDatabase", name.toString(), [&] {
+    writeConflictRetry(opCtx, "dropDatabase", toStringForLogging(name), [&] {
         storageEngine->dropDatabase(opCtx, name).transitional_ignore();
     });
 }
diff --git a/src/mongo/db/catalog/drop_database.cpp b/src/mongo/db/catalog/drop_database.cpp
index ed431fb0370..9e157623bf8 100644
--- a/src/mongo/db/catalog/drop_database.cpp
+++ b/src/mongo/db/catalog/drop_database.cpp
@@ -104,7 +104,7 @@ void _finishDropDatabase(OperationContext* opCtx,
         IndexBuildsCoordinator::get(opCtx)->assertNoBgOpInProgForDb(dbName);
     }
 
-    writeConflictRetry(opCtx, "dropDatabase_database", dbName.toString(), [&] {
+    writeConflictRetry(opCtx, "dropDatabase_database", toStringForLogging(dbName), [&] {
         // We need to replicate the dropDatabase oplog entry and clear the collection catalog in the
         // same transaction. This is to prevent stepdown from interrupting between these two
         // operations and leaving this node in an inconsistent state.
diff --git a/src/mongo/db/commands/dbcheck.cpp b/src/mongo/db/commands/dbcheck.cpp
index e329d1d781a..3fce4b46a11 100644
--- a/src/mongo/db/commands/dbcheck.cpp
+++ b/src/mongo/db/commands/dbcheck.cpp
@@ -292,7 +292,7 @@ std::shared_ptr<const CollectionCatalog> getConsistentCatalogAndSnapshot(Operati
 class DbCheckJob : public BackgroundJob {
 public:
     DbCheckJob(const DatabaseName& dbName, std::unique_ptr<DbCheckRun> run)
-        : BackgroundJob(true), _done(false), _dbName(dbName.toString()), _run(std::move(run)) {}
+        : BackgroundJob(true), _done(false), _run(std::move(run)) {}
 
 protected:
     virtual std::string name() const override {
@@ -506,7 +506,6 @@ private:
 
     // Set if the job cannot proceed.
     bool _done;
-    std::string _dbName;
     std::unique_ptr<DbCheckRun> _run;
 
     StatusWith<BatchStats> _runBatch(OperationContext* opCtx,
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index 775a08afea9..16f3f9c457d 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -292,7 +292,7 @@ public:
         BSONObj query = BSON("files_id" << jsobj["filemd5"] << "n" << GTE << n);
         BSONObj sort = BSON("files_id" << 1 << "n" << 1);
 
-        return writeConflictRetry(opCtx, "filemd5", dbName.toString(), [&] {
+        return writeConflictRetry(opCtx, "filemd5", toStringForLogging(dbName), [&] {
             auto findCommand = std::make_unique<FindCommandRequest>(nss);
             findCommand->setFilter(query.getOwned());
             findCommand->setSort(sort.getOwned());
diff --git a/src/mongo/db/commands/explain_cmd.cpp b/src/mongo/db/commands/explain_cmd.cpp
index c10e8250aed..7c815373005 100644
--- a/src/mongo/db/commands/explain_cmd.cpp
+++ b/src/mongo/db/commands/explain_cmd.cpp
@@ -185,7 +185,7 @@ std::unique_ptr<CommandInvocation> CmdExplain::parse(OperationContext* opCtx,
                 str::stream() << "Mismatched $db in explain command. Expected "
                               << dbName.toStringForErrorMsg() << " but got "
                               << innerDbName.toStringForErrorMsg(),
-                innerDb.checkAndGetStringData() == dbName.toString());
+                innerDbName == dbName);
     }
     auto explainedCommand = CommandHelpers::findCommand(explainedObj.firstElementFieldName());
     uassert(ErrorCodes::CommandNotFound,
diff --git a/src/mongo/db/commands/list_databases_common.h b/src/mongo/db/commands/list_databases_common.h
index 942f1e76aff..26a3ff7ea1d 100644
--- a/src/mongo/db/commands/list_databases_common.h
+++ b/src/mongo/db/commands/list_databases_common.h
@@ -93,7 +93,7 @@ int64_t setReplyItems(OperationContext* opCtx,
 
     for (const auto& dbName : dbNames) {
         if (authorizedDatabases &&
-            !as->isAuthorizedForAnyActionOnAnyResourceInDB(dbName.toString())) {
+            !as->isAuthorizedForAnyActionOnAnyResourceInDB(DatabaseNameUtil::serialize(dbName))) {
             // We don't have listDatabases on the cluster or find on this database.
             continue;
         }
@@ -118,7 +118,7 @@ int64_t setReplyItems(OperationContext* opCtx,
                 continue;
             }
 
-            writeConflictRetry(opCtx, "sizeOnDisk", dbName.toString(), [&] {
+            writeConflictRetry(opCtx, "sizeOnDisk", toStringForLogging(dbName), [&] {
                 size = storageEngine->sizeOnDiskForDb(opCtx, dbName);
             });
             item.setSizeOnDisk(size);
diff --git a/src/mongo/db/database_name.h b/src/mongo/db/database_name.h
index 8f111118890..bde0341dd59 100644
--- a/src/mongo/db/database_name.h
+++ b/src/mongo/db/database_name.h
@@ -152,10 +152,6 @@ public:
         return _data.size() == kDataOffset;
     }
 
-    std::string toString() const {
-        return db().toString();
-    }
-
     std::string toStringWithTenantId() const {
         if (_hasTenantId()) {
             return str::stream() << TenantId{OID::from(&_data[kDataOffset])} << "_" << db();
@@ -187,6 +183,15 @@ public:
         return dbName.toStringWithTenantId();
     }
 
+    /**
+     * This function returns the DatabaseName as a string, ignoring the tenantId.
+     *
+     * MUST only be used for tests.
+     */
+    std::string toString_forTest() const {
+        return toString();
+    }
+
     bool equalCaseInsensitive(const DatabaseName& other) const {
         return StringData{_data.data() + kDataOffset, _data.size() - kDataOffset}
             .equalCaseInsensitive(
@@ -287,6 +292,10 @@ private:
         }
     }
 
+    std::string toString() const {
+        return db().toString();
+    }
+
     static constexpr size_t kDataOffset = sizeof(uint8_t);
     static constexpr uint8_t kTenantIdMask = 0x80;
     static constexpr uint8_t kDatabaseNameOffsetEndMask = 0x7F;
diff --git a/src/mongo/db/database_name_test.cpp b/src/mongo/db/database_name_test.cpp
index c0981ac8798..d173700a22f 100644
--- a/src/mongo/db/database_name_test.cpp
+++ b/src/mongo/db/database_name_test.cpp
@@ -44,14 +44,14 @@ TEST(DatabaseNameTest, MultitenancySupportDisabled) {
 
     ASSERT(!dbnWithoutTenant1.tenantId());
     ASSERT_EQUALS(std::string("a"), dbnWithoutTenant1.db());
-    ASSERT_EQUALS(std::string("a"), dbnWithoutTenant1.toString());
+    ASSERT_EQUALS(std::string("a"), dbnWithoutTenant1.toString_forTest());
 
     TenantId tenantId(OID::gen());
     DatabaseName dbnWithTenant = DatabaseName::createDatabaseName_forTest(tenantId, "a");
     ASSERT(dbnWithTenant.tenantId());
     ASSERT_EQUALS(tenantId, *dbnWithTenant.tenantId());
     ASSERT_EQUALS(std::string("a"), dbnWithTenant.db());
-    ASSERT_EQUALS(std::string("a"), dbnWithTenant.toString());
+    ASSERT_EQUALS(std::string("a"), dbnWithoutTenant1.toString_forTest());
     ASSERT_EQUALS(std::string(tenantId.toString() + "_a"),
                   dbnWithTenant.toStringWithTenantId_forTest());
 }
@@ -63,14 +63,14 @@ TEST(DatabaseNameTest, MultitenancySupportEnabledTenantIDNotRequired) {
     DatabaseName dbnWithoutTenant = DatabaseName::createDatabaseName_forTest(boost::none, "a");
     ASSERT(!dbnWithoutTenant.tenantId());
     ASSERT_EQUALS(std::string("a"), dbnWithoutTenant.db());
-    ASSERT_EQUALS(std::string("a"), dbnWithoutTenant.toString());
+    ASSERT_EQUALS(std::string("a"), dbnWithoutTenant.toString_forTest());
 
     TenantId tenantId(OID::gen());
     DatabaseName dbnWithTenant = DatabaseName::createDatabaseName_forTest(tenantId, "a");
     ASSERT(dbnWithTenant.tenantId());
     ASSERT_EQUALS(tenantId, *dbnWithTenant.tenantId());
     ASSERT_EQUALS(std::string("a"), dbnWithTenant.db());
-    ASSERT_EQUALS(std::string("a"), dbnWithTenant.toString());
+    ASSERT_EQUALS(std::string("a"), dbnWithTenant.toString_forTest());
     ASSERT_EQUALS(std::string(tenantId.toString() + "_a"),
                   dbnWithTenant.toStringWithTenantId_forTest());
 }
@@ -165,21 +165,21 @@ TEST(DatabaseNameTest, EmptyDbString) {
     DatabaseName empty{};
     ASSERT_EQ(empty.db(), "");
     ASSERT_FALSE(empty.tenantId());
-    ASSERT_EQ(empty.toString(), "");
+    ASSERT_EQ(empty.toString_forTest(), "");
     ASSERT_EQ(empty.toStringWithTenantId_forTest(), "");
 
     DatabaseName emptyFromStringData =
         DatabaseName::createDatabaseName_forTest(boost::none, StringData());
     ASSERT_EQ(emptyFromStringData.db(), "");
     ASSERT_FALSE(emptyFromStringData.tenantId());
-    ASSERT_EQ(emptyFromStringData.toString(), "");
+    ASSERT_EQ(emptyFromStringData.toString_forTest(), "");
     ASSERT_EQ(emptyFromStringData.toStringWithTenantId_forTest(), "");
 
     TenantId tenantId(OID::gen());
     DatabaseName emptyWithTenantId = DatabaseName::createDatabaseName_forTest(tenantId, "");
     ASSERT_EQ(emptyWithTenantId.db(), "");
     ASSERT(emptyWithTenantId.tenantId());
-    ASSERT_EQ(emptyWithTenantId.toString(), "");
+    ASSERT_EQ(emptyWithTenantId.toString_forTest(), "");
     ASSERT_EQ(emptyWithTenantId.toStringWithTenantId_forTest(),
               fmt::format("{}_", tenantId.toString()));
 }
diff --git a/src/mongo/db/namespace_string_test.cpp b/src/mongo/db/namespace_string_test.cpp
index 11e32926be9..c489db13caf 100644
--- a/src/mongo/db/namespace_string_test.cpp
+++ b/src/mongo/db/namespace_string_test.cpp
@@ -446,7 +446,7 @@ TEST(NamespaceStringTest, NSSWithTenantId) {
         ASSERT_EQ(empty.toStringWithTenantId_forTest(), "");
         ASSERT_EQ(empty.dbName().db(), "");
         ASSERT_EQ(empty.dbName().tenantId(), boost::none);
-        ASSERT_EQ(empty.dbName().toString(), "");
+        ASSERT_EQ(empty.dbName().toString_forTest(), "");
         ASSERT_EQ(empty.dbName().toStringWithTenantId_forTest(), "");
     }
 
@@ -463,7 +463,7 @@ TEST(NamespaceStringTest, NSSWithTenantId) {
         ASSERT_EQ(emptyWithTenant.dbName().db(), "");
         ASSERT(emptyWithTenant.dbName().tenantId());
         ASSERT_EQ(emptyWithTenant.dbName().tenantId(), tenantId);
-        ASSERT_EQ(emptyWithTenant.dbName().toString(), "");
+        ASSERT_EQ(emptyWithTenant.dbName().toString_forTest(), "");
         ASSERT_EQ(emptyWithTenant.dbName().toStringWithTenantId_forTest(),
                   "{}_"_format(tenantId.toString()));
     }
@@ -478,7 +478,7 @@ TEST(NamespaceStringTest, NSSWithTenantId) {
         ASSERT_EQ(dbWithoutColl.toStringWithTenantId_forTest(), "foo");
         ASSERT_EQ(dbWithoutColl.dbName().db(), "foo");
         ASSERT_FALSE(dbWithoutColl.dbName().tenantId());
-        ASSERT_EQ(dbWithoutColl.dbName().toString(), "foo");
+        ASSERT_EQ(dbWithoutColl.dbName().toString_forTest(), "foo");
         ASSERT_EQ(dbWithoutColl.dbName().toStringWithTenantId_forTest(), "foo");
     }
 
@@ -495,7 +495,7 @@ TEST(NamespaceStringTest, NSSWithTenantId) {
         ASSERT_EQ(dbWithoutCollWithTenant.dbName().db(), "foo");
         ASSERT(dbWithoutCollWithTenant.dbName().tenantId());
         ASSERT_EQ(dbWithoutCollWithTenant.dbName().tenantId(), tenantId);
-        ASSERT_EQ(dbWithoutCollWithTenant.dbName().toString(), "foo");
+        ASSERT_EQ(dbWithoutCollWithTenant.dbName().toString_forTest(), "foo");
         ASSERT_EQ(dbWithoutCollWithTenant.dbName().toStringWithTenantId_forTest(),
                   fmt::format("{}_foo", tenantId.toString()));
     }
diff --git a/src/mongo/db/pipeline/document_source_graph_lookup_test.cpp b/src/mongo/db/pipeline/document_source_graph_lookup_test.cpp
index cf1d045a35a..a5582782104 100644
--- a/src/mongo/db/pipeline/document_source_graph_lookup_test.cpp
+++ b/src/mongo/db/pipeline/document_source_graph_lookup_test.cpp
@@ -995,7 +995,7 @@ TEST_F(DocumentSourceGraphLookupServerlessTest,
                                                                    flagStatus);
 
         NamespaceString nss = NamespaceString::createNamespaceString_forTest(
-            boost::none, expCtx->ns.dbName().toString(), _targetColl);
+            boost::none, expCtx->ns.dbName().toString_forTest(), _targetColl);
         auto liteParsedLookup =
             DocumentSourceGraphLookUp::LiteParsed::parse(nss, originalBSON.firstElement());
         auto namespaceSet = liteParsedLookup->getInvolvedNamespaces();
diff --git a/src/mongo/db/pipeline/document_source_lookup_test.cpp b/src/mongo/db/pipeline/document_source_lookup_test.cpp
index 1f404764392..0166fce3d8d 100644
--- a/src/mongo/db/pipeline/document_source_lookup_test.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup_test.cpp
@@ -1661,7 +1661,7 @@ TEST_F(DocumentSourceLookUpServerlessTest,
                         << "lookup1"));
 
     NamespaceString nss = NamespaceString::createNamespaceString_forTest(
-        boost::none, expCtx->ns.dbName().toString(), _targetColl);
+        boost::none, expCtx->ns.dbName().toString_forTest(), _targetColl);
 
     for (bool flagStatus : {false, true}) {
         RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID",
diff --git a/src/mongo/db/pipeline/document_source_merge_test.cpp b/src/mongo/db/pipeline/document_source_merge_test.cpp
index 5cd838401da..5eafa0330ef 100644
--- a/src/mongo/db/pipeline/document_source_merge_test.cpp
+++ b/src/mongo/db/pipeline/document_source_merge_test.cpp
@@ -1083,7 +1083,7 @@ TEST_F(DocumentSourceMergeServerlessTest,
         ASSERT_EQ(*mergeSource->getOutputNs().tenantId(), *expCtx->ns.tenantId());
 
         // Assert the tenantId is not included in the serialized namespace.
-        auto dbField = flagStatus ? expCtx->ns.dbName().toString()
+        auto dbField = flagStatus ? expCtx->ns.dbName().toString_forTest()
                                   : expCtx->ns.dbName().toStringWithTenantId_forTest();
         auto expectedDoc = Document{{"db", dbField}, {"coll", _targetColl}};
 
@@ -1111,7 +1111,7 @@ TEST_F(DocumentSourceMergeServerlessTest,
         ASSERT(mergeSource->getOutputNs().tenantId());
         ASSERT_EQ(*mergeSource->getOutputNs().tenantId(), *expCtx->ns.tenantId());
 
-        auto dbField = flagStatus ? expCtx->ns.dbName().toString()
+        auto dbField = flagStatus ? expCtx->ns.dbName().toString_forTest()
                                   : expCtx->ns.dbName().toStringWithTenantId_forTest();
         auto expectedDoc = Document{{"db", dbField}, {"coll", _targetColl}};
 
diff --git a/src/mongo/db/pipeline/document_source_out.cpp b/src/mongo/db/pipeline/document_source_out.cpp
index 987f9a04088..8763cb0874c 100644
--- a/src/mongo/db/pipeline/document_source_out.cpp
+++ b/src/mongo/db/pipeline/document_source_out.cpp
@@ -180,7 +180,7 @@ void DocumentSourceOut::initialize() {
     // consultation.
     _tempNs = NamespaceStringUtil::parseNamespaceFromRequest(
         getOutputNs().tenantId(),
-        str::stream() << getOutputNs().dbName().toString() << "."
+        str::stream() << getOutputNs().dbName().db() << "."
                       << NamespaceString::kOutTmpCollectionPrefix << UUID::gen());
 
     // Save the original collection options and index specs so we can check they didn't change
diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp
index 8cfe6307ed2..75267be9c55 100644
--- a/src/mongo/db/repl/storage_interface_impl.cpp
+++ b/src/mongo/db/repl/storage_interface_impl.cpp
@@ -405,14 +405,15 @@ Status StorageInterfaceImpl::dropReplicatedDatabases(OperationContext* opCtx) {
             hasLocalDatabase = true;
             continue;
         }
-        writeConflictRetry(opCtx, "dropReplicatedDatabases", dbName.toString(), [&] {
+        writeConflictRetry(opCtx, "dropReplicatedDatabases", toStringForLogging(dbName), [&] {
             if (auto db = databaseHolder->getDb(opCtx, dbName)) {
                 WriteUnitOfWork wuow(opCtx);
                 databaseHolder->dropDb(opCtx, db);
                 wuow.commit();
             } else {
                 // This is needed since dropDatabase can't be rolled back.
-                // This is safe be replaced by "invariant(db);dropDatabase(opCtx, db);" once fixed.
+                // This is safe be replaced by "invariant(db);dropDatabase(opCtx, db);" once
+                // fixed.
                 LOGV2(21755,
                       "dropReplicatedDatabases - database disappeared after retrieving list of "
                       "database names but before drop: {dbName}",
diff --git a/src/mongo/db/repl/tenant_oplog_applier_test.cpp b/src/mongo/db/repl/tenant_oplog_applier_test.cpp
index 5e1b9a0ca57..638e961eca7 100644
--- a/src/mongo/db/repl/tenant_oplog_applier_test.cpp
+++ b/src/mongo/db/repl/tenant_oplog_applier_test.cpp
@@ -491,7 +491,7 @@ TEST_F(TenantOplogApplierTest, ApplyInsert_DatabaseMissing) {
 }
 
 TEST_F(TenantOplogApplierTest, ApplyInsert_CollectionMissing) {
-    createDatabase(_opCtx.get(), _dbName.toString());
+    createDatabase(_opCtx.get(), _dbName.toString_forTest());
     auto entry = makeInsertOplogEntry(1,
                                       NamespaceString::createNamespaceString_forTest(
                                           _dbName.toStringWithTenantId_forTest(), "bar"),
@@ -816,7 +816,7 @@ TEST_F(TenantOplogApplierTest, ApplyDelete_DatabaseMissing) {
 }
 
 TEST_F(TenantOplogApplierTest, ApplyDelete_CollectionMissing) {
-    createDatabase(_opCtx.get(), _dbName.toString());
+    createDatabase(_opCtx.get(), _dbName.toString_forTest());
     auto entry = makeOplogEntry(OpTypeEnum::kDelete,
                                 NamespaceString::createNamespaceString_forTest(
                                     _dbName.toStringWithTenantId_forTest(), "bar"),
@@ -1281,7 +1281,7 @@ TEST_F(TenantOplogApplierTest, ApplyCollModCommand_IndexNotFound) {
 }
 
 TEST_F(TenantOplogApplierTest, ApplyCollModCommand_CollectionMissing) {
-    createDatabase(_opCtx.get(), _dbName.toString());
+    createDatabase(_opCtx.get(), _dbName.toString_forTest());
     NamespaceString nss = NamespaceString::createNamespaceString_forTest(
         _dbName.toStringWithTenantId_forTest(), "bar");
     UUID uuid(UUID::gen());
diff --git a/src/mongo/db/s/database_sharding_state.cpp b/src/mongo/db/s/database_sharding_state.cpp
index c43b8463e39..38c0db9159a 100644
--- a/src/mongo/db/s/database_sharding_state.cpp
+++ b/src/mongo/db/s/database_sharding_state.cpp
@@ -159,7 +159,8 @@ std::vector<DatabaseName> DatabaseShardingState::getDatabaseNames(OperationConte
 
 void DatabaseShardingState::assertMatchingDbVersion(OperationContext* opCtx,
                                                     const DatabaseName& dbName) {
-    const auto receivedVersion = OperationShardingState::get(opCtx).getDbVersion(dbName.toString());
+    const auto receivedVersion =
+        OperationShardingState::get(opCtx).getDbVersion(DatabaseNameUtil::serialize(dbName));
     if (!receivedVersion) {
         return;
     }
@@ -179,7 +180,8 @@ void DatabaseShardingState::assertMatchingDbVersion(OperationContext* opCtx,
         const auto optCritSecReason = scopedDss->getCriticalSectionReason();
 
         uassert(
-            StaleDbRoutingVersion(dbName.toString(), receivedVersion, boost::none, critSecSignal),
+            StaleDbRoutingVersion(
+                DatabaseNameUtil::serialize(dbName), receivedVersion, boost::none, critSecSignal),
             str::stream() << "The critical section for the database "
                           << dbName.toStringForErrorMsg()
                           << " is acquired with reason: " << scopedDss->getCriticalSectionReason(),
@@ -187,13 +189,15 @@ void DatabaseShardingState::assertMatchingDbVersion(OperationContext* opCtx,
     }
 
     const auto wantedVersion = scopedDss->getDbVersion(opCtx);
-    uassert(StaleDbRoutingVersion(dbName.toString(), receivedVersion, boost::none),
-            str::stream() << "No cached info for the database " << dbName.toStringForErrorMsg(),
-            wantedVersion);
-
-    uassert(StaleDbRoutingVersion(dbName.toString(), receivedVersion, *wantedVersion),
-            str::stream() << "Version mismatch for the database " << dbName.toStringForErrorMsg(),
-            receivedVersion == *wantedVersion);
+    uassert(
+        StaleDbRoutingVersion(DatabaseNameUtil::serialize(dbName), receivedVersion, boost::none),
+        str::stream() << "No cached info for the database " << dbName.toStringForErrorMsg(),
+        wantedVersion);
+
+    uassert(
+        StaleDbRoutingVersion(DatabaseNameUtil::serialize(dbName), receivedVersion, *wantedVersion),
+        str::stream() << "Version mismatch for the database " << dbName.toStringForErrorMsg(),
+        receivedVersion == *wantedVersion);
 }
 
 void DatabaseShardingState::assertIsPrimaryShardForDb(OperationContext* opCtx,
@@ -206,7 +210,8 @@ void DatabaseShardingState::assertIsPrimaryShardForDb(OperationContext* opCtx,
         return;
     }
 
-    auto expectedDbVersion = OperationShardingState::get(opCtx).getDbVersion(dbName.toString());
+    auto expectedDbVersion =
+        OperationShardingState::get(opCtx).getDbVersion(DatabaseNameUtil::serialize(dbName));
 
     uassert(ErrorCodes::IllegalOperation,
             str::stream() << "Received request without the version for the database "
diff --git a/src/mongo/db/s/flush_database_cache_updates_command.cpp b/src/mongo/db/s/flush_database_cache_updates_command.cpp
index 63278e043be..8489914dcb9 100644
--- a/src/mongo/db/s/flush_database_cache_updates_command.cpp
+++ b/src/mongo/db/s/flush_database_cache_updates_command.cpp
@@ -66,8 +66,9 @@ Status insertDatabaseEntryForBackwardCompatibility(OperationContext* opCtx,
 
     DBDirectClient client(opCtx);
     auto commandResponse = client.runCommand([&] {
-        auto dbMetadata =
-            DatabaseType(dbName.toString(), ShardId::kConfigServerId, DatabaseVersion::makeFixed());
+        auto dbMetadata = DatabaseType(DatabaseNameUtil::serialize(dbName),
+                                       ShardId::kConfigServerId,
+                                       DatabaseVersion::makeFixed());
 
         write_ops::InsertCommandRequest insertOp(NamespaceString::kShardConfigDatabasesNamespace);
         insertOp.setDocuments({dbMetadata.toBSON()});
diff --git a/src/mongo/db/s/migration_destination_manager.cpp b/src/mongo/db/s/migration_destination_manager.cpp
index c81b3bed882..179910677b7 100644
--- a/src/mongo/db/s/migration_destination_manager.cpp
+++ b/src/mongo/db/s/migration_destination_manager.cpp
@@ -839,10 +839,9 @@ MigrationDestinationManager::IndexesAndIdIndex MigrationDestinationManager::getC
     auto indexes = uassertStatusOK(
         fromShard->runExhaustiveCursorCommand(opCtx,
                                               ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                                              nssOrUUID.dbName().toString(),
+                                              DatabaseNameUtil::serialize(nssOrUUID.dbName()),
                                               cmd,
                                               Milliseconds(-1)));
-
     for (auto&& spec : indexes.docs) {
         if (spec[IndexDescriptor::kClusteredFieldName]) {
             // The 'clustered' index is implicitly created upon clustered collection creation.
@@ -885,7 +884,7 @@ MigrationDestinationManager::getCollectionOptions(OperationContext* opCtx,
     auto infosRes = uassertStatusOK(
         fromShard->runExhaustiveCursorCommand(opCtx,
                                               ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                                              nssOrUUID.dbName().toString(),
+                                              DatabaseNameUtil::serialize(nssOrUUID.dbName()),
                                               cmd,
                                               Milliseconds(-1)));
 
diff --git a/src/mongo/db/s/migration_util.cpp b/src/mongo/db/s/migration_util.cpp
index eb169c07f6a..e930e656923 100644
--- a/src/mongo/db/s/migration_util.cpp
+++ b/src/mongo/db/s/migration_util.cpp
@@ -171,7 +171,7 @@ void sendWriteCommandToRecipient(OperationContext* opCtx,
     auto response = recipientShard->runCommandWithFixedRetryAttempts(
         opCtx,
         ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-        cmd.getDbName().toString(),
+        DatabaseNameUtil::serialize(cmd.getDbName()),
         cmdBSON,
         Shard::RetryPolicy::kIdempotent);
 
diff --git a/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp b/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp
index 9b198e212f8..624d7c4c775 100644
--- a/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp
+++ b/src/mongo/db/s/move_primary/move_primary_database_cloner.cpp
@@ -104,8 +104,8 @@ void MovePrimaryDatabaseCloner::calculateListCatalogEntriesForDonor() {
     Pipeline::SourceContainer stages;
     stages.emplace_back(DocumentSourceListCatalog::createFromBson(
         BSON("$listCatalog" << BSONObj()).firstElement(), expCtx));
-    stages.emplace_back(
-        DocumentSourceMatch::create(BSON("db" << BSON("$eq" << _dbName.toString())), expCtx));
+    stages.emplace_back(DocumentSourceMatch::create(
+        BSON("db" << BSON("$eq" << DatabaseNameUtil::serialize(_dbName))), expCtx));
     stages.emplace_back(DocumentSourceMatch::create(BSON("type" << BSON("$eq"
                                                                         << "collection")),
                                                     expCtx));
@@ -134,7 +134,7 @@ void MovePrimaryDatabaseCloner::calculateListCatalogEntriesForRecipient() {
 
     auto opCtx = _opCtxHolder.get();
     auto recipientShardedColls = _catalogClient->getAllShardedCollectionsForDb(
-        opCtx, _dbName.toString(), repl::ReadConcernLevel::kMajorityReadConcern);
+        opCtx, DatabaseNameUtil::serialize(_dbName), repl::ReadConcernLevel::kMajorityReadConcern);
 
     stdx::unordered_set<NamespaceString> shardedCollsSet;
     std::copy(recipientShardedColls.begin(),
diff --git a/src/mongo/db/s/move_primary/move_primary_donor_service.cpp b/src/mongo/db/s/move_primary/move_primary_donor_service.cpp
index 18caf3d27ff..8bf7ec7c41b 100644
--- a/src/mongo/db/s/move_primary/move_primary_donor_service.cpp
+++ b/src/mongo/db/s/move_primary/move_primary_donor_service.cpp
@@ -347,7 +347,7 @@ std::shared_ptr<MovePrimaryDonor> MovePrimaryDonor::create(OperationContext* opC
     auto service = registry->lookupServiceByName(MovePrimaryDonorService::kServiceName);
     MovePrimaryCommonMetadata metadata;
     metadata.setMigrationId(UUID::gen());
-    metadata.setDatabaseName(NamespaceString{dbName.toString()});
+    metadata.setDatabaseName(NamespaceString{dbName});
     metadata.setFromShardName(ShardingState::get(opCtx)->shardId());
     metadata.setToShardName(toShard.toString());
     MovePrimaryDonorDocument document;
diff --git a/src/mongo/db/s/move_primary/move_primary_recipient_service.cpp b/src/mongo/db/s/move_primary/move_primary_recipient_service.cpp
index 8879b3719e3..e608c42e939 100644
--- a/src/mongo/db/s/move_primary/move_primary_recipient_service.cpp
+++ b/src/mongo/db/s/move_primary/move_primary_recipient_service.cpp
@@ -614,8 +614,10 @@ MovePrimaryRecipientService::MovePrimaryRecipient::_getUnshardedCollections(
     const auto fromShard =
         uassertStatusOK(shardRegistry->getShard(opCtx, _metadata.getFromShardName().toString()));
 
-    auto collectionsToCloneWithStatus = _cloner->getListOfCollections(
-        opCtx, getDatabaseName().dbName().toString(), fromShard->getConnString().toString());
+    auto collectionsToCloneWithStatus =
+        _cloner->getListOfCollections(opCtx,
+                                      DatabaseNameUtil::serialize(getDatabaseName().dbName()),
+                                      fromShard->getConnString().toString());
     auto collectionsToClone = uassertStatusOK(collectionsToCloneWithStatus);
 
     const auto allCollections = [&] {
diff --git a/src/mongo/db/s/move_primary_coordinator.cpp b/src/mongo/db/s/move_primary_coordinator.cpp
index 095269dae72..fbf4cbf3eb9 100644
--- a/src/mongo/db/s/move_primary_coordinator.cpp
+++ b/src/mongo/db/s/move_primary_coordinator.cpp
@@ -69,7 +69,7 @@ MovePrimaryCoordinator::MovePrimaryCoordinator(ShardingDDLCoordinatorService* se
       _csReason([&] {
           BSONObjBuilder builder;
           builder.append("command", "movePrimary");
-          builder.append("db", _dbName.toString());
+          builder.append("db", DatabaseNameUtil::serialize(_dbName));
           builder.append("to", _doc.getToShardId());
           return builder.obj();
       }()) {}
@@ -481,7 +481,7 @@ void MovePrimaryCoordinator::logChange(OperationContext* opCtx,
         details.append("error", status.toString());
     }
     ShardingLogging::get(opCtx)->logChange(
-        opCtx, "movePrimary.{}"_format(what), _dbName.toString(), details.obj());
+        opCtx, "movePrimary.{}"_format(what), DatabaseNameUtil::serialize(_dbName), details.obj());
 }
 
 std::vector<NamespaceString> MovePrimaryCoordinator::getUnshardedCollections(
@@ -510,7 +510,9 @@ std::vector<NamespaceString> MovePrimaryCoordinator::getUnshardedCollections(
 
     const auto shardedCollections = [&] {
         auto colls = Grid::get(opCtx)->catalogClient()->getAllShardedCollectionsForDb(
-            opCtx, _dbName.toString(), repl::ReadConcernLevel::kMajorityReadConcern);
+            opCtx,
+            DatabaseNameUtil::serialize(_dbName),
+            repl::ReadConcernLevel::kMajorityReadConcern);
 
         std::sort(colls.begin(), colls.end());
         return colls;
@@ -544,7 +546,7 @@ void MovePrimaryCoordinator::assertNoOrphanedDataOnRecipient(
         const auto listResponse = uassertStatusOK(
             toShard->runExhaustiveCursorCommand(opCtx,
                                                 ReadPreferenceSetting(ReadPreference::PrimaryOnly),
-                                                _dbName.toString(),
+                                                DatabaseNameUtil::serialize(_dbName),
                                                 listCommand,
                                                 Milliseconds(-1)));
 
@@ -581,7 +583,7 @@ std::vector<NamespaceString> MovePrimaryCoordinator::cloneDataToRecipient(
 
     const auto cloneCommand = [&] {
         BSONObjBuilder commandBuilder;
-        commandBuilder.append("_shardsvrCloneCatalogData", _dbName.toString());
+        commandBuilder.append("_shardsvrCloneCatalogData", DatabaseNameUtil::serialize(_dbName));
         commandBuilder.append("from", fromShard->getConnString().toString());
         return CommandHelpers::appendMajorityWriteConcern(commandBuilder.obj());
     }();
@@ -652,14 +654,14 @@ void MovePrimaryCoordinator::assertChangedMetadataOnConfig(
     OperationContext* opCtx, const DatabaseVersion& preCommitDbVersion) const {
     const auto postCommitDbType = [&]() {
         const auto config = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-        auto findResponse = uassertStatusOK(
-            config->exhaustiveFindOnConfig(opCtx,
-                                           ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-                                           repl::ReadConcernLevel::kMajorityReadConcern,
-                                           NamespaceString::kConfigDatabasesNamespace,
-                                           BSON(DatabaseType::kNameFieldName << _dbName.toString()),
-                                           BSONObj(),
-                                           1));
+        auto findResponse = uassertStatusOK(config->exhaustiveFindOnConfig(
+            opCtx,
+            ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+            repl::ReadConcernLevel::kMajorityReadConcern,
+            NamespaceString::kConfigDatabasesNamespace,
+            BSON(DatabaseType::kNameFieldName << DatabaseNameUtil::serialize(_dbName)),
+            BSONObj(),
+            1));
 
         const auto databases = std::move(findResponse.docs);
         uassert(ErrorCodes::IncompatibleShardingMetadata,
diff --git a/src/mongo/db/s/range_deleter_service.cpp b/src/mongo/db/s/range_deleter_service.cpp
index 2c8af8ec2b8..c475da0836e 100644
--- a/src/mongo/db/s/range_deleter_service.cpp
+++ b/src/mongo/db/s/range_deleter_service.cpp
@@ -56,7 +56,9 @@ BSONObj getShardKeyPattern(OperationContext* opCtx,
         boost::optional<NamespaceString> optNss;
         {
             AutoGetCollection collection(
-                opCtx, NamespaceStringOrUUID{dbName.toString(), collectionUuid}, MODE_IS);
+                opCtx,
+                NamespaceStringOrUUID{DatabaseNameUtil::serialize(dbName), collectionUuid},
+                MODE_IS);
 
             auto optMetadata = CollectionShardingRuntime::assertCollectionLockedAndAcquireShared(
                                    opCtx, collection.getNss())
diff --git a/src/mongo/db/s/range_deletion_util.cpp b/src/mongo/db/s/range_deletion_util.cpp
index c926f3b7b18..b629d67f280 100644
--- a/src/mongo/db/s/range_deletion_util.cpp
+++ b/src/mongo/db/s/range_deletion_util.cpp
@@ -292,7 +292,8 @@ Status deleteRangeInBatches(OperationContext* opCtx,
             int numDeleted;
             const auto nss = [&]() {
                 try {
-                    const auto nssOrUuid = NamespaceStringOrUUID{dbName.toString(), collectionUuid};
+                    const auto nssOrUuid =
+                        NamespaceStringOrUUID{DatabaseNameUtil::serialize(dbName), collectionUuid};
                     const auto collection =
                         acquireCollection(opCtx,
                                           {nssOrUuid,
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 4cfa38852c0..b6f3ad8533e 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -846,7 +846,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
 }
 
 std::string StorageEngineImpl::getFilesystemPathForDb(const DatabaseName& dbName) const {
-    return _catalog->getFilesystemPathForDb(dbName.toString());
+    return _catalog->getFilesystemPathForDb(DatabaseNameUtil::serializeForCatalog(dbName));
 }
 
 void StorageEngineImpl::cleanShutdown(ServiceContext* svcCtx) {
diff --git a/src/mongo/s/commands/cluster_explain_cmd.cpp b/src/mongo/s/commands/cluster_explain_cmd.cpp
index 8b8f03db407..5dbcb614859 100644
--- a/src/mongo/s/commands/cluster_explain_cmd.cpp
+++ b/src/mongo/s/commands/cluster_explain_cmd.cpp
@@ -170,7 +170,7 @@ std::unique_ptr<CommandInvocation> ClusterExplainCmd::parse(OperationContext* op
         IDLParserContext(ExplainCommandRequest::kCommandName,
                          APIParameters::get(opCtx).getAPIStrict().value_or(false)),
         request.body);
-    std::string dbName = cmdObj.getDbName().toString();
+    std::string dbName = DatabaseNameUtil::serialize(cmdObj.getDbName());
     ExplainOptions::Verbosity verbosity = cmdObj.getVerbosity();
     // This is the nested command which we are explaining. We need to propagate generic
     // arguments into the inner command since it is what is passed to the virtual
diff --git a/src/mongo/s/commands/cluster_user_management_commands.cpp b/src/mongo/s/commands/cluster_user_management_commands.cpp
index 42512a64b3e..616c47a2f1a 100644
--- a/src/mongo/s/commands/cluster_user_management_commands.cpp
+++ b/src/mongo/s/commands/cluster_user_management_commands.cpp
@@ -240,7 +240,7 @@ public:
             BSONObjBuilder builder;
             const bool ok = Grid::get(opCtx)->catalogClient()->runUserManagementReadCommand(
                 opCtx,
-                cmd.getDbName().toString(),
+                DatabaseNameUtil::serialize(cmd.getDbName()),
                 applyReadWriteConcern(
                     opCtx,
                     this,
diff --git a/src/mongo/s/query_analysis_client.cpp b/src/mongo/s/query_analysis_client.cpp
index 5651d403209..9085a4e6e4c 100644
--- a/src/mongo/s/query_analysis_client.cpp
+++ b/src/mongo/s/query_analysis_client.cpp
@@ -73,7 +73,7 @@ void QueryAnalysisClient::setTaskExecutor(ServiceContext* service,
 bool QueryAnalysisClient::_canAcceptWrites(OperationContext* opCtx, const DatabaseName& dbName) {
     repl::ReplicationStateTransitionLockGuard rstl(opCtx, MODE_IX);
     return mongo::repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesForDatabase(
-        opCtx, dbName.toString());
+        opCtx, DatabaseNameUtil::serialize(dbName));
 }
 
 BSONObj QueryAnalysisClient::_executeCommandOnPrimaryLocal(
@@ -103,7 +103,7 @@ BSONObj QueryAnalysisClient::_executeCommandOnPrimaryRemote(
     invariant(executor, "Failed to run command since the executor has not been initialized");
 
     executor::RemoteCommandRequest request(
-        std::move(hostAndPort), dbName.toString(), cmdObj, opCtx);
+        std::move(hostAndPort), DatabaseNameUtil::serialize(dbName), cmdObj, opCtx);
     auto [promise, future] = makePromiseFuture<executor::TaskExecutor::RemoteCommandCallbackArgs>();
     auto promisePtr = std::make_shared<Promise<executor::TaskExecutor::RemoteCommandCallbackArgs>>(
         std::move(promise));
diff --git a/src/mongo/shell/encrypted_dbclient_base.cpp b/src/mongo/shell/encrypted_dbclient_base.cpp
index 6cd1572be6d..e27671e254f 100644
--- a/src/mongo/shell/encrypted_dbclient_base.cpp
+++ b/src/mongo/shell/encrypted_dbclient_base.cpp
@@ -167,7 +167,7 @@ BSONObj EncryptedDBClientBase::encryptDecryptCommand(const BSONObj& object,
     }
     invariant(frameStack.size() == 1);
     // Append '$db' which shouldn't contain tenantid.
-    frameStack.top().second.append("$db", dbName.toString());
+    frameStack.top().second.append("$db", dbName.toString_forTest());
     // If encrypt request, append '$tenant' which contains tenantid.
     if (encrypt && dbName.tenantId() && !object.hasField("$tenant")) {
         dbName.tenantId()->serializeToBSON("$tenant", &frameStack.top().second);
diff --git a/src/mongo/util/database_name_util.cpp b/src/mongo/util/database_name_util.cpp
index bc2b999cc5e..a79255578eb 100644
--- a/src/mongo/util/database_name_util.cpp
+++ b/src/mongo/util/database_name_util.cpp
@@ -60,6 +60,11 @@ std::string DatabaseNameUtil::serializeForStorage(const DatabaseName& dbName,
     return dbName.toStringWithTenantId();
 }
 
+std::string DatabaseNameUtil::serializeForCatalog(const DatabaseName& dbName,
+                                                  const SerializationContext& context) {
+    return dbName.toStringWithTenantId();
+}
+
 std::string DatabaseNameUtil::serializeForCommands(const DatabaseName& dbName,
                                                    const SerializationContext& context) {
     // tenantId came from either a $tenant field or security token
@@ -88,7 +93,6 @@ std::string DatabaseNameUtil::serializeForCommands(const DatabaseName& dbName,
     }
 }
 
-
 DatabaseName DatabaseNameUtil::parseDbNameFromStringExpectTenantIdInMultitenancyMode(
     StringData dbName) {
     if (!gMultitenancySupport) {
@@ -196,4 +200,12 @@ DatabaseName DatabaseNameUtil::deserializeForCommands(boost::optional<TenantId>
     return dbName;
 }
 
+DatabaseName DatabaseNameUtil::deserializeForCatalog(StringData db,
+                                                     const SerializationContext& context) {
+    // TenantId always prefix in the passed `db` for durable catalog. This method below checks for
+    // multitenancy and will either return a DatabaseName with (tenantId, nonPrefixedDb) or
+    // (none, prefixedDb).
+    return DatabaseNameUtil::parseDbNameFromStringExpectTenantIdInMultitenancyMode(db);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/util/database_name_util.h b/src/mongo/util/database_name_util.h
index 352f24a9dd8..d58d6f02c8c 100644
--- a/src/mongo/util/database_name_util.h
+++ b/src/mongo/util/database_name_util.h
@@ -67,6 +67,13 @@ public:
         const DatabaseName& dbName, const SerializationContext& context = SerializationContext());
 
     /**
+     * To be used only for durable catalog. We always include the tenantId as prefixed in a
+     * databasename for the catalog.
+     */
+    static std::string serializeForCatalog(
+        const DatabaseName& dbName, const SerializationContext& context = SerializationContext());
+
+    /**
      * Deserializes StringData dbName to a DatabaseName object.
      *
      * If multitenancySupport is enabled and featureFlagRequireTenantID is enabled, then a
@@ -105,6 +112,12 @@ public:
         StringData db,
         const SerializationContext& context = SerializationContext());
 
+    /**
+     * To be used only by the storage catalog.
+     */
+    static DatabaseName deserializeForCatalog(
+        StringData db, const SerializationContext& context = SerializationContext());
+
 private:
     static DatabaseName parseDbNameFromStringExpectTenantIdInMultitenancyMode(StringData dbName);
 };
diff --git a/src/mongo/util/database_name_util_test.cpp b/src/mongo/util/database_name_util_test.cpp
index d573ee33434..1c292a9c2e5 100644
--- a/src/mongo/util/database_name_util_test.cpp
+++ b/src/mongo/util/database_name_util_test.cpp
@@ -273,6 +273,46 @@ TEST(DatabaseNameUtilTest, SerializeExpectPrefixTrue_CommandReply) {
     }
 }
 
+TEST(DatabaseNameUtilTest, Serialize_StorageCatalog) {
+    TenantId tenantId(OID::gen());
+    const std::string dbnString = "foo";
+    const std::string dbnPrefixString = str::stream() << tenantId.toString() << "_" << dbnString;
+
+    {
+        RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", false);
+        {  // No prefix, no tenantId.
+            // request --> { ns: database.coll }
+            auto dbName = DatabaseNameUtil::deserializeForCatalog(dbnString);
+            ASSERT_EQ(dbName.tenantId(), boost::none);
+            ASSERT_EQ(DatabaseNameUtil::serializeForCatalog(dbName), dbnString);
+        }
+
+        {  // Has prefix, no tenantId. Storage catalog always returns prefixed dbname.
+            // request --> { ns: tenantId_database.coll }
+            auto dbName = DatabaseNameUtil::deserializeForCatalog(dbnPrefixString);
+            ASSERT_EQ(dbName.tenantId(), boost::none);
+            ASSERT_EQ(DatabaseNameUtil::serializeForCatalog(dbName), dbnPrefixString);
+        }
+    }
+
+    {
+        RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
+        {  // No prefix, no tenantId.
+            // request --> { ns: database.coll }
+            auto dbName = DatabaseNameUtil::deserializeForCatalog(dbnString);
+            ASSERT_EQ(dbName.tenantId(), boost::none);
+            ASSERT_EQ(DatabaseNameUtil::serializeForCatalog(dbName), dbnString);
+        }
+
+        {  // Has prefix, no tenantId. Storage catalog always returns prefixed dbname.
+            // request --> { ns: tenantId_database.coll }
+            auto dbName = DatabaseNameUtil::deserializeForCatalog(dbnPrefixString);
+            ASSERT_EQ(dbName.tenantId(), tenantId);
+            ASSERT_EQ(DatabaseNameUtil::serializeForCatalog(dbName), dbnPrefixString);
+        }
+    }
+}
+
 TEST(DatabaseNameUtilTest, DeserializeMissingExpectPrefix_CommandRequest) {
     RAIIServerParameterControllerForTest multitenanyController("multitenancySupport", true);
     TenantId tenantId(OID::gen());
@@ -299,7 +339,7 @@ TEST(DatabaseNameUtilTest, DeserializeMissingExpectPrefix_CommandRequest) {
         auto dbName =
             DatabaseNameUtil::deserializeForCommands(boost::none, dbnPrefixString, ctxt_noTenantId);
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnString);
     }
 
     {  // No prefix, has tenantId.
@@ -307,7 +347,7 @@ TEST(DatabaseNameUtilTest, DeserializeMissingExpectPrefix_CommandRequest) {
         auto dbName =
             DatabaseNameUtil::deserializeForCommands(tenantId, dbnString, ctxt_withTenantId);
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnString);
     }
 
     {  // Has prefix, has tenantId.  *** we shouldn't see this from Atlas Proxy
@@ -315,7 +355,7 @@ TEST(DatabaseNameUtilTest, DeserializeMissingExpectPrefix_CommandRequest) {
         auto dbName =
             DatabaseNameUtil::deserializeForCommands(tenantId, dbnPrefixString, ctxt_withTenantId);
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnPrefixString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnPrefixString);
     }
 }
 
@@ -350,7 +390,7 @@ TEST(DatabaseNameUtilTest, DeserializeExpectPrefixFalse_CommandRequest) {
         // can't expect dbName.toString == dbnPrefixString as we will still attempt to parse the
         // prefix as usual.
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnString);
     }
 
     {  // No prefix, has tenantId.
@@ -358,7 +398,7 @@ TEST(DatabaseNameUtilTest, DeserializeExpectPrefixFalse_CommandRequest) {
         auto dbName =
             DatabaseNameUtil::deserializeForCommands(tenantId, dbnString, ctxt_withTenantId);
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnString);
     }
 
     {  // Has prefix, has tenantId.  *** we shouldn't see this from Atlas Proxy
@@ -366,7 +406,7 @@ TEST(DatabaseNameUtilTest, DeserializeExpectPrefixFalse_CommandRequest) {
         auto dbName =
             DatabaseNameUtil::deserializeForCommands(tenantId, dbnPrefixString, ctxt_withTenantId);
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnPrefixString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnPrefixString);
     }
 }
 
@@ -398,7 +438,7 @@ TEST(DatabaseNameUtilTest, DeserializeExpectPrefixTrue_CommandRequest) {
         auto dbName =
             DatabaseNameUtil::deserializeForCommands(boost::none, dbnPrefixString, ctxt_noTenantId);
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnString);
     }
 
     {  // No prefix, has tenantId.  *** we shouldn't see this from Atlas Proxy
@@ -414,7 +454,7 @@ TEST(DatabaseNameUtilTest, DeserializeExpectPrefixTrue_CommandRequest) {
         auto dbName =
             DatabaseNameUtil::deserializeForCommands(tenantId, dbnPrefixString, ctxt_withTenantId);
         ASSERT_EQ(dbName.tenantId(), tenantId);
-        ASSERT_EQ(dbName.toString(), dbnString);
+        ASSERT_EQ(dbName.toString_forTest(), dbnString);
     }
 }