SERVER-64404 Add collection-level `$listCatalog`

8 files changed, 100 insertions, 23 deletions

diff --git a/src/mongo/db/pipeline/aggregate_command.idl b/src/mongo/db/pipeline/aggregate_command.idl
index 98ab3572924..0e69d47c6ba 100644
--- a/src/mongo/db/pipeline/aggregate_command.idl
+++ b/src/mongo/db/pipeline/aggregate_command.idl
@@ -126,6 +126,10 @@ commands:
                     action_type: listCachedAndActiveUsers
                 - privilege: # $listCatalog
                     agg_stage: listCatalog
+                    resource_pattern: exact_namespace
+                    action_type: [listCollections, listIndexes]
+                - privilege: # $listCatalog
+                    agg_stage: listCatalog
                     resource_pattern: cluster
                     action_type: listDatabases
                 - privilege: # $listCatalog
diff --git a/src/mongo/db/pipeline/document_source_list_catalog.cpp b/src/mongo/db/pipeline/document_source_list_catalog.cpp
index a1163c8b2b3..1e4e9fde9f1 100644
--- a/src/mongo/db/pipeline/document_source_list_catalog.cpp
+++ b/src/mongo/db/pipeline/document_source_list_catalog.cpp
@@ -47,7 +47,7 @@ REGISTER_DOCUMENT_SOURCE_WITH_MIN_VERSION(listCatalog,
                                           DocumentSourceListCatalog::LiteParsed::parse,
                                           DocumentSourceListCatalog::createFromBson,
                                           AllowedWithApiStrict::kNeverInVersion1,
-                                          multiversion::FeatureCompatibilityVersion::kVersion_5_3);
+                                          multiversion::FeatureCompatibilityVersion::kVersion_6_0);
 
 const char* DocumentSourceListCatalog::getSourceName() const {
     return kStageName.rawData();
@@ -60,22 +60,34 @@ PrivilegeVector DocumentSourceListCatalog::LiteParsed::requiredPrivileges(
     // See builtin_roles.cpp.
     ActionSet listCollectionsAndIndexesActions{ActionType::listCollections,
                                                ActionType::listIndexes};
-    return {Privilege(ResourcePattern::forClusterResource(), ActionType::listDatabases),
-            Privilege(ResourcePattern::forAnyNormalResource(), listCollectionsAndIndexesActions),
-            Privilege(ResourcePattern::forCollectionName("system.js"),
-                      listCollectionsAndIndexesActions),
-            Privilege(ResourcePattern::forAnySystemBuckets(), listCollectionsAndIndexesActions)};
+    return _ns.isCollectionlessAggregateNS()
+        ? PrivilegeVector{Privilege(ResourcePattern::forClusterResource(),
+                                    ActionType::listDatabases),
+                          Privilege(ResourcePattern::forAnyNormalResource(),
+                                    listCollectionsAndIndexesActions),
+                          Privilege(ResourcePattern::forCollectionName("system.js"),
+                                    listCollectionsAndIndexesActions),
+                          Privilege(ResourcePattern::forAnySystemBuckets(),
+                                    listCollectionsAndIndexesActions)}
+        : PrivilegeVector{
+              Privilege(ResourcePattern::forExactNamespace(_ns), listCollectionsAndIndexesActions)};
 }
 
 DocumentSource::GetNextResult DocumentSourceListCatalog::doGetNext() {
-    if (!_catalogDocsInitialized) {
-        _catalogDocs = pExpCtx->mongoProcessInterface->listCatalog(pExpCtx->opCtx);
-        _catalogDocsInitialized = true;
+    if (!_catalogDocs) {
+        if (pExpCtx->ns.isCollectionlessAggregateNS()) {
+            _catalogDocs = pExpCtx->mongoProcessInterface->listCatalog(pExpCtx->opCtx);
+        } else if (auto catalogDoc = pExpCtx->mongoProcessInterface->getCatalogEntry(pExpCtx->opCtx,
+                                                                                     pExpCtx->ns)) {
+            _catalogDocs = {{std::move(*catalogDoc)}};
+        } else {
+            _catalogDocs.emplace();
+        }
     }
 
-    if (!_catalogDocs.empty()) {
-        Document doc{_catalogDocs.front()};
-        _catalogDocs.pop_front();
+    if (!_catalogDocs->empty()) {
+        Document doc{std::move(_catalogDocs->front())};
+        _catalogDocs->pop_front();
         return doc;
     }
 
@@ -94,9 +106,10 @@ intrusive_ptr<DocumentSource> DocumentSourceListCatalog::createFromBson(
 
     const NamespaceString& nss = pExpCtx->ns;
 
-    uassert(ErrorCodes::InvalidNamespace,
-            "$listCatalog must be run against the 'admin' database with {aggregate: 1}",
-            nss.db() == NamespaceString::kAdminDb && nss.isCollectionlessAggregateNS());
+    uassert(
+        ErrorCodes::InvalidNamespace,
+        "Collectionless $listCatalog must be run against the 'admin' database with {aggregate: 1}",
+        nss.db() == NamespaceString::kAdminDb || !nss.isCollectionlessAggregateNS());
 
     uassert(ErrorCodes::QueryFeatureNotAllowed,
             fmt::format("The {} aggregation stage is not enabled", kStageName),
diff --git a/src/mongo/db/pipeline/document_source_list_catalog.h b/src/mongo/db/pipeline/document_source_list_catalog.h
index c0cb3829a00..c3bcb52fe77 100644
--- a/src/mongo/db/pipeline/document_source_list_catalog.h
+++ b/src/mongo/db/pipeline/document_source_list_catalog.h
@@ -51,11 +51,11 @@ public:
     public:
         static std::unique_ptr<LiteParsed> parse(const NamespaceString& nss,
                                                  const BSONElement& spec) {
-            return std::make_unique<LiteParsed>(spec.fieldName());
+            return std::make_unique<LiteParsed>(spec.fieldName(), nss);
         }
 
-        explicit LiteParsed(std::string parseTimeName)
-            : LiteParsedDocumentSource(std::move(parseTimeName)) {}
+        explicit LiteParsed(std::string parseTimeName, NamespaceString ns)
+            : LiteParsedDocumentSource(std::move(parseTimeName)), _ns(std::move(ns)) {}
 
         stdx::unordered_set<NamespaceString> getInvolvedNamespaces() const final {
             return stdx::unordered_set<NamespaceString>();
@@ -68,6 +68,9 @@ public:
         bool isInitialSource() const final {
             return true;
         }
+
+    private:
+        NamespaceString _ns;
     };
 
     // virtuals from DocumentSource
@@ -84,7 +87,7 @@ public:
                                      LookupRequirement::kAllowed,
                                      UnionRequirement::kAllowed);
 
-        constraints.isIndependentOfAnyCollection = true;
+        constraints.isIndependentOfAnyCollection = pExpCtx->ns.isCollectionlessAggregateNS();
         constraints.requiresInputDocSource = false;
         return constraints;
     }
@@ -100,8 +103,7 @@ private:
     DocumentSourceListCatalog(const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
     GetNextResult doGetNext() final;
 
-    bool _catalogDocsInitialized = false;
-    std::deque<BSONObj> _catalogDocs;
+    boost::optional<std::deque<BSONObj>> _catalogDocs;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
index cf6225dddac..1532cfbf4ed 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.cpp
@@ -119,6 +119,7 @@ void assertIgnorePrepareConflictsBehavior(const boost::intrusive_ptr<ExpressionC
  * <db>.system.views namespaces found.
  */
 void listDurableCatalog(OperationContext* opCtx,
+                        StringData shardName,
                         std::deque<BSONObj>* docs,
                         std::vector<NamespaceStringOrUUID>* systemViewsNamespaces) {
     auto durableCatalog = DurableCatalog::get(opCtx);
@@ -146,6 +147,9 @@ void listDurableCatalog(OperationContext* opCtx,
         builder.append("db", ns.db());
         builder.append("name", ns.coll());
         builder.append("type", "collection");
+        if (!shardName.empty()) {
+            builder.append("shard", shardName);
+        }
         builder.appendElements(obj);
         docs->push_back(builder.obj());
     }
@@ -224,7 +228,7 @@ std::deque<BSONObj> CommonMongodProcessInterface::listCatalog(OperationContext*
         std::vector<NamespaceStringOrUUID> systemViewsNamespaces;
         {
             Lock::GlobalLock globalLock(opCtx, MODE_IS);
-            listDurableCatalog(opCtx, &docs, &systemViewsNamespaces);
+            listDurableCatalog(opCtx, getShardName(opCtx), &docs, &systemViewsNamespaces);
         }
 
         if (systemViewsNamespaces.empty()) {
@@ -257,7 +261,8 @@ std::deque<BSONObj> CommonMongodProcessInterface::listCatalog(OperationContext*
         // we read it, we should discard this set of results and retry from the top (with the
         // global read lock) of this loop.
         std::vector<NamespaceStringOrUUID> systemViewsNamespacesFromSecondCatalogRead;
-        listDurableCatalog(opCtx, &docs, &systemViewsNamespacesFromSecondCatalogRead);
+        listDurableCatalog(
+            opCtx, getShardName(opCtx), &docs, &systemViewsNamespacesFromSecondCatalogRead);
         if (!std::equal(
                 systemViewsNamespaces.cbegin(),
                 systemViewsNamespaces.cend(),
@@ -288,6 +293,9 @@ std::deque<BSONObj> CommonMongodProcessInterface::listCatalog(OperationContext*
                 } else {
                     builder.append("type", "view");
                 }
+                if (auto shardName = getShardName(opCtx); !shardName.empty()) {
+                    builder.append("shard", shardName);
+                }
                 builder.appendAs(obj["_id"], "ns");
                 builder.appendElements(obj);
                 docs.push_back(builder.obj());
@@ -298,6 +306,37 @@ std::deque<BSONObj> CommonMongodProcessInterface::listCatalog(OperationContext*
     }
 }
 
+boost::optional<BSONObj> CommonMongodProcessInterface::getCatalogEntry(
+    OperationContext* opCtx, const NamespaceString& ns) const {
+    Lock::GlobalLock globalLock{opCtx, MODE_IS};
+
+    auto rs = DurableCatalog::get(opCtx)->getRecordStore();
+    if (!rs) {
+        return boost::none;
+    }
+
+    auto cursor = rs->getCursor(opCtx);
+    while (auto record = cursor->next()) {
+        auto obj = record->data.toBson();
+        if (NamespaceString{obj.getStringField("ns")} != ns) {
+            continue;
+        }
+
+        BSONObjBuilder builder;
+        builder.append("db", ns.db());
+        builder.append("name", ns.coll());
+        builder.append("type", "collection");
+        if (auto shardName = getShardName(opCtx); !shardName.empty()) {
+            builder.append("shard", shardName);
+        }
+        builder.appendElements(obj);
+
+        return builder.obj();
+    }
+
+    return boost::none;
+}
+
 void CommonMongodProcessInterface::appendLatencyStats(OperationContext* opCtx,
                                                       const NamespaceString& nss,
                                                       bool includeHistograms,
diff --git a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h
index 108c3cbdf39..79b455e8e6d 100644
--- a/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/common_mongod_process_interface.h
@@ -59,6 +59,9 @@ public:
 
     std::deque<BSONObj> listCatalog(OperationContext* opCtx) const final;
 
+    boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
+                                             const NamespaceString& ns) const final;
+
     void appendLatencyStats(OperationContext* opCtx,
                             const NamespaceString& nss,
                             bool includeHistograms,
diff --git a/src/mongo/db/pipeline/process_interface/mongo_process_interface.h b/src/mongo/db/pipeline/process_interface/mongo_process_interface.h
index 371021f492f..89e47af3d74 100644
--- a/src/mongo/db/pipeline/process_interface/mongo_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/mongo_process_interface.h
@@ -196,6 +196,12 @@ public:
     virtual std::deque<BSONObj> listCatalog(OperationContext* opCtx) const = 0;
 
     /**
+     * Returns the catalog entry for the given namespace, if it exists.
+     */
+    virtual boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
+                                                     const NamespaceString& ns) const = 0;
+
+    /**
      * Appends operation latency statistics for collection "nss" to "builder"
      */
     virtual void appendLatencyStats(OperationContext* opCtx,
diff --git a/src/mongo/db/pipeline/process_interface/mongos_process_interface.h b/src/mongo/db/pipeline/process_interface/mongos_process_interface.h
index f3ae8cae7b3..dba29a99e3d 100644
--- a/src/mongo/db/pipeline/process_interface/mongos_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/mongos_process_interface.h
@@ -102,6 +102,11 @@ public:
         MONGO_UNREACHABLE;
     }
 
+    boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
+                                             const NamespaceString& ns) const final {
+        MONGO_UNREACHABLE;
+    }
+
     void appendLatencyStats(OperationContext* opCtx,
                             const NamespaceString& nss,
                             bool includeHistograms,
diff --git a/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h b/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h
index 1e4f0dccb85..eaabb1f64a9 100644
--- a/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h
+++ b/src/mongo/db/pipeline/process_interface/stub_mongo_process_interface.h
@@ -93,6 +93,11 @@ public:
         MONGO_UNREACHABLE;
     }
 
+    boost::optional<BSONObj> getCatalogEntry(OperationContext* opCtx,
+                                             const NamespaceString& ns) const override {
+        MONGO_UNREACHABLE;
+    }
+
     void appendLatencyStats(OperationContext* opCtx,
                             const NamespaceString& nss,
                             bool includeHistograms,