SERVER-75456 Fix establishConsistentCollection when used concurrently with rename with dropTarget=true

2 files changed, 303 insertions, 7 deletions

diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 8981dceb535..3af63c65d47 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -940,19 +940,47 @@ const Collection* CollectionCatalog::_openCollectionAtLatestByNamespaceOrUUID(
     // If the catalog entry is not found in our snapshot then the collection is being dropped and we
     // can observe the drop. Lookups by this namespace or uuid should not find a collection.
     if (catalogEntry.isEmpty()) {
+        // If we performed this lookup by UUID we could be in a case where we're looking up
+        // concurrently with a rename with dropTarget=true where the UUID that we use is the target
+        // that got dropped. If that rename has committed we need to put the correct collection
+        // under open collection for this namespace. We can detect this case by comparing the
+        // catalogId with what is pending for this namespace.
+        if (nssOrUUID.uuid()) {
+            const std::shared_ptr<Collection>& pending = *_pendingCommitNamespaces.find(nss);
+            if (pending && pending->getCatalogId() != catalogId) {
+                openedCollections.store(nullptr, boost::none, uuid);
+                openedCollections.store(pending, nss, pending->uuid());
+                return nullptr;
+            }
+        }
         openedCollections.store(nullptr, nss, uuid);
         return nullptr;
     }
 
     // When trying to open the latest collection by namespace and the catalog entry has a different
-    // namespace in our snapshot, then there is a rename operation concurrent with this call. We
-    // need to store entries under uncommitted catalog changes for two namespaces (rename 'from' and
-    // 'to') so we can make sure lookups by UUID is supported and will return a Collection with its
-    // namespace in sync with the storage snapshot.
+    // namespace in our snapshot, then there is a rename operation concurrent with this call.
     NamespaceString nsInDurableCatalog = DurableCatalog::getNamespaceFromCatalogEntry(catalogEntry);
     if (nssOrUUID.nss() && nss != nsInDurableCatalog) {
-        // Like above, the correct instance is either in the catalog or under pending. First
-        // lookup in pending by UUID to determine if it contains the right namespace.
+        // There are two types of rename depending on the dropTarget flag.
+        if (pendingCollection && latestCollection &&
+            pendingCollection->getCatalogId() != latestCollection->getCatalogId()) {
+            // When there is a rename with dropTarget=true the two possible choices for the
+            // collection we need to observe are different logical collections, they have different
+            // UUID and catalogId. In this case storing a single entry in open collections is
+            // sufficient. We know that the instance we are looking for must be under
+            // 'latestCollection' as we used the catalogId from 'pendingCollection' when fetching
+            // durable catalog entry and the namespace in it did not match the namespace for
+            // 'pendingCollection' (the rename has not been comitted yet)
+            openedCollections.store(latestCollection, nss, latestCollection->uuid());
+            return latestCollection.get();
+        }
+
+        // For a regular rename of the same logical collection with dropTarget=false have the same
+        // UUID and catalogId for the two choices. In this case we need to store entries under open
+        // collections for two namespaces (rename 'from' and 'to') so we can make sure lookups by
+        // UUID is supported and will return a Collection with its namespace in sync with the
+        // storage snapshot. Like above, the correct instance is either in the catalog or under
+        // pending. First lookup in pending by UUID to determine if it contains the right namespace.
         const std::shared_ptr<Collection>* pending = _pendingCommitUUIDs.find(uuid);
         invariant(pending);
         const auto& pendingCollectionByUUID = *pending;
@@ -979,7 +1007,18 @@ const Collection* CollectionCatalog::_openCollectionAtLatestByNamespaceOrUUID(
     if (nssOrUUID.uuid() && latestCollection && pendingCollection &&
         latestCollection->ns() != pendingCollection->ns()) {
         if (latestCollection->ns() == nsInDurableCatalog) {
-            openedCollections.store(nullptr, pendingCollection->ns(), boost::none);
+            // If this is a rename with dropTarget=true and we're looking up with the 'from' UUID
+            // before the rename committed, the namespace would correspond to a valid collection
+            // that we need to store under open collections.
+            auto latestCollectionByNamespace =
+                _getCollectionByNamespace(opCtx, pendingCollection->ns());
+            if (latestCollectionByNamespace) {
+                openedCollections.store(latestCollectionByNamespace,
+                                        latestCollectionByNamespace->ns(),
+                                        latestCollectionByNamespace->uuid());
+            } else {
+                openedCollections.store(nullptr, pendingCollection->ns(), boost::none);
+            }
             openedCollections.store(latestCollection, nsInDurableCatalog, uuid);
             return latestCollection.get();
         } else {
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 3b2f3dde976..73ed389ae4d 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -3358,6 +3358,263 @@ TEST_F(CollectionCatalogTimestampTest,
         });
 }
 
+TEST_F(CollectionCatalogTimestampTest,
+       ConcurrentRenameCollectionWithDropTargetAndOpenCollectionBeforeCommit) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString originalNss = NamespaceString::createNamespaceString_forTest("a.b");
+    const NamespaceString targetNss = NamespaceString::createNamespaceString_forTest("a.c");
+    const Timestamp createOriginalCollectionTs = Timestamp(10, 10);
+    const Timestamp createTargetCollectionTs = Timestamp(15, 15);
+    const Timestamp renameCollectionTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), originalNss, createOriginalCollectionTs);
+    createCollection(opCtx.get(), targetNss, createTargetCollectionTs);
+
+    // We expect to find the UUID for the original collection
+    UUID uuid = CollectionCatalog::get(opCtx.get())
+                    ->lookupCollectionByNamespace(opCtx.get(), originalNss)
+                    ->uuid();
+
+    // When the snapshot is opened right before the rename is committed to the durable catalog, and
+    // the openCollection looks for the targetNss, we find the target collection.
+    concurrentRenameCollectionAndEstablishConsistentCollection(
+        opCtx.get(), originalNss, targetNss, targetNss, renameCollectionTs, true, true, 0, [&]() {
+            // Verify that we can find the original Collection when we search by original UUID.
+            auto coll =
+                CollectionCatalog::get(opCtx.get())->lookupCollectionByUUID(opCtx.get(), uuid);
+            ASSERT(coll);
+            ASSERT_EQ(coll->ns(), originalNss);
+
+            ASSERT_EQ(CollectionCatalog::get(opCtx.get())->lookupNSSByUUID(opCtx.get(), uuid),
+                      originalNss);
+        });
+}
+
+TEST_F(CollectionCatalogTimestampTest,
+       ConcurrentRenameCollectionWithDropTargetAndOpenCollectionAfterCommit) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString originalNss = NamespaceString::createNamespaceString_forTest("a.b");
+    const NamespaceString targetNss = NamespaceString::createNamespaceString_forTest("a.c");
+    const Timestamp createOriginalCollectionTs = Timestamp(10, 10);
+    const Timestamp createTargetCollectionTs = Timestamp(15, 15);
+    const Timestamp renameCollectionTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), originalNss, createOriginalCollectionTs);
+    createCollection(opCtx.get(), targetNss, createTargetCollectionTs);
+
+    // We expect to find the UUID for the target collection
+    UUID uuid = CollectionCatalog::get(opCtx.get())
+                    ->lookupCollectionByNamespace(opCtx.get(), targetNss)
+                    ->uuid();
+
+    // When the snapshot is opened right after the rename is committed to the durable catalog, and
+    // the openCollection looks for the targetNss, we find the original collection.
+    concurrentRenameCollectionAndEstablishConsistentCollection(
+        opCtx.get(), originalNss, targetNss, targetNss, renameCollectionTs, false, true, 0, [&]() {
+            // Verify that search by UUID is as expected and returns the target collection
+            auto coll =
+                CollectionCatalog::get(opCtx.get())->lookupCollectionByUUID(opCtx.get(), uuid);
+            ASSERT(coll);
+            ASSERT_EQ(coll->ns(), targetNss);
+
+            ASSERT_EQ(CollectionCatalog::get(opCtx.get())->lookupNSSByUUID(opCtx.get(), uuid),
+                      targetNss);
+        });
+}
+
+TEST_F(CollectionCatalogTimestampTest,
+       ConcurrentRenameCollectionWithDropTargetAndOpenCollectionWithOriginalUUIDBeforeCommit) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString originalNss = NamespaceString::createNamespaceString_forTest("a.b");
+    const NamespaceString targetNss = NamespaceString::createNamespaceString_forTest("a.c");
+    const Timestamp createOriginalCollectionTs = Timestamp(10, 10);
+    const Timestamp createTargetCollectionTs = Timestamp(15, 15);
+    const Timestamp renameCollectionTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), originalNss, createOriginalCollectionTs);
+    createCollection(opCtx.get(), targetNss, createTargetCollectionTs);
+
+    // We expect to find the UUID for the original collection
+    UUID originalUUID = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), originalNss)
+                            ->uuid();
+    UUID targetUUID = CollectionCatalog::get(opCtx.get())
+                          ->lookupCollectionByNamespace(opCtx.get(), targetNss)
+                          ->uuid();
+    NamespaceStringOrUUID uuidWithDbName(originalNss.dbName(), originalUUID);
+
+    // When the snapshot is opened right before the rename is committed to the durable catalog, and
+    // the openCollection looks for the original UUID, we should find the original collection
+    concurrentRenameCollectionAndEstablishConsistentCollection(
+        opCtx.get(),
+        originalNss,
+        targetNss,
+        uuidWithDbName,
+        renameCollectionTs,
+        true,
+        true,
+        0,
+        [&]() {
+            // Verify that we can find the original Collection when we search by namespace as rename
+            // has not committed yet.
+            auto coll = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), originalNss);
+            ASSERT(coll);
+            ASSERT_EQ(coll->uuid(), originalUUID);
+
+            // Verify that we can find the target Collection when we search by namespace as rename
+            // has not committed yet.
+            coll = CollectionCatalog::get(opCtx.get())
+                       ->lookupCollectionByNamespace(opCtx.get(), targetNss);
+            ASSERT(coll);
+            ASSERT_EQ(coll->uuid(), targetUUID);
+        });
+}
+
+TEST_F(CollectionCatalogTimestampTest,
+       ConcurrentRenameCollectionWithDropTargetAndOpenCollectionWithOriginalUUIDAfterCommit) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString originalNss = NamespaceString::createNamespaceString_forTest("a.b");
+    const NamespaceString targetNss = NamespaceString::createNamespaceString_forTest("a.c");
+    const Timestamp createOriginalCollectionTs = Timestamp(10, 10);
+    const Timestamp createTargetCollectionTs = Timestamp(15, 15);
+    const Timestamp renameCollectionTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), originalNss, createOriginalCollectionTs);
+    createCollection(opCtx.get(), targetNss, createTargetCollectionTs);
+
+    // We expect to find the UUID for the original collection
+    UUID uuid = CollectionCatalog::get(opCtx.get())
+                    ->lookupCollectionByNamespace(opCtx.get(), originalNss)
+                    ->uuid();
+    NamespaceStringOrUUID uuidWithDbName(originalNss.dbName(), uuid);
+
+    // When the snapshot is opened right after the rename is committed to the durable catalog, and
+    // the openCollection looks for the newNss, no collection instance should be returned.
+    concurrentRenameCollectionAndEstablishConsistentCollection(
+        opCtx.get(),
+        originalNss,
+        targetNss,
+        uuidWithDbName,
+        renameCollectionTs,
+        false,
+        true,
+        0,
+        [&]() {
+            // Verify that we cannot find the Collection when we search by the original namespace.
+            auto coll = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), originalNss);
+            ASSERT(!coll);
+
+            // Verify that we can find the original Collection UUID when we search by namespace.
+            coll = CollectionCatalog::get(opCtx.get())
+                       ->lookupCollectionByNamespace(opCtx.get(), targetNss);
+            ASSERT(coll);
+            ASSERT_EQ(coll->uuid(), uuid);
+        });
+}
+
+TEST_F(CollectionCatalogTimestampTest,
+       ConcurrentRenameCollectionWithDropTargetAndOpenCollectionWithTargetUUIDBeforeCommit) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString originalNss = NamespaceString::createNamespaceString_forTest("a.b");
+    const NamespaceString targetNss = NamespaceString::createNamespaceString_forTest("a.c");
+    const Timestamp createOriginalCollectionTs = Timestamp(10, 10);
+    const Timestamp createTargetCollectionTs = Timestamp(15, 15);
+    const Timestamp renameCollectionTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), originalNss, createOriginalCollectionTs);
+    createCollection(opCtx.get(), targetNss, createTargetCollectionTs);
+
+    UUID originalUUID = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), originalNss)
+                            ->uuid();
+    UUID targetUUID = CollectionCatalog::get(opCtx.get())
+                          ->lookupCollectionByNamespace(opCtx.get(), targetNss)
+                          ->uuid();
+    NamespaceStringOrUUID uuidWithDbName(originalNss.dbName(), targetUUID);
+
+    // When the snapshot is opened right before the rename is committed to the durable catalog, and
+    // the openCollection looks for the original UUID, we should find the original collection
+    concurrentRenameCollectionAndEstablishConsistentCollection(
+        opCtx.get(),
+        originalNss,
+        targetNss,
+        uuidWithDbName,
+        renameCollectionTs,
+        true,
+        true,
+        0,
+        [&]() {
+            // Verify that we can find the original Collection when we search by namespace as rename
+            // has not committed yet.
+            auto coll = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), originalNss);
+            ASSERT(coll);
+            ASSERT_EQ(coll->uuid(), originalUUID);
+
+            // Verify that we can find the target Collection when we search by namespace as rename
+            // has not committed yet.
+            coll = CollectionCatalog::get(opCtx.get())
+                       ->lookupCollectionByNamespace(opCtx.get(), targetNss);
+            ASSERT(coll);
+            ASSERT_EQ(coll->uuid(), targetUUID);
+        });
+}
+
+TEST_F(CollectionCatalogTimestampTest,
+       ConcurrentRenameCollectionWithDropTargetAndOpenCollectionWithTargetUUIDAfterCommit) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    const NamespaceString originalNss = NamespaceString::createNamespaceString_forTest("a.b");
+    const NamespaceString targetNss = NamespaceString::createNamespaceString_forTest("a.c");
+    const Timestamp createOriginalCollectionTs = Timestamp(10, 10);
+    const Timestamp createTargetCollectionTs = Timestamp(15, 15);
+    const Timestamp renameCollectionTs = Timestamp(20, 20);
+
+    createCollection(opCtx.get(), originalNss, createOriginalCollectionTs);
+    createCollection(opCtx.get(), targetNss, createTargetCollectionTs);
+
+    // We expect to find the UUID for the original collection
+    UUID originalUUID = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), originalNss)
+                            ->uuid();
+    UUID targetUUID = CollectionCatalog::get(opCtx.get())
+                          ->lookupCollectionByNamespace(opCtx.get(), targetNss)
+                          ->uuid();
+    NamespaceStringOrUUID uuidWithDbName(originalNss.dbName(), targetUUID);
+
+    // When the snapshot is opened right after the rename is committed to the durable catalog, and
+    // the openCollection looks for the newNss, no collection instance should be returned.
+    concurrentRenameCollectionAndEstablishConsistentCollection(
+        opCtx.get(),
+        originalNss,
+        targetNss,
+        uuidWithDbName,
+        renameCollectionTs,
+        false,
+        false,
+        0,
+        [&]() {
+            // Verify that we can find the original Collection UUID when we search by namespace.
+            auto coll = CollectionCatalog::get(opCtx.get())
+                            ->lookupCollectionByNamespace(opCtx.get(), targetNss);
+            ASSERT(coll);
+            ASSERT_EQ(coll->uuid(), originalUUID);
+        });
+}
+
 TEST_F(CollectionCatalogTimestampTest, ConcurrentCreateIndexAndOpenCollectionBeforeCommit) {
     RAIIServerParameterControllerForTest featureFlagController(
         "featureFlagPointInTimeCatalogLookups", true);