SERVER-63101 Have CollectionCatalog APIs return a set TenantDatabaseName

12 files changed, 110 insertions, 51 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 5111b44c000..ac907b5a8ae 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -279,6 +279,7 @@ env.Library(
         '$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',
@@ -637,6 +638,7 @@ 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_stats.cpp b/src/mongo/db/catalog/catalog_stats.cpp
index 38e38ab6d1b..79d5ed0dfd2 100644
--- a/src/mongo/db/catalog/catalog_stats.cpp
+++ b/src/mongo/db/catalog/catalog_stats.cpp
@@ -81,9 +81,9 @@ public:
 
         const auto viewCatalogDbNames = catalog->getViewCatalogDbNames();
         if (const auto viewCatalog = ViewCatalog::get(opCtx)) {
-            for (const auto& dbName : viewCatalogDbNames) {
+            for (const auto& tenantDbName : viewCatalogDbNames) {
                 try {
-                    const auto viewStats = viewCatalog->getStats(dbName);
+                    const auto viewStats = viewCatalog->getStats(tenantDbName.dbName());
                     if (!viewStats) {
                         // The database may have been dropped between listing the database names and
                         // looking up the view catalog.
@@ -97,7 +97,7 @@ public:
                     LOGV2_DEBUG(5578400,
                                 2,
                                 "Failed to collect view catalog statistics",
-                                "db"_attr = dbName);
+                                "db"_attr = tenantDbName);
                 }
             }
         }
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index e6201238930..01668af845f 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -895,14 +895,16 @@ std::vector<NamespaceString> CollectionCatalog::getAllCollectionNamesFromDb(
     return ret;
 }
 
-std::vector<std::string> CollectionCatalog::getAllDbNames() const {
-    std::vector<std::string> ret;
+std::vector<TenantDatabaseName> CollectionCatalog::getAllDbNames() const {
+    std::vector<TenantDatabaseName> ret;
     auto maxUuid = UUID::parse("FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF").getValue();
     auto iter = _orderedCollections.upper_bound(std::make_pair("", maxUuid));
     while (iter != _orderedCollections.end()) {
         auto dbName = iter->first.first;
         if (iter->second->isCommitted()) {
-            ret.push_back(dbName);
+            // TODO SERVER-61988: Once iter->first.first has TenantDatabaseName object, we need not
+            // create the TenantDatabaseName here.
+            ret.push_back(TenantDatabaseName(boost::none, dbName));
         } else {
             // If the first collection found for `dbName` is not yet committed, increment the
             // iterator to find the next visible collection (possibly under a different `dbName`).
@@ -941,7 +943,9 @@ CollectionCatalog::Stats CollectionCatalog::getStats() const {
 CollectionCatalog::ViewCatalogSet CollectionCatalog::getViewCatalogDbNames() const {
     ViewCatalogSet results;
     for (const auto& dbNameViewSetPair : _views) {
-        results.insert(dbNameViewSetPair.first);
+        // TODO SERVER-61988: This should insert dbNameViewSetPair.first instead of creating a new
+        // TenantDatabaseName object.
+        results.insert(TenantDatabaseName(boost::none, dbNameViewSetPair.first));
     }
     return results;
 }
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index 413378233e0..31c1e969bb9 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -309,7 +309,7 @@ public:
      *
      * Unlike DatabaseHolder::getNames(), this does not return databases that are empty.
      */
-    std::vector<std::string> getAllDbNames() const;
+    std::vector<TenantDatabaseName> getAllDbNames() const;
 
     /**
      * Sets 'newProfileSettings' as the profiling settings for the database 'dbName'.
@@ -361,7 +361,7 @@ public:
     /**
      * Returns a set of databases, by name, that have view catalogs.
      */
-    using ViewCatalogSet = StringSet;
+    using ViewCatalogSet = absl::flat_hash_set<TenantDatabaseName>;
     ViewCatalogSet getViewCatalogDbNames() const;
 
     /**
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index bc5269e25ea..22f097048b3 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -574,8 +574,12 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
     std::sort(res.begin(), res.end());
     ASSERT(res == dCollList);
 
-    std::vector<std::string> dbNames = {"dbA", "dbB", "dbC", "dbD", "testdb"};
-    ASSERT(catalog.getAllDbNames() == dbNames);
+    std::vector<TenantDatabaseName> tenantDbNames = {TenantDatabaseName(boost::none, "dbA"),
+                                                     TenantDatabaseName(boost::none, "dbB"),
+                                                     TenantDatabaseName(boost::none, "dbC"),
+                                                     TenantDatabaseName(boost::none, "dbD"),
+                                                     TenantDatabaseName(boost::none, "testdb")};
+    ASSERT(catalog.getAllDbNames() == tenantDbNames);
 
     catalog.deregisterAllCollectionsAndViews();
 }
@@ -628,8 +632,11 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommitt
     auto res = catalog.getAllCollectionNamesFromDb(opCtx.get(), "dbA");
     ASSERT(res.empty());
 
-    std::vector<std::string> dbNames = {"dbB", "dbC", "dbD", "testdb"};
-    ASSERT(catalog.getAllDbNames() == dbNames);
+    std::vector<TenantDatabaseName> tenantDbNames = {TenantDatabaseName(boost::none, "dbB"),
+                                                     TenantDatabaseName(boost::none, "dbC"),
+                                                     TenantDatabaseName(boost::none, "dbD"),
+                                                     TenantDatabaseName(boost::none, "testdb")};
+    ASSERT(catalog.getAllDbNames() == tenantDbNames);
 
     // One dbName with both visible and invisible collections is still visible.
     std::vector<NamespaceString> dbDNss = {d1Coll, d2Coll, d3Coll};
@@ -648,7 +655,7 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommitt
         std::sort(res.begin(), res.end());
         ASSERT(res == dCollList);
 
-        ASSERT(catalog.getAllDbNames() == dbNames);
+        ASSERT(catalog.getAllDbNames() == tenantDbNames);
         invisibleCollD->setCommitted(true);
     }
 
@@ -661,7 +668,7 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommitt
         invisibleColl->setCommitted(false);
     }
 
-    std::vector<std::string> dbList = {"testdb"};
+    std::vector<TenantDatabaseName> dbList = {TenantDatabaseName(boost::none, "testdb")};
     ASSERT(catalog.getAllDbNames() == dbList);
 
     catalog.deregisterAllCollectionsAndViews();
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 7551a05e8a6..297c2dc1597 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -368,6 +368,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/exec/sbe/query_sbe_abt',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
         '$BUILD_DIR/mongo/db/index_commands_idl',
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/ops/write_ops_exec',
         '$BUILD_DIR/mongo/db/pipeline/aggregation_request_helper',
         '$BUILD_DIR/mongo/db/pipeline/process_interface/mongo_process_interface',
@@ -379,6 +380,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/repl/replica_set_messages',
         '$BUILD_DIR/mongo/db/repl/tenant_migration_access_blocker',
         '$BUILD_DIR/mongo/db/rw_concern_d',
+        '$BUILD_DIR/mongo/db/server_feature_flags',
         '$BUILD_DIR/mongo/db/stats/counters',
         '$BUILD_DIR/mongo/db/stats/server_read_concern_write_concern_metrics',
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
diff --git a/src/mongo/db/commands/validate_db_metadata_cmd.cpp b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
index 999865ba5b3..13b832c3257 100644
--- a/src/mongo/db/commands/validate_db_metadata_cmd.cpp
+++ b/src/mongo/db/commands/validate_db_metadata_cmd.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/commands/validate_db_metadata_common.h"
 #include "mongo/db/commands/validate_db_metadata_gen.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/multitenancy.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/logv2/log.h"
 namespace mongo {
@@ -119,19 +120,21 @@ public:
 
             // If there is no database name present in the input, run validation against all the
             // databases.
-            auto dbNames = validateCmdRequest.getDb()
-                ? std::vector<std::string>{validateCmdRequest.getDb()->toString()}
+            auto tenantDbNames = validateCmdRequest.getDb()
+                ? std::vector<TenantDatabaseName>{TenantDatabaseName(
+                      getActiveTenant(opCtx), validateCmdRequest.getDb()->toString())}
                 : collectionCatalog->getAllDbNames();
 
-            for (const auto& dbName : dbNames) {
-                AutoGetDb autoDb(opCtx, dbName, LockMode::MODE_IS);
+            for (const auto& tenantDbName : tenantDbNames) {
+                AutoGetDb autoDb(opCtx, tenantDbName.dbName(), LockMode::MODE_IS);
                 if (!autoDb.getDb()) {
                     continue;
                 }
 
                 if (validateCmdRequest.getCollection()) {
-                    if (!_validateNamespace(
-                            opCtx, NamespaceString(dbName, *validateCmdRequest.getCollection()))) {
+                    if (!_validateNamespace(opCtx,
+                                            NamespaceString(tenantDbName.dbName(),
+                                                            *validateCmdRequest.getCollection()))) {
                         return;
                     }
                     continue;
@@ -139,13 +142,15 @@ 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, dbName)) {
-                    viewCatalog->iterate(dbName, [this, opCtx](const ViewDefinition& view) {
-                        return _validateView(opCtx, view);
-                    });
+                if (auto viewCatalog =
+                        DatabaseHolder::get(opCtx)->getViewCatalog(opCtx, tenantDbName.dbName())) {
+                    viewCatalog->iterate(tenantDbName.dbName(),
+                                         [this, opCtx](const ViewDefinition& view) {
+                                             return _validateView(opCtx, view);
+                                         });
                 }
 
-                for (auto collIt = collectionCatalog->begin(opCtx, dbName);
+                for (auto collIt = collectionCatalog->begin(opCtx, tenantDbName.dbName());
                      collIt != collectionCatalog->end(opCtx);
                      ++collIt) {
                     if (!_validateNamespace(
diff --git a/src/mongo/db/repl/idempotency_test_fixture.cpp b/src/mongo/db/repl/idempotency_test_fixture.cpp
index 938ccb8749e..08a0d446c85 100644
--- a/src/mongo/db/repl/idempotency_test_fixture.cpp
+++ b/src/mongo/db/repl/idempotency_test_fixture.cpp
@@ -354,14 +354,15 @@ std::string IdempotencyTest::computeDataHash(const CollectionPtr& collection) {
 std::vector<CollectionState> IdempotencyTest::validateAllCollections() {
     std::vector<CollectionState> collStates;
     auto catalog = CollectionCatalog::get(_opCtx.get());
-    auto dbs = catalog->getAllDbNames();
-    for (auto& db : dbs) {
+    auto tenantDbNames = catalog->getAllDbNames();
+    for (auto& tenantDbName : tenantDbNames) {
         // Skip local database.
-        if (db != "local") {
+        if (tenantDbName.dbName() != "local") {
             std::vector<NamespaceString> collectionNames;
             {
-                Lock::DBLock lk(_opCtx.get(), db, MODE_S);
-                collectionNames = catalog->getAllCollectionNamesFromDb(_opCtx.get(), db);
+                Lock::DBLock lk(_opCtx.get(), tenantDbName.dbName(), MODE_S);
+                collectionNames =
+                    catalog->getAllCollectionNamesFromDb(_opCtx.get(), tenantDbName.dbName());
             }
             for (const auto& nss : collectionNames) {
                 collStates.push_back(validate(nss));
diff --git a/src/mongo/db/s/migration_util.cpp b/src/mongo/db/s/migration_util.cpp
index aeab3994ee9..24ed3c8147d 100644
--- a/src/mongo/db/s/migration_util.cpp
+++ b/src/mongo/db/s/migration_util.cpp
@@ -612,13 +612,15 @@ void submitOrphanRanges(OperationContext* opCtx, const NamespaceString& nss, con
 
 void submitOrphanRangesForCleanup(OperationContext* opCtx) {
     auto catalog = CollectionCatalog::get(opCtx);
-    const auto& dbs = catalog->getAllDbNames();
+    const auto& tenantDbNames = catalog->getAllDbNames();
 
-    for (const auto& dbName : dbs) {
-        if (dbName == NamespaceString::kLocalDb)
+    for (const auto& tenantDbName : tenantDbNames) {
+        if (tenantDbName.dbName() == NamespaceString::kLocalDb)
             continue;
 
-        for (auto collIt = catalog->begin(opCtx, dbName); collIt != catalog->end(opCtx); ++collIt) {
+        for (auto collIt = catalog->begin(opCtx, tenantDbName.dbName());
+             collIt != catalog->end(opCtx);
+             ++collIt) {
             auto uuid = collIt.uuid().get();
             auto nss = catalog->lookupNSSByUUID(opCtx, uuid).get();
             LOGV2_DEBUG(22034,
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index 16435c9cac8..de771eff66e 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -774,14 +774,7 @@ RecoveryUnit* StorageEngineImpl::newRecoveryUnit() {
 }
 
 std::vector<TenantDatabaseName> StorageEngineImpl::listDatabases() const {
-    // TODO SERVER-61988 Return instead the result of CollectionCatalog::getAllDbNames() directly.
-    std::vector<TenantDatabaseName> tenantDbNames;
-    std::vector<std::string> dbNames =
-        CollectionCatalog::get(getGlobalServiceContext())->getAllDbNames();
-    for (auto dbName : dbNames) {
-        tenantDbNames.push_back(TenantDatabaseName(boost::none, dbName));
-    }
-    return tenantDbNames;
+    return CollectionCatalog::get(getGlobalServiceContext())->getAllDbNames();
 }
 
 Status StorageEngineImpl::closeDatabase(OperationContext* opCtx, StringData db) {
@@ -791,9 +784,10 @@ Status StorageEngineImpl::closeDatabase(OperationContext* opCtx, StringData db)
 
 Status StorageEngineImpl::dropDatabase(OperationContext* opCtx, StringData db) {
     auto catalog = CollectionCatalog::get(opCtx);
+    const TenantDatabaseName tenantDbName(boost::none, db);
     {
-        auto dbs = catalog->getAllDbNames();
-        if (std::count(dbs.begin(), dbs.end(), db.toString()) == 0) {
+        auto tenantDbNames = catalog->getAllDbNames();
+        if (std::count(tenantDbNames.begin(), tenantDbNames.end(), tenantDbName) == 0) {
             return Status(ErrorCodes::NamespaceNotFound, "db not found to drop");
         }
     }
diff --git a/src/mongo/db/tenant_database_name.h b/src/mongo/db/tenant_database_name.h
index eebe8301618..615e832c735 100644
--- a/src/mongo/db/tenant_database_name.h
+++ b/src/mongo/db/tenant_database_name.h
@@ -78,12 +78,12 @@ public:
         return fullName();
     }
 
-    friend bool operator==(const TenantDatabaseName& l, const TenantDatabaseName& r) {
-        return l.tenantId() == r.tenantId() && l.dbName() == r.dbName();
-    }
-
-    friend bool operator!=(const TenantDatabaseName& l, const TenantDatabaseName& r) {
-        return !(l == r);
+    /**
+     * Returns -1, 0, or 1 if 'this' is less, equal, or greater than 'other' in
+     * lexicographical order.
+     */
+    int compare(const TenantDatabaseName& other) const {
+        return fullName().compare(other.fullName());
     }
 
     template <typename H>
@@ -100,4 +100,29 @@ private:
     std::string _dbName;
     boost::optional<std::string> _tenantDbName;
 };
+
+inline bool operator==(const TenantDatabaseName& lhs, const TenantDatabaseName& rhs) {
+    return lhs.compare(rhs) == 0;
+}
+
+inline bool operator!=(const TenantDatabaseName& lhs, const TenantDatabaseName& rhs) {
+    return !(lhs == rhs);
+}
+
+inline bool operator<(const TenantDatabaseName& lhs, const TenantDatabaseName& rhs) {
+    return lhs.compare(rhs) < 0;
+}
+
+inline bool operator>(const TenantDatabaseName& lhs, const TenantDatabaseName& rhs) {
+    return rhs < lhs;
+}
+
+inline bool operator<=(const TenantDatabaseName& lhs, const TenantDatabaseName& rhs) {
+    return !(lhs > rhs);
+}
+
+inline bool operator>=(const TenantDatabaseName& lhs, const TenantDatabaseName& rhs) {
+    return !(lhs < rhs);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/tenant_database_name_test.cpp b/src/mongo/db/tenant_database_name_test.cpp
index b2aa8632404..4c91d9da8c1 100644
--- a/src/mongo/db/tenant_database_name_test.cpp
+++ b/src/mongo/db/tenant_database_name_test.cpp
@@ -130,5 +130,22 @@ TEST(TenantDatabaseNameTest, VerifyHashFunction) {
     ASSERT_EQUALS(dbMap[tdn3], "value no tenant a2");
 }
 
+TEST(TenantDatabaseNameTest, VerifyCompareFunction) {
+    TenantId tenantId1 = TenantId(OID::gen());
+    TenantId tenantId2 = TenantId(OID::gen());
+
+    // OID's generated by the same process are monotonically increasing.
+    ASSERT(tenantId1 < tenantId2);
+
+    TenantDatabaseName tdn1a = TenantDatabaseName(tenantId1, "a");
+    TenantDatabaseName tdn1b = TenantDatabaseName(tenantId1, "b");
+    TenantDatabaseName tdn2a = TenantDatabaseName(tenantId2, "a");
+    TenantDatabaseName tdn3a = TenantDatabaseName(boost::none, "a");
+
+    ASSERT(tdn1a < tdn1b);
+    ASSERT(tdn1b < tdn2a);
+    ASSERT(tdn3a != tdn1a);
+    ASSERT(tdn1a != tdn2a);
+}
 }  // namespace
 }  // namespace mongo