SERVER-60567 Acquire AutoGetDB before parsing and verify collection existence at parse time in SBE command

4 files changed, 41 insertions, 14 deletions

diff --git a/etc/backports_required_for_multiversion_tests.yml b/etc/backports_required_for_multiversion_tests.yml
index cc5e6e1206a..04558db2995 100644
--- a/etc/backports_required_for_multiversion_tests.yml
+++ b/etc/backports_required_for_multiversion_tests.yml
@@ -102,6 +102,8 @@ last-continuous:
       test_file: jstests/aggregation/range.js
     - ticket: SERVER-59923
       test_file: jstests/sharding/test_resharding_test_fixture_shutdown_retry_needed.js
+    - ticket: SERVER-60567
+      test_file: jstests/core/sbe/sbe_cmd.js
 
   # Tests that should only be excluded from particular suites should be listed under that suite.
   suites:
@@ -364,6 +366,8 @@ last-lts:
       test_file: jstests/aggregation/range.js
     - ticket: SERVER-59923
       test_file: jstests/sharding/test_resharding_test_fixture_shutdown_retry_needed.js
+    - ticket: SERVER-60567
+      test_file: jstests/core/sbe/sbe_cmd.js
 
   # Tests that should only be excluded from particular suites should be listed under that suite.
   suites:
diff --git a/jstests/core/sbe/sbe_cmd.js b/jstests/core/sbe/sbe_cmd.js
index 289e48820ac..21e68f894f8 100644
--- a/jstests/core/sbe/sbe_cmd.js
+++ b/jstests/core/sbe/sbe_cmd.js
@@ -66,4 +66,16 @@ assertSbeCommandWorked({query: {b: 1}});
 assertSbeCommandWorked({query: {a: 1, c: 3}});
 // Test query: {a: 1, c: 3} with projection {_id: 0}.
 assertSbeCommandWorked({query: {a: 1, c: 3}, projection: {_id: 0}});
+
+// Verify that the sbe command can detect if a collection has been dropped.
+const explain = coll.find().explain();
+assert(explain.queryPlanner.winningPlan.hasOwnProperty("slotBasedPlan"), explain);
+const slotBasedPlan = explain.queryPlanner.winningPlan.slotBasedPlan.stages;
+
+// The command response should be OK as long as the collection exists.
+assert(db._sbe(slotBasedPlan));
+
+// After the collection is dropped, the parser should detect that the collection doesn't exist.
+assert(coll.drop());
+assert.throwsWithCode(() => db._sbe(slotBasedPlan), 6056700);
 }());
diff --git a/src/mongo/db/exec/sbe/parser/parser.cpp b/src/mongo/db/exec/sbe/parser/parser.cpp
index 417a751541e..1f0bb90d357 100644
--- a/src/mongo/db/exec/sbe/parser/parser.cpp
+++ b/src/mongo/db/exec/sbe/parser/parser.cpp
@@ -1894,13 +1894,23 @@ CollectionUUID Parser::getCollectionUuid(const std::string& collName) {
         return uuid.getValue();
     }
 
+    auto catalog = CollectionCatalog::get(_opCtx);
+
     // Try to parse the collection name as a UUID directly, otherwise fallback to lookup by
     // NamespaceString.
     auto parsedUuid = UUID::parse(collName);
     if (parsedUuid.isOK()) {
-        return parsedUuid.getValue();
+        auto uuid = parsedUuid.getValue();
+
+        // Verify that the UUID corresponds to an existing collection.
+        auto collPtr = catalog->lookupCollectionByUUID(_opCtx, uuid);
+        uassert(6056700,
+                str::stream() << "SBE command parser could not find collection: " << collName,
+                collPtr);
+        return uuid;
     }
-    auto uuid = CollectionCatalog::get(_opCtx)->lookupUUIDByNSS(_opCtx, NamespaceString{collName});
+
+    auto uuid = catalog->lookupUUIDByNSS(_opCtx, NamespaceString{collName});
     uassert(5162900,
             str::stream() << "SBE command parser could not find collection: " << collName,
             uuid);
diff --git a/src/mongo/db/exec/sbe_cmd.cpp b/src/mongo/db/exec/sbe_cmd.cpp
index a2c929354ba..5468c9a54e7 100644
--- a/src/mongo/db/exec/sbe_cmd.cpp
+++ b/src/mongo/db/exec/sbe_cmd.cpp
@@ -83,6 +83,19 @@ public:
         uassertStatusOK(CursorRequest::parseCommandCursorOptions(
             cmdObj, query_request_helper::kDefaultBatchSize, &batchSize));
 
+        // Acquire the global lock, acquire a snapshot, and stash our version of the collection
+        // catalog. The is necessary because SBE plan executors currently use the external lock
+        // policy. Note that this must be done before parsing because the parser may look up
+        // collections in the stashed catalog.
+        //
+        // Unlike the similar 'AutoGetCollection*' variants of this db_raii object, this will not
+        // handle re-establishing a view of the catalog which is consistent with the new storage
+        // snapshot after a yield. For this reason, the SBE command cannot yield.
+        //
+        // Also, this will not handle all read concerns. This is ok because the SBE command is
+        // experimental and need not support the various read concerns.
+        AutoGetDbForReadLockFree autoGet{opCtx, dbname};
+
         auto env = std::make_unique<sbe::RuntimeEnvironment>();
         sbe::Parser parser(env.get());
         auto root = parser.parse(opCtx, dbname, cmdObj["sbe"].String());
@@ -106,18 +119,6 @@ public:
             data.outputs.set(stage_builder::PlanStageSlots::kRecordId, *recordIdSlot);
         }
 
-        // Acquire the global lock, acquire a snapshot, and stash our version of the collection
-        // catalog. The is necessary because SBE plan executors currently use the external lock
-        // policy.
-        //
-        // Unlike the similar 'AutoGetCollection*' variants of this db_raii object, this will not
-        // handle re-establishing a view of the catalog which is consistent with the new storage
-        // snapshot after a yield. For this reason, the SBE command cannot yield.
-        //
-        // Also, this will not handle all read concerns. This is ok because the SBE command is
-        // experimental and need not support the various read concerns.
-        AutoGetDbForReadLockFree autoGet{opCtx, dbname};
-
         root->attachToOperationContext(opCtx);
         exec = uassertStatusOK(plan_executor_factory::make(opCtx,
                                                            std::move(cq),