SERVER-29370 Update UUIDs when featureCompatibilityVersion is changed

This patch ensures collections have UUIDs when
featureCompatibilityVersion is 3.6 and collections do not have UUIDs
when featureCompatibilityVersion is 3.4 (or below).

19 files changed, 470 insertions, 23 deletions

diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index b58b1ea96bf..c4dd2a4c4b6 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog_entry.h"
+#include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/client.h"
 #include "mongo/db/db_raii.h"
@@ -259,13 +260,12 @@ void setCollectionOptionFlag(OperationContext* opCtx,
     invariant(newOptions.flagsSet);
 }
 
-}  // namespace
-}  // namespace mongo
-
-mongo::Status mongo::collMod(OperationContext* opCtx,
-                             const NamespaceString& nss,
-                             const BSONObj& cmdObj,
-                             BSONObjBuilder* result) {
+Status _collModInternal(OperationContext* opCtx,
+                        const NamespaceString& nss,
+                        const BSONObj& cmdObj,
+                        BSONObjBuilder* result,
+                        bool upgradeUUID,
+                        OptionalCollectionUUID uuid) {
     StringData dbName = nss.db();
     AutoGetDb autoDb(opCtx, dbName, MODE_X);
     Database* const db = autoDb.getDb();
@@ -389,6 +389,18 @@ mongo::Status mongo::collMod(OperationContext* opCtx,
     if (!cmr.noPadding.eoo())
         setCollectionOptionFlag(opCtx, coll, cmr.noPadding, result);
 
+    // Modify collection UUID if we are upgrading or downgrading. This is a no-op if we have
+    // already upgraded or downgraded.
+    if (upgradeUUID) {
+        if (uuid && !coll->uuid()) {
+            CollectionCatalogEntry* cce = coll->getCatalogEntry();
+            cce->addUUID(opCtx, uuid.get(), coll);
+        } else if (!uuid && coll->uuid()) {
+            CollectionCatalogEntry* cce = coll->getCatalogEntry();
+            cce->removeUUID(opCtx);
+        }
+        coll->refreshUUID(opCtx);
+    }
 
     // Only observe non-view collMods, as view operations are observed as operations on the
     // system.views collection.
@@ -399,3 +411,145 @@ mongo::Status mongo::collMod(OperationContext* opCtx,
 
     return Status::OK();
 }
+
+void _updateDBSchemaVersion(OperationContext* opCtx,
+                            const std::string& dbname,
+                            bool needUUIDAdded) {
+    // Iterate through all collections of database dbname and make necessary UUID changes.
+    std::vector<NamespaceString> collNamespaceStrings;
+    {
+        AutoGetDb autoDb(opCtx, dbname, MODE_X);
+        Database* const db = autoDb.getDb();
+        for (auto collectionIt = db->begin(); collectionIt != db->end(); ++collectionIt) {
+            Collection* coll = *collectionIt;
+            collNamespaceStrings.push_back(coll->ns());
+        }
+    }
+    for (auto& collNSS : collNamespaceStrings) {
+        // Skip system.namespaces until SERVER-30095 is addressed.
+        if (collNSS.coll() == "system.namespaces") {
+            continue;
+        }
+        // Skip all non-replicated collections.
+        if (collNSS.db() == "local" || collNSS.coll() == "system.profile") {
+            continue;
+        }
+        AutoGetDb autoDb(opCtx, dbname, MODE_X);
+        Database* const db = autoDb.getDb();
+        Collection* coll = db->getCollection(opCtx, collNSS);
+        BSONObjBuilder collModObjBuilder;
+        collModObjBuilder.append("collMod", coll->ns().coll());
+        BSONObj collModObj = collModObjBuilder.done();
+        OptionalCollectionUUID uuid = boost::none;
+        if (needUUIDAdded) {
+            uuid = UUID::gen();
+        }
+        if ((needUUIDAdded && !coll->uuid()) || (!needUUIDAdded && coll->uuid())) {
+            uassertStatusOK(collModForUUIDUpgrade(opCtx, coll->ns(), collModObj, uuid));
+        }
+    }
+}
+
+void _updateDBSchemaVersionNonReplicated(OperationContext* opCtx,
+                                         const std::string& dbname,
+                                         bool needUUIDAdded) {
+    // Iterate through all collections if we're in the "local" database.
+    std::vector<NamespaceString> collNamespaceStrings;
+    if (dbname == "local") {
+        AutoGetDb autoDb(opCtx, dbname, MODE_X);
+        Database* const db = autoDb.getDb();
+        for (auto collectionIt = db->begin(); collectionIt != db->end(); ++collectionIt) {
+            Collection* coll = *collectionIt;
+            collNamespaceStrings.push_back(coll->ns());
+        }
+    } else {
+        // If we're not in the "local" database, the only non-replicated collection
+        // is system.profile, if present.
+        collNamespaceStrings.push_back(NamespaceString(dbname, "system.profile"));
+    }
+    for (auto& collNSS : collNamespaceStrings) {
+        // Skip system.namespaces until SERVER-30095 is addressed.
+        if (collNSS.coll() == "system.namespaces") {
+            continue;
+        }
+        AutoGetDb autoDb(opCtx, dbname, MODE_X);
+        Database* const db = autoDb.getDb();
+        Collection* coll = db->getCollection(opCtx, collNSS);
+        if (!coll) {
+            // This will only occur if we incorrectly assumed there was a
+            // system.profile collection present.
+            return;
+        }
+        BSONObjBuilder collModObjBuilder;
+        collModObjBuilder.append("collMod", coll->ns().coll());
+        BSONObj collModObj = collModObjBuilder.done();
+        OptionalCollectionUUID uuid = boost::none;
+        if (needUUIDAdded) {
+            uuid = UUID::gen();
+        }
+        if ((needUUIDAdded && !coll->uuid()) || (!needUUIDAdded && coll->uuid())) {
+            BSONObjBuilder resultWeDontCareAbout;
+            uassertStatusOK(_collModInternal(
+                opCtx, coll->ns(), collModObj, &resultWeDontCareAbout, /*upgradeUUID*/ true, uuid));
+        }
+    }
+}
+
+void updateUUIDSchemaVersionNonReplicated(OperationContext* opCtx, bool upgrade) {
+    if (!enableCollectionUUIDs) {
+        return;
+    }
+    // Update UUIDs on all collections of all non-replicated databases.
+    std::vector<std::string> dbNames;
+    StorageEngine* storageEngine = opCtx->getServiceContext()->getGlobalStorageEngine();
+    {
+        Lock::GlobalLock lk(opCtx, MODE_IS, UINT_MAX);
+        storageEngine->listDatabases(&dbNames);
+    }
+    for (auto it = dbNames.begin(); it != dbNames.end(); ++it) {
+        auto dbName = *it;
+        _updateDBSchemaVersionNonReplicated(opCtx, dbName, upgrade);
+    }
+}
+}  // namespace
+
+Status collMod(OperationContext* opCtx,
+               const NamespaceString& nss,
+               const BSONObj& cmdObj,
+               BSONObjBuilder* result) {
+    return _collModInternal(
+        opCtx, nss, cmdObj, result, /*upgradeUUID*/ false, /*UUID*/ boost::none);
+}
+
+Status collModForUUIDUpgrade(OperationContext* opCtx,
+                             const NamespaceString& nss,
+                             const BSONObj& cmdObj,
+                             OptionalCollectionUUID uuid) {
+    BSONObjBuilder resultWeDontCareAbout;
+    // First update all non-replicated collection UUIDs.
+    updateUUIDSchemaVersionNonReplicated(opCtx, !!uuid);
+    return _collModInternal(opCtx, nss, cmdObj, &resultWeDontCareAbout, /*upgradeUUID*/ true, uuid);
+}
+
+void updateUUIDSchemaVersion(OperationContext* opCtx, bool upgrade) {
+    if (!enableCollectionUUIDs) {
+        return;
+    }
+    // Update UUIDs on all collections of all databases.
+    std::vector<std::string> dbNames;
+    StorageEngine* storageEngine = opCtx->getServiceContext()->getGlobalStorageEngine();
+    {
+        Lock::GlobalLock lk(opCtx, MODE_IS, UINT_MAX);
+        storageEngine->listDatabases(&dbNames);
+    }
+
+    for (auto it = dbNames.begin(); it != dbNames.end(); ++it) {
+        auto dbName = *it;
+        _updateDBSchemaVersion(opCtx, dbName, upgrade);
+    }
+    const WriteConcernOptions writeConcern(WriteConcernOptions::kMajority,
+                                           WriteConcernOptions::SyncMode::UNSET,
+                                           /*timeout*/ INT_MAX);
+    repl::getGlobalReplicationCoordinator()->awaitReplicationOfLastOpForClient(opCtx, writeConcern);
+}
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/coll_mod.h b/src/mongo/db/catalog/coll_mod.h
index 456103f4c2f..8a6eda14f46 100644
--- a/src/mongo/db/catalog/coll_mod.h
+++ b/src/mongo/db/catalog/coll_mod.h
@@ -28,6 +28,7 @@
 
 #include "mongo/base/status.h"
 #include "mongo/base/status_with.h"
+#include "mongo/db/catalog/collection_options.h"
 
 namespace mongo {
 class BSONObj;
@@ -36,6 +37,8 @@ class Collection;
 class NamespaceString;
 class OperationContext;
 
+void updateUUIDSchemaVersion(OperationContext* opCtx, bool upgrade);
+
 /**
  * Performs the collection modification described in "cmdObj" on the collection "ns".
  */
@@ -43,4 +46,14 @@ Status collMod(OperationContext* opCtx,
                const NamespaceString& ns,
                const BSONObj& cmdObj,
                BSONObjBuilder* result);
+
+/*
+ * Adds uuid to the collection "ns" if uuid exists and removes any existing UUID from
+ * the collection "ns" if uuid is boost::none. This is called in circumstances where
+ * we may be upgrading or downgrading and we need to update the UUID.
+ */
+Status collModForUUIDUpgrade(OperationContext* opCtx,
+                             const NamespaceString& ns,
+                             const BSONObj& cmdObj,
+                             OptionalCollectionUUID uuid);
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 39d61c71477..d1f08edc0e8 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -215,6 +215,8 @@ public:
         virtual const NamespaceString& ns() const = 0;
         virtual OptionalCollectionUUID uuid() const = 0;
 
+        virtual void refreshUUID(OperationContext* opCtx) = 0;
+
         virtual const IndexCatalog* getIndexCatalog() const = 0;
         virtual IndexCatalog* getIndexCatalog() = 0;
 
@@ -390,6 +392,10 @@ public:
         return this->_impl().uuid();
     }
 
+    inline void refreshUUID(OperationContext* opCtx) {
+        return this->_impl().refreshUUID(opCtx);
+    }
+
     inline const IndexCatalog* getIndexCatalog() const {
         return this->_impl().getIndexCatalog();
     }
diff --git a/src/mongo/db/catalog/collection_catalog_entry.h b/src/mongo/db/catalog/collection_catalog_entry.h
index 535a1b6e09a..cbf206b0c13 100644
--- a/src/mongo/db/catalog/collection_catalog_entry.h
+++ b/src/mongo/db/catalog/collection_catalog_entry.h
@@ -35,6 +35,7 @@
 #include "mongo/db/index/multikey_paths.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/record_id.h"
+#include "mongo/db/server_options.h"
 #include "mongo/db/storage/kv/kv_prefix.h"
 
 namespace mongo {
@@ -132,6 +133,22 @@ public:
                                  const BSONObj& validator,
                                  StringData validationLevel,
                                  StringData validationAction) = 0;
+    /**
+     * Assigns a new UUID to this collection. This is to be called when the schemaVersion is set
+     * to 3.6 and there are collections that still do not have UUIDs.
+     */
+    virtual void addUUID(OperationContext* opCtx, CollectionUUID uuid, Collection* coll) = 0;
+    /**
+     * Removes the UUID from this collection. This is to be called when the schemaVersion is set
+     * to 3.4 and there are collections that still have UUIDs.
+     */
+    virtual void removeUUID(OperationContext* opCtx) = 0;
+
+    /**
+     * Compare the UUID argument to the UUID obtained from the metadata. Return true if they
+     * are equal, false otherwise.
+     */
+    virtual bool isEqualToMetadataUUID(OperationContext* opCtx, OptionalCollectionUUID uuid) = 0;
 
     /**
      * Updates size of a capped Collection.
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index 5505b1c5c4c..92ba2076dec 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -80,6 +80,11 @@ public:
         return _uuid;
     }
 
+    void refreshUUID(OperationContext* opCtx) final {
+        auto options = getCatalogEntry()->getCollectionOptions(opCtx);
+        _uuid = options.uuid;
+    }
+
     const IndexCatalog* getIndexCatalog() const final {
         return &_indexCatalog;
     }
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index 97c59bdb8e9..e92c81924e2 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -86,6 +86,10 @@ public:
         std::abort();
     }
 
+    void refreshUUID(OperationContext* opCtx) {
+        std::abort();
+    }
+
     const IndexCatalog* getIndexCatalog() const {
         std::abort();
     }
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index b4c7a989a4b..c82948ddefa 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -722,8 +722,10 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
     invariant(!options.isView());
 
     CollectionOptions optionsWithUUID = options;
-    if (enableCollectionUUIDs && !optionsWithUUID.uuid)
+    if (enableCollectionUUIDs && !optionsWithUUID.uuid &&
+        serverGlobalParams.featureCompatibility.isSchemaVersion36.load() == true) {
         optionsWithUUID.uuid.emplace(CollectionUUID::gen());
+    }
 
     NamespaceString nss(ns);
     _checkCanCreateCollection(opCtx, nss, optionsWithUUID);
diff --git a/src/mongo/db/commands/feature_compatibility_version.cpp b/src/mongo/db/commands/feature_compatibility_version.cpp
index b8a54bead96..a33d48da76f 100644
--- a/src/mongo/db/commands/feature_compatibility_version.cpp
+++ b/src/mongo/db/commands/feature_compatibility_version.cpp
@@ -220,6 +220,11 @@ void FeatureCompatibilityVersion::setIfCleanStartup(OperationContext* opCtx,
             // and we have just created an empty "admin" database. Therefore, it is safe to create
             // the "admin.system.version" collection.
             invariant(autoDB.justCreated());
+
+            // We update the value of the isSchemaVersion36 server parameter so the
+            // admin.system.version collection gets a UUID.
+            serverGlobalParams.featureCompatibility.isSchemaVersion36.store(true);
+
             uassertStatusOK(storageInterface->createCollection(opCtx, nss, {}));
         }
 
@@ -247,6 +252,10 @@ void FeatureCompatibilityVersion::onInsertOrUpdate(const BSONObj& doc) {
     auto newVersion = uassertStatusOK(FeatureCompatibilityVersion::parse(doc));
     log() << "setting featureCompatibilityVersion to " << toString(newVersion);
     serverGlobalParams.featureCompatibility.version.store(newVersion);
+
+    serverGlobalParams.featureCompatibility.isSchemaVersion36.store(
+        serverGlobalParams.featureCompatibility.version.load() ==
+        ServerGlobalParams::FeatureCompatibility::Version::k36);
 }
 
 void FeatureCompatibilityVersion::onDelete(const BSONObj& doc) {
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 002cf2d1bce..f552c87478c 100644
--- a/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
+++ b/src/mongo/db/commands/set_feature_compatibility_version_command.cpp
@@ -29,9 +29,19 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/coll_mod.h"
+#include "mongo/db/catalog/database.h"
+#include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/commands/feature_compatibility_version_command_parser.h"
+#include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/db_raii.h"
+#include "mongo/db/repl/replication_coordinator.h"
+#include "mongo/db/repl/replication_coordinator_global.h"
+#include "mongo/db/server_options.h"
+#include "mongo/rpc/get_status_from_command_result.h"
+#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 
@@ -82,15 +92,46 @@ public:
         return Status::OK();
     }
 
+    bool isFCVUpgrade(StringData version) {
+        const auto existingVersion = FeatureCompatibilityVersion::toString(
+            serverGlobalParams.featureCompatibility.version.load());
+        return version == FeatureCompatibilityVersionCommandParser::kVersion36 &&
+            existingVersion == FeatureCompatibilityVersionCommandParser::kVersion34;
+    }
+
     bool run(OperationContext* opCtx,
              const std::string& dbname,
              const BSONObj& cmdObj,
              BSONObjBuilder& result) {
         const auto version = uassertStatusOK(
             FeatureCompatibilityVersionCommandParser::extractVersionFromCommand(getName(), cmdObj));
+        auto existingVersion = FeatureCompatibilityVersion::toString(
+                                   serverGlobalParams.featureCompatibility.version.load())
+                                   .toString();
+
+        // Wait for majority commit in case we're upgrading simultaneously with another session.
+        if (version == existingVersion) {
+            const WriteConcernOptions writeConcern(WriteConcernOptions::kMajority,
+                                                   WriteConcernOptions::SyncMode::UNSET,
+                                                   /*timeout*/ INT_MAX);
+            repl::getGlobalReplicationCoordinator()->awaitReplicationOfLastOpForClient(
+                opCtx, writeConcern);
+        }
+
+        if (version != existingVersion && isFCVUpgrade(version)) {
+            serverGlobalParams.featureCompatibility.isSchemaVersion36.store(true);
+            updateUUIDSchemaVersion(opCtx, /*upgrade*/ true);
+            existingVersion = version;
+        }
 
         FeatureCompatibilityVersion::set(opCtx, version);
 
+        // If version and existingVersion are still not equal, we must be downgrading.
+        if (version != existingVersion) {
+            serverGlobalParams.featureCompatibility.isSchemaVersion36.store(false);
+            updateUUIDSchemaVersion(opCtx, /*upgrade*/ false);
+        }
+
         return true;
     }
 
diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index ad2d9422734..c8229712fe5 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -388,6 +388,11 @@ void repairDatabasesAndCheckVersion(OperationContext* opCtx) {
                         fassertFailedNoTrace(40283);
                     }
                     serverGlobalParams.featureCompatibility.version.store(version.getValue());
+
+                    // Update schemaVersion parameter.
+                    serverGlobalParams.featureCompatibility.isSchemaVersion36.store(
+                        serverGlobalParams.featureCompatibility.version.load() ==
+                        ServerGlobalParams::FeatureCompatibility::Version::k36);
                 }
             }
         }
diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp
index 31836bd69e1..bfe68266c81 100644
--- a/src/mongo/db/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer_impl.cpp
@@ -32,12 +32,14 @@
 
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/auth/authorization_manager_global.h"
+#include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/namespace_uuid_cache.h"
 #include "mongo/db/catalog/uuid_catalog.h"
 #include "mongo/db/commands/feature_compatibility_version.h"
+#include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
@@ -47,6 +49,7 @@
 #include "mongo/db/server_options.h"
 #include "mongo/db/views/durable_view_catalog.h"
 #include "mongo/scripting/engine.h"
+#include "mongo/util/assert_util.h"
 
 namespace mongo {
 namespace {
@@ -310,6 +313,26 @@ void OpObserverImpl::onCollMod(OperationContext* opCtx,
     }
 
     getGlobalAuthorizationManager()->logOp(opCtx, "c", cmdNss, cmdObj, nullptr);
+
+    // Make sure the UUID values in the Collection metadata, the Collection object,
+    // and the UUID catalog are all present and equal if uuid exists and do not exist
+    // if uuid does not exist.
+    invariant(opCtx->lockState()->isDbLockedForMode(nss.db(), MODE_X));
+    Database* db = dbHolder().get(opCtx, nss.db());
+    // Some unit tests call the op observer on an unregistered Database.
+    if (!db) {
+        return;
+    }
+    Collection* coll = db->getCollection(opCtx, nss.ns());
+    invariant(coll->uuid() == uuid);
+    CollectionCatalogEntry* entry = coll->getCatalogEntry();
+    invariant(entry->isEqualToMetadataUUID(opCtx, uuid));
+
+    if (uuid) {
+        UUIDCatalog& catalog = UUIDCatalog::get(opCtx->getServiceContext());
+        Collection* catalogColl = catalog.lookupCollectionByUUID(uuid.get());
+        invariant(catalogColl && catalogColl->uuid() == uuid);
+    }
 }
 
 void OpObserverImpl::onDropDatabase(OperationContext* opCtx, const std::string& dbName) {
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index e086d0fc9e6..88eb0d2ccea 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -741,8 +741,23 @@ std::map<std::string, ApplyOpMetadata> opsMap = {
          const BSONElement& ui,
          BSONObj& cmd,
          const OpTime& opTime) -> Status {
-          BSONObjBuilder resultWeDontCareAbout;
-          return collMod(opCtx, parseUUIDorNs(opCtx, ns, ui, cmd), cmd, &resultWeDontCareAbout);
+          // Get UUID from cmd, if it exists.
+          OptionalCollectionUUID uuid;
+          NamespaceString nss;
+          if (ui.eoo()) {
+              uuid = boost::none;
+              nss = parseNs(ns, cmd);
+          } else {
+              uuid = uassertStatusOK(UUID::parse(ui));
+              // We need to see whether a collection with UUID ui exists before attempting to do
+              // a collMod on it. This is because we add UUIDs during upgrade to
+              // featureCompatibilityVersion 3.6 with a collMod command, so the collection will
+              // not have a UUID at the time we attempt to look it up by UUID.
+              auto& catalog = UUIDCatalog::get(opCtx);
+              nss = catalog.lookupCollectionByUUID(uuid.get()) ? catalog.lookupNSSByUUID(uuid.get())
+                                                               : parseNs(ns, cmd);
+          }
+          return collModForUUIDUpgrade(opCtx, nss, cmd, uuid);
       },
       {ErrorCodes::IndexNotFound, ErrorCodes::NamespaceNotFound}}},
     {"dropDatabase",
diff --git a/src/mongo/db/repl/rs_rollback_test.cpp b/src/mongo/db/repl/rs_rollback_test.cpp
index b36dbb08789..430fc143579 100644
--- a/src/mongo/db/repl/rs_rollback_test.cpp
+++ b/src/mongo/db/repl/rs_rollback_test.cpp
@@ -337,7 +337,9 @@ TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionDoesNotExist) {
 
 TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsNonCapped) {
     createOplog(_opCtx.get());
-    auto coll = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto coll = _createCollection(_opCtx.get(), "test.t", options);
     _testRollbackDelete(
         _opCtx.get(), _coordinator, _replicationProcess.get(), coll->uuid().get(), BSONObj());
     ASSERT_EQUALS(
@@ -349,6 +351,7 @@ TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsNonCapped
 TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsCapped) {
     createOplog(_opCtx.get());
     CollectionOptions options;
+    options.uuid = UUID::gen();
     options.capped = true;
     auto coll = _createCollection(_opCtx.get(), "test.t", options);
     ASSERT_EQUALS(
@@ -359,7 +362,9 @@ TEST_F(RSRollbackTest, RollbackDeleteNoDocumentAtSourceCollectionExistsCapped) {
 
 TEST_F(RSRollbackTest, RollbackDeleteRestoreDocument) {
     createOplog(_opCtx.get());
-    auto coll = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto coll = _createCollection(_opCtx.get(), "test.t", options);
     BSONObj doc = BSON("_id" << 0 << "a" << 1);
     _testRollbackDelete(
         _opCtx.get(), _coordinator, _replicationProcess.get(), coll->uuid().get(), doc);
@@ -415,7 +420,9 @@ TEST_F(RSRollbackTest, RollbackInsertDocumentWithNoId) {
 
 TEST_F(RSRollbackTest, RollbackCreateIndexCommand) {
     createOplog(_opCtx.get());
-    auto collection = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto collection = _createCollection(_opCtx.get(), "test.t", options);
     auto indexSpec = BSON("ns"
                           << "test.t"
                           << "key"
@@ -486,7 +493,9 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommand) {
 
 TEST_F(RSRollbackTest, RollbackCreateIndexCommandIndexNotInCatalog) {
     createOplog(_opCtx.get());
-    auto collection = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto collection = _createCollection(_opCtx.get(), "test.t", options);
     auto indexSpec = BSON("ns"
                           << "test.t"
                           << "key"
@@ -595,7 +604,9 @@ TEST_F(RSRollbackTest, RollbackCreateIndexCommandMissingNamespace) {
 
 TEST_F(RSRollbackTest, RollbackDropIndexCommandWithOneIndex) {
     createOplog(_opCtx.get());
-    auto collection = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto collection = _createCollection(_opCtx.get(), "test.t", options);
     {
         Lock::DBLock dbLock(_opCtx.get(), "test", MODE_S);
         auto indexCatalog = collection->getIndexCatalog();
@@ -655,7 +666,9 @@ TEST_F(RSRollbackTest, RollbackDropIndexCommandWithOneIndex) {
 
 TEST_F(RSRollbackTest, RollbackDropIndexCommandWithMultipleIndexes) {
     createOplog(_opCtx.get());
-    auto collection = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto collection = _createCollection(_opCtx.get(), "test.t", options);
     {
         Lock::DBLock dbLock(_opCtx.get(), "test", MODE_S);
         auto indexCatalog = collection->getIndexCatalog();
@@ -868,7 +881,9 @@ TEST_F(RSRollbackTest, RollbackDropCollectionCommand) {
 
     OpTime dropTime = OpTime(Timestamp(2, 0), 5);
     auto dpns = NamespaceString("test.t").makeDropPendingNamespace(dropTime);
-    auto coll = _createCollection(_opCtx.get(), dpns, CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto coll = _createCollection(_opCtx.get(), dpns, options);
     _dropPendingCollectionReaper->addDropPendingNamespace(dropTime, dpns);
 
     auto commonOperation =
@@ -922,7 +937,9 @@ TEST_F(RSRollbackTest, RollbackDropCollectionCommand) {
 
 TEST_F(RSRollbackTest, RollbackCollModCommandFailsIfRBIDChangesWhileSyncingCollectionMetadata) {
     createOplog(_opCtx.get());
-    auto coll = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto coll = _createCollection(_opCtx.get(), "test.t", options);
 
     auto commonOperation =
         std::make_pair(BSON("ts" << Timestamp(Seconds(1), 0) << "h" << 1LL), RecordId(1));
@@ -1025,12 +1042,14 @@ OpTime getOpTimeFromOplogEntry(const BSONObj& entry) {
 TEST_F(RSRollbackTest, RollbackApplyOpsCommand) {
     createOplog(_opCtx.get());
     Collection* coll = nullptr;
+    CollectionOptions options;
+    options.uuid = UUID::gen();
     {
         AutoGetOrCreateDb autoDb(_opCtx.get(), "test", MODE_X);
         mongo::WriteUnitOfWork wuow(_opCtx.get());
         coll = autoDb.getDb()->getCollection(_opCtx.get(), "test.t");
         if (!coll) {
-            coll = autoDb.getDb()->createCollection(_opCtx.get(), "test.t");
+            coll = autoDb.getDb()->createCollection(_opCtx.get(), "test.t", options);
         }
         ASSERT(coll);
         OpDebug* const nullOpDebug = nullptr;
@@ -1140,7 +1159,7 @@ TEST_F(RSRollbackTest, RollbackApplyOpsCommand) {
         mutable std::multiset<int> searchedIds;
     } rollbackSource(std::unique_ptr<OplogInterface>(new OplogInterfaceMock({commonOperation})));
 
-    _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    _createCollection(_opCtx.get(), "test.t", options);
     ASSERT_OK(syncRollback(_opCtx.get(),
                            OplogInterfaceMock({applyOpsOperation, commonOperation}),
                            rollbackSource,
@@ -1167,7 +1186,9 @@ TEST_F(RSRollbackTest, RollbackApplyOpsCommand) {
 
 TEST_F(RSRollbackTest, RollbackCreateCollectionCommand) {
     createOplog(_opCtx.get());
-    auto coll = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto coll = _createCollection(_opCtx.get(), "test.t", options);
 
     auto commonOperation =
         std::make_pair(BSON("ts" << Timestamp(Seconds(1), 0) << "h" << 1LL), RecordId(1));
@@ -1201,7 +1222,9 @@ TEST_F(RSRollbackTest, RollbackCreateCollectionCommand) {
 
 TEST_F(RSRollbackTest, RollbackCollectionModificationCommand) {
     createOplog(_opCtx.get());
-    auto coll = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto coll = _createCollection(_opCtx.get(), "test.t", options);
 
     auto commonOperation =
         std::make_pair(BSON("ts" << Timestamp(Seconds(1), 0) << "h" << 1LL), RecordId(1));
@@ -1248,7 +1271,9 @@ TEST_F(RSRollbackTest, RollbackCollectionModificationCommand) {
 
 TEST_F(RSRollbackTest, RollbackCollectionModificationCommandInvalidCollectionOptions) {
     createOplog(_opCtx.get());
-    auto coll = _createCollection(_opCtx.get(), "test.t", CollectionOptions());
+    CollectionOptions options;
+    options.uuid = UUID::gen();
+    auto coll = _createCollection(_opCtx.get(), "test.t", options);
 
     auto commonOperation =
         std::make_pair(BSON("ts" << Timestamp(Seconds(1), 0) << "h" << 1LL), RecordId(1));
diff --git a/src/mongo/db/server_options.h b/src/mongo/db/server_options.h
index c8775b6a56e..b24fb513c7e 100644
--- a/src/mongo/db/server_options.h
+++ b/src/mongo/db/server_options.h
@@ -165,6 +165,10 @@ struct ServerGlobalParams {
         // Read-only parameter featureCompatibilityVersion.
         AtomicWord<Version> version{Version::k34};
 
+        // Read-only global isSchemaVersion36. This determines whether to give Collections UUIDs
+        // upon creation.
+        AtomicWord<bool> isSchemaVersion36{false};
+
         // Feature validation differs depending on the role of a mongod in a replica set or
         // master/slave configuration. Masters/primaries can accept user-initiated writes and
         // validate based on the feature compatibility version. A secondary/slave (which is not also
diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index 88af38faf43..d6f2a541df9 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -26,6 +26,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/index_names',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/storage/bson_collection_catalog_entry',
+        '$BUILD_DIR/mongo/db/catalog/uuid_catalog',
         'kv_prefix',
         ],
     )
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
index c775ada1a23..d354af2bb05 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
@@ -32,6 +32,7 @@
 
 #include "mongo/db/storage/kv/kv_collection_catalog_entry.h"
 
+#include "mongo/db/catalog/uuid_catalog.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
@@ -230,6 +231,46 @@ void KVCollectionCatalogEntry::updateTTLSetting(OperationContext* opCtx,
     _catalog->putMetaData(opCtx, ns().toString(), md);
 }
 
+void KVCollectionCatalogEntry::addUUID(OperationContext* opCtx,
+                                       CollectionUUID uuid,
+                                       Collection* coll) {
+    // Add a UUID to CollectionOptions if a UUID does not yet exist.
+    MetaData md = _getMetaData(opCtx);
+    if (!md.options.uuid) {
+        md.options.uuid = uuid;
+        _catalog->putMetaData(opCtx, ns().toString(), md);
+        UUIDCatalog& catalog = UUIDCatalog::get(opCtx->getServiceContext());
+        catalog.onCreateCollection(opCtx, coll, uuid);
+    } else {
+        fassert(40564, md.options.uuid.get() == uuid);
+    }
+}
+
+void KVCollectionCatalogEntry::removeUUID(OperationContext* opCtx) {
+    // Remove the UUID from CollectionOptions if a UUID exists.
+    MetaData md = _getMetaData(opCtx);
+    if (md.options.uuid) {
+        CollectionUUID uuid = md.options.uuid.get();
+        md.options.uuid = boost::none;
+        _catalog->putMetaData(opCtx, ns().toString(), md);
+        UUIDCatalog& catalog = UUIDCatalog::get(opCtx->getServiceContext());
+        Collection* coll = catalog.lookupCollectionByUUID(uuid);
+        if (coll) {
+            catalog.onDropCollection(opCtx, uuid);
+        }
+    }
+}
+
+bool KVCollectionCatalogEntry::isEqualToMetadataUUID(OperationContext* opCtx,
+                                                     OptionalCollectionUUID uuid) {
+    MetaData md = _getMetaData(opCtx);
+    if (uuid) {
+        return md.options.uuid && md.options.uuid.get() == uuid.get();
+    } else {
+        return !md.options.uuid;
+    }
+}
+
 void KVCollectionCatalogEntry::updateFlags(OperationContext* opCtx, int newValue) {
     MetaData md = _getMetaData(opCtx);
     md.options.flags = newValue;
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
index 21f31691739..6ff063eb388 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
@@ -33,6 +33,7 @@
 #include <memory>
 
 #include "mongo/db/catalog/collection_catalog_entry.h"
+#include "mongo/db/server_options.h"
 #include "mongo/db/storage/bson_collection_catalog_entry.h"
 #include "mongo/db/storage/record_store.h"
 
@@ -80,6 +81,12 @@ public:
 
     void updateCappedSize(OperationContext*, long long int) final;
 
+    void addUUID(OperationContext* opCtx, CollectionUUID uuid, Collection* coll) final;
+
+    void removeUUID(OperationContext* opCtx) final;
+
+    bool isEqualToMetadataUUID(OperationContext* opCtx, OptionalCollectionUUID uuid) final;
+
     RecordStore* getRecordStore() {
         return _recordStore.get();
     }
diff --git a/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.cpp b/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.cpp
index b37ca245298..65916d02de2 100644
--- a/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.cpp
+++ b/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.cpp
@@ -34,6 +34,7 @@
 
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/database_holder.h"
+#include "mongo/db/catalog/uuid_catalog.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/ops/update.h"
 #include "mongo/db/record_id.h"
@@ -412,6 +413,73 @@ void NamespaceDetailsCollectionCatalogEntry::updateFlags(OperationContext* opCtx
     _updateSystemNamespaces(opCtx, BSON("$set" << BSON("options.flags" << newValue)));
 }
 
+void NamespaceDetailsCollectionCatalogEntry::addUUID(OperationContext* opCtx,
+                                                     CollectionUUID uuid,
+                                                     Collection* coll) {
+    // Add a UUID to CollectionOptions if a UUID does not yet exist.
+    if (ns().coll() == "system.namespaces") {
+        return;
+    }
+    RecordData namespaceData;
+    invariant(_namespacesRecordStore->findRecord(opCtx, _namespacesRecordId, &namespaceData));
+
+    auto namespacesBson = namespaceData.releaseToBson();
+
+    if (namespacesBson["options"].isABSONObj() && !namespacesBson["options"].Obj()["uuid"].eoo()) {
+        fassert(40565, UUID::parse(namespacesBson["options"].Obj()["uuid"]).getValue() == uuid);
+    } else {
+        _updateSystemNamespaces(opCtx, BSON("$set" << BSON("options.uuid" << uuid)));
+        UUIDCatalog& catalog = UUIDCatalog::get(opCtx->getServiceContext());
+        catalog.onCreateCollection(opCtx, coll, uuid);
+    }
+}
+
+void NamespaceDetailsCollectionCatalogEntry::removeUUID(OperationContext* opCtx) {
+    // Remove the UUID from CollectionOptions if a UUID exists.
+    if (ns().coll() == "system.namespaces") {
+        return;
+    }
+    RecordData namespaceData;
+    invariant(_namespacesRecordStore->findRecord(opCtx, _namespacesRecordId, &namespaceData));
+    auto namespacesBson = namespaceData.releaseToBson();
+    if (!namespacesBson["options"].isABSONObj()) {
+        return;
+    }
+    auto optionsObj = namespacesBson["options"].Obj();
+
+    if (!optionsObj["uuid"].eoo()) {
+        CollectionUUID uuid = UUID::parse(optionsObj["uuid"]).getValue();
+        _updateSystemNamespaces(opCtx,
+                                BSON("$unset" << BSON("options.uuid"
+                                                      << "")));
+        UUIDCatalog& catalog = UUIDCatalog::get(opCtx->getServiceContext());
+        Collection* coll = catalog.lookupCollectionByUUID(uuid);
+        if (coll) {
+            catalog.onDropCollection(opCtx, uuid);
+        }
+    }
+}
+
+bool NamespaceDetailsCollectionCatalogEntry::isEqualToMetadataUUID(OperationContext* opCtx,
+                                                                   OptionalCollectionUUID uuid) {
+    if (ns().coll() != "system.namespaces") {
+        RecordData namespaceData;
+        invariant(_namespacesRecordStore->findRecord(opCtx, _namespacesRecordId, &namespaceData));
+
+        auto namespacesBson = namespaceData.releaseToBson();
+        if (uuid && namespacesBson["options"].isABSONObj()) {
+            auto optionsObj = namespacesBson["options"].Obj();
+            return !optionsObj["uuid"].eoo() &&
+                UUID::parse(optionsObj["uuid"]).getValue() == uuid.get();
+        } else {
+            return !uuid && (!namespacesBson["options"].isABSONObj() ||
+                             namespacesBson["options"].Obj()["uuid"].eoo());
+        }
+    } else {
+        return true;
+    }
+}
+
 void NamespaceDetailsCollectionCatalogEntry::updateValidator(OperationContext* opCtx,
                                                              const BSONObj& validator,
                                                              StringData validationLevel,
diff --git a/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.h b/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.h
index aff16b22093..c5226f3934e 100644
--- a/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.h
+++ b/src/mongo/db/storage/mmap_v1/catalog/namespace_details_collection_entry.h
@@ -33,6 +33,7 @@
 #include "mongo/base/string_data.h"
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/catalog/collection_catalog_entry.h"
+#include "mongo/db/server_options.h"
 #include "mongo/db/storage/mmap_v1/diskloc.h"
 
 namespace mongo {
@@ -97,6 +98,12 @@ public:
 
     void updateFlags(OperationContext* opCtx, int newValue) final;
 
+    void addUUID(OperationContext* opCtx, CollectionUUID uuid, Collection* coll) final;
+
+    void removeUUID(OperationContext* opCtx) final;
+
+    bool isEqualToMetadataUUID(OperationContext* opCtx, OptionalCollectionUUID uuid) final;
+
     void updateValidator(OperationContext* opCtx,
                          const BSONObj& validator,
                          StringData validationLevel,