SERVER-42560 Simplify passing of metadata between DocumentSource and PlanStage

63 files changed, 711 insertions, 765 deletions

diff --git a/src/mongo/db/clientcursor.cpp b/src/mongo/db/clientcursor.cpp
index 3e93171254b..2e87deb2fd0 100644
--- a/src/mongo/db/clientcursor.cpp
+++ b/src/mongo/db/clientcursor.cpp
@@ -93,6 +93,7 @@ ClientCursor::ClientCursor(ClientCursorParams params,
       _originatingPrivileges(std::move(params.originatingPrivileges)),
       _queryOptions(params.queryOptions),
       _lockPolicy(params.lockPolicy),
+      _needsMerge(params.needsMerge),
       _exec(std::move(params.exec)),
       _operationUsingCursor(operationUsingCursor),
       _lastUseDate(now),
diff --git a/src/mongo/db/clientcursor.h b/src/mongo/db/clientcursor.h
index 4bc29f4d3a7..2f5663012d8 100644
--- a/src/mongo/db/clientcursor.h
+++ b/src/mongo/db/clientcursor.h
@@ -82,7 +82,8 @@ struct ClientCursorParams {
                        repl::ReadConcernArgs readConcernArgs,
                        BSONObj originatingCommandObj,
                        LockPolicy lockPolicy,
-                       PrivilegeVector originatingPrivileges)
+                       PrivilegeVector originatingPrivileges,
+                       bool needsMerge)
         : exec(std::move(planExecutor)),
           nss(std::move(nss)),
           writeConcernOptions(std::move(writeConcernOptions)),
@@ -92,7 +93,8 @@ struct ClientCursorParams {
                            : 0),
           originatingCommandObj(originatingCommandObj.getOwned()),
           lockPolicy(lockPolicy),
-          originatingPrivileges(std::move(originatingPrivileges)) {
+          originatingPrivileges(std::move(originatingPrivileges)),
+          needsMerge(needsMerge) {
         while (authenticatedUsersIter.more()) {
             authenticatedUsers.emplace_back(authenticatedUsersIter.next());
         }
@@ -121,6 +123,7 @@ struct ClientCursorParams {
     BSONObj originatingCommandObj;
     const LockPolicy lockPolicy;
     PrivilegeVector originatingPrivileges;
+    const bool needsMerge;
 };
 
 /**
@@ -172,6 +175,10 @@ public:
         return _writeConcernOptions;
     }
 
+    bool needsMerge() const {
+        return _needsMerge;
+    }
+
     /**
      * Returns a pointer to the underlying query plan executor. All cursors manage a PlanExecutor,
      * so this method never returns a null pointer.
@@ -401,6 +408,13 @@ private:
 
     const ClientCursorParams::LockPolicy _lockPolicy;
 
+    // The value of a flag specified on the originating command which indicates whether the result
+    // of this cursor will be consumed by a merging node (mongos or a mongod selected to perform a
+    // merge). Note that this flag is only set for aggregate() commands, and not for find()
+    // commands. It is therefore possible that 'needsMerge' is false when in fact there will be a
+    // merge performed.
+    const bool _needsMerge;
+
     // Unused maxTime budget for this cursor.
     Microseconds _leftoverMaxTimeMicros = Microseconds::max();
 
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index 4800ed178d9..9da9269284f 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -491,12 +491,13 @@ public:
             CursorResponseBuilder::Options options;
             options.isInitialResponse = true;
             CursorResponseBuilder firstBatch(result, options);
-            BSONObj obj;
+            Document doc;
             PlanExecutor::ExecState state = PlanExecutor::ADVANCED;
             std::uint64_t numResults = 0;
             while (!FindCommon::enoughForFirstBatch(originalQR, numResults) &&
-                   PlanExecutor::ADVANCED == (state = exec->getNext(&obj, nullptr))) {
+                   PlanExecutor::ADVANCED == (state = exec->getNext(&doc, nullptr))) {
                 // If we can't fit this result inside the current batch, then we stash it for later.
+                BSONObj obj = doc.toBson();
                 if (!FindCommon::haveSpaceForNext(obj, numResults, firstBatch.bytesUsed())) {
                     exec->enqueue(obj);
                     break;
@@ -512,7 +513,7 @@ public:
                 firstBatch.abandon();
 
                 // We should always have a valid status member object at this point.
-                auto status = WorkingSetCommon::getMemberObjectStatus(obj);
+                auto status = WorkingSetCommon::getMemberObjectStatus(doc);
                 invariant(!status.isOK());
                 warning() << "Plan executor error during find command: "
                           << PlanExecutor::statestr(state) << ", status: " << status
@@ -535,7 +536,8 @@ public:
                      repl::ReadConcernArgs::get(opCtx),
                      _request.body,
                      ClientCursorParams::LockPolicy::kLockExternally,
-                     {Privilege(ResourcePattern::forExactNamespace(nss), ActionType::find)}});
+                     {Privilege(ResourcePattern::forExactNamespace(nss), ActionType::find)},
+                     expCtx->needsMerge});
                 cursorId = pinnedCursor.getCursor()->cursorid();
 
                 invariant(!exec);
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index 295c8e0553d..62a5d38cd0a 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -277,10 +277,15 @@ public:
             // If an awaitData getMore is killed during this process due to our max time expiring at
             // an interrupt point, we just continue as normal and return rather than reporting a
             // timeout to the user.
-            BSONObj obj;
+            Document doc;
             try {
                 while (!FindCommon::enoughForGetMore(request.batchSize.value_or(0), *numResults) &&
-                       PlanExecutor::ADVANCED == (*state = exec->getNext(&obj, nullptr))) {
+                       PlanExecutor::ADVANCED == (*state = exec->getNext(&doc, nullptr))) {
+                    auto* expCtx = exec->getExpCtx().get();
+                    BSONObj obj = cursor->needsMerge()
+                        ? doc.toBsonWithMetaData(expCtx ? expCtx->use42ChangeStreamSortKeys : false)
+                        : doc.toBson();
+
                     // If adding this object will cause us to exceed the message size limit, then we
                     // stash it for later.
                     if (!FindCommon::haveSpaceForNext(obj, *numResults, nextBatch->bytesUsed())) {
@@ -305,7 +310,7 @@ public:
             switch (*state) {
                 case PlanExecutor::FAILURE: {
                     // We should always have a valid status member object at this point.
-                    auto status = WorkingSetCommon::getMemberObjectStatus(obj);
+                    auto status = WorkingSetCommon::getMemberObjectStatus(doc);
                     invariant(!status.isOK());
                     // Log an error message and then perform the cleanup.
                     warning() << "GetMore command executor error: "
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index dd89bf02821..398c4f4fb36 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -362,14 +362,15 @@ public:
                 opCtx, std::move(ws), std::move(root), nullptr, PlanExecutor::NO_YIELD, cursorNss));
 
             for (long long objCount = 0; objCount < batchSize; objCount++) {
-                BSONObj next;
-                PlanExecutor::ExecState state = exec->getNext(&next, nullptr);
+                Document nextDoc;
+                PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr);
                 if (state == PlanExecutor::IS_EOF) {
                     break;
                 }
                 invariant(state == PlanExecutor::ADVANCED);
 
                 // If we can't fit this result inside the current batch, then we stash it for later.
+                BSONObj next = nextDoc.toBson();
                 if (!FindCommon::haveSpaceForNext(next, objCount, firstBatch.len())) {
                     exec->enqueue(next);
                     break;
@@ -387,15 +388,18 @@ public:
 
         auto pinnedCursor = CursorManager::get(opCtx)->registerCursor(
             opCtx,
-            {std::move(exec),
-             cursorNss,
-             AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
-             opCtx->getWriteConcern(),
-             repl::ReadConcernArgs::get(opCtx),
-             jsobj,
-             ClientCursorParams::LockPolicy::kLocksInternally,
-             uassertStatusOK(AuthorizationSession::get(opCtx->getClient())
-                                 ->checkAuthorizedToListCollections(dbname, jsobj))});
+            {
+                std::move(exec),
+                cursorNss,
+                AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
+                opCtx->getWriteConcern(),
+                repl::ReadConcernArgs::get(opCtx),
+                jsobj,
+                ClientCursorParams::LockPolicy::kLocksInternally,
+                uassertStatusOK(AuthorizationSession::get(opCtx->getClient())
+                                    ->checkAuthorizedToListCollections(dbname, jsobj)),
+                false  // needsMerge always 'false' for listCollections.
+            });
 
         appendCursorResponseObject(
             pinnedCursor.getCursor()->cursorid(), cursorNss.ns(), firstBatch.arr(), &result);
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index e830325e456..b7c26eb3309 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -199,13 +199,14 @@ public:
                 opCtx, std::move(ws), std::move(root), nullptr, PlanExecutor::NO_YIELD, nss));
 
             for (long long objCount = 0; objCount < batchSize; objCount++) {
-                BSONObj next;
-                PlanExecutor::ExecState state = exec->getNext(&next, nullptr);
+                Document nextDoc;
+                PlanExecutor::ExecState state = exec->getNext(&nextDoc, nullptr);
                 if (state == PlanExecutor::IS_EOF) {
                     break;
                 }
                 invariant(state == PlanExecutor::ADVANCED);
 
+                BSONObj next = nextDoc.toBson();
                 // If we can't fit this result inside the current batch, then we stash it for later.
                 if (!FindCommon::haveSpaceForNext(next, objCount, firstBatch.len())) {
                     exec->enqueue(next);
@@ -227,14 +228,17 @@ public:
 
         const auto pinnedCursor = CursorManager::get(opCtx)->registerCursor(
             opCtx,
-            {std::move(exec),
-             nss,
-             AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
-             opCtx->getWriteConcern(),
-             repl::ReadConcernArgs::get(opCtx),
-             cmdObj,
-             ClientCursorParams::LockPolicy::kLocksInternally,
-             {Privilege(ResourcePattern::forExactNamespace(nss), ActionType::listIndexes)}});
+            {
+                std::move(exec),
+                nss,
+                AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
+                opCtx->getWriteConcern(),
+                repl::ReadConcernArgs::get(opCtx),
+                cmdObj,
+                ClientCursorParams::LockPolicy::kLocksInternally,
+                {Privilege(ResourcePattern::forExactNamespace(nss), ActionType::listIndexes)},
+                false  // needsMerge always 'false' for listIndexes.
+            });
 
         appendCursorResponseObject(
             pinnedCursor.getCursor()->cursorid(), nss.ns(), firstBatch.arr(), &result);
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 5847e6aabed..97a2b86f11d 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -172,15 +172,15 @@ bool handleCursorCommand(OperationContext* opCtx,
     auto exec = cursor->getExecutor();
     invariant(exec);
 
-    BSONObj next;
     bool stashedResult = false;
     for (int objCount = 0; objCount < batchSize; objCount++) {
         // The initial getNext() on a PipelineProxyStage may be very expensive so we don't
         // do it when batchSize is 0 since that indicates a desire for a fast return.
         PlanExecutor::ExecState state;
+        Document nextDoc;
 
         try {
-            state = exec->getNext(&next, nullptr);
+            state = exec->getNext(&nextDoc, nullptr);
         } catch (const ExceptionFor<ErrorCodes::CloseChangeStream>&) {
             // This exception is thrown when a $changeStream stage encounters an event
             // that invalidates the cursor. We should close the cursor and return without
@@ -201,7 +201,7 @@ bool handleCursorCommand(OperationContext* opCtx,
 
         if (PlanExecutor::ADVANCED != state) {
             // We should always have a valid status member object at this point.
-            auto status = WorkingSetCommon::getMemberObjectStatus(next);
+            auto status = WorkingSetCommon::getMemberObjectStatus(nextDoc);
             invariant(!status.isOK());
             warning() << "Aggregate command executor error: " << PlanExecutor::statestr(state)
                       << ", status: " << status
@@ -212,8 +212,13 @@ bool handleCursorCommand(OperationContext* opCtx,
 
         // If adding this object will cause us to exceed the message size limit, then we stash it
         // for later.
+
+        auto* expCtx = exec->getExpCtx().get();
+        BSONObj next = expCtx->needsMerge
+            ? nextDoc.toBsonWithMetaData(expCtx ? expCtx->use42ChangeStreamSortKeys : false)
+            : nextDoc.toBson();
         if (!FindCommon::haveSpaceForNext(next, objCount, responseBuilder.bytesUsed())) {
-            exec->enqueue(next);
+            exec->enqueue(nextDoc);
             stashedResult = true;
             break;
         }
@@ -720,7 +725,8 @@ Status runAggregate(OperationContext* opCtx,
             repl::ReadConcernArgs::get(opCtx),
             cmdObj,
             lockPolicy,
-            privileges);
+            privileges,
+            expCtx->needsMerge);
         if (expCtx->tailableMode == TailableModeEnum::kTailable) {
             cursorParams.setTailable(true);
         } else if (expCtx->tailableMode == TailableModeEnum::kTailableAndAwaitData) {
diff --git a/src/mongo/db/commands/test_commands.cpp b/src/mongo/db/commands/test_commands.cpp
index 183d4cf4095..3e94951bb5b 100644
--- a/src/mongo/db/commands/test_commands.cpp
+++ b/src/mongo/db/commands/test_commands.cpp
@@ -161,7 +161,7 @@ public:
                 opCtx, fullNs.ns(), collection, PlanExecutor::NO_YIELD, InternalPlanner::BACKWARD);
 
             for (int i = 0; i < n + 1; ++i) {
-                PlanExecutor::ExecState state = exec->getNext(nullptr, &end);
+                PlanExecutor::ExecState state = exec->getNext(static_cast<BSONObj*>(nullptr), &end);
                 if (PlanExecutor::ADVANCED != state) {
                     uasserted(ErrorCodes::IllegalOperation,
                               str::stream() << "invalid n, collection contains fewer than " << n
diff --git a/src/mongo/db/exec/document_value/document.cpp b/src/mongo/db/exec/document_value/document.cpp
index 245d58f5a40..2f6660bd568 100644
--- a/src/mongo/db/exec/document_value/document.cpp
+++ b/src/mongo/db/exec/document_value/document.cpp
@@ -493,6 +493,10 @@ constexpr StringData Document::metaFieldSearchHighlights;
 BSONObj Document::toBsonWithMetaData(bool use42ChangeStreamSortKeys) const {
     BSONObjBuilder bb;
     toBson(&bb);
+    if (!metadata()) {
+        return bb.obj();
+    }
+
     if (metadata().hasTextScore())
         bb.append(metaFieldTextScore, metadata().getTextScore());
     if (metadata().hasRandVal())
diff --git a/src/mongo/db/exec/document_value/document_metadata_fields.cpp b/src/mongo/db/exec/document_value/document_metadata_fields.cpp
index 68b0a0da0d0..c7c2e4d202d 100644
--- a/src/mongo/db/exec/document_value/document_metadata_fields.cpp
+++ b/src/mongo/db/exec/document_value/document_metadata_fields.cpp
@@ -234,4 +234,37 @@ Value DocumentMetadataFields::deserializeSortKey(bool isSingleElementKey,
     return Value{std::move(keys)};
 }
 
+const char* DocumentMetadataFields::typeNameToDebugString(DocumentMetadataFields::MetaType type) {
+    switch (type) {
+        case DocumentMetadataFields::kGeoNearDist:
+            return "$geoNear distance";
+        case DocumentMetadataFields::kGeoNearPoint:
+            return "$geoNear point";
+        case DocumentMetadataFields::kIndexKey:
+            return "index key";
+        case DocumentMetadataFields::kRandVal:
+            return "rand val";
+        case DocumentMetadataFields::kRecordId:
+            return "record ID";
+        case DocumentMetadataFields::kSearchHighlights:
+            return "$searchBeta highlights";
+        case DocumentMetadataFields::kSearchScore:
+            return "$searchBeta score";
+        case DocumentMetadataFields::kSortKey:
+            return "sort key";
+        case DocumentMetadataFields::kTextScore:
+            return "text score";
+        default:
+            MONGO_UNREACHABLE;
+    }
+}  // namespace
+
+
+std::ostream& operator<<(std::ostream& stream, DocumentMetadataFields::MetaType type) {
+    return stream << DocumentMetadataFields::typeNameToDebugString(type);
+}
+
+StringBuilder& operator<<(StringBuilder& stream, DocumentMetadataFields::MetaType type) {
+    return stream << DocumentMetadataFields::typeNameToDebugString(type);
+}
 }  // namespace mongo
diff --git a/src/mongo/db/exec/document_value/document_metadata_fields.h b/src/mongo/db/exec/document_value/document_metadata_fields.h
index f5104c847de..ef4627b14c6 100644
--- a/src/mongo/db/exec/document_value/document_metadata_fields.h
+++ b/src/mongo/db/exec/document_value/document_metadata_fields.h
@@ -36,7 +36,6 @@
 #include "mongo/db/record_id.h"
 
 namespace mongo {
-
 /**
  * This class represents the metadata that the query execution engine can associate with a
  * particular intermediate result (either index key or document) passing between execution stages.
@@ -55,7 +54,8 @@ namespace mongo {
 class DocumentMetadataFields {
 public:
     enum MetaType : char {
-        kGeoNearDist,
+        // Start from 1 so that these values can be stored in a bitset.
+        kGeoNearDist = 1,
         kGeoNearPoint,
         kIndexKey,
         kRandVal,
@@ -91,6 +91,11 @@ public:
     static Value deserializeSortKey(bool isSingleElementKey, const BSONObj& bsonSortKey);
 
     /**
+     * Given a metadata type, return a (debug) string representation.
+     */
+    static const char* typeNameToDebugString(DocumentMetadataFields::MetaType type);
+
+    /**
      * Constructs a new DocumentMetadataFields in an uninitialized state.
      */
     DocumentMetadataFields() = default;
@@ -324,4 +329,9 @@ private:
     std::unique_ptr<MetadataHolder> _holder;
 };
 
+using QueryMetadataBitSet = std::bitset<DocumentMetadataFields::MetaType::kNumFields>;
+
+// Prints the metadata's name to the given stream.
+std::ostream& operator<<(std::ostream& stream, DocumentMetadataFields::MetaType type);
+StringBuilder& operator<<(StringBuilder& sb, DocumentMetadataFields::MetaType type);
 }  // namespace mongo
diff --git a/src/mongo/db/exec/working_set_common.cpp b/src/mongo/db/exec/working_set_common.cpp
index ef1fa322c42..571e7a25b4f 100644
--- a/src/mongo/db/exec/working_set_common.cpp
+++ b/src/mongo/db/exec/working_set_common.cpp
@@ -100,7 +100,7 @@ bool WorkingSetCommon::fetch(OperationContext* opCtx,
     return true;
 }
 
-BSONObj WorkingSetCommon::buildMemberStatusObject(const Status& status) {
+Document WorkingSetCommon::buildMemberStatusObject(const Status& status) {
     BSONObjBuilder bob;
     bob.append("ok", status.isOK() ? 1.0 : 0.0);
     bob.append("code", status.code());
@@ -109,7 +109,7 @@ BSONObj WorkingSetCommon::buildMemberStatusObject(const Status& status) {
         extraInfo->serialize(&bob);
     }
 
-    return bob.obj();
+    return Document{bob.obj()};
 }
 
 WorkingSetID WorkingSetCommon::allocateStatusMember(WorkingSet* ws, const Status& status) {
@@ -117,14 +117,19 @@ WorkingSetID WorkingSetCommon::allocateStatusMember(WorkingSet* ws, const Status
 
     WorkingSetID wsid = ws->allocate();
     WorkingSetMember* member = ws->get(wsid);
-    member->resetDocument(SnapshotId(), buildMemberStatusObject(status));
+    member->doc = {SnapshotId(), buildMemberStatusObject(status)};
     member->transitionToOwnedObj();
 
     return wsid;
 }
 
+bool WorkingSetCommon::isValidStatusMemberObject(const Document& obj) {
+    return !obj["ok"].missing() && obj["code"].getType() == BSONType::NumberInt &&
+        obj["errmsg"].getType() == BSONType::String;
+}
+
 bool WorkingSetCommon::isValidStatusMemberObject(const BSONObj& obj) {
-    return obj.hasField("ok") && obj["code"].type() == NumberInt && obj["errmsg"].type() == String;
+    return isValidStatusMemberObject(Document{obj});
 }
 
 boost::optional<Document> WorkingSetCommon::getStatusMemberDocument(const WorkingSet& ws,
@@ -136,8 +141,8 @@ boost::optional<Document> WorkingSetCommon::getStatusMemberDocument(const Workin
     if (!member->hasOwnedObj()) {
         return boost::none;
     }
-    BSONObj obj = member->doc.value().toBson();
-    if (!isValidStatusMemberObject(obj)) {
+
+    if (!isValidStatusMemberObject(member->doc.value())) {
         return boost::none;
     }
     return member->doc.value();
@@ -150,17 +155,22 @@ Status WorkingSetCommon::getMemberObjectStatus(const BSONObj& memberObj) {
                   memberObj);
 }
 
+Status WorkingSetCommon::getMemberObjectStatus(const Document& doc) {
+    return getMemberObjectStatus(doc.toBson());
+}
+
 Status WorkingSetCommon::getMemberStatus(const WorkingSetMember& member) {
     invariant(member.hasObj());
     return getMemberObjectStatus(member.doc.value().toBson());
 }
 
 std::string WorkingSetCommon::toStatusString(const BSONObj& obj) {
-    if (!isValidStatusMemberObject(obj)) {
+    Document doc{obj};
+    if (!isValidStatusMemberObject(doc)) {
         Status unknownStatus(ErrorCodes::UnknownError, "no details available");
         return unknownStatus.toString();
     }
-    return getMemberObjectStatus(obj).toString();
+    return getMemberObjectStatus(doc).toString();
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/exec/working_set_common.h b/src/mongo/db/exec/working_set_common.h
index 15657a87725..0ebeb2ae65c 100644
--- a/src/mongo/db/exec/working_set_common.h
+++ b/src/mongo/db/exec/working_set_common.h
@@ -56,9 +56,9 @@ public:
                       unowned_ptr<SeekableRecordCursor> cursor);
 
     /**
-     * Build a BSONObj which represents a Status to return in a WorkingSet.
+     * Build a Document which represents a Status to return in a WorkingSet.
      */
-    static BSONObj buildMemberStatusObject(const Status& status);
+    static Document buildMemberStatusObject(const Status& status);
 
     /**
      * Allocate a new WSM and initialize it with
@@ -75,6 +75,7 @@ public:
     /**
      * Returns true if object was created by allocateStatusMember().
      */
+    static bool isValidStatusMemberObject(const Document& obj);
     static bool isValidStatusMemberObject(const BSONObj& obj);
 
     /**
@@ -89,6 +90,7 @@ public:
      * Assumes isValidStatusMemberObject().
      */
     static Status getMemberObjectStatus(const BSONObj& memberObj);
+    static Status getMemberObjectStatus(const Document& memberObj);
 
     /**
      * Returns status from working set member created with allocateStatusMember().
diff --git a/src/mongo/db/pipeline/dependencies.cpp b/src/mongo/db/pipeline/dependencies.cpp
index c9482c9f0e1..7bc9532ed6f 100644
--- a/src/mongo/db/pipeline/dependencies.cpp
+++ b/src/mongo/db/pipeline/dependencies.cpp
@@ -29,6 +29,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/exec/document_value/document_metadata_fields.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/pipeline/dependencies.h"
 #include "mongo/db/pipeline/field_path.h"
@@ -36,50 +37,17 @@
 
 namespace mongo {
 
-constexpr DepsTracker::MetadataAvailable DepsTracker::kAllGeoNearDataAvailable;
-
-bool DepsTracker::_appendMetaProjections(BSONObjBuilder* projectionBuilder) const {
-    if (_needTextScore) {
-        projectionBuilder->append(Document::metaFieldTextScore,
-                                  BSON("$meta"
-                                       << "textScore"));
-    }
-    if (_needSortKey) {
-        projectionBuilder->append(Document::metaFieldSortKey,
-                                  BSON("$meta"
-                                       << "sortKey"));
-    }
-    if (_needGeoNearDistance) {
-        projectionBuilder->append(Document::metaFieldGeoNearDistance,
-                                  BSON("$meta"
-                                       << "geoNearDistance"));
-    }
-    if (_needGeoNearPoint) {
-        projectionBuilder->append(Document::metaFieldGeoNearPoint,
-                                  BSON("$meta"
-                                       << "geoNearPoint"));
-    }
-    return (_needTextScore || _needSortKey || _needGeoNearDistance || _needGeoNearPoint);
-}
-
-BSONObj DepsTracker::toProjection() const {
+BSONObj DepsTracker::toProjectionWithoutMetadata() const {
     BSONObjBuilder bb;
 
-    const bool needsMetadata = _appendMetaProjections(&bb);
-
     if (needWholeDocument) {
         return bb.obj();
     }
 
     if (fields.empty()) {
-        if (needsMetadata) {
-            // We only need metadata, but there is no easy way to express this in the query
-            // projection language. We use $noFieldsNeeded with a meta-projection since this is an
-            // inclusion projection which will exclude all existing fields but add the metadata.
-            bb.append("_id", 0);
-            bb.append("$__INTERNAL_QUERY_PROJECTION_RESERVED", 1);
-        }
-        // We either need nothing (as we would if this was logically a count), or only the metadata.
+        // We need no user-level fields (as we would if this was logically a count). Since there is
+        // no way of expressing a projection that indicates no depencies, we return an empty
+        // projection.
         return bb.obj();
     }
 
@@ -103,10 +71,11 @@ BSONObj DepsTracker::toProjection() const {
 
         last = field + '.';
 
-        // We should only have dependencies on fields that are valid in aggregation. Create a
-        // FieldPath to check this.
-        FieldPath fieldPath(field);
-
+        {
+            // Check that the field requested is a valid field name in the agg language. This
+            // constructor will throw if it isn't.
+            FieldPath fp(field);
+        }
         bb.append(field, 1);
     }
 
@@ -118,173 +87,21 @@ BSONObj DepsTracker::toProjection() const {
     return bb.obj();
 }
 
-// ParsedDeps::_fields is a simple recursive look-up table. For each field:
-//      If the value has type==Bool, the whole field is needed
-//      If the value has type==Object, the fields in the subobject are needed
-//      All other fields should be missing which means not needed
-boost::optional<ParsedDeps> DepsTracker::toParsedDeps() const {
-    MutableDocument md;
+void DepsTracker::setNeedsMetadata(DocumentMetadataFields::MetaType type, bool required) {
+    // For everything but sortKey/randval metadata, check that it's available. A pipeline can
+    // generate those types of metadata.
 
-    if (needWholeDocument || _needTextScore) {
-        // can't use ParsedDeps in this case
-        return boost::none;
+    if (type != DocumentMetadataFields::MetaType::kSortKey &&
+        type != DocumentMetadataFields::MetaType::kRandVal) {
+        uassert(40218,
+                str::stream() << "pipeline requires " << type
+                              << " metadata, but it is not available",
+                !required || isMetadataAvailable(type));
     }
 
-    std::string last;
-    for (const auto& field : fields) {
-        if (!last.empty() && str::startsWith(field, last)) {
-            // we are including a parent of *it so we don't need to include this field
-            // explicitly. In fact, if we included this field, the parent wouldn't be fully
-            // included.  This logic relies on on set iterators going in lexicographic order so
-            // that a string is always directly before of all fields it prefixes.
-            continue;
-        }
-        last = field + '.';
-        md.setNestedField(field, Value(true));
-    }
-
-    return ParsedDeps(md.freeze());
-}
-
-bool DepsTracker::getNeedsMetadata(MetadataType type) const {
-    switch (type) {
-        case MetadataType::TEXT_SCORE:
-            return _needTextScore;
-        case MetadataType::SORT_KEY:
-            return _needSortKey;
-        case MetadataType::GEO_NEAR_DISTANCE:
-            return _needGeoNearDistance;
-        case MetadataType::GEO_NEAR_POINT:
-            return _needGeoNearPoint;
-    }
-    MONGO_UNREACHABLE;
-}
-
-bool DepsTracker::isMetadataAvailable(MetadataType type) const {
-    switch (type) {
-        case MetadataType::TEXT_SCORE:
-            return _metadataAvailable & MetadataAvailable::kTextScore;
-        case MetadataType::SORT_KEY:
-            MONGO_UNREACHABLE;
-        case MetadataType::GEO_NEAR_DISTANCE:
-            return _metadataAvailable & MetadataAvailable::kGeoNearDistance;
-        case MetadataType::GEO_NEAR_POINT:
-            return _metadataAvailable & MetadataAvailable::kGeoNearPoint;
-    }
-    MONGO_UNREACHABLE;
-}
-
-void DepsTracker::setNeedsMetadata(MetadataType type, bool required) {
-    switch (type) {
-        case MetadataType::TEXT_SCORE:
-            uassert(40218,
-                    "pipeline requires text score metadata, but there is no text score available",
-                    !required || isMetadataAvailable(type));
-            _needTextScore = required;
-            return;
-        case MetadataType::SORT_KEY:
-            invariant(required || !_needSortKey);
-            _needSortKey = required;
-            return;
-        case MetadataType::GEO_NEAR_DISTANCE:
-            uassert(50860,
-                    "pipeline requires $geoNear distance metadata, but it is not available",
-                    !required || isMetadataAvailable(type));
-            invariant(required || !_needGeoNearDistance);
-            _needGeoNearDistance = required;
-            return;
-        case MetadataType::GEO_NEAR_POINT:
-            uassert(50859,
-                    "pipeline requires $geoNear point metadata, but it is not available",
-                    !required || isMetadataAvailable(type));
-            invariant(required || !_needGeoNearPoint);
-            _needGeoNearPoint = required;
-            return;
-    }
-    MONGO_UNREACHABLE;
-}
-
-std::vector<DepsTracker::MetadataType> DepsTracker::getAllRequiredMetadataTypes() const {
-    std::vector<MetadataType> reqs;
-    if (_needTextScore) {
-        reqs.push_back(MetadataType::TEXT_SCORE);
-    }
-    if (_needSortKey) {
-        reqs.push_back(MetadataType::SORT_KEY);
-    }
-    if (_needGeoNearDistance) {
-        reqs.push_back(MetadataType::GEO_NEAR_DISTANCE);
-    }
-    if (_needGeoNearPoint) {
-        reqs.push_back(MetadataType::GEO_NEAR_POINT);
-    }
-    return reqs;
-}
-
-namespace {
-// Mutually recursive with arrayHelper
-Document documentHelper(const BSONObj& bson, const Document& neededFields, int nFieldsNeeded = -1);
-
-// Handles array-typed values for ParsedDeps::extractFields
-Value arrayHelper(const BSONObj& bson, const Document& neededFields) {
-    BSONObjIterator it(bson);
-
-    std::vector<Value> values;
-    while (it.more()) {
-        BSONElement bsonElement(it.next());
-        if (bsonElement.type() == Object) {
-            Document sub = documentHelper(bsonElement.embeddedObject(), neededFields);
-            values.push_back(Value(sub));
-        }
-
-        if (bsonElement.type() == Array) {
-            values.push_back(arrayHelper(bsonElement.embeddedObject(), neededFields));
-        }
-    }
-
-    return Value(std::move(values));
-}
-
-// Handles object-typed values including the top-level for ParsedDeps::extractFields
-Document documentHelper(const BSONObj& bson, const Document& neededFields, int nFieldsNeeded) {
-    // We cache the number of top level fields, so don't need to re-compute it every time. For
-    // sub-documents, just scan for the number of fields.
-    if (nFieldsNeeded == -1) {
-        nFieldsNeeded = neededFields.size();
-    }
-    MutableDocument md(nFieldsNeeded);
-
-    BSONObjIterator it(bson);
-    while (it.more() && nFieldsNeeded > 0) {
-        auto bsonElement = it.next();
-        StringData fieldName = bsonElement.fieldNameStringData();
-        Value isNeeded = neededFields[fieldName];
-
-        if (isNeeded.missing())
-            continue;
-
-        --nFieldsNeeded;  // Found a needed field.
-        if (isNeeded.getType() == Bool) {
-            md.addField(fieldName, Value(bsonElement));
-        } else {
-            dassert(isNeeded.getType() == Object);
-
-            if (bsonElement.type() == BSONType::Object) {
-                md.addField(
-                    fieldName,
-                    Value(documentHelper(bsonElement.embeddedObject(), isNeeded.getDocument())));
-            } else if (bsonElement.type() == BSONType::Array) {
-                md.addField(fieldName,
-                            arrayHelper(bsonElement.embeddedObject(), isNeeded.getDocument()));
-            }
-        }
-    }
-
-    return md.freeze();
-}
-}  // namespace
-
-Document ParsedDeps::extractFields(const BSONObj& input) const {
-    return documentHelper(input, _fields, _nFields);
+    // If the metadata type is not required, then it should not be recorded as a metadata
+    // dependency.
+    invariant(required || !_metadataDeps[type]);
+    _metadataDeps[type] = required;
 }
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/dependencies.h b/src/mongo/db/pipeline/dependencies.h
index 27e2efc41fd..eff5610b089 100644
--- a/src/mongo/db/pipeline/dependencies.h
+++ b/src/mongo/db/pipeline/dependencies.h
@@ -37,7 +37,6 @@
 #include "mongo/db/pipeline/variables.h"
 
 namespace mongo {
-class ParsedDeps;
 
 /**
  * This struct allows components in an agg pipeline to report what they need from their input.
@@ -69,56 +68,40 @@ struct DepsTracker {
     };
 
     /**
-     * Represents the type of metadata a pipeline might request.
+     * Represents a state where all geo metadata is available.
      */
-    enum class MetadataType {
-        // The score associated with a text match.
-        TEXT_SCORE,
-
-        // The key to use for sorting.
-        SORT_KEY,
-
-        // The computed distance for a near query.
-        GEO_NEAR_DISTANCE,
-
-        // The point used in the computation of the GEO_NEAR_DISTANCE.
-        GEO_NEAR_POINT,
-    };
+    static constexpr auto kAllGeoNearData = QueryMetadataBitSet(
+        (1 << DocumentMetadataFields::kGeoNearDist) | (1 << DocumentMetadataFields::kGeoNearPoint));
 
     /**
-     * Represents what metadata is available on documents that are input to the pipeline.
+     * Represents a state where all metadata is available.
      */
-    enum MetadataAvailable {
-        kNoMetadata = 0,
-        kTextScore = 1 << 1,
-        kGeoNearDistance = 1 << 2,
-        kGeoNearPoint = 1 << 3,
-    };
+    static constexpr auto kAllMetadata =
+        QueryMetadataBitSet(~(1 << DocumentMetadataFields::kNumFields));
 
     /**
-     * Represents a state where all geo metadata is available.
+     * Represents a state where only text score metadata is available.
      */
-    static constexpr auto kAllGeoNearDataAvailable =
-        MetadataAvailable(MetadataAvailable::kGeoNearDistance | MetadataAvailable::kGeoNearPoint);
+    static constexpr auto kOnlyTextScore =
+        QueryMetadataBitSet(1 << DocumentMetadataFields::kTextScore);
 
     /**
-     * Represents a state where all metadata is available.
+     * Represents a state where no metadata is available.
      */
-    static constexpr auto kAllMetadataAvailable =
-        MetadataAvailable(kTextScore | kGeoNearDistance | kGeoNearPoint);
+    static constexpr auto kNoMetadata = QueryMetadataBitSet();
+
 
-    DepsTracker(MetadataAvailable metadataAvailable = kNoMetadata)
+    DepsTracker(QueryMetadataBitSet metadataAvailable = kNoMetadata)
         : _metadataAvailable(metadataAvailable) {}
 
     /**
-     * Returns a projection object covering the dependencies tracked by this class.
+     * Returns a projection object covering the non-metadata dependencies tracked by this class, or
+     * empty BSONObj if the entire document is required.
      */
-    BSONObj toProjection() const;
-
-    boost::optional<ParsedDeps> toParsedDeps() const;
+    BSONObj toProjectionWithoutMetadata() const;
 
     bool hasNoRequirements() const {
-        return fields.empty() && !needWholeDocument && !_needTextScore;
+        return fields.empty() && !needWholeDocument && !_metadataDeps.any();
     }
 
     /**
@@ -134,7 +117,7 @@ struct DepsTracker {
     /**
      * Returns a value with bits set indicating the types of metadata available.
      */
-    MetadataAvailable getMetadataAvailable() const {
+    QueryMetadataBitSet getMetadataAvailable() const {
         return _metadataAvailable;
     }
 
@@ -143,7 +126,9 @@ struct DepsTracker {
      * illegal to call this with MetadataType::SORT_KEY, since the sort key will always be available
      * if needed.
      */
-    bool isMetadataAvailable(MetadataType type) const;
+    bool isMetadataAvailable(DocumentMetadataFields::MetaType type) const {
+        return _metadataAvailable[type];
+    }
 
     /**
      * Sets whether or not metadata 'type' is required. Throws if 'required' is true but that
@@ -151,58 +136,49 @@ struct DepsTracker {
      *
      * Except for MetadataType::SORT_KEY, once 'type' is required, it cannot be unset.
      */
-    void setNeedsMetadata(MetadataType type, bool required);
+    void setNeedsMetadata(DocumentMetadataFields::MetaType type, bool required);
 
     /**
      * Returns true if the DepsTracker requires that metadata of type 'type' is present.
      */
-    bool getNeedsMetadata(MetadataType type) const;
+    bool getNeedsMetadata(DocumentMetadataFields::MetaType type) const {
+        return _metadataDeps[type];
+    }
 
     /**
      * Returns true if there exists a type of metadata required by the DepsTracker.
      */
     bool getNeedsAnyMetadata() const {
-        return _needTextScore || _needSortKey || _needGeoNearDistance || _needGeoNearPoint;
+        return _metadataDeps.any();
     }
 
     /**
-     * Returns a vector containing all the types of metadata required by this DepsTracker.
+     * Return all of the metadata dependencies.
      */
-    std::vector<MetadataType> getAllRequiredMetadataTypes() const;
-
-    std::set<std::string> fields;    // Names of needed fields in dotted notation.
-    std::set<Variables::Id> vars;    // IDs of referenced variables.
-    bool needWholeDocument = false;  // If true, ignore 'fields'; the whole document is needed.
+    QueryMetadataBitSet& metadataDeps() {
+        return _metadataDeps;
+    }
+    const QueryMetadataBitSet& metadataDeps() const {
+        return _metadataDeps;
+    }
 
-private:
     /**
-     * Appends the meta projections for the sort key and/or text score to 'bb' if necessary. Returns
-     * true if either type of metadata was needed, and false otherwise.
+     * Request that all metadata in the given QueryMetadataBitSet be added as dependencies.
      */
-    bool _appendMetaProjections(BSONObjBuilder* bb) const;
-
-    MetadataAvailable _metadataAvailable;
-
-    // Each member variable influences a different $meta projection.
-    bool _needTextScore = false;        // {$meta: "textScore"}
-    bool _needSortKey = false;          // {$meta: "sortKey"}
-    bool _needGeoNearDistance = false;  // {$meta: "geoNearDistance"}
-    bool _needGeoNearPoint = false;     // {$meta: "geoNearPoint"}
-};
+    void requestMetadata(const QueryMetadataBitSet& metadata) {
+        _metadataDeps |= metadata;
+    }
 
-/**
- * This class is designed to quickly extract the needed fields from a BSONObj into a Document.
- * It should only be created by a call to DepsTracker::ParsedDeps
- */
-class ParsedDeps {
-public:
-    Document extractFields(const BSONObj& input) const;
+    std::set<std::string> fields;    // Names of needed fields in dotted notation.
+    std::set<Variables::Id> vars;    // IDs of referenced variables.
+    bool needWholeDocument = false;  // If true, ignore 'fields'; the whole document is needed.
 
 private:
-    friend struct DepsTracker;  // so it can call constructor
-    explicit ParsedDeps(Document&& fields) : _fields(std::move(fields)), _nFields(_fields.size()) {}
+    // Represents all metadata available to the pipeline.
+    QueryMetadataBitSet _metadataAvailable;
 
-    Document _fields;
-    int _nFields;  // Cache the number of top-level fields needed.
+    // Represents which metadata is used by the pipeline. This is populated while performing
+    // dependency analysis.
+    QueryMetadataBitSet _metadataDeps;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/dependencies_test.cpp b/src/mongo/db/pipeline/dependencies_test.cpp
index de11708289d..f5b9c5a37a5 100644
--- a/src/mongo/db/pipeline/dependencies_test.cpp
+++ b/src/mongo/db/pipeline/dependencies_test.cpp
@@ -44,9 +44,6 @@ namespace {
 using std::set;
 using std::string;
 
-static const BSONObj metaTextScore = BSON("$meta"
-                                          << "textScore");
-
 template <size_t ArrayLen>
 set<string> arrayToSet(const char* (&array)[ArrayLen]) {
     set<string> out;
@@ -55,53 +52,64 @@ set<string> arrayToSet(const char* (&array)[ArrayLen]) {
     return out;
 }
 
+TEST(DependenciesTest, CheckClassConstants) {
+    ASSERT_TRUE(DepsTracker::kAllGeoNearData[DocumentMetadataFields::kGeoNearPoint]);
+    ASSERT_TRUE(DepsTracker::kAllGeoNearData[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_EQ(DepsTracker::kAllGeoNearData.count(), 2);
+    ASSERT_TRUE(DepsTracker::kAllMetadata.all());
+    ASSERT_EQ(DepsTracker::kOnlyTextScore.count(), 1);
+    ASSERT_TRUE(DepsTracker::kOnlyTextScore[DocumentMetadataFields::kTextScore]);
+}
+
 TEST(DependenciesToProjectionTest, ShouldIncludeAllFieldsAndExcludeIdIfNotSpecified) {
     const char* array[] = {"a", "b"};
     DepsTracker deps;
     deps.fields = arrayToSet(array);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON("a" << 1 << "b" << 1 << "_id" << 0));
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSON("a" << 1 << "b" << 1 << "_id" << 0));
 }
 
 TEST(DependenciesToProjectionTest, ShouldIncludeFieldEvenIfSuffixOfAnotherIncludedField) {
     const char* array[] = {"a", "ab"};
     DepsTracker deps;
     deps.fields = arrayToSet(array);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON("a" << 1 << "ab" << 1 << "_id" << 0));
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(),
+                      BSON("a" << 1 << "ab" << 1 << "_id" << 0));
 }
 
 TEST(DependenciesToProjectionTest, ShouldNotIncludeSubFieldIfTopLevelAlreadyIncluded) {
     const char* array[] = {"a", "b", "a.b"};  // a.b included by a
     DepsTracker deps;
     deps.fields = arrayToSet(array);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON("a" << 1 << "b" << 1 << "_id" << 0));
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSON("a" << 1 << "b" << 1 << "_id" << 0));
 }
 
 TEST(DependenciesToProjectionTest, ShouldIncludeIdIfNeeded) {
     const char* array[] = {"a", "_id"};
     DepsTracker deps;
     deps.fields = arrayToSet(array);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON("a" << 1 << "_id" << 1));
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSON("a" << 1 << "_id" << 1));
 }
 
 TEST(DependenciesToProjectionTest, ShouldIncludeEntireIdEvenIfOnlyASubFieldIsNeeded) {
     const char* array[] = {"a", "_id.a"};  // still include whole _id (SERVER-7502)
     DepsTracker deps;
     deps.fields = arrayToSet(array);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON("a" << 1 << "_id" << 1));
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSON("a" << 1 << "_id" << 1));
 }
 
 TEST(DependenciesToProjectionTest, ShouldNotIncludeSubFieldOfIdIfIdIncluded) {
     const char* array[] = {"a", "_id", "_id.a"};  // handle both _id and subfield
     DepsTracker deps;
     deps.fields = arrayToSet(array);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON("a" << 1 << "_id" << 1));
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSON("a" << 1 << "_id" << 1));
 }
 
 TEST(DependenciesToProjectionTest, ShouldIncludeFieldPrefixedById) {
     const char* array[] = {"a", "_id", "_id_a"};  // _id prefixed but non-subfield
     DepsTracker deps;
     deps.fields = arrayToSet(array);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON("_id_a" << 1 << "a" << 1 << "_id" << 1));
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(),
+                      BSON("_id_a" << 1 << "a" << 1 << "_id" << 1));
 }
 
 TEST(DependenciesToProjectionTest, ShouldOutputEmptyObjectIfEntireDocumentNeeded) {
@@ -109,55 +117,59 @@ TEST(DependenciesToProjectionTest, ShouldOutputEmptyObjectIfEntireDocumentNeeded
     DepsTracker deps;
     deps.fields = arrayToSet(array);
     deps.needWholeDocument = true;
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSONObj());
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSONObj());
 }
 
 TEST(DependenciesToProjectionTest, ShouldOnlyRequestTextScoreIfEntireDocumentAndTextScoreNeeded) {
     const char* array[] = {"a"};  // needTextScore with needWholeDocument
-    DepsTracker deps(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker deps(DepsTracker::kOnlyTextScore);
     deps.fields = arrayToSet(array);
     deps.needWholeDocument = true;
-    deps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON(Document::metaFieldTextScore << metaTextScore));
+    deps.setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSONObj());
+    ASSERT_EQ(deps.metadataDeps().count(), 1u);
+    ASSERT_TRUE(deps.metadataDeps()[DocumentMetadataFields::kTextScore]);
 }
 
 TEST(DependenciesToProjectionTest,
      ShouldRequireFieldsAndTextScoreIfTextScoreNeededWithoutWholeDocument) {
     const char* array[] = {"a"};  // needTextScore without needWholeDocument
-    DepsTracker deps(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker deps(DepsTracker::kOnlyTextScore);
     deps.fields = arrayToSet(array);
-    deps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
-    ASSERT_BSONOBJ_EQ(
-        deps.toProjection(),
-        BSON(Document::metaFieldTextScore << metaTextScore << "a" << 1 << "_id" << 0));
+    deps.setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSON("a" << 1 << "_id" << 0));
+    ASSERT_EQ(deps.metadataDeps().count(), 1u);
+    ASSERT_TRUE(deps.metadataDeps()[DocumentMetadataFields::kTextScore]);
 }
 
 TEST(DependenciesToProjectionTest, ShouldProduceEmptyObjectIfThereAreNoDependencies) {
-    DepsTracker deps(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker deps(DepsTracker::kOnlyTextScore);
     deps.fields = {};
     deps.needWholeDocument = false;
-    deps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, false);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSONObj());
+    deps.setNeedsMetadata(DocumentMetadataFields::kTextScore, false);
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSONObj());
 }
 
-TEST(DependenciesToProjectionTest, ShouldAttemptToExcludeOtherFieldsIfOnlyTextScoreIsNeeded) {
-    DepsTracker deps(DepsTracker::MetadataAvailable::kTextScore);
+TEST(DependenciesToProjectionTest, ShouldReturnEmptyObjectIfOnlyTextScoreIsNeeded) {
+    DepsTracker deps(DepsTracker::kOnlyTextScore);
     deps.fields = {};
     deps.needWholeDocument = false;
-    deps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(),
-                      BSON(Document::metaFieldTextScore << metaTextScore << "_id" << 0
-                                                        << "$__INTERNAL_QUERY_PROJECTION_RESERVED"
-                                                        << 1));
+    deps.setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSONObj());
+
+    ASSERT_EQ(deps.metadataDeps().count(), 1u);
+    ASSERT_TRUE(deps.metadataDeps()[DocumentMetadataFields::kTextScore]);
 }
 
 TEST(DependenciesToProjectionTest,
      ShouldRequireTextScoreIfNoFieldsPresentButWholeDocumentIsNeeded) {
-    DepsTracker deps(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker deps(DepsTracker::kOnlyTextScore);
     deps.fields = {};
     deps.needWholeDocument = true;
-    deps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
-    ASSERT_BSONOBJ_EQ(deps.toProjection(), BSON(Document::metaFieldTextScore << metaTextScore));
+    deps.setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
+    ASSERT_BSONOBJ_EQ(deps.toProjectionWithoutMetadata(), BSONObj());
+    ASSERT_EQ(deps.metadataDeps().count(), 1u);
+    ASSERT_TRUE(deps.metadataDeps()[DocumentMetadataFields::kTextScore]);
 }
 
 }  // namespace
diff --git a/src/mongo/db/pipeline/document_source_add_fields_test.cpp b/src/mongo/db/pipeline/document_source_add_fields_test.cpp
index da57f5f7847..e20820feb0d 100644
--- a/src/mongo/db/pipeline/document_source_add_fields_test.cpp
+++ b/src/mongo/db/pipeline/document_source_add_fields_test.cpp
@@ -147,7 +147,7 @@ TEST_F(AddFieldsTest, ShouldAddReferencedFieldsToDependencies) {
     auto addFields = DocumentSourceAddFields::create(
         fromjson("{a: true, x: '$b', y: {$and: ['$c','$d']}, z: {$meta: 'textScore'}}"),
         getExpCtx());
-    DepsTracker dependencies(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker dependencies(DepsTracker::kOnlyTextScore);
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, addFields->getDependencies(&dependencies));
     ASSERT_EQUALS(3U, dependencies.fields.size());
 
@@ -164,7 +164,7 @@ TEST_F(AddFieldsTest, ShouldAddReferencedFieldsToDependencies) {
     ASSERT_EQUALS(1U, dependencies.fields.count("c"));
     ASSERT_EQUALS(1U, dependencies.fields.count("d"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(AddFieldsTest, ShouldPropagatePauses) {
diff --git a/src/mongo/db/pipeline/document_source_bucket_auto_test.cpp b/src/mongo/db/pipeline/document_source_bucket_auto_test.cpp
index 5f450c1f792..9c68c05a50c 100644
--- a/src/mongo/db/pipeline/document_source_bucket_auto_test.cpp
+++ b/src/mongo/db/pipeline/document_source_bucket_auto_test.cpp
@@ -450,19 +450,19 @@ TEST_F(BucketAutoTests, ShouldAddDependenciesOfGroupByFieldAndComputedFields) {
     ASSERT_EQUALS(1U, dependencies.fields.count("b"));
 
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(BucketAutoTests, ShouldNeedTextScoreInDependenciesFromGroupByField) {
     auto bucketAuto =
         createBucketAuto(fromjson("{$bucketAuto : {groupBy : {$meta: 'textScore'}, buckets : 2}}"));
 
-    DepsTracker dependencies(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker dependencies(DepsTracker::kOnlyTextScore);
     ASSERT_EQUALS(DepsTracker::State::EXHAUSTIVE_ALL, bucketAuto->getDependencies(&dependencies));
     ASSERT_EQUALS(0U, dependencies.fields.size());
 
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(BucketAutoTests, ShouldNeedTextScoreInDependenciesFromOutputField) {
@@ -470,7 +470,7 @@ TEST_F(BucketAutoTests, ShouldNeedTextScoreInDependenciesFromOutputField) {
         createBucketAuto(fromjson("{$bucketAuto : {groupBy : '$x', buckets : 2, output: {avg : "
                                   "{$avg : {$meta : 'textScore'}}}}}"));
 
-    DepsTracker dependencies(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker dependencies(DepsTracker::kOnlyTextScore);
     ASSERT_EQUALS(DepsTracker::State::EXHAUSTIVE_ALL, bucketAuto->getDependencies(&dependencies));
     ASSERT_EQUALS(1U, dependencies.fields.size());
 
@@ -478,7 +478,7 @@ TEST_F(BucketAutoTests, ShouldNeedTextScoreInDependenciesFromOutputField) {
     ASSERT_EQUALS(1U, dependencies.fields.count("x"));
 
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(BucketAutoTests, SerializesDefaultAccumulatorIfOutputFieldIsNotSpecified) {
diff --git a/src/mongo/db/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index eb01db2a2ad..e1193e33cb8 100644
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -73,13 +73,6 @@ DocumentSource::GetNextResult DocumentSourceCursor::doGetNext() {
     return std::move(out);
 }
 
-Document DocumentSourceCursor::transformBSONObjToDocument(const BSONObj& obj) const {
-    // Aggregation assumes ownership of underlying BSON.
-    return _dependencies ? _dependencies->extractFields(obj)
-                         : (_inputHasMetadata ? Document::fromBsonWithMetaData(obj.getOwned())
-                                              : Document(obj.getOwned()));
-}
-
 void DocumentSourceCursor::loadBatch() {
     if (!_exec || _exec->isDisposed()) {
         // No more documents.
@@ -92,7 +85,7 @@ void DocumentSourceCursor::loadBatch() {
     }
 
     PlanExecutor::ExecState state;
-    BSONObj resultObj;
+    Document resultObj;
     {
         AutoGetCollectionForRead autoColl(pExpCtx->opCtx, _exec->nss());
         uassertStatusOK(repl::ReplicationCoordinator::get(pExpCtx->opCtx)
@@ -108,7 +101,7 @@ void DocumentSourceCursor::loadBatch() {
                 if (_shouldProduceEmptyDocs) {
                     _currentBatch.push_back(Document());
                 } else {
-                    _currentBatch.push_back(transformBSONObjToDocument(resultObj));
+                    _currentBatch.push_back(transformDoc(resultObj.getOwned()));
                 }
 
                 if (_limit) {
@@ -228,9 +221,6 @@ Value DocumentSourceCursor::serialize(boost::optional<ExplainOptions::Verbosity>
     if (_limit)
         out["limit"] = Value(_limit->getLimit());
 
-    if (!_projection.isEmpty())
-        out["fields"] = Value(_projection);
-
     BSONObjBuilder explainStatsBuilder;
 
     {
diff --git a/src/mongo/db/pipeline/document_source_cursor.h b/src/mongo/db/pipeline/document_source_cursor.h
index fa28dd114a0..60495193b93 100644
--- a/src/mongo/db/pipeline/document_source_cursor.h
+++ b/src/mongo/db/pipeline/document_source_cursor.h
@@ -115,21 +115,6 @@ public:
     }
 
     /**
-     * Informs this object of projection and dependency information.
-     *
-     * @param projection The projection that has been passed down to the query system.
-     * @param deps The output of DepsTracker::toParsedDeps.
-     * @param inputHasMetadata Indicates whether the input BSON object contains metadata fields.
-     */
-    void setProjection(const BSONObj& projection,
-                       const boost::optional<ParsedDeps>& deps,
-                       bool inputHasMetadata) {
-        _projection = projection;
-        _dependencies = deps;
-        _inputHasMetadata = inputHasMetadata;
-    }
-
-    /**
      * Returns the limit associated with this cursor, or -1 if there is no limit.
      */
     long long getLimit() const {
@@ -182,9 +167,11 @@ protected:
      * If '_shouldProduceEmptyDocs' is false, this function hook is called on each 'obj' returned by
      * '_exec' when loading a batch and returns a Document to be added to '_currentBatch'.
      *
-     * The default implementation is a dependency-aware BSONObj-to-Document transformation.
+     * The default implementation is the identity function.
      */
-    virtual Document transformBSONObjToDocument(const BSONObj& obj) const;
+    virtual Document transformDoc(Document&& doc) const {
+        return std::move(doc);
+    }
 
 private:
     /**
@@ -213,10 +200,7 @@ private:
     // BSONObj members must outlive _projection and cursor.
     BSONObj _query;
     BSONObj _sort;
-    BSONObj _projection;
     bool _shouldProduceEmptyDocs = false;
-    bool _inputHasMetadata = false;
-    boost::optional<ParsedDeps> _dependencies;
     boost::intrusive_ptr<DocumentSourceLimit> _limit;
     long long _docsAddedToBatches;  // for _limit enforcement
 
diff --git a/src/mongo/db/pipeline/document_source_facet.cpp b/src/mongo/db/pipeline/document_source_facet.cpp
index 34bd1d9a4bd..71e9571f869 100644
--- a/src/mongo/db/pipeline/document_source_facet.cpp
+++ b/src/mongo/db/pipeline/document_source_facet.cpp
@@ -305,14 +305,14 @@ DepsTracker::State DocumentSourceFacet::getDependencies(DepsTracker* deps) const
 
         // The text score is the only type of metadata that could be needed by $facet.
         deps->setNeedsMetadata(
-            DepsTracker::MetadataType::TEXT_SCORE,
-            deps->getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE) ||
-                subDepsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+            DocumentMetadataFields::kTextScore,
+            deps->getNeedsMetadata(DocumentMetadataFields::kTextScore) ||
+                subDepsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 
         // If there are variables defined at this stage's scope, there may be dependencies upon
         // them in subsequent pipelines. Keep enumerating.
-        if (deps->needWholeDocument &&
-            deps->getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE) && !scopeHasVariables) {
+        if (deps->needWholeDocument && deps->getNeedsMetadata(DocumentMetadataFields::kTextScore) &&
+            !scopeHasVariables) {
             break;
         }
     }
diff --git a/src/mongo/db/pipeline/document_source_facet_test.cpp b/src/mongo/db/pipeline/document_source_facet_test.cpp
index 5d094fe5bcf..980786689a8 100644
--- a/src/mongo/db/pipeline/document_source_facet_test.cpp
+++ b/src/mongo/db/pipeline/document_source_facet_test.cpp
@@ -618,8 +618,7 @@ TEST_F(DocumentSourceFacetTest, ShouldUnionDependenciesOfInnerPipelines) {
     auto needsA = DocumentSourceNeedsA::create();
     auto firstPipeline = unittest::assertGet(Pipeline::createFacetPipeline({needsA}, ctx));
 
-    auto firstPipelineDeps =
-        firstPipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    auto firstPipelineDeps = firstPipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_FALSE(firstPipelineDeps.needWholeDocument);
     ASSERT_EQ(firstPipelineDeps.fields.size(), 1UL);
     ASSERT_EQ(firstPipelineDeps.fields.count("a"), 1UL);
@@ -627,8 +626,7 @@ TEST_F(DocumentSourceFacetTest, ShouldUnionDependenciesOfInnerPipelines) {
     auto needsB = DocumentSourceNeedsB::create();
     auto secondPipeline = unittest::assertGet(Pipeline::createFacetPipeline({needsB}, ctx));
 
-    auto secondPipelineDeps =
-        secondPipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    auto secondPipelineDeps = secondPipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_FALSE(secondPipelineDeps.needWholeDocument);
     ASSERT_EQ(secondPipelineDeps.fields.size(), 1UL);
     ASSERT_EQ(secondPipelineDeps.fields.count("b"), 1UL);
@@ -638,10 +636,10 @@ TEST_F(DocumentSourceFacetTest, ShouldUnionDependenciesOfInnerPipelines) {
     facets.emplace_back("needsB", std::move(secondPipeline));
     auto facetStage = DocumentSourceFacet::create(std::move(facets), ctx);
 
-    DepsTracker deps(DepsTracker::MetadataAvailable::kNoMetadata);
+    DepsTracker deps(DepsTracker::kNoMetadata);
     ASSERT_EQ(facetStage->getDependencies(&deps), DepsTracker::State::EXHAUSTIVE_ALL);
     ASSERT_FALSE(deps.needWholeDocument);
-    ASSERT_FALSE(deps.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_FALSE(deps.getNeedsMetadata(DocumentMetadataFields::kTextScore));
     ASSERT_EQ(deps.fields.size(), 2UL);
     ASSERT_EQ(deps.fields.count("a"), 1UL);
     ASSERT_EQ(deps.fields.count("b"), 1UL);
@@ -676,10 +674,10 @@ TEST_F(DocumentSourceFacetTest, ShouldRequireWholeDocumentIfAnyPipelineRequiresW
     facets.emplace_back("needsWholeDocument", std::move(secondPipeline));
     auto facetStage = DocumentSourceFacet::create(std::move(facets), ctx);
 
-    DepsTracker deps(DepsTracker::MetadataAvailable::kNoMetadata);
+    DepsTracker deps(DepsTracker::kNoMetadata);
     ASSERT_EQ(facetStage->getDependencies(&deps), DepsTracker::State::EXHAUSTIVE_ALL);
     ASSERT_TRUE(deps.needWholeDocument);
-    ASSERT_FALSE(deps.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_FALSE(deps.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 /**
@@ -688,7 +686,7 @@ TEST_F(DocumentSourceFacetTest, ShouldRequireWholeDocumentIfAnyPipelineRequiresW
 class DocumentSourceNeedsOnlyTextScore : public DocumentSourcePassthrough {
 public:
     DepsTracker::State getDependencies(DepsTracker* deps) const override {
-        deps->setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
+        deps->setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
         return DepsTracker::State::EXHAUSTIVE_ALL;
     }
     static boost::intrusive_ptr<DocumentSourceNeedsOnlyTextScore> create() {
@@ -715,10 +713,10 @@ TEST_F(DocumentSourceFacetTest, ShouldRequireTextScoreIfAnyPipelineRequiresTextS
     facets.emplace_back("needsTextScore", std::move(thirdPipeline));
     auto facetStage = DocumentSourceFacet::create(std::move(facets), ctx);
 
-    DepsTracker deps(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker deps(DepsTracker::kOnlyTextScore);
     ASSERT_EQ(facetStage->getDependencies(&deps), DepsTracker::State::EXHAUSTIVE_ALL);
     ASSERT_TRUE(deps.needWholeDocument);
-    ASSERT_TRUE(deps.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_TRUE(deps.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceFacetTest, ShouldThrowIfAnyPipelineRequiresTextScoreButItIsNotAvailable) {
@@ -735,7 +733,7 @@ TEST_F(DocumentSourceFacetTest, ShouldThrowIfAnyPipelineRequiresTextScoreButItIs
     facets.emplace_back("needsTextScore", std::move(secondPipeline));
     auto facetStage = DocumentSourceFacet::create(std::move(facets), ctx);
 
-    DepsTracker deps(DepsTracker::MetadataAvailable::kNoMetadata);
+    DepsTracker deps(DepsTracker::kNoMetadata);
     ASSERT_THROWS(facetStage->getDependencies(&deps), AssertionException);
 }
 
diff --git a/src/mongo/db/pipeline/document_source_geo_near.cpp b/src/mongo/db/pipeline/document_source_geo_near.cpp
index 61f04889ffa..5081865eeba 100644
--- a/src/mongo/db/pipeline/document_source_geo_near.cpp
+++ b/src/mongo/db/pipeline/document_source_geo_near.cpp
@@ -225,8 +225,8 @@ DepsTracker::State DocumentSourceGeoNear::getDependencies(DepsTracker* deps) con
     // produced by this stage, and we could inform the query system that it need not include it in
     // its response. For now, assume that we require the entire document as well as the appropriate
     // geoNear metadata.
-    deps->setNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_DISTANCE, true);
-    deps->setNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_POINT, needsGeoNearPoint());
+    deps->setNeedsMetadata(DocumentMetadataFields::kGeoNearDist, true);
+    deps->setNeedsMetadata(DocumentMetadataFields::kGeoNearPoint, needsGeoNearPoint());
 
     deps->needWholeDocument = true;
     return DepsTracker::State::EXHAUSTIVE_FIELDS;
diff --git a/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp b/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
index 306bc5f6d3e..0eb6fd5b04b 100644
--- a/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
@@ -83,14 +83,14 @@ const char* DocumentSourceGeoNearCursor::getSourceName() const {
     return DocumentSourceGeoNearCursor::kStageName.rawData();
 }
 
-Document DocumentSourceGeoNearCursor::transformBSONObjToDocument(const BSONObj& obj) const {
-    MutableDocument output(Document::fromBsonWithMetaData(obj));
+Document DocumentSourceGeoNearCursor::transformDoc(Document&& objInput) const {
+    MutableDocument output(std::move(objInput));
 
     // Scale the distance by the requested factor.
     invariant(output.peek().metadata().hasGeoNearDistance(),
               str::stream()
                   << "Query returned a document that is unexpectedly missing the geoNear distance: "
-                  << obj.jsonString());
+                  << output.peek().toString());
     const auto distance = output.peek().metadata().getGeoNearDistance() * _distanceMultiplier;
 
     output.setNestedField(_distanceField, Value(distance));
@@ -99,7 +99,7 @@ Document DocumentSourceGeoNearCursor::transformBSONObjToDocument(const BSONObj&
             output.peek().metadata().hasGeoNearPoint(),
             str::stream()
                 << "Query returned a document that is unexpectedly missing the geoNear point: "
-                << obj.jsonString());
+                << output.peek().toString());
         output.setNestedField(*_locationField, output.peek().metadata().getGeoNearPoint());
     }
 
diff --git a/src/mongo/db/pipeline/document_source_geo_near_cursor.h b/src/mongo/db/pipeline/document_source_geo_near_cursor.h
index 5022c716d2e..084e8b76bbe 100644
--- a/src/mongo/db/pipeline/document_source_geo_near_cursor.h
+++ b/src/mongo/db/pipeline/document_source_geo_near_cursor.h
@@ -82,7 +82,7 @@ private:
     /**
      * Transforms 'obj' into a Document, calculating the distance.
      */
-    Document transformBSONObjToDocument(const BSONObj& obj) const final;
+    Document transformDoc(Document&& obj) const override final;
 
     // The output field in which to store the computed distance.
     FieldPath _distanceField;
diff --git a/src/mongo/db/pipeline/document_source_group_test.cpp b/src/mongo/db/pipeline/document_source_group_test.cpp
index 5de0e26078a..e2fdfe4f107 100644
--- a/src/mongo/db/pipeline/document_source_group_test.cpp
+++ b/src/mongo/db/pipeline/document_source_group_test.cpp
@@ -813,7 +813,7 @@ public:
         ASSERT_EQUALS(1U, dependencies.fields.count("u"));
         ASSERT_EQUALS(1U, dependencies.fields.count("v"));
         ASSERT_EQUALS(false, dependencies.needWholeDocument);
-        ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+        ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
     }
 };
 
diff --git a/src/mongo/db/pipeline/document_source_limit_test.cpp b/src/mongo/db/pipeline/document_source_limit_test.cpp
index 67d4954274c..85ccf9a811c 100644
--- a/src/mongo/db/pipeline/document_source_limit_test.cpp
+++ b/src/mongo/db/pipeline/document_source_limit_test.cpp
@@ -116,7 +116,7 @@ TEST_F(DocumentSourceLimitTest, ShouldNotIntroduceAnyDependencies) {
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, limit->getDependencies(&dependencies));
     ASSERT_EQUALS(0U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceLimitTest, ShouldPropagatePauses) {
diff --git a/src/mongo/db/pipeline/document_source_lookup.cpp b/src/mongo/db/pipeline/document_source_lookup.cpp
index 575d3335ba4..e11a44fcf32 100644
--- a/src/mongo/db/pipeline/document_source_lookup.cpp
+++ b/src/mongo/db/pipeline/document_source_lookup.cpp
@@ -731,7 +731,7 @@ DepsTracker::State DocumentSourceLookUp::getDependencies(DepsTracker* deps) cons
         // subpipeline for the top-level pipeline. So without knowledge of what metadata is in fact
         // available, we "lie" and say that all metadata is available to avoid tripping any
         // assertions.
-        DepsTracker subDeps(DepsTracker::kAllMetadataAvailable);
+        DepsTracker subDeps(DepsTracker::kAllMetadata);
 
         // Get the subpipeline dependencies. Subpipeline stages may reference both 'let' variables
         // declared by this $lookup and variables declared externally.
diff --git a/src/mongo/db/pipeline/document_source_match.cpp b/src/mongo/db/pipeline/document_source_match.cpp
index 44697476cf1..494b3535879 100644
--- a/src/mongo/db/pipeline/document_source_match.cpp
+++ b/src/mongo/db/pipeline/document_source_match.cpp
@@ -486,7 +486,7 @@ DepsTracker::State DocumentSourceMatch::getDependencies(DepsTracker* deps) const
         // A $text aggregation field should return EXHAUSTIVE_FIELDS, since we don't necessarily
         // know what field it will be searching without examining indices.
         deps->needWholeDocument = true;
-        deps->setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
+        deps->setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
         return DepsTracker::State::EXHAUSTIVE_FIELDS;
     }
 
@@ -504,8 +504,9 @@ void DocumentSourceMatch::rebuild(BSONObj filter) {
     _expression = uassertStatusOK(MatchExpressionParser::parse(
         _predicate, pExpCtx, ExtensionsCallbackNoop(), Pipeline::kAllowedMatcherFeatures));
     _isTextQuery = isTextQuery(_predicate);
-    _dependencies = DepsTracker(_isTextQuery ? DepsTracker::MetadataAvailable::kTextScore
-                                             : DepsTracker::MetadataAvailable::kNoMetadata);
+    _dependencies =
+        DepsTracker(_isTextQuery ? QueryMetadataBitSet().set(DocumentMetadataFields::kTextScore)
+                                 : DepsTracker::kNoMetadata);
     getDependencies(&_dependencies);
 }
 
diff --git a/src/mongo/db/pipeline/document_source_match_test.cpp b/src/mongo/db/pipeline/document_source_match_test.cpp
index de5eb9cbc46..3d42a81f91d 100644
--- a/src/mongo/db/pipeline/document_source_match_test.cpp
+++ b/src/mongo/db/pipeline/document_source_match_test.cpp
@@ -220,15 +220,15 @@ TEST_F(DocumentSourceMatchTest, ShouldAddDependenciesOfAllBranchesOfOrClause) {
     ASSERT_EQUALS(1U, dependencies.fields.count("x.y"));
     ASSERT_EQUALS(2U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, TextSearchShouldRequireWholeDocumentAndTextScore) {
     auto match = DocumentSourceMatch::create(fromjson("{$text: {$search: 'hello'} }"), getExpCtx());
-    DepsTracker dependencies(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker dependencies(DepsTracker::kOnlyTextScore);
     ASSERT_EQUALS(DepsTracker::State::EXHAUSTIVE_FIELDS, match->getDependencies(&dependencies));
     ASSERT_EQUALS(true, dependencies.needWholeDocument);
-    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldOnlyAddOuterFieldAsDependencyOfImplicitEqualityPredicate) {
@@ -239,7 +239,7 @@ TEST_F(DocumentSourceMatchTest, ShouldOnlyAddOuterFieldAsDependencyOfImplicitEqu
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldOnlyAddOuterFieldAsDependencyOfClausesWithinElemMatch) {
@@ -250,7 +250,7 @@ TEST_F(DocumentSourceMatchTest, ShouldOnlyAddOuterFieldAsDependencyOfClausesWith
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest,
@@ -267,7 +267,7 @@ TEST_F(DocumentSourceMatchTest,
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest,
@@ -278,7 +278,7 @@ TEST_F(DocumentSourceMatchTest,
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, match->getDependencies(&dependencies));
     ASSERT_EQUALS(0U, dependencies.fields.size());
     ASSERT_EQUALS(true, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest,
@@ -289,7 +289,7 @@ TEST_F(DocumentSourceMatchTest,
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, match->getDependencies(&dependencies1));
     ASSERT_EQUALS(0U, dependencies1.fields.size());
     ASSERT_EQUALS(true, dependencies1.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies1.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies1.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 
     query = fromjson("{a: {$_internalSchemaObjectMatch: {$_internalSchemaMaxProperties: 1}}}");
     match = DocumentSourceMatch::create(query, getExpCtx());
@@ -298,7 +298,7 @@ TEST_F(DocumentSourceMatchTest,
     ASSERT_EQUALS(1U, dependencies2.fields.size());
     ASSERT_EQUALS(1U, dependencies2.fields.count("a"));
     ASSERT_EQUALS(false, dependencies2.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies2.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies2.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest,
@@ -311,7 +311,7 @@ TEST_F(DocumentSourceMatchTest,
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, match->getDependencies(&dependencies));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(true, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest,
@@ -322,7 +322,7 @@ TEST_F(DocumentSourceMatchTest,
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, match->getDependencies(&dependencies));
     ASSERT_EQUALS(0U, dependencies.fields.size());
     ASSERT_EQUALS(true, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithInternalSchemaType) {
@@ -333,7 +333,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithIntern
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithInternalSchemaCond) {
@@ -346,7 +346,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithIntern
     ASSERT_EQUALS(1U, dependencies.fields.count("b"));
     ASSERT_EQUALS(1U, dependencies.fields.count("c"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithInternalSchemaXor) {
@@ -359,7 +359,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithIntern
     ASSERT_EQUALS(1U, dependencies.fields.count("b"));
     ASSERT_EQUALS(1U, dependencies.fields.count("c"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithEmptyJSONSchema) {
@@ -369,7 +369,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithEmptyJ
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, match->getDependencies(&dependencies));
     ASSERT_EQUALS(0U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithJSONSchemaProperties) {
@@ -380,7 +380,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForClausesWithJSONSc
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForMultiplePredicatesWithJSONSchema) {
@@ -392,7 +392,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddCorrectDependenciesForMultiplePredicate
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.count("b"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddOuterFieldToDependenciesIfElemMatchContainsNoFieldNames) {
@@ -403,7 +403,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddOuterFieldToDependenciesIfElemMatchCont
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddNotClausesFieldAsDependency) {
@@ -413,7 +413,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddNotClausesFieldAsDependency) {
     ASSERT_EQUALS(1U, dependencies.fields.count("b"));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ShouldAddDependenciesOfEachNorClause) {
@@ -425,7 +425,7 @@ TEST_F(DocumentSourceMatchTest, ShouldAddDependenciesOfEachNorClause) {
     ASSERT_EQUALS(1U, dependencies.fields.count("b.c"));
     ASSERT_EQUALS(2U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, CommentShouldNotAddAnyDependencies) {
@@ -434,7 +434,7 @@ TEST_F(DocumentSourceMatchTest, CommentShouldNotAddAnyDependencies) {
     ASSERT_EQUALS(DepsTracker::State::SEE_NEXT, match->getDependencies(&dependencies));
     ASSERT_EQUALS(0U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, ClauseAndedWithCommentShouldAddDependencies) {
@@ -445,7 +445,7 @@ TEST_F(DocumentSourceMatchTest, ClauseAndedWithCommentShouldAddDependencies) {
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceMatchTest, MultipleMatchStagesShouldCombineIntoOne) {
diff --git a/src/mongo/db/pipeline/document_source_project_test.cpp b/src/mongo/db/pipeline/document_source_project_test.cpp
index 9480c80602b..9c04482340e 100644
--- a/src/mongo/db/pipeline/document_source_project_test.cpp
+++ b/src/mongo/db/pipeline/document_source_project_test.cpp
@@ -171,7 +171,7 @@ TEST_F(ProjectStageTest, InclusionShouldAddDependenciesOfIncludedAndComputedFiel
         fromjson("{a: true, x: '$b', y: {$and: ['$c','$d']}, z: {$meta: 'textScore'}}"),
         getExpCtx(),
         "$project"_sd);
-    DepsTracker dependencies(DepsTracker::MetadataAvailable::kTextScore);
+    DepsTracker dependencies(DepsTracker::kOnlyTextScore);
     ASSERT_EQUALS(DepsTracker::State::EXHAUSTIVE_FIELDS, project->getDependencies(&dependencies));
     ASSERT_EQUALS(5U, dependencies.fields.size());
 
@@ -188,7 +188,7 @@ TEST_F(ProjectStageTest, InclusionShouldAddDependenciesOfIncludedAndComputedFiel
     ASSERT_EQUALS(1U, dependencies.fields.count("c"));
     ASSERT_EQUALS(1U, dependencies.fields.count("d"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(true, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(ProjectStageTest, ExclusionShouldNotAddDependencies) {
@@ -200,7 +200,7 @@ TEST_F(ProjectStageTest, ExclusionShouldNotAddDependencies) {
 
     ASSERT_EQUALS(0U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(ProjectStageTest, InclusionProjectionReportsIncludedPathsFromGetModifiedPaths) {
@@ -446,7 +446,7 @@ TEST_F(UnsetTest, UnsetShouldNotAddDependencies) {
 
     ASSERT_EQUALS(0U, dependencies.fields.size());
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(UnsetTest, UnsetReportsExcludedPathsAsModifiedPaths) {
diff --git a/src/mongo/db/pipeline/document_source_replace_root_test.cpp b/src/mongo/db/pipeline/document_source_replace_root_test.cpp
index 227331d6897..7fac5040d32 100644
--- a/src/mongo/db/pipeline/document_source_replace_root_test.cpp
+++ b/src/mongo/db/pipeline/document_source_replace_root_test.cpp
@@ -272,7 +272,7 @@ TEST_F(ReplaceRootBasics, OnlyDependentFieldIsNewRoot) {
 
     // Should not need any other fields.
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(ReplaceRootBasics, ReplaceRootModifiesAllFields) {
diff --git a/src/mongo/db/pipeline/document_source_sequential_document_cache.cpp b/src/mongo/db/pipeline/document_source_sequential_document_cache.cpp
index 8cdfba006dd..971971ae19b 100644
--- a/src/mongo/db/pipeline/document_source_sequential_document_cache.cpp
+++ b/src/mongo/db/pipeline/document_source_sequential_document_cache.cpp
@@ -115,7 +115,7 @@ Pipeline::SourceContainer::iterator DocumentSourceSequentialDocumentCache::doOpt
     // elsewhere. So without knowledge of what metadata is in fact available, here
     // we "lie" and say that all metadata is available to avoid tripping any
     // assertions.
-    DepsTracker deps(DepsTracker::kAllMetadataAvailable);
+    DepsTracker deps(DepsTracker::kAllMetadata);
 
     // Iterate through the pipeline stages until we find one which references an external variable.
     for (; prefixSplit != container->end(); ++prefixSplit) {
diff --git a/src/mongo/db/pipeline/document_source_sort.cpp b/src/mongo/db/pipeline/document_source_sort.cpp
index 82b476c791f..d306c4e8248 100644
--- a/src/mongo/db/pipeline/document_source_sort.cpp
+++ b/src/mongo/db/pipeline/document_source_sort.cpp
@@ -158,7 +158,7 @@ DepsTracker::State DocumentSourceSort::getDependencies(DepsTracker* deps) const
     }
     if (pExpCtx->needsMerge) {
         // Include the sort key if we will merge several sorted streams later.
-        deps->setNeedsMetadata(DepsTracker::MetadataType::SORT_KEY, true);
+        deps->setNeedsMetadata(DocumentMetadataFields::kSortKey, true);
     }
 
     return DepsTracker::State::SEE_NEXT;
diff --git a/src/mongo/db/pipeline/document_source_sort_test.cpp b/src/mongo/db/pipeline/document_source_sort_test.cpp
index 67d909c54d2..88e909d326f 100644
--- a/src/mongo/db/pipeline/document_source_sort_test.cpp
+++ b/src/mongo/db/pipeline/document_source_sort_test.cpp
@@ -173,7 +173,7 @@ TEST_F(DocumentSourceSortTest, Dependencies) {
     ASSERT_EQUALS(1U, dependencies.fields.count("a"));
     ASSERT_EQUALS(1U, dependencies.fields.count("b.c"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(DocumentSourceSortTest, ReportsNoPathsModified) {
diff --git a/src/mongo/db/pipeline/document_source_unwind_test.cpp b/src/mongo/db/pipeline/document_source_unwind_test.cpp
index 00d0b3390b4..41d76c428b0 100644
--- a/src/mongo/db/pipeline/document_source_unwind_test.cpp
+++ b/src/mongo/db/pipeline/document_source_unwind_test.cpp
@@ -680,7 +680,7 @@ TEST_F(UnwindStageTest, AddsUnwoundPathToDependencies) {
     ASSERT_EQUALS(1U, dependencies.fields.size());
     ASSERT_EQUALS(1U, dependencies.fields.count("x.y.z"));
     ASSERT_EQUALS(false, dependencies.needWholeDocument);
-    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(UnwindStageTest, ShouldPropagatePauses) {
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index 61abc3f009b..6d307929e3e 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -2634,22 +2634,13 @@ Value ExpressionMeta::evaluate(const Document& root, Variables* variables) const
 }
 
 void ExpressionMeta::_doAddDependencies(DepsTracker* deps) const {
-    if (_metaType == MetaType::kTextScore) {
-        deps->setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
-
+    if (_metaType == MetaType::kSearchScore || _metaType == MetaType::kSearchHighlights) {
         // We do not add the dependencies for SEARCH_SCORE or SEARCH_HIGHLIGHTS because those values
         // are not stored in the collection (or in mongod at all).
-    } else if (_metaType == MetaType::kGeoNearDist) {
-        deps->setNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_DISTANCE, true);
-    } else if (_metaType == MetaType::kGeoNearPoint) {
-        deps->setNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_POINT, true);
-    } else if (_metaType == MetaType::kRecordId) {
-        // TODO: SERVER-42560 handle passing of metadata between PlanStage and DS layers.
-    } else if (_metaType == MetaType::kIndexKey) {
-        // TODO: SERVER-42560 handle passing of metadata between PlanStage and DS layers.
-    } else if (_metaType == MetaType::kSortKey) {
-        deps->setNeedsMetadata(DepsTracker::MetadataType::SORT_KEY, true);
+        return;
     }
+
+    deps->setNeedsMetadata(_metaType, true);
 }
 
 /* ----------------------- ExpressionMod ---------------------------- */
diff --git a/src/mongo/db/pipeline/expression_test.cpp b/src/mongo/db/pipeline/expression_test.cpp
index 0b3b8be1179..859ed6cc22e 100644
--- a/src/mongo/db/pipeline/expression_test.cpp
+++ b/src/mongo/db/pipeline/expression_test.cpp
@@ -262,7 +262,7 @@ protected:
         }
         ASSERT_BSONOBJ_EQ(expectedDependencies, dependenciesBson.arr());
         ASSERT_EQUALS(false, dependencies.needWholeDocument);
-        ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+        ASSERT_EQUALS(false, dependencies.getNeedsAnyMetadata());
     }
 
     void assertContents(const intrusive_ptr<Testable>& expr, const BSONArray& expectedContents) {
@@ -1985,7 +1985,7 @@ public:
         ASSERT_EQUALS(1U, dependencies.fields.size());
         ASSERT_EQUALS(1U, dependencies.fields.count("a.b"));
         ASSERT_EQUALS(false, dependencies.needWholeDocument);
-        ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+        ASSERT_EQUALS(false, dependencies.getNeedsAnyMetadata());
     }
 };
 
@@ -2491,7 +2491,7 @@ public:
         expression->addDependencies(&dependencies);
         ASSERT_EQUALS(0U, dependencies.fields.size());
         ASSERT_EQUALS(false, dependencies.needWholeDocument);
-        ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+        ASSERT_EQUALS(false, dependencies.getNeedsAnyMetadata());
     }
 };
 
@@ -3017,7 +3017,7 @@ public:
         ASSERT_EQUALS(1U, dependencies.fields.size());
         ASSERT_EQUALS(1U, dependencies.fields.count("a.b"));
         ASSERT_EQUALS(false, dependencies.needWholeDocument);
-        ASSERT_EQUALS(false, dependencies.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+        ASSERT_EQUALS(false, dependencies.getNeedsAnyMetadata());
     }
 };
 
diff --git a/src/mongo/db/pipeline/field_path.cpp b/src/mongo/db/pipeline/field_path.cpp
index 2d891ce7b10..6eb9fe46d0b 100644
--- a/src/mongo/db/pipeline/field_path.cpp
+++ b/src/mongo/db/pipeline/field_path.cpp
@@ -47,17 +47,12 @@ const StringDataSet kAllowedDollarPrefixedFields = {
 
     // Metadata fields.
 
-    // TODO SERVER-42560: It may be possible to eliminate some of these, if they're only used for
-    // creating the "dependency" projection. Some of them ($dis and $sortKey) may be used in
-    // sharded queries and are necessary.
+    // This is necessary for sharded query execution of find() commands. mongos may attach a
+    // $sortKey field to the projection sent to shards so that it can merge the results correctly.
     "$sortKey",
-    "$pt",
-    "$dis",
-    "$textScore",
-    "$recordId",
 
-    // Used internally for forcing projections to be of a certain type.
-    "$__INTERNAL_QUERY_PROJECTION_RESERVED"};
+    // This is necessary for the "showRecordId" feature.
+    "$recordId"};
 
 }  // namespace
 
diff --git a/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp b/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp
index 1e1b4b4256e..cc7271be016 100644
--- a/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp
+++ b/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp
@@ -149,7 +149,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldNotAddAnyDependencies) {
 
     ASSERT_EQ(deps.fields.size(), 0UL);
     ASSERT_FALSE(deps.needWholeDocument);
-    ASSERT_FALSE(deps.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_FALSE(deps.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldReportExcludedFieldsAsModified) {
diff --git a/src/mongo/db/pipeline/pipeline.cpp b/src/mongo/db/pipeline/pipeline.cpp
index 7642b1ef49a..7e9fa39c66d 100644
--- a/src/mongo/db/pipeline/pipeline.cpp
+++ b/src/mongo/db/pipeline/pipeline.cpp
@@ -501,7 +501,7 @@ void Pipeline::addFinalSource(intrusive_ptr<DocumentSource> source) {
     _sources.push_back(source);
 }
 
-DepsTracker Pipeline::getDependencies(DepsTracker::MetadataAvailable metadataAvailable) const {
+DepsTracker Pipeline::getDependencies(QueryMetadataBitSet metadataAvailable) const {
     DepsTracker deps(metadataAvailable);
     const bool scopeHasVariables = pCtx->variablesParseState.hasDefinedVariables();
     bool skipFieldsAndMetadataDeps = false;
@@ -533,9 +533,7 @@ DepsTracker Pipeline::getDependencies(DepsTracker::MetadataAvailable metadataAva
         }
 
         if (!knowAllMeta) {
-            for (auto&& req : localDeps.getAllRequiredMetadataTypes()) {
-                deps.setNeedsMetadata(req, true);
-            }
+            deps.requestMetadata(localDeps.metadataDeps());
             knowAllMeta = status & DepsTracker::State::EXHAUSTIVE_META;
         }
 
@@ -549,15 +547,15 @@ DepsTracker Pipeline::getDependencies(DepsTracker::MetadataAvailable metadataAva
     if (!knowAllFields)
         deps.needWholeDocument = true;  // don't know all fields we need
 
-    if (metadataAvailable & DepsTracker::MetadataAvailable::kTextScore) {
+    if (metadataAvailable[DocumentMetadataFields::kTextScore]) {
         // If there is a text score, assume we need to keep it if we can't prove we don't. If we are
         // the first half of a pipeline which has been split, future stages might need it.
         if (!knowAllMeta) {
-            deps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
+            deps.setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
         }
     } else {
         // If there is no text score available, then we don't need to ask for it.
-        deps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, false);
+        deps.setNeedsMetadata(DocumentMetadataFields::kTextScore, false);
     }
 
     return deps;
diff --git a/src/mongo/db/pipeline/pipeline.h b/src/mongo/db/pipeline/pipeline.h
index 1af188f4b15..39ba6372687 100644
--- a/src/mongo/db/pipeline/pipeline.h
+++ b/src/mongo/db/pipeline/pipeline.h
@@ -251,7 +251,7 @@ public:
      * Returns the dependencies needed by this pipeline. 'metadataAvailable' should reflect what
      * metadata is present on documents that are input to the front of the pipeline.
      */
-    DepsTracker getDependencies(DepsTracker::MetadataAvailable metadataAvailable) const;
+    DepsTracker getDependencies(QueryMetadataBitSet metadataAvailable) const;
 
     const SourceContainer& getSources() const {
         return _sources;
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index c327187ecd3..a523066d2a8 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -193,6 +193,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
     const intrusive_ptr<ExpressionContext>& pExpCtx,
     BSONObj queryObj,
     BSONObj projectionObj,
+    const QueryMetadataBitSet& metadataRequested,
     BSONObj sortObj,
     boost::optional<std::string> groupIdForDistinctScan,
     const AggregationRequest* aggRequest,
@@ -232,6 +233,9 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
         return {cq.getStatus()};
     }
 
+    // Mark the metadata that's requested by the pipeline on the CQ.
+    cq.getValue()->requestAdditionalMetadata(metadataRequested);
+
     if (groupIdForDistinctScan) {
         // When the pipeline includes a $group that groups by a single field
         // (groupIdForDistinctScan), we use getExecutorDistinct() to attempt to get an executor that
@@ -266,13 +270,6 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
     return getExecutorFind(opCtx, collection, std::move(cq.getValue()), permitYield, plannerOpts);
 }
 
-BSONObj removeSortKeyMetaProjection(BSONObj projectionObj) {
-    if (!projectionObj[Document::metaFieldSortKey]) {
-        return projectionObj;
-    }
-    return projectionObj.removeField(Document::metaFieldSortKey);
-}
-
 /**
  * Examines the indexes in 'collection' and returns the field name of a geo-indexed field suitable
  * for use in $geoNear. 2d indexes are given priority over 2dsphere indexes.
@@ -366,7 +363,7 @@ PipelineD::buildInnerQueryExecutor(Collection* collection,
 
                 // TODO SERVER-37453 this should no longer be necessary when we no don't need locks
                 // to destroy a PlanExecutor.
-                auto deps = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+                auto deps = pipeline->getDependencies(DepsTracker::kNoMetadata);
                 auto attachExecutorCallback =
                     [deps](Collection* collection,
                            std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> exec,
@@ -479,10 +476,10 @@ PipelineD::buildInnerQueryExecutorGeneric(Collection* collection,
 
     // Find the set of fields in the source documents depended on by this pipeline.
     DepsTracker deps = pipeline->getDependencies(DocumentSourceMatch::isTextQuery(queryObj)
-                                                     ? DepsTracker::MetadataAvailable::kTextScore
-                                                     : DepsTracker::MetadataAvailable::kNoMetadata);
+                                                     ? DepsTracker::kOnlyTextScore
+                                                     : DepsTracker::kNoMetadata);
 
-    BSONObj projForQuery = deps.toProjection();
+    BSONObj projForQuery = deps.toProjectionWithoutMetadata();
 
     boost::intrusive_ptr<DocumentSourceSort> sortStage;
     boost::intrusive_ptr<DocumentSourceGroup> groupStage;
@@ -536,8 +533,7 @@ PipelineD::buildInnerQueryExecutorGeneric(Collection* collection,
                                       Pipeline* pipeline) {
         auto cursor = DocumentSourceCursor::create(
             collection, std::move(exec), pipeline->getContext(), trackOplogTS);
-        addCursorSource(
-            pipeline, std::move(cursor), std::move(deps), queryObj, sortObj, projForQuery);
+        addCursorSource(pipeline, std::move(cursor), std::move(deps), queryObj, sortObj);
     };
     return std::make_pair(std::move(attachExecutorCallback), std::move(exec));
 }
@@ -557,7 +553,7 @@ PipelineD::buildInnerQueryExecutorGeoNear(Collection* collection,
     const auto geoNearStage = dynamic_cast<DocumentSourceGeoNear*>(sources.front().get());
     invariant(geoNearStage);
 
-    auto deps = pipeline->getDependencies(DepsTracker::kAllGeoNearDataAvailable);
+    auto deps = pipeline->getDependencies(DepsTracker::kAllGeoNearData);
 
     // If the user specified a "key" field, use that field to satisfy the "near" query. Otherwise,
     // look for a geo-indexed field in 'collection' that can.
@@ -569,7 +565,7 @@ PipelineD::buildInnerQueryExecutorGeoNear(Collection* collection,
     // Create a PlanExecutor whose query is the "near" predicate on 'nearFieldName' combined with
     // the optional "query" argument in the $geoNear stage.
     BSONObj fullQuery = geoNearStage->asNearQuery(nearFieldName);
-    BSONObj proj = deps.toProjection();
+    BSONObj proj = deps.toProjectionWithoutMetadata();
     BSONObj sortFromQuerySystem;
     auto exec = uassertStatusOK(prepareExecutor(expCtx->opCtx,
                                                 collection,
@@ -627,7 +623,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
     // The query system has the potential to use an index to provide a non-blocking sort and/or to
     // use the projection to generate a covered plan. If this is possible, it is more efficient to
     // let the query system handle those parts of the pipeline. If not, it is more efficient to use
-    // a $sort and/or a ParsedDeps object. Thus, we will determine whether the query system can
+    // a $sort and/or a $project. Thus, we will determine whether the query system can
     // provide a non-blocking sort or a covered projection before we commit to a PlanExecutor.
     //
     // To determine if the query system can provide a non-blocking sort, we pass the
@@ -674,6 +670,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
                                                       expCtx,
                                                       queryObj,
                                                       *projectionObj,
+                                                      deps.metadataDeps(),
                                                       sortObj ? *sortObj : emptySort,
                                                       rewrittenGroupStage->groupId(),
                                                       aggRequest,
@@ -711,16 +708,17 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
     const BSONObj metaSortProjection = BSON("$sortKey" << BSON("$meta"
                                                                << "sortKey"));
 
-    // The only way to get meta information (e.g. the text score) is to let the query system handle
-    // the projection. In all other cases, unless the query system can do an index-covered
-    // projection and avoid going to the raw record at all, it is faster to have ParsedDeps filter
-    // the fields we need.
+    // TODO SERVER-42905: It should be possible to push down all eligible projections to the query
+    // layer.  This code assumes that metadata is passed from the query layer to the DocumentSource
+    // layer via a projection, which is no longer true.
     if (!deps.getNeedsAnyMetadata()) {
         plannerOpts |= QueryPlannerParams::NO_UNCOVERED_PROJECTIONS;
     }
 
     SortPattern userSortPattern(*sortObj, expCtx);
     if (sortStage && canSortBePushedDown(userSortPattern)) {
+        QueryMetadataBitSet needsSortKey;
+        needsSortKey.set(DocumentMetadataFields::MetaType::kSortKey);
         // See if the query system can provide a non-blocking sort.
         auto swExecutorSort =
             attemptToGetExecutor(opCtx,
@@ -728,7 +726,8 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
                                  nss,
                                  expCtx,
                                  queryObj,
-                                 expCtx->needsMerge ? metaSortProjection : emptyProjection,
+                                 BSONObj(),  // empty projection
+                                 expCtx->needsMerge ? needsSortKey : DepsTracker::kNoMetadata,
                                  *sortObj,
                                  boost::none, /* groupIdForDistinctScan */
                                  aggRequest,
@@ -744,6 +743,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
                                      expCtx,
                                      queryObj,
                                      *projectionObj,
+                                     deps.metadataDeps(),
                                      *sortObj,
                                      boost::none, /* groupIdForDistinctScan */
                                      aggRequest,
@@ -780,13 +780,15 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
         }
     }
 
-    // Either there's no sort or the query system can't provide a non-blocking sort.
+    // Either there was no $sort stage, or the query system could not provide a non-blocking
+    // sort.
     *sortObj = BSONObj();
 
-    *projectionObj = removeSortKeyMetaProjection(*projectionObj);
-    const auto metadataRequired = deps.getAllRequiredMetadataTypes();
-    if (metadataRequired.size() == 1 &&
-        metadataRequired.front() == DepsTracker::MetadataType::SORT_KEY) {
+    // Since the DocumentSource layer will perform the sort, remove any dependencies we have on the
+    // query layer for a sort key.
+    QueryMetadataBitSet metadataDepsWithoutSortKey = deps.metadataDeps();
+    metadataDepsWithoutSortKey[DocumentMetadataFields::kSortKey] = false;
+    if (!metadataDepsWithoutSortKey.any()) {
         // A sort key requirement would have prevented us from being able to add this parameter
         // before, but now we know the query system won't cover the sort, so we will be able to
         // compute the sort key ourselves during the $sort stage, and thus don't need a query
@@ -801,6 +803,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
                                                expCtx,
                                                queryObj,
                                                *projectionObj,
+                                               metadataDepsWithoutSortKey,
                                                *sortObj,
                                                boost::none, /* groupIdForDistinctScan */
                                                aggRequest,
@@ -814,7 +817,8 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
             "Failed to determine whether query system can provide a covered projection");
     }
 
-    // The query system couldn't provide a covered or simple uncovered projection.
+    // The query system couldn't provide a covered or simple uncovered projection. Do no projections
+    // and request no metadata from the query layer.
     *projectionObj = BSONObj();
     // If this doesn't work, nothing will.
     return attemptToGetExecutor(opCtx,
@@ -823,6 +827,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
                                 expCtx,
                                 queryObj,
                                 *projectionObj,
+                                DepsTracker::kNoMetadata,
                                 *sortObj,
                                 boost::none, /* groupIdForDistinctScan */
                                 aggRequest,
@@ -834,8 +839,7 @@ void PipelineD::addCursorSource(Pipeline* pipeline,
                                 boost::intrusive_ptr<DocumentSourceCursor> cursor,
                                 DepsTracker deps,
                                 const BSONObj& queryObj,
-                                const BSONObj& sortObj,
-                                const BSONObj& projectionObj) {
+                                const BSONObj& sortObj) {
     // Add the cursor to the pipeline first so that it's correctly disposed of as part of the
     // pipeline if an exception is thrown during this method.
     pipeline->addInitialSource(cursor);
@@ -845,19 +849,6 @@ void PipelineD::addCursorSource(Pipeline* pipeline,
     if (deps.hasNoRequirements()) {
         cursor->shouldProduceEmptyDocs();
     }
-
-    if (!projectionObj.isEmpty()) {
-        cursor->setProjection(projectionObj, boost::none, deps.getNeedsAnyMetadata());
-    } else {
-        // There may be fewer dependencies now if the sort was covered.
-        if (!sortObj.isEmpty()) {
-            deps = pipeline->getDependencies(DocumentSourceMatch::isTextQuery(queryObj)
-                                                 ? DepsTracker::MetadataAvailable::kTextScore
-                                                 : DepsTracker::MetadataAvailable::kNoMetadata);
-        }
-
-        cursor->setProjection(deps.toProjection(), deps.toParsedDeps(), deps.getNeedsAnyMetadata());
-    }
 }
 
 Timestamp PipelineD::getLatestOplogTimestamp(const Pipeline* pipeline) {
diff --git a/src/mongo/db/pipeline/pipeline_d.h b/src/mongo/db/pipeline/pipeline_d.h
index 606c91a1067..9433198c4d9 100644
--- a/src/mongo/db/pipeline/pipeline_d.h
+++ b/src/mongo/db/pipeline/pipeline_d.h
@@ -202,8 +202,7 @@ private:
                                 boost::intrusive_ptr<DocumentSourceCursor> cursor,
                                 DepsTracker deps,
                                 const BSONObj& queryObj = BSONObj(),
-                                const BSONObj& sortObj = BSONObj(),
-                                const BSONObj& projectionObj = BSONObj());
+                                const BSONObj& sortObj = BSONObj());
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/pipeline_test.cpp b/src/mongo/db/pipeline/pipeline_test.cpp
index cc92f5958f2..3da457141dd 100644
--- a/src/mongo/db/pipeline/pipeline_test.cpp
+++ b/src/mongo/db/pipeline/pipeline_test.cpp
@@ -2793,13 +2793,13 @@ using PipelineDependenciesTest = AggregationContextFixture;
 TEST_F(PipelineDependenciesTest, EmptyPipelineShouldRequireWholeDocument) {
     auto pipeline = unittest::assertGet(Pipeline::create({}, getExpCtx()));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_TRUE(depsTracker.needWholeDocument);
-    ASSERT_FALSE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_FALSE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 
-    depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kTextScore);
+    depsTracker = pipeline->getDependencies(DepsTracker::kOnlyTextScore);
     ASSERT_TRUE(depsTracker.needWholeDocument);
-    ASSERT_TRUE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_TRUE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 //
@@ -2861,7 +2861,7 @@ public:
 class DocumentSourceNeedsOnlyTextScore : public DocumentSourceDependencyDummy {
 public:
     DepsTracker::State getDependencies(DepsTracker* deps) const final {
-        deps->setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, true);
+        deps->setNeedsMetadata(DocumentMetadataFields::kTextScore, true);
         return DepsTracker::State::EXHAUSTIVE_META;
     }
 
@@ -2887,15 +2887,15 @@ TEST_F(PipelineDependenciesTest, ShouldRequireWholeDocumentIfAnyStageDoesNotSupp
     auto notSupported = DocumentSourceDependenciesNotSupported::create();
     auto pipeline = unittest::assertGet(Pipeline::create({needsASeeNext, notSupported}, ctx));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_TRUE(depsTracker.needWholeDocument);
     // The inputs did not have a text score available, so we should not require a text score.
-    ASSERT_FALSE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_FALSE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 
     // Now in the other order.
     pipeline = unittest::assertGet(Pipeline::create({notSupported, needsASeeNext}, ctx));
 
-    depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    depsTracker = pipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_TRUE(depsTracker.needWholeDocument);
 }
 
@@ -2904,7 +2904,7 @@ TEST_F(PipelineDependenciesTest, ShouldRequireWholeDocumentIfNoStageReturnsExhau
     auto needsASeeNext = DocumentSourceNeedsASeeNext::create();
     auto pipeline = unittest::assertGet(Pipeline::create({needsASeeNext}, ctx));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_TRUE(depsTracker.needWholeDocument);
 }
 
@@ -2914,7 +2914,7 @@ TEST_F(PipelineDependenciesTest, ShouldNotRequireWholeDocumentIfAnyStageReturnsE
     auto needsOnlyB = DocumentSourceNeedsOnlyB::create();
     auto pipeline = unittest::assertGet(Pipeline::create({needsASeeNext, needsOnlyB}, ctx));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_FALSE(depsTracker.needWholeDocument);
     ASSERT_EQ(depsTracker.fields.size(), 2UL);
     ASSERT_EQ(depsTracker.fields.count("a"), 1UL);
@@ -2927,9 +2927,9 @@ TEST_F(PipelineDependenciesTest, ShouldNotAddAnyRequiredFieldsAfterFirstStageWit
     auto needsASeeNext = DocumentSourceNeedsASeeNext::create();
     auto pipeline = unittest::assertGet(Pipeline::create({needsOnlyB, needsASeeNext}, ctx));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kNoMetadata);
     ASSERT_FALSE(depsTracker.needWholeDocument);
-    ASSERT_FALSE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_FALSE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 
     // 'needsOnlyB' claims to know all its field dependencies, so we shouldn't add any from
     // 'needsASeeNext'.
@@ -2941,8 +2941,8 @@ TEST_F(PipelineDependenciesTest, ShouldNotRequireTextScoreIfThereIsNoScoreAvaila
     auto ctx = getExpCtx();
     auto pipeline = unittest::assertGet(Pipeline::create({}, ctx));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata);
-    ASSERT_FALSE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kNoMetadata);
+    ASSERT_FALSE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(PipelineDependenciesTest, ShouldThrowIfTextScoreIsNeededButNotPresent) {
@@ -2950,21 +2950,20 @@ TEST_F(PipelineDependenciesTest, ShouldThrowIfTextScoreIsNeededButNotPresent) {
     auto needsText = DocumentSourceNeedsOnlyTextScore::create();
     auto pipeline = unittest::assertGet(Pipeline::create({needsText}, ctx));
 
-    ASSERT_THROWS(pipeline->getDependencies(DepsTracker::MetadataAvailable::kNoMetadata),
-                  AssertionException);
+    ASSERT_THROWS(pipeline->getDependencies(DepsTracker::kNoMetadata), AssertionException);
 }
 
 TEST_F(PipelineDependenciesTest, ShouldRequireTextScoreIfAvailableAndNoStageReturnsExhaustiveMeta) {
     auto ctx = getExpCtx();
     auto pipeline = unittest::assertGet(Pipeline::create({}, ctx));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kTextScore);
-    ASSERT_TRUE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kOnlyTextScore);
+    ASSERT_TRUE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 
     auto needsASeeNext = DocumentSourceNeedsASeeNext::create();
     pipeline = unittest::assertGet(Pipeline::create({needsASeeNext}, ctx));
-    depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kTextScore);
-    ASSERT_TRUE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    depsTracker = pipeline->getDependencies(DepsTracker::kOnlyTextScore);
+    ASSERT_TRUE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 TEST_F(PipelineDependenciesTest, ShouldNotRequireTextScoreIfAvailableButDefinitelyNotNeeded) {
@@ -2973,11 +2972,11 @@ TEST_F(PipelineDependenciesTest, ShouldNotRequireTextScoreIfAvailableButDefinite
     auto needsText = DocumentSourceNeedsOnlyTextScore::create();
     auto pipeline = unittest::assertGet(Pipeline::create({stripsTextScore, needsText}, ctx));
 
-    auto depsTracker = pipeline->getDependencies(DepsTracker::MetadataAvailable::kTextScore);
+    auto depsTracker = pipeline->getDependencies(DepsTracker::kOnlyTextScore);
 
     // 'stripsTextScore' claims that no further stage will need metadata information, so we
     // shouldn't have the text score as a dependency.
-    ASSERT_FALSE(depsTracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE));
+    ASSERT_FALSE(depsTracker.getNeedsMetadata(DocumentMetadataFields::kTextScore));
 }
 
 }  // namespace Dependencies
diff --git a/src/mongo/db/query/canonical_query.cpp b/src/mongo/db/query/canonical_query.cpp
index d3590e9c722..fdf06b7c849 100644
--- a/src/mongo/db/query/canonical_query.cpp
+++ b/src/mongo/db/query/canonical_query.cpp
@@ -271,9 +271,12 @@ Status CanonicalQuery::init(OperationContext* opCtx,
         if (!newParserStatus.isOK()) {
             return newParserStatus;
         }
+
+        _metadataDeps = _proj->metadataDeps();
     }
 
-    if (_proj && _proj->wantSortKey() && _qr->getSort().isEmpty()) {
+    if (_proj && _proj->metadataDeps()[DocumentMetadataFields::kSortKey] &&
+        _qr->getSort().isEmpty()) {
         return Status(ErrorCodes::BadValue, "cannot use sortKey $meta projection without a sort");
     }
 
diff --git a/src/mongo/db/query/canonical_query.h b/src/mongo/db/query/canonical_query.h
index 3cacc3b4c9d..39ed4e927f5 100644
--- a/src/mongo/db/query/canonical_query.h
+++ b/src/mongo/db/query/canonical_query.h
@@ -133,6 +133,20 @@ public:
     }
 
     /**
+     * Returns a bitset indicating what metadata has been requested in the query.
+     */
+    const QueryMetadataBitSet& metadataDeps() const {
+        return _metadataDeps;
+    }
+
+    /**
+     * Allows callers to request metadata in addition to that needed as part of the query.
+     */
+    void requestAdditionalMetadata(const QueryMetadataBitSet& additionalDeps) {
+        _metadataDeps |= additionalDeps;
+    }
+
+    /**
      * Compute the "shape" of this query by encoding the match, projection and sort, and stripping
      * out the appropriate values.
      */
@@ -210,6 +224,9 @@ private:
 
     boost::optional<projection_ast::Projection> _proj;
 
+    // Keeps track of what metadata has been explicitly requested.
+    QueryMetadataBitSet _metadataDeps;
+
     std::unique_ptr<CollatorInterface> _collator;
 
     bool _canHaveNoopMatchNodes = false;
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index bf2d41c52a8..12b5f4073ee 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -181,9 +181,11 @@ void generateBatch(int ntoreturn,
                    PlanExecutor::ExecState* state) {
     PlanExecutor* exec = cursor->getExecutor();
 
-    BSONObj obj;
+    Document doc;
     while (!FindCommon::enoughForGetMore(ntoreturn, *numResults) &&
-           PlanExecutor::ADVANCED == (*state = exec->getNext(&obj, nullptr))) {
+           PlanExecutor::ADVANCED == (*state = exec->getNext(&doc, nullptr))) {
+        BSONObj obj = doc.toBson();
+
         // If we can't fit this result inside the current batch, then we stash it for later.
         if (!FindCommon::haveSpaceForNext(obj, *numResults, bb->len())) {
             exec->enqueue(obj);
@@ -204,7 +206,7 @@ void generateBatch(int ntoreturn,
             error() << "getMore executor error, stats: "
                     << redact(Explain::getWinningPlanStats(exec));
             // We should always have a valid status object by this point.
-            auto status = WorkingSetCommon::getMemberObjectStatus(obj);
+            auto status = WorkingSetCommon::getMemberObjectStatus(doc);
             invariant(!status.isOK());
             uassertStatusOK(status);
         }
@@ -684,7 +686,10 @@ std::string runQuery(OperationContext* opCtx,
         curOp.setPlanSummary_inlock(Explain::getPlanSummary(exec.get()));
     }
 
-    while (PlanExecutor::ADVANCED == (state = exec->getNext(&obj, nullptr))) {
+    Document doc;
+    while (PlanExecutor::ADVANCED == (state = exec->getNext(&doc, nullptr))) {
+        obj = doc.toBson();
+
         // If we can't fit this result inside the current batch, then we stash it for later.
         if (!FindCommon::haveSpaceForNext(obj, numResults, bb.len())) {
             exec->enqueue(obj);
@@ -709,7 +714,7 @@ std::string runQuery(OperationContext* opCtx,
     if (PlanExecutor::FAILURE == state) {
         error() << "Plan executor error during find: " << PlanExecutor::statestr(state)
                 << ", stats: " << redact(Explain::getWinningPlanStats(exec.get()));
-        uassertStatusOKWithContext(WorkingSetCommon::getMemberObjectStatus(obj),
+        uassertStatusOKWithContext(WorkingSetCommon::getMemberObjectStatus(doc),
                                    "Executor error during OP_QUERY find");
         MONGO_UNREACHABLE;
     }
@@ -727,14 +732,17 @@ std::string runQuery(OperationContext* opCtx,
         // Allocate a new ClientCursor and register it with the cursor manager.
         ClientCursorPin pinnedCursor = CursorManager::get(opCtx)->registerCursor(
             opCtx,
-            {std::move(exec),
-             nss,
-             AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
-             opCtx->getWriteConcern(),
-             readConcernArgs,
-             upconvertedQuery,
-             ClientCursorParams::LockPolicy::kLockExternally,
-             {Privilege(ResourcePattern::forExactNamespace(nss), ActionType::find)}});
+            {
+                std::move(exec),
+                nss,
+                AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
+                opCtx->getWriteConcern(),
+                readConcernArgs,
+                upconvertedQuery,
+                ClientCursorParams::LockPolicy::kLockExternally,
+                {Privilege(ResourcePattern::forExactNamespace(nss), ActionType::find)},
+                false  // needsMerge always 'false' for find().
+            });
         ccId = pinnedCursor.getCursor()->cursorid();
 
         LOG(5) << "caching executor with cursorid " << ccId << " after returning " << numResults
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index 2a2a3559040..29cc73c097a 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -390,8 +390,8 @@ StatusWith<PrepareExecutionResult> prepareExecution(OperationContext* opCtx,
                 std::move(root));
         }
 
-        // Add a SortKeyGeneratorStage if there is a $meta sortKey projection.
-        if (canonicalQuery->getProj() && canonicalQuery->getProj()->wantSortKey()) {
+        // Add a SortKeyGeneratorStage if the query requested sortKey metadata.
+        if (canonicalQuery->metadataDeps()[DocumentMetadataFields::kSortKey]) {
             root = std::make_unique<SortKeyGeneratorStage>(
                 canonicalQuery->getExpCtx(),
                 std::move(root),
@@ -675,8 +675,10 @@ StatusWith<unique_ptr<PlanStage>> applyProjection(OperationContext* opCtx,
                 "cannot use a positional projection and return the new document"};
     }
 
+    cq->requestAdditionalMetadata(proj.metadataDeps());
+
     // $meta sortKey is not allowed to be projected in findAndModify commands.
-    if (proj.wantSortKey()) {
+    if (cq->metadataDeps()[DocumentMetadataFields::kSortKey]) {
         return {ErrorCodes::BadValue,
                 "Cannot use a $meta sortKey projection in findAndModify commands."};
     }
diff --git a/src/mongo/db/query/plan_executor.h b/src/mongo/db/query/plan_executor.h
index a0bf2344a6b..178652b2594 100644
--- a/src/mongo/db/query/plan_executor.h
+++ b/src/mongo/db/query/plan_executor.h
@@ -289,6 +289,11 @@ public:
      */
     virtual OperationContext* getOpCtx() const = 0;
 
+    /**
+     * Return the ExpressionContext that the plan is currently executing with.
+     */
+    virtual const boost::intrusive_ptr<ExpressionContext>& getExpCtx() const = 0;
+
     //
     // Methods that just pass down to the PlanStage tree.
     //
@@ -352,10 +357,19 @@ public:
      * For write operations, the return depends on the particulars of the write stage.
      *
      * If a YIELD_AUTO policy is set, then this method may yield.
+     *
+     * The Documents returned by this method may not be owned. If the caller wants to ensure a
+     * returned Document is preserved across a yield, getOwned() should be called.
      */
+    virtual ExecState getNextSnapshotted(Snapshotted<Document>* objOut, RecordId* dlOut) = 0;
     virtual ExecState getNextSnapshotted(Snapshotted<BSONObj>* objOut, RecordId* dlOut) = 0;
 
-    virtual ExecState getNext(BSONObj* objOut, RecordId* dlOut) = 0;
+    virtual ExecState getNext(Document* objOut, RecordId* dlOut) = 0;
+
+    /**
+     * Will perform the Document -> BSON conversion for the caller.
+     */
+    virtual ExecState getNext(BSONObj* out, RecordId* dlOut) = 0;
 
     /**
      * Returns 'true' if the plan is done producing results (or writing), 'false' otherwise.
@@ -421,6 +435,7 @@ public:
      * If used in combination with getNextSnapshotted(), then the SnapshotId associated with
      * 'obj' will be null when 'obj' is dequeued.
      */
+    virtual void enqueue(const Document& obj) = 0;
     virtual void enqueue(const BSONObj& obj) = 0;
 
     virtual bool isMarkedAsKilled() const = 0;
@@ -442,8 +457,9 @@ public:
     virtual BSONObj getPostBatchResumeToken() const = 0;
 
     /**
-     * Turns a BSONObj representing an error status produced by getNext() into a Status.
+     * Turns a Document representing an error status produced by getNext() into a Status.
      */
+    virtual Status getMemberObjectStatus(const Document& memberObj) const = 0;
     virtual Status getMemberObjectStatus(const BSONObj& memberObj) const = 0;
 };
 
diff --git a/src/mongo/db/query/plan_executor_impl.cpp b/src/mongo/db/query/plan_executor_impl.cpp
index 3d0629d5d30..46807db4359 100644
--- a/src/mongo/db/query/plan_executor_impl.cpp
+++ b/src/mongo/db/query/plan_executor_impl.cpp
@@ -324,6 +324,10 @@ OperationContext* PlanExecutorImpl::getOpCtx() const {
     return _opCtx;
 }
 
+const boost::intrusive_ptr<ExpressionContext>& PlanExecutorImpl::getExpCtx() const {
+    return _expCtx;
+}
+
 void PlanExecutorImpl::saveState() {
     invariant(_currentState == kUsable || _currentState == kSaved);
 
@@ -377,7 +381,16 @@ void PlanExecutorImpl::reattachToOperationContext(OperationContext* opCtx) {
 }
 
 PlanExecutor::ExecState PlanExecutorImpl::getNext(BSONObj* objOut, RecordId* dlOut) {
-    Snapshotted<BSONObj> snapshotted;
+    Document doc;
+    const auto state = getNext(&doc, dlOut);
+    if (objOut) {
+        *objOut = doc.toBson();
+    }
+    return state;
+}
+
+PlanExecutor::ExecState PlanExecutorImpl::getNext(Document* objOut, RecordId* dlOut) {
+    Snapshotted<Document> snapshotted;
     ExecState state = _getNextImpl(objOut ? &snapshotted : nullptr, dlOut);
 
     if (objOut) {
@@ -387,13 +400,25 @@ PlanExecutor::ExecState PlanExecutorImpl::getNext(BSONObj* objOut, RecordId* dlO
     return state;
 }
 
-PlanExecutor::ExecState PlanExecutorImpl::getNextSnapshotted(Snapshotted<BSONObj>* objOut,
+PlanExecutor::ExecState PlanExecutorImpl::getNextSnapshotted(Snapshotted<Document>* objOut,
                                                              RecordId* dlOut) {
     // Detaching from the OperationContext means that the returned snapshot ids could be invalid.
     invariant(!_everDetachedFromOperationContext);
     return _getNextImpl(objOut, dlOut);
 }
 
+PlanExecutor::ExecState PlanExecutorImpl::getNextSnapshotted(Snapshotted<BSONObj>* objOut,
+                                                             RecordId* dlOut) {
+    // Detaching from the OperationContext means that the returned snapshot ids could be invalid.
+    invariant(!_everDetachedFromOperationContext);
+    Snapshotted<Document> docOut;
+    const auto status = _getNextImpl(&docOut, dlOut);
+    if (objOut) {
+        *objOut = {docOut.snapshotId(), docOut.value().toBson()};
+    }
+    return status;
+}
+
 bool PlanExecutorImpl::_shouldListenForInserts() {
     return _cq && _cq->getQueryRequest().isTailableAndAwaitData() &&
         awaitDataState(_opCtx).shouldWaitForInserts && _opCtx->checkForInterruptNoAssert().isOK() &&
@@ -438,7 +463,7 @@ std::shared_ptr<CappedInsertNotifier> PlanExecutorImpl::_getCappedInsertNotifier
 }
 
 PlanExecutor::ExecState PlanExecutorImpl::_waitForInserts(CappedInsertNotifierData* notifierData,
-                                                          Snapshotted<BSONObj>* errorObj) {
+                                                          Snapshotted<Document>* errorObj) {
     invariant(notifierData->notifier);
 
     // The notifier wait() method will not wait unless the version passed to it matches the
@@ -463,19 +488,19 @@ PlanExecutor::ExecState PlanExecutorImpl::_waitForInserts(CappedInsertNotifierDa
     }
 
     if (errorObj) {
-        *errorObj = Snapshotted<BSONObj>(SnapshotId(),
-                                         WorkingSetCommon::buildMemberStatusObject(yieldResult));
+        *errorObj = Snapshotted<Document>(SnapshotId(),
+                                          WorkingSetCommon::buildMemberStatusObject(yieldResult));
     }
     return FAILURE;
 }
 
-PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<BSONObj>* objOut,
+PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<Document>* objOut,
                                                        RecordId* dlOut) {
     if (MONGO_unlikely(planExecutorAlwaysFails.shouldFail())) {
         Status status(ErrorCodes::InternalError,
                       str::stream() << "PlanExecutor hit planExecutorAlwaysFails fail point");
         *objOut =
-            Snapshotted<BSONObj>(SnapshotId(), WorkingSetCommon::buildMemberStatusObject(status));
+            Snapshotted<Document>(SnapshotId(), WorkingSetCommon::buildMemberStatusObject(status));
 
         return PlanExecutor::FAILURE;
     }
@@ -483,8 +508,8 @@ PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<BSONObj>* obj
     invariant(_currentState == kUsable);
     if (isMarkedAsKilled()) {
         if (nullptr != objOut) {
-            *objOut = Snapshotted<BSONObj>(SnapshotId(),
-                                           WorkingSetCommon::buildMemberStatusObject(_killStatus));
+            *objOut = Snapshotted<Document>(SnapshotId(),
+                                            WorkingSetCommon::buildMemberStatusObject(_killStatus));
         }
         return PlanExecutor::FAILURE;
     }
@@ -517,7 +542,7 @@ PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<BSONObj>* obj
             auto yieldStatus = _yieldPolicy->yieldOrInterrupt();
             if (!yieldStatus.isOK()) {
                 if (objOut) {
-                    *objOut = Snapshotted<BSONObj>(
+                    *objOut = Snapshotted<Document>(
                         SnapshotId(), WorkingSetCommon::buildMemberStatusObject(yieldStatus));
                 }
                 return PlanExecutor::FAILURE;
@@ -542,15 +567,11 @@ PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<BSONObj>* obj
                     } else {
                         // TODO: currently snapshot ids are only associated with documents, and
                         // not with index keys.
-                        *objOut = Snapshotted<BSONObj>(SnapshotId(), member->keyData[0].keyData);
+                        *objOut = Snapshotted<Document>(SnapshotId(),
+                                                        Document{member->keyData[0].keyData});
                     }
                 } else if (member->hasObj()) {
-                    *objOut = Snapshotted<BSONObj>(
-                        member->doc.snapshotId(),
-                        member->metadata() && member->doc.value().metadata()
-                            ? member->doc.value().toBsonWithMetaData(
-                                  _expCtx ? _expCtx->use42ChangeStreamSortKeys : false)
-                            : member->doc.value().toBson());
+                    *objOut = member->doc;
                 } else {
                     _workingSet->free(id);
                     hasRequestedData = false;
@@ -567,6 +588,12 @@ PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<BSONObj>* obj
             }
 
             if (hasRequestedData) {
+                // transfer the metadata from the WSM to Document.
+                invariant(objOut);
+                MutableDocument md(std::move(objOut->value()));
+                md.setMetadata(member->releaseMetadata());
+                objOut->setValue(md.freeze());
+
                 _workingSet->free(id);
                 return PlanExecutor::ADVANCED;
             }
@@ -609,9 +636,8 @@ PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<BSONObj>* obj
 
             if (nullptr != objOut) {
                 invariant(WorkingSet::INVALID_ID != id);
-                BSONObj statusObj =
-                    WorkingSetCommon::getStatusMemberDocument(*_workingSet, id)->toBson();
-                *objOut = Snapshotted<BSONObj>(SnapshotId(), statusObj);
+                auto statusObj = WorkingSetCommon::getStatusMemberDocument(*_workingSet, id);
+                *objOut = Snapshotted<Document>(SnapshotId(), *statusObj);
             }
 
             return PlanExecutor::FAILURE;
@@ -643,7 +669,7 @@ void PlanExecutorImpl::dispose(OperationContext* opCtx) {
 
 Status PlanExecutorImpl::executePlan() {
     invariant(_currentState == kUsable);
-    BSONObj obj;
+    Document obj;
     PlanExecutor::ExecState state = PlanExecutor::ADVANCED;
     while (PlanExecutor::ADVANCED == state) {
         state = this->getNext(&obj, nullptr);
@@ -666,10 +692,14 @@ Status PlanExecutorImpl::executePlan() {
 }
 
 
-void PlanExecutorImpl::enqueue(const BSONObj& obj) {
+void PlanExecutorImpl::enqueue(const Document& obj) {
     _stash.push(obj.getOwned());
 }
 
+void PlanExecutorImpl::enqueue(const BSONObj& obj) {
+    enqueue(Document{obj});
+}
+
 bool PlanExecutorImpl::isMarkedAsKilled() const {
     return !_killStatus.isOK();
 }
@@ -701,8 +731,11 @@ BSONObj PlanExecutorImpl::getPostBatchResumeToken() const {
     return {};
 }
 
-Status PlanExecutorImpl::getMemberObjectStatus(const BSONObj& memberObj) const {
+Status PlanExecutorImpl::getMemberObjectStatus(const Document& memberObj) const {
     return WorkingSetCommon::getMemberObjectStatus(memberObj);
 }
 
+Status PlanExecutorImpl::getMemberObjectStatus(const BSONObj& memberObj) const {
+    return WorkingSetCommon::getMemberObjectStatus(memberObj);
+}
 }  // namespace mongo
diff --git a/src/mongo/db/query/plan_executor_impl.h b/src/mongo/db/query/plan_executor_impl.h
index 629f66c6474..8d796e33ca4 100644
--- a/src/mongo/db/query/plan_executor_impl.h
+++ b/src/mongo/db/query/plan_executor_impl.h
@@ -61,17 +61,21 @@ public:
     CanonicalQuery* getCanonicalQuery() const final;
     const NamespaceString& nss() const final;
     OperationContext* getOpCtx() const final;
+    const boost::intrusive_ptr<ExpressionContext>& getExpCtx() const final;
     void saveState() final;
     void restoreState() final;
     void detachFromOperationContext() final;
     void reattachToOperationContext(OperationContext* opCtx) final;
     void restoreStateWithoutRetrying() final;
+    ExecState getNextSnapshotted(Snapshotted<Document>* objOut, RecordId* dlOut) final;
     ExecState getNextSnapshotted(Snapshotted<BSONObj>* objOut, RecordId* dlOut) final;
-    ExecState getNext(BSONObj* objOut, RecordId* dlOut) final;
+    ExecState getNext(Document* objOut, RecordId* dlOut) final;
+    ExecState getNext(BSONObj* out, RecordId* dlOut) final;
     bool isEOF() final;
     Status executePlan() final;
     void markAsKilled(Status killStatus) final;
     void dispose(OperationContext* opCtx) final;
+    void enqueue(const Document& obj) final;
     void enqueue(const BSONObj& obj) final;
     bool isMarkedAsKilled() const final;
     Status getKillStatus() final;
@@ -79,6 +83,8 @@ public:
     bool isDetached() const final;
     Timestamp getLatestOplogTimestamp() const final;
     BSONObj getPostBatchResumeToken() const final;
+
+    Status getMemberObjectStatus(const Document& memberObj) const final;
     Status getMemberObjectStatus(const BSONObj& memberObj) const final;
 
 private:
@@ -142,12 +148,12 @@ private:
      * describing the error.
      */
     ExecState _waitForInserts(CappedInsertNotifierData* notifierData,
-                              Snapshotted<BSONObj>* errorObj);
+                              Snapshotted<Document>* errorObj);
 
     /**
      * Common implementation for getNext() and getNextSnapshotted().
      */
-    ExecState _getNextImpl(Snapshotted<BSONObj>* objOut, RecordId* dlOut);
+    ExecState _getNextImpl(Snapshotted<Document>* objOut, RecordId* dlOut);
 
     // The OperationContext that we're executing within. This can be updated if necessary by using
     // detachFromOperationContext() and reattachToOperationContext().
@@ -181,7 +187,7 @@ private:
     // A stash of results generated by this plan that the user of the PlanExecutor didn't want
     // to consume yet. We empty the queue before retrieving further results from the plan
     // stages.
-    std::queue<BSONObj> _stash;
+    std::queue<Document> _stash;
 
     enum { kUsable, kSaved, kDetached, kDisposed } _currentState = kUsable;
 
diff --git a/src/mongo/db/query/planner_access.cpp b/src/mongo/db/query/planner_access.cpp
index eb837ff2121..ace8c86c703 100644
--- a/src/mongo/db/query/planner_access.cpp
+++ b/src/mongo/db/query/planner_access.cpp
@@ -285,20 +285,17 @@ std::unique_ptr<QuerySolutionNode> QueryPlannerAccess::makeLeafNode(
             auto ret = std::make_unique<GeoNear2DNode>(index);
             ret->nq = &nearExpr->getData();
             ret->baseBounds.fields.resize(index.keyPattern.nFields());
-            if (nullptr != query.getProj()) {
-                ret->addPointMeta = query.getProj()->wantGeoNearPoint();
-                ret->addDistMeta = query.getProj()->wantGeoNearDistance();
-            }
+            ret->addPointMeta = query.metadataDeps()[DocumentMetadataFields::kGeoNearPoint];
+            ret->addDistMeta = query.metadataDeps()[DocumentMetadataFields::kGeoNearDist];
 
             return std::move(ret);
         } else {
             auto ret = std::make_unique<GeoNear2DSphereNode>(index);
             ret->nq = &nearExpr->getData();
             ret->baseBounds.fields.resize(index.keyPattern.nFields());
-            if (nullptr != query.getProj()) {
-                ret->addPointMeta = query.getProj()->wantGeoNearPoint();
-                ret->addDistMeta = query.getProj()->wantGeoNearDistance();
-            }
+            ret->addPointMeta = query.metadataDeps()[DocumentMetadataFields::kGeoNearPoint];
+            ret->addDistMeta = query.metadataDeps()[DocumentMetadataFields::kGeoNearDist];
+
             return std::move(ret);
         }
     } else if (MatchExpression::TEXT == expr->matchType()) {
diff --git a/src/mongo/db/query/planner_analysis.cpp b/src/mongo/db/query/planner_analysis.cpp
index 3646422184f..2e6ae124505 100644
--- a/src/mongo/db/query/planner_analysis.cpp
+++ b/src/mongo/db/query/planner_analysis.cpp
@@ -328,7 +328,7 @@ auto produceCoveredKeyObj(QuerySolutionNode* solnRoot) {
  */
 std::unique_ptr<QuerySolutionNode> addSortKeyGeneratorStageIfNeeded(
     const CanonicalQuery& query, bool hasSortStage, std::unique_ptr<QuerySolutionNode> solnRoot) {
-    if (!hasSortStage && query.getProj() && query.getProj()->wantSortKey()) {
+    if (!hasSortStage && query.metadataDeps()[DocumentMetadataFields::kSortKey]) {
         auto keyGenNode = std::make_unique<SortKeyGeneratorNode>();
         keyGenNode->sortSpec = query.getQueryRequest().getSort();
         keyGenNode->children.push_back(solnRoot.release());
@@ -803,6 +803,9 @@ std::unique_ptr<QuerySolution> QueryPlannerAnalysis::analyzeDataAccess(
         if (solnRoot->fetched() && params.options & QueryPlannerParams::NO_UNCOVERED_PROJECTIONS)
             return nullptr;
     } else {
+        // Even if there's no projection, the client may want sort key metadata.
+        solnRoot = addSortKeyGeneratorStageIfNeeded(query, hasSortStage, std::move(solnRoot));
+
         // If there's no projection, we must fetch, as the user wants the entire doc.
         if (!solnRoot->fetched() && !(params.options & QueryPlannerParams::IS_COUNT)) {
             FetchNode* fetch = new FetchNode();
diff --git a/src/mongo/db/query/projection.cpp b/src/mongo/db/query/projection.cpp
index 6ad92a8859f..d2a4ee89264 100644
--- a/src/mongo/db/query/projection.cpp
+++ b/src/mongo/db/query/projection.cpp
@@ -43,7 +43,7 @@ namespace {
  * context.
  */
 struct DepsAnalysisData {
-    DepsTracker fieldDependencyTracker{DepsTracker::kAllMetadataAvailable};
+    DepsTracker fieldDependencyTracker{DepsTracker::kAllMetadata};
 
     void addRequiredField(const std::string& fieldName) {
         fieldDependencyTracker.fields.insert(fieldName);
@@ -188,11 +188,7 @@ auto analyzeProjection(ProjectionPathASTNode* root, ProjectType type) {
         deps.requiresDocument = true;
     }
 
-    deps.needsTextScore = tracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE);
-    deps.needsGeoPoint = tracker.getNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_POINT);
-    deps.needsGeoDistance = tracker.getNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_DISTANCE);
-    deps.needsSortKey = tracker.getNeedsMetadata(DepsTracker::MetadataType::SORT_KEY);
-
+    deps.metadataRequested = tracker.metadataDeps();
     return deps;
 }
 }  // namespace
diff --git a/src/mongo/db/query/projection.h b/src/mongo/db/query/projection.h
index 8067b08591d..91afe2b931f 100644
--- a/src/mongo/db/query/projection.h
+++ b/src/mongo/db/query/projection.h
@@ -29,6 +29,7 @@
 
 #pragma once
 
+#include "mongo/db/exec/document_value/document_metadata_fields.h"
 #include "mongo/db/query/projection_ast.h"
 #include "mongo/util/str.h"
 
@@ -47,12 +48,9 @@ struct ProjectionDependencies {
     // Which fields are necessary to perform the projection, or boost::none if all are required.
     boost::optional<std::vector<std::string>> requiredFields;
 
-    bool needsGeoDistance = false;
-    bool needsGeoPoint = false;
-    bool needsSortKey = false;
-    bool needsTextScore = false;
-
     bool hasDottedPath = false;
+
+    QueryMetadataBitSet metadataRequested;
 };
 
 /**
@@ -95,23 +93,8 @@ public:
         return *_deps.requiredFields;
     }
 
-    /**
-     * Does the projection want geoNear metadata?  If so any geoNear stage should include them.
-     */
-    bool wantGeoNearDistance() const {
-        return _deps.needsGeoDistance;
-    }
-
-    bool wantGeoNearPoint() const {
-        return _deps.needsGeoPoint;
-    }
-
-    bool wantSortKey() const {
-        return _deps.needsSortKey;
-    }
-
-    bool wantTextScore() const {
-        return _deps.needsTextScore;
+    const QueryMetadataBitSet& metadataDeps() const {
+        return _deps.metadataRequested;
     }
 
     /**
@@ -135,8 +118,8 @@ public:
      * on top-level fields, has no positional projection, and doesn't require the sort key.
      */
     bool isSimple() const {
-        return !_deps.hasDottedPath && !_deps.requiresMatchDetails && !_deps.needsSortKey &&
-            !_deps.requiresDocument;
+        return !_deps.hasDottedPath && !_deps.requiresMatchDetails &&
+            !_deps.metadataRequested.any() && !_deps.requiresDocument;
     }
 
 private:
diff --git a/src/mongo/db/query/projection_test.cpp b/src/mongo/db/query/projection_test.cpp
index b11b00bd270..c0044173cd4 100644
--- a/src/mongo/db/query/projection_test.cpp
+++ b/src/mongo/db/query/projection_test.cpp
@@ -217,12 +217,12 @@ TEST(QueryProjectionTest, InvalidPositionalProjectionDefaultPathMatchExpression)
 TEST(QueryProjectionTest, ProjectionDefaults) {
     auto proj = createProjection("{}", "{}");
 
-    ASSERT_FALSE(proj.wantSortKey());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
     ASSERT_TRUE(proj.requiresDocument());
     ASSERT_FALSE(proj.requiresMatchDetails());
-    ASSERT_FALSE(proj.wantGeoNearDistance());
-    ASSERT_FALSE(proj.wantGeoNearPoint());
-    ASSERT_FALSE(proj.wantTextScore());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearPoint]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kTextScore]);
 }
 
 TEST(QueryProjectionTest, SortKeyMetaProjectionInExclusionProjection) {
@@ -230,11 +230,11 @@ TEST(QueryProjectionTest, SortKeyMetaProjectionInExclusionProjection) {
     auto proj = createProjection("{}", "{foo: {$meta: 'sortKey'}}");
 
     ASSERT_BSONOBJ_EQ(proj.getProjObj(), fromjson("{foo: {$meta: 'sortKey'}}"));
-    ASSERT_TRUE(proj.wantSortKey());
+    ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
 
     ASSERT_FALSE(proj.requiresMatchDetails());
-    ASSERT_FALSE(proj.wantGeoNearDistance());
-    ASSERT_FALSE(proj.wantGeoNearPoint());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearPoint]);
     ASSERT_TRUE(proj.requiresDocument());
 }
 
@@ -242,11 +242,11 @@ TEST(QueryProjectionTest, SortKeyMetaProjectionInExclusionProjectionWithOtherFie
     auto proj = createProjection("{}", "{a: 0, foo: {$meta: 'sortKey'}}");
 
     ASSERT_BSONOBJ_EQ(proj.getProjObj(), fromjson("{a: 0, foo: {$meta: 'sortKey'}}"));
-    ASSERT_TRUE(proj.wantSortKey());
+    ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
 
     ASSERT_FALSE(proj.requiresMatchDetails());
-    ASSERT_FALSE(proj.wantGeoNearDistance());
-    ASSERT_FALSE(proj.wantGeoNearPoint());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearPoint]);
     ASSERT_TRUE(proj.requiresDocument());
 }
 
@@ -254,11 +254,11 @@ TEST(QueryProjectionTest, SortKeyMetaProjectionInInclusionProjection) {
     auto proj = createProjection("{}", "{a: 1, foo: {$meta: 'sortKey'}}");
 
     ASSERT_BSONOBJ_EQ(proj.getProjObj(), fromjson("{a: 1, foo: {$meta: 'sortKey'}}"));
-    ASSERT_TRUE(proj.wantSortKey());
+    ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
 
     ASSERT_FALSE(proj.requiresMatchDetails());
-    ASSERT_FALSE(proj.wantGeoNearDistance());
-    ASSERT_FALSE(proj.wantGeoNearPoint());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearPoint]);
     ASSERT_FALSE(proj.requiresDocument());
 }
 
@@ -266,12 +266,12 @@ TEST(QueryProjectionTest, SortKeyMetaProjectionDoesNotRequireDocument) {
     auto proj = createProjection("{}", "{a: 1, foo: {$meta: 'sortKey'}, _id: 0}");
 
     ASSERT_BSONOBJ_EQ(proj.getProjObj(), fromjson("{a: 1, foo: {$meta: 'sortKey'}, _id: 0}"));
-    ASSERT_TRUE(proj.wantSortKey());
+    ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
 
     ASSERT_FALSE(proj.requiresDocument());
     ASSERT_FALSE(proj.requiresMatchDetails());
-    ASSERT_FALSE(proj.wantGeoNearDistance());
-    ASSERT_FALSE(proj.wantGeoNearPoint());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearPoint]);
 }
 
 TEST(QueryProjectionTest, SortKeyMetaAndSlice) {
@@ -279,12 +279,12 @@ TEST(QueryProjectionTest, SortKeyMetaAndSlice) {
 
     ASSERT_BSONOBJ_EQ(proj.getProjObj(),
                       fromjson("{a: 1, foo: {$meta: 'sortKey'}, _id: 0, b: {$slice: 1}}"));
-    ASSERT_TRUE(proj.wantSortKey());
+    ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
     ASSERT_TRUE(proj.requiresDocument());
 
     ASSERT_FALSE(proj.requiresMatchDetails());
-    ASSERT_FALSE(proj.wantGeoNearDistance());
-    ASSERT_FALSE(proj.wantGeoNearPoint());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearPoint]);
 }
 
 TEST(QueryProjectionTest, SortKeyMetaAndElemMatch) {
@@ -293,12 +293,12 @@ TEST(QueryProjectionTest, SortKeyMetaAndElemMatch) {
 
     ASSERT_BSONOBJ_EQ(proj.getProjObj(),
                       fromjson("{a: 1, foo: {$meta: 'sortKey'}, _id: 0, b: {$elemMatch: {a: 1}}}"));
-    ASSERT_TRUE(proj.wantSortKey());
+    ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
     ASSERT_TRUE(proj.requiresDocument());
 
     ASSERT_FALSE(proj.requiresMatchDetails());
-    ASSERT_FALSE(proj.wantGeoNearDistance());
-    ASSERT_FALSE(proj.wantGeoNearPoint());
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearDist]);
+    ASSERT_FALSE(proj.metadataDeps()[DocumentMetadataFields::kGeoNearPoint]);
 }
 
 //
diff --git a/src/mongo/db/query/stage_builder.cpp b/src/mongo/db/query/stage_builder.cpp
index 0beb57d2024..b70ca49a5a3 100644
--- a/src/mongo/db/query/stage_builder.cpp
+++ b/src/mongo/db/query/stage_builder.cpp
@@ -265,7 +265,7 @@ std::unique_ptr<PlanStage> buildStages(OperationContext* opCtx,
             // practice, this means that it is illegal to use the StageBuilder on a QuerySolution
             // created by planning a query that contains "no-op" expressions.
             params.query = static_cast<FTSQueryImpl&>(*node->ftsQuery);
-            params.wantTextScore = (cq.getProj() && cq.getProj()->wantTextScore());
+            params.wantTextScore = cq.metadataDeps()[DocumentMetadataFields::kTextScore];
             return std::make_unique<TextStage>(opCtx, params, ws, node->filter.get());
         }
         case STAGE_SHARDING_FILTER: {
diff --git a/src/mongo/db/repl/rollback_impl.cpp b/src/mongo/db/repl/rollback_impl.cpp
index 4c670a82a2b..f88a6cdd751 100644
--- a/src/mongo/db/repl/rollback_impl.cpp
+++ b/src/mongo/db/repl/rollback_impl.cpp
@@ -587,7 +587,8 @@ void RollbackImpl::_correctRecordStoreCounts(OperationContext* opCtx) {
                 opCtx, PlanExecutor::INTERRUPT_ONLY, Collection::ScanDirection::kForward);
             long long countFromScan = 0;
             PlanExecutor::ExecState state;
-            while (PlanExecutor::ADVANCED == (state = exec->getNext(nullptr, nullptr))) {
+            while (PlanExecutor::ADVANCED ==
+                   (state = exec->getNext(static_cast<BSONObj*>(nullptr), nullptr))) {
                 ++countFromScan;
             }
             if (PlanExecutor::IS_EOF != state) {
diff --git a/src/mongo/dbtests/cursor_manager_test.cpp b/src/mongo/dbtests/cursor_manager_test.cpp
index 90def084bba..a4c3d835079 100644
--- a/src/mongo/dbtests/cursor_manager_test.cpp
+++ b/src/mongo/dbtests/cursor_manager_test.cpp
@@ -82,14 +82,17 @@ public:
     }
 
     ClientCursorParams makeParams(OperationContext* opCtx) {
-        return {makeFakePlanExecutor(opCtx),
-                kTestNss,
-                {},
-                opCtx->getWriteConcern(),
-                repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-                BSONObj(),
-                ClientCursorParams::LockPolicy::kLocksInternally,
-                PrivilegeVector()};
+        return {
+            makeFakePlanExecutor(opCtx),
+            kTestNss,
+            {},
+            opCtx->getWriteConcern(),
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        };
     }
 
     ClientCursorPin makeCursor(OperationContext* opCtx) {
@@ -133,14 +136,17 @@ TEST_F(CursorManagerTest, ShouldBeAbleToKillPinnedCursor) {
 
     auto cursorPin = cursorManager->registerCursor(
         pinningOpCtx,
-        {makeFakePlanExecutor(),
-         kTestNss,
-         {},
-         {},
-         repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-         BSONObj(),
-         ClientCursorParams::LockPolicy::kLocksInternally,
-         PrivilegeVector()});
+        {
+            makeFakePlanExecutor(),
+            kTestNss,
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
 
     auto cursorId = cursorPin.getCursor()->cursorid();
     ASSERT_OK(cursorManager->killCursor(_opCtx.get(), cursorId, shouldAudit));
@@ -160,14 +166,17 @@ TEST_F(CursorManagerTest, ShouldBeAbleToKillPinnedCursorMultiClient) {
     // Pin the cursor from one client.
     auto cursorPin = cursorManager->registerCursor(
         pinningOpCtx,
-        {makeFakePlanExecutor(),
-         kTestNss,
-         {},
-         {},
-         repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-         BSONObj(),
-         ClientCursorParams::LockPolicy::kLocksInternally,
-         PrivilegeVector()});
+        {
+            makeFakePlanExecutor(),
+            kTestNss,
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
 
     auto cursorId = cursorPin.getCursor()->cursorid();
 
@@ -196,15 +205,19 @@ TEST_F(CursorManagerTest, InactiveCursorShouldTimeout) {
     CursorManager* cursorManager = useCursorManager();
     auto clock = useClock();
 
-    cursorManager->registerCursor(_opCtx.get(),
-                                  {makeFakePlanExecutor(),
-                                   NamespaceString{"test.collection"},
-                                   {},
-                                   {},
-                                   repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-                                   BSONObj(),
-                                   ClientCursorParams::LockPolicy::kLocksInternally,
-                                   PrivilegeVector()});
+    cursorManager->registerCursor(
+        _opCtx.get(),
+        {
+            makeFakePlanExecutor(),
+            NamespaceString{"test.collection"},
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
 
     ASSERT_EQ(0UL, cursorManager->timeoutCursors(_opCtx.get(), Date_t()));
 
@@ -212,15 +225,19 @@ TEST_F(CursorManagerTest, InactiveCursorShouldTimeout) {
     ASSERT_EQ(1UL, cursorManager->timeoutCursors(_opCtx.get(), clock->now()));
     ASSERT_EQ(0UL, cursorManager->numCursors());
 
-    cursorManager->registerCursor(_opCtx.get(),
-                                  {makeFakePlanExecutor(),
-                                   NamespaceString{"test.collection"},
-                                   {},
-                                   {},
-                                   repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-                                   BSONObj(),
-                                   ClientCursorParams::LockPolicy::kLocksInternally,
-                                   PrivilegeVector()});
+    cursorManager->registerCursor(
+        _opCtx.get(),
+        {
+            makeFakePlanExecutor(),
+            NamespaceString{"test.collection"},
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
     ASSERT_EQ(1UL, cursorManager->timeoutCursors(_opCtx.get(), Date_t::max()));
     ASSERT_EQ(0UL, cursorManager->numCursors());
 }
@@ -234,14 +251,17 @@ TEST_F(CursorManagerTest, InactivePinnedCursorShouldNotTimeout) {
 
     auto cursorPin = cursorManager->registerCursor(
         _opCtx.get(),
-        {makeFakePlanExecutor(),
-         NamespaceString{"test.collection"},
-         {},
-         {},
-         repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-         BSONObj(),
-         ClientCursorParams::LockPolicy::kLocksInternally,
-         PrivilegeVector()});
+        {
+            makeFakePlanExecutor(),
+            NamespaceString{"test.collection"},
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
 
     // The pin is still in scope, so it should not time out.
     clock->advance(getDefaultCursorTimeoutMillis());
@@ -259,14 +279,17 @@ TEST_F(CursorManagerTest, MarkedAsKilledCursorsShouldBeDeletedOnCursorPin) {
 
     auto cursorPin = cursorManager->registerCursor(
         _opCtx.get(),
-        {makeFakePlanExecutor(),
-         NamespaceString{"test.collection"},
-         {},
-         {},
-         repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-         BSONObj(),
-         ClientCursorParams::LockPolicy::kLocksInternally,
-         PrivilegeVector()});
+        {
+            makeFakePlanExecutor(),
+            NamespaceString{"test.collection"},
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
     auto cursorId = cursorPin->cursorid();
 
     // A cursor will stay alive, but be marked as killed, if it is interrupted with a code other
@@ -293,14 +316,17 @@ TEST_F(CursorManagerTest, InactiveKilledCursorsShouldTimeout) {
 
     auto cursorPin = cursorManager->registerCursor(
         _opCtx.get(),
-        {makeFakePlanExecutor(),
-         NamespaceString{"test.collection"},
-         {},
-         {},
-         repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-         BSONObj(),
-         ClientCursorParams::LockPolicy::kLocksInternally,
-         PrivilegeVector()});
+        {
+            makeFakePlanExecutor(),
+            NamespaceString{"test.collection"},
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
 
     // A cursor will stay alive, but be marked as killed, if it is interrupted with a code other
     // than ErrorCodes::Interrupted or ErrorCodes::CursorKilled and then unpinned.
@@ -326,28 +352,35 @@ TEST_F(CursorManagerTest, UsingACursorShouldUpdateTimeOfLastUse) {
     // Register a cursor which we will look at again.
     auto cursorPin = cursorManager->registerCursor(
         _opCtx.get(),
-        {makeFakePlanExecutor(),
-         kTestNss,
-         {},
-         {},
-         repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-         BSONObj(),
-         ClientCursorParams::LockPolicy::kLocksInternally,
-         PrivilegeVector()});
+        {
+            makeFakePlanExecutor(),
+            kTestNss,
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
     auto usedCursorId = cursorPin.getCursor()->cursorid();
     cursorPin.release();
 
     // Register a cursor to immediately forget about, to make sure it will time out on a normal
     // schedule.
-    cursorManager->registerCursor(_opCtx.get(),
-                                  {makeFakePlanExecutor(),
-                                   kTestNss,
-                                   {},
-                                   {},
-                                   repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-                                   BSONObj(),
-                                   ClientCursorParams::LockPolicy::kLocksInternally,
-                                   PrivilegeVector()});
+    cursorManager->registerCursor(
+        _opCtx.get(),
+        {
+            makeFakePlanExecutor(),
+            kTestNss,
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
 
     // Advance the clock to simulate time passing.
     clock->advance(Milliseconds(1));
@@ -378,14 +411,17 @@ TEST_F(CursorManagerTest, CursorShouldNotTimeOutUntilIdleForLongEnoughAfterBeing
     // Register a cursor which we will look at again.
     auto cursorPin = cursorManager->registerCursor(
         _opCtx.get(),
-        {makeFakePlanExecutor(),
-         kTestNss,
-         {},
-         {},
-         repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
-         BSONObj(),
-         ClientCursorParams::LockPolicy::kLocksInternally,
-         PrivilegeVector()});
+        {
+            makeFakePlanExecutor(),
+            kTestNss,
+            {},
+            {},
+            repl::ReadConcernArgs(repl::ReadConcernLevel::kLocalReadConcern),
+            BSONObj(),
+            ClientCursorParams::LockPolicy::kLocksInternally,
+            PrivilegeVector(),
+            false  // needsMerge
+        });
 
     // Advance the clock to simulate time passing.
     clock->advance(getDefaultCursorTimeoutMillis() + Milliseconds(1));
diff --git a/src/mongo/dbtests/query_stage_sort.cpp b/src/mongo/dbtests/query_stage_sort.cpp
index 5676ba90157..3e52f8e622a 100644
--- a/src/mongo/dbtests/query_stage_sort.cpp
+++ b/src/mongo/dbtests/query_stage_sort.cpp
@@ -583,7 +583,8 @@ public:
             &_opCtx, std::move(ws), std::move(fetchStage), coll, PlanExecutor::NO_YIELD);
         auto exec = std::move(statusWithPlanExecutor.getValue());
 
-        PlanExecutor::ExecState runnerState = exec->getNext(nullptr, nullptr);
+        PlanExecutor::ExecState runnerState =
+            exec->getNext(static_cast<BSONObj*>(nullptr), nullptr);
         ASSERT_EQUALS(PlanExecutor::FAILURE, runnerState);
     }
 };
diff --git a/src/mongo/dbtests/query_stage_tests.cpp b/src/mongo/dbtests/query_stage_tests.cpp
index f7c7dcf85ac..a81bee3b634 100644
--- a/src/mongo/dbtests/query_stage_tests.cpp
+++ b/src/mongo/dbtests/query_stage_tests.cpp
@@ -102,7 +102,8 @@ public:
 
         int count = 0;
         PlanExecutor::ExecState state;
-        for (RecordId dl; PlanExecutor::ADVANCED == (state = exec->getNext(nullptr, &dl));) {
+        for (RecordId dl; PlanExecutor::ADVANCED ==
+             (state = exec->getNext(static_cast<BSONObj*>(nullptr), &dl));) {
             ++count;
         }
         ASSERT_EQUALS(PlanExecutor::IS_EOF, state);
diff --git a/src/mongo/s/query/cluster_aggregation_planner.cpp b/src/mongo/s/query/cluster_aggregation_planner.cpp
index d3c899755cb..1c03e375e00 100644
--- a/src/mongo/s/query/cluster_aggregation_planner.cpp
+++ b/src/mongo/s/query/cluster_aggregation_planner.cpp
@@ -210,7 +210,7 @@ void propagateDocLimitToShards(Pipeline* shardPipe, Pipeline* mergePipe) {
  * Documents.
  */
 void limitFieldsSentFromShardsToMerger(Pipeline* shardPipe, Pipeline* mergePipe) {
-    DepsTracker mergeDeps(mergePipe->getDependencies(DepsTracker::kAllMetadataAvailable));
+    DepsTracker mergeDeps(mergePipe->getDependencies(DepsTracker::kAllMetadata));
     if (mergeDeps.needWholeDocument)
         return;  // the merge needs all fields, so nothing we can do.
 
@@ -218,10 +218,6 @@ void limitFieldsSentFromShardsToMerger(Pipeline* shardPipe, Pipeline* mergePipe)
     if (mergeDeps.fields.empty())
         mergeDeps.fields.insert("_id");
 
-    // Remove metadata from dependencies since it automatically flows through projection and we
-    // don't want to project it in to the document.
-    mergeDeps.setNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE, false);
-
     // HEURISTIC: only apply optimization if none of the shard stages have an exhaustive list of
     // field dependencies. While this may not be 100% ideal in all cases, it is simple and
     // avoids the worst cases by ensuring that:
@@ -232,13 +228,14 @@ void limitFieldsSentFromShardsToMerger(Pipeline* shardPipe, Pipeline* mergePipe)
     // 2) Optimization IS NOT applied immediately following a $project or $group since it would
     //    add an unnecessary project (and therefore a deep-copy).
     for (auto&& source : shardPipe->getSources()) {
-        DepsTracker dt(DepsTracker::kAllMetadataAvailable);
+        DepsTracker dt(DepsTracker::kAllMetadata);
         if (source->getDependencies(&dt) & DepsTracker::State::EXHAUSTIVE_FIELDS)
             return;
     }
     // if we get here, add the project.
     boost::intrusive_ptr<DocumentSource> project = DocumentSourceProject::createFromBson(
-        BSON("$project" << mergeDeps.toProjection()).firstElement(), shardPipe->getContext());
+        BSON("$project" << mergeDeps.toProjectionWithoutMetadata()).firstElement(),
+        shardPipe->getContext());
     shardPipe->pushBack(project);
 }