SERVER-42649 Use Value instead of BSONObj in the in-memory sort key

48 files changed, 597 insertions, 434 deletions

diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index c6ce95ebf55..7dc58def146 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -358,7 +358,7 @@ public:
             }
 
             exec = uassertStatusOK(PlanExecutor::make(
-                opCtx, std::move(ws), std::move(root), cursorNss, PlanExecutor::NO_YIELD));
+                opCtx, std::move(ws), std::move(root), nullptr, PlanExecutor::NO_YIELD, cursorNss));
 
             for (long long objCount = 0; objCount < batchSize; objCount++) {
                 BSONObj next;
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 68d2465e48a..e830325e456 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -196,7 +196,7 @@ public:
             }
 
             exec = uassertStatusOK(PlanExecutor::make(
-                opCtx, std::move(ws), std::move(root), nss, PlanExecutor::NO_YIELD));
+                opCtx, std::move(ws), std::move(root), nullptr, PlanExecutor::NO_YIELD, nss));
 
             for (long long objCount = 0; objCount < batchSize; objCount++) {
                 BSONObj next;
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 5fc6cadacb3..98a332ff145 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -456,6 +456,8 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> createOuterPipelineProxyExe
     const NamespaceString& nss,
     std::unique_ptr<Pipeline, PipelineDeleter> pipeline,
     bool hasChangeStream) {
+    boost::intrusive_ptr<ExpressionContext> expCtx(pipeline->getContext());
+
     // Transfer ownership of the Pipeline to the PipelineProxyStage.
     auto ws = std::make_unique<WorkingSet>();
     auto proxy = hasChangeStream
@@ -467,8 +469,8 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> createOuterPipelineProxyExe
     // invalidations. The Pipeline may contain PlanExecutors which *are* yielding
     // PlanExecutors and which *are* registered with their respective collection's
     // CursorManager
-    return uassertStatusOK(
-        PlanExecutor::make(opCtx, std::move(ws), std::move(proxy), nss, PlanExecutor::NO_YIELD));
+    return uassertStatusOK(PlanExecutor::make(
+        std::move(expCtx), std::move(ws), std::move(proxy), nullptr, PlanExecutor::NO_YIELD, nss));
 }
 }  // namespace
 
diff --git a/src/mongo/db/exec/SConscript b/src/mongo/db/exec/SConscript
index 06fdd5c3238..021474174f2 100644
--- a/src/mongo/db/exec/SConscript
+++ b/src/mongo/db/exec/SConscript
@@ -34,6 +34,7 @@ sortExecutorEnv.Library(
     target="sort_executor",
     source=[
         "sort_executor.cpp",
+        "sort_key_comparator.cpp",
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/query/sort_pattern',
diff --git a/src/mongo/db/exec/change_stream_proxy.cpp b/src/mongo/db/exec/change_stream_proxy.cpp
index 7f5da7c214e..c10bccf4cd4 100644
--- a/src/mongo/db/exec/change_stream_proxy.cpp
+++ b/src/mongo/db/exec/change_stream_proxy.cpp
@@ -56,7 +56,7 @@ boost::optional<Document> ChangeStreamProxyStage::getNext() {
         // the latest event observed in the oplog, the latter via its sort key metadata field.
         _validateResumeToken(*next);
         _latestOplogTimestamp = PipelineD::getLatestOplogTimestamp(_pipeline.get());
-        _postBatchResumeToken = next->metadata().getSortKey();
+        _postBatchResumeToken = next->metadata().getSortKey().getDocument().toBson();
         _setSpeculativeReadTimestamp();
         return next;
     }
@@ -85,7 +85,7 @@ void ChangeStreamProxyStage::_validateResumeToken(const Document& event) const {
     auto eventBSON = event.toBson();
     auto resumeToken = event.metadata().getSortKey();
     auto idField = eventBSON.getObjectField("_id");
-    invariant(!resumeToken.isEmpty());
+    invariant(!resumeToken.missing());
     uassert(ErrorCodes::ChangeStreamFatalError,
             str::stream() << "Encountered an event whose _id field, which contains the resume "
                              "token, was modified by the pipeline. Modifying the _id field of an "
@@ -94,7 +94,8 @@ void ChangeStreamProxyStage::_validateResumeToken(const Document& event) const {
                              "Expected: "
                           << BSON("_id" << resumeToken) << " but found: "
                           << (eventBSON["_id"] ? BSON("_id" << eventBSON["_id"]) : BSONObj()),
-            idField.binaryEqual(resumeToken));
+            (resumeToken.getType() == BSONType::Object) &&
+                idField.binaryEqual(resumeToken.getDocument().toBson()));
 }
 
 void ChangeStreamProxyStage::_setSpeculativeReadTimestamp() {
diff --git a/src/mongo/db/exec/ensure_sorted.cpp b/src/mongo/db/exec/ensure_sorted.cpp
index e4c0acc052b..aaec72bf377 100644
--- a/src/mongo/db/exec/ensure_sorted.cpp
+++ b/src/mongo/db/exec/ensure_sorted.cpp
@@ -46,9 +46,8 @@ EnsureSortedStage::EnsureSortedStage(OperationContext* opCtx,
                                      BSONObj pattern,
                                      WorkingSet* ws,
                                      std::unique_ptr<PlanStage> child)
-    : PlanStage(kStageType, opCtx), _ws(ws) {
+    : PlanStage(kStageType, opCtx), _ws(ws), _sortKeyComparator(pattern) {
     _children.emplace_back(std::move(child));
-    _pattern = FindCommon::transformSortSpec(pattern);
 }
 
 bool EnsureSortedStage::isEOF() {
@@ -63,16 +62,15 @@ PlanStage::StageState EnsureSortedStage::doWork(WorkingSetID* out) {
         // SortKeyGeneratorStage descendent in the execution tree.
         WorkingSetMember* member = _ws->get(*out);
         auto curSortKey = member->metadata().getSortKey();
-        invariant(!curSortKey.isEmpty());
+        invariant(!curSortKey.missing());
 
-        if (!_prevSortKey.isEmpty() && !isInOrder(_prevSortKey, curSortKey)) {
+        if (!_prevSortKey.missing() && !isInOrder(_prevSortKey, curSortKey)) {
             // 'member' is out of order. Drop it from the result set.
             _ws->free(*out);
             ++_specificStats.nDropped;
             return PlanStage::NEED_TIME;
         }
 
-        invariant(curSortKey.isOwned());
         _prevSortKey = curSortKey;
         return PlanStage::ADVANCED;
     }
@@ -93,10 +91,10 @@ const SpecificStats* EnsureSortedStage::getSpecificStats() const {
     return &_specificStats;
 }
 
-bool EnsureSortedStage::isInOrder(const BSONObj& lhsSortKey, const BSONObj& rhsSortKey) const {
+bool EnsureSortedStage::isInOrder(const Value& lhsKey, const Value& rhsKey) const {
     // No need to compare with a collator, since the sort keys were extracted by the
     // SortKeyGenerator, which has already mapped strings to their comparison keys.
-    return lhsSortKey.woCompare(rhsSortKey, _pattern, /*considerFieldName*/ false) <= 0;
+    return _sortKeyComparator(lhsKey, rhsKey) <= 0;
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/exec/ensure_sorted.h b/src/mongo/db/exec/ensure_sorted.h
index 3e3915b6a78..423647eeacf 100644
--- a/src/mongo/db/exec/ensure_sorted.h
+++ b/src/mongo/db/exec/ensure_sorted.h
@@ -30,6 +30,7 @@
 #pragma once
 
 #include "mongo/db/exec/plan_stage.h"
+#include "mongo/db/exec/sort_key_comparator.h"
 
 namespace mongo {
 
@@ -65,15 +66,15 @@ private:
      * Returns whether the result with the lhsSortKey should come before the result with the
      * rhsSortKey in sort order.
      */
-    bool isInOrder(const BSONObj& lhsSortKey, const BSONObj& rhsSortKey) const;
+    bool isInOrder(const Value& lhsSortKey, const Value& rhsSortKey) const;
 
     WorkingSet* _ws;
 
-    // The pattern that we're sorting by.
-    BSONObj _pattern;
+    // Comparator that is aware of the pattern that we're sorting by.
+    SortKeyComparator _sortKeyComparator;
 
     // The sort key of the previous result.
-    BSONObj _prevSortKey;
+    Value _prevSortKey;
 
     EnsureSortedStats _specificStats;
 };
diff --git a/src/mongo/db/exec/projection.cpp b/src/mongo/db/exec/projection.cpp
index 8faa479e9d9..57239c83759 100644
--- a/src/mongo/db/exec/projection.cpp
+++ b/src/mongo/db/exec/projection.cpp
@@ -55,7 +55,8 @@ BSONObj indexKey(const WorkingSetMember& member) {
 }
 
 BSONObj sortKey(const WorkingSetMember& member) {
-    return member.metadata().getSortKey();
+    return DocumentMetadataFields::serializeSortKey(member.metadata().isSingleElementKey(),
+                                                    member.metadata().getSortKey());
 }
 
 double geoDistance(const WorkingSetMember& member) {
diff --git a/src/mongo/db/exec/return_key.cpp b/src/mongo/db/exec/return_key.cpp
index 8f0e185b554..7b8bf6cbf77 100644
--- a/src/mongo/db/exec/return_key.cpp
+++ b/src/mongo/db/exec/return_key.cpp
@@ -82,7 +82,10 @@ Status ReturnKeyStage::_extractIndexKey(WorkingSetMember* member) {
     }
 
     auto indexKey = member->metadata().hasIndexKey() ? member->metadata().getIndexKey() : BSONObj();
-    auto sortKey = member->metadata().hasSortKey() ? member->metadata().getSortKey() : BSONObj();
+    auto sortKey = member->metadata().hasSortKey()
+        ? DocumentMetadataFields::serializeSortKey(member->metadata().isSingleElementKey(),
+                                                   member->metadata().getSortKey())
+        : BSONObj();
     BSONObjBuilder bob;
 
     if (!indexKey.isEmpty()) {
diff --git a/src/mongo/db/exec/sort.cpp b/src/mongo/db/exec/sort.cpp
index b98870858bb..c7c3cbe3ea7 100644
--- a/src/mongo/db/exec/sort.cpp
+++ b/src/mongo/db/exec/sort.cpp
@@ -53,17 +53,16 @@ using std::vector;
 // static
 const char* SortStage::kStageType = "SORT";
 
-SortStage::WorkingSetComparator::WorkingSetComparator(BSONObj p) : pattern(p) {}
-
 bool SortStage::WorkingSetComparator::operator()(const SortableDataItem& lhs,
                                                  const SortableDataItem& rhs) const {
-    // False means ignore field names.
-    int result = lhs.sortKey.woCompare(rhs.sortKey, pattern, false);
-    if (0 != result) {
-        return result < 0;
+    int cmp = sortKeyComparator(lhs.sortKey, rhs.sortKey);
+    if (cmp != 0) {
+        return cmp < 0;
+    } else {
+        // Indexes use recordId as a final tiebreaker, and we do the same here, so that SortStage
+        // outputs documents in the same order that an index would have.
+        return lhs.recordId < rhs.recordId;
     }
-    // Indices use RecordId as an additional sort key so we must as well.
-    return lhs.recordId < rhs.recordId;
 }
 
 SortStage::SortStage(boost::intrusive_ptr<ExpressionContext> expCtx,
@@ -87,12 +86,12 @@ SortStage::SortStage(boost::intrusive_ptr<ExpressionContext> expCtx,
     _children.emplace_back(std::move(child));
 
     BSONObj sortComparator = FindCommon::transformSortSpec(_pattern);
-    _sortKeyComparator = std::make_unique<WorkingSetComparator>(sortComparator);
+    _workingSetComparator = std::make_unique<WorkingSetComparator>(sortComparator);
 
     // If limit > 1, we need to initialize _dataSet here to maintain ordered set of data items while
     // fetching from the child stage.
     if (_limit > 1) {
-        const WorkingSetComparator& cmp = *_sortKeyComparator;
+        const WorkingSetComparator& cmp = *_workingSetComparator;
         _dataSet.reset(new SortableDataItemSet(cmp));
     }
 }
@@ -226,7 +225,7 @@ void SortStage::addToBuffer(const SortableDataItem& item) {
             return;
         }
         wsidToFree = item.wsid;
-        const WorkingSetComparator& cmp = *_sortKeyComparator;
+        const WorkingSetComparator& cmp = *_workingSetComparator;
         // Compare new item with existing item in vector.
         if (cmp(item, _data[0])) {
             wsidToFree = _data[0].wsid;
@@ -250,7 +249,7 @@ void SortStage::addToBuffer(const SortableDataItem& item) {
         wsidToFree = item.wsid;
         SortableDataItemSet::const_iterator lastItemIt = --(_dataSet->end());
         const SortableDataItem& lastItem = *lastItemIt;
-        const WorkingSetComparator& cmp = *_sortKeyComparator;
+        const WorkingSetComparator& cmp = *_workingSetComparator;
         if (cmp(item, lastItem)) {
             _memUsage -= _ws->get(lastItem.wsid)->getMemUsage();
             _memUsage += member->getMemUsage();
@@ -275,7 +274,7 @@ void SortStage::addToBuffer(const SortableDataItem& item) {
 
 void SortStage::sortBuffer() {
     if (_limit == 0) {
-        const WorkingSetComparator& cmp = *_sortKeyComparator;
+        const WorkingSetComparator& cmp = *_workingSetComparator;
         std::sort(_data.begin(), _data.end(), cmp);
     } else if (_limit == 1) {
         // Buffer contains either 0 or 1 item so it is already in a sorted state.
diff --git a/src/mongo/db/exec/sort.h b/src/mongo/db/exec/sort.h
index 5a964e8bbc6..53957917581 100644
--- a/src/mongo/db/exec/sort.h
+++ b/src/mongo/db/exec/sort.h
@@ -34,6 +34,7 @@
 
 #include "mongo/db/exec/plan_stage.h"
 #include "mongo/db/exec/sort_executor.h"
+#include "mongo/db/exec/sort_key_comparator.h"
 #include "mongo/db/exec/sort_key_generator.h"
 #include "mongo/db/exec/working_set.h"
 #include "mongo/db/record_id.h"
@@ -94,7 +95,7 @@ private:
     // Collection of working set members to sort with their respective sort key.
     struct SortableDataItem {
         WorkingSetID wsid;
-        BSONObj sortKey;
+        Value sortKey;
         // Since we must replicate the behavior of a covered sort as much as possible we use the
         // RecordId to break sortKey ties.
         // See sorta.js.
@@ -108,11 +109,11 @@ private:
     // We are comparing keys generated by the SortKeyGenerator, which are already ordered with
     // respect the collation. Therefore, we explicitly avoid comparing using a collator here.
     struct WorkingSetComparator {
-        explicit WorkingSetComparator(BSONObj p);
+        explicit WorkingSetComparator(const BSONObj& pattern) : sortKeyComparator(pattern) {}
 
         bool operator()(const SortableDataItem& lhs, const SortableDataItem& rhs) const;
 
-        BSONObj pattern;
+        SortKeyComparator sortKeyComparator;
     };
 
     /**
@@ -131,7 +132,7 @@ private:
 
     // Comparator for data buffer
     // Initialization follows sort key generator
-    std::unique_ptr<WorkingSetComparator> _sortKeyComparator;
+    std::unique_ptr<WorkingSetComparator> _workingSetComparator;
 
     // The data we buffer and sort.
     // _data will contain sorted data when all data is gathered
diff --git a/src/mongo/db/exec/sort_executor.cpp b/src/mongo/db/exec/sort_executor.cpp
index cbcfaa78b1c..613a1b2a0c4 100644
--- a/src/mongo/db/exec/sort_executor.cpp
+++ b/src/mongo/db/exec/sort_executor.cpp
@@ -60,44 +60,6 @@ SortExecutor::SortExecutor(SortPattern sortPattern,
       _tempDir(std::move(tempDir)),
       _diskUseAllowed(allowDiskUse) {}
 
-int SortExecutor::Comparator::operator()(const DocumentSorter::Data& lhs,
-                                         const DocumentSorter::Data& rhs) const {
-    Value lhsKey = lhs.first;
-    Value rhsKey = rhs.first;
-    // DocumentSourceSort::populate() has already guaranteed that the sort key is non-empty.
-    // However, the tricky part is deciding what to do if none of the sort keys are present. In that
-    // case, consider the document "less".
-    //
-    // Note that 'comparator' must use binary comparisons here, as both 'lhs' and 'rhs' are
-    // collation comparison keys.
-    ValueComparator comparator;
-    const size_t n = _sort.size();
-    if (n == 1) {  // simple fast case
-        if (_sort[0].isAscending)
-            return comparator.compare(lhsKey, rhsKey);
-        else
-            return -comparator.compare(lhsKey, rhsKey);
-    }
-
-    // compound sort
-    for (size_t i = 0; i < n; i++) {
-        int cmp = comparator.compare(lhsKey[i], rhsKey[i]);
-        if (cmp) {
-            /* if necessary, adjust the return value by the key ordering */
-            if (!_sort[i].isAscending)
-                cmp = -cmp;
-
-            return cmp;
-        }
-    }
-
-    /*
-      If we got here, everything matched (or didn't exist), so we'll
-      consider the documents equal for purposes of this sort.
-    */
-    return 0;
-}
-
 boost::optional<Document> SortExecutor::getNextDoc() {
     auto wsm = getNextWsm();
     if (!wsm) {
diff --git a/src/mongo/db/exec/sort_executor.h b/src/mongo/db/exec/sort_executor.h
index 86ce3fc2687..3e3beca9863 100644
--- a/src/mongo/db/exec/sort_executor.h
+++ b/src/mongo/db/exec/sort_executor.h
@@ -29,6 +29,7 @@
 
 #pragma once
 
+#include "mongo/db/exec/sort_key_comparator.h"
 #include "mongo/db/exec/working_set.h"
 #include "mongo/db/pipeline/document.h"
 #include "mongo/db/pipeline/expression.h"
@@ -101,11 +102,13 @@ private:
     using DocumentSorter = Sorter<Value, WorkingSetMember>;
     class Comparator {
     public:
-        Comparator(const SortPattern& sortPattern) : _sort(sortPattern) {}
-        int operator()(const DocumentSorter::Data& lhs, const DocumentSorter::Data& rhs) const;
+        Comparator(const SortPattern& sortPattern) : _sortKeyComparator(sortPattern) {}
+        int operator()(const DocumentSorter::Data& lhs, const DocumentSorter::Data& rhs) const {
+            return _sortKeyComparator(lhs.first, rhs.first);
+        }
 
     private:
-        const SortPattern& _sort;
+        SortKeyComparator _sortKeyComparator;
     };
 
     SortOptions makeSortOptions() const;
diff --git a/src/mongo/db/exec/sort_key_comparator.cpp b/src/mongo/db/exec/sort_key_comparator.cpp
new file mode 100644
index 00000000000..b48ea5c1f99
--- /dev/null
+++ b/src/mongo/db/exec/sort_key_comparator.cpp
@@ -0,0 +1,85 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/exec/sort_key_comparator.h"
+
+namespace mongo {
+
+SortKeyComparator::SortKeyComparator(const SortPattern& sortPattern) {
+    _pattern.reserve(sortPattern.size());
+    std::transform(sortPattern.begin(),
+                   sortPattern.end(),
+                   std::back_inserter(_pattern),
+                   [](const SortPattern::SortPatternPart& part) {
+                       return part.isAscending ? SortDirection::kAscending
+                                               : SortDirection::kDescending;
+                   });
+}
+
+int SortKeyComparator::operator()(const Value& lhsKey, const Value& rhsKey) const {
+    // Note that 'comparator' must use binary comparisons here, as both 'lhs' and 'rhs' are
+    // collation comparison keys.
+    ValueComparator comparator;
+    const size_t n = _pattern.size();
+    if (n == 1) {  // simple fast case
+        if (_pattern[0] == SortDirection::kAscending)
+            return comparator.compare(lhsKey, rhsKey);
+        else
+            return -comparator.compare(lhsKey, rhsKey);
+    }
+
+    // compound sort
+    for (size_t i = 0; i < n; i++) {
+        int cmp = comparator.compare(lhsKey[i], rhsKey[i]);
+        if (cmp) {
+            // If necessary, adjust the return value by the key ordering.
+            if (_pattern[i] == SortDirection::kDescending)
+                cmp = -cmp;
+
+            return cmp;
+        }
+    }
+
+    // If we got here, everything matched (or didn't exist), so we'll consider the documents equal
+    // for purposes of this sort.
+    return 0;
+}
+
+SortKeyComparator::SortKeyComparator(const BSONObj& sortPattern) {
+    _pattern.reserve(sortPattern.nFields());
+    std::transform(sortPattern.begin(),
+                   sortPattern.end(),
+                   std::back_inserter(_pattern),
+                   [](const BSONElement& part) {
+                       return (part.number() >= 0) ? SortDirection::kAscending
+                                                   : SortDirection::kDescending;
+                   });
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/exec/sort_key_comparator.h b/src/mongo/db/exec/sort_key_comparator.h
new file mode 100644
index 00000000000..74ecb89c793
--- /dev/null
+++ b/src/mongo/db/exec/sort_key_comparator.h
@@ -0,0 +1,58 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <vector>
+
+#include "mongo/db/pipeline/value.h"
+#include "mongo/db/query/sort_pattern.h"
+
+namespace mongo {
+
+/**
+ * This class is used to compare "sort keys," which are the values used to determine the order of
+ * documents returned by a query that requests a sort. When executing a query with a blocking sort,
+ * a SortKeyGenerator stage creates a sort key for each document based on the requested sort
+ * pattern, and a sort stage orders the documents using the sort keys and this comparator.
+ */
+class SortKeyComparator {
+public:
+    SortKeyComparator(const SortPattern& sortPattern);
+    SortKeyComparator(const BSONObj& sortPattern);
+    int operator()(const Value& lhsKey, const Value& rhsKey) const;
+
+private:
+    // The comparator does not need the entire sort pattern, just the sort direction for each
+    // component.
+    enum class SortDirection { kDescending, kAscending };
+    std::vector<SortDirection> _pattern;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/exec/sort_key_generator.cpp b/src/mongo/db/exec/sort_key_generator.cpp
index 2439da67cb3..f8c81395b8a 100644
--- a/src/mongo/db/exec/sort_key_generator.cpp
+++ b/src/mongo/db/exec/sort_key_generator.cpp
@@ -75,7 +75,8 @@ PlanStage::StageState SortKeyGeneratorStage::doWork(WorkingSetID* out) {
         }
 
         // Add the sort key to the WSM as metadata.
-        member->metadata().setSortKey(sortKey.getValue());
+        member->metadata().setSortKey(std::move(sortKey.getValue()),
+                                      _sortKeyGen.isSingleElementKey());
 
         return PlanStage::ADVANCED;
     }
diff --git a/src/mongo/db/index/SConscript b/src/mongo/db/index/SConscript
index 0e5a1ead912..2de54d9fbf8 100644
--- a/src/mongo/db/index/SConscript
+++ b/src/mongo/db/index/SConscript
@@ -163,8 +163,9 @@ env.CppUnitTest(
     ],
     LIBDEPS=[
         'key_generator',
-        "$BUILD_DIR/mongo/db/matcher/expressions",
+        '$BUILD_DIR/mongo/db/matcher/expressions',
         '$BUILD_DIR/mongo/db/mongohasher',
+        '$BUILD_DIR/mongo/db/pipeline/document_value_test_util',
         '$BUILD_DIR/mongo/db/query/collation/collator_interface_mock',
         '$BUILD_DIR/mongo/db/query/query_test_service_context',
     ],
diff --git a/src/mongo/db/index/sort_key_generator.cpp b/src/mongo/db/index/sort_key_generator.cpp
index 634e0ff2d69..97abfe097a7 100644
--- a/src/mongo/db/index/sort_key_generator.cpp
+++ b/src/mongo/db/index/sort_key_generator.cpp
@@ -80,19 +80,25 @@ SortKeyGenerator::SortKeyGenerator(SortPattern sortPattern, const CollatorInterf
 // will be able to use the Document overload of computeSortKeyFromDocument, and it will be able to
 // store the text score with the Document instead of in a separate SortKeyGenerator::Metadata
 // object.
-StatusWith<BSONObj> SortKeyGenerator::computeSortKey(const WorkingSetMember& wsm) const {
+StatusWith<Value> SortKeyGenerator::computeSortKey(const WorkingSetMember& wsm) const {
     if (wsm.hasObj()) {
         SortKeyGenerator::Metadata metadata;
         if (_sortHasMeta && wsm.metadata().hasTextScore()) {
             metadata.textScore = wsm.metadata().getTextScore();
         }
-        return computeSortKeyFromDocument(wsm.doc.value().toBson(), &metadata);
+        auto statusWithSortKeyObj = computeSortKeyFromDocument(wsm.doc.value().toBson(), &metadata);
+        if (!statusWithSortKeyObj.isOK()) {
+            return statusWithSortKeyObj.getStatus();
+        }
+
+        return DocumentMetadataFields::deserializeSortKey(isSingleElementKey(),
+                                                          statusWithSortKeyObj.getValue());
     }
 
     return computeSortKeyFromIndexKey(wsm);
 }
 
-StatusWith<BSONObj> SortKeyGenerator::computeSortKeyFromIndexKey(
+StatusWith<Value> SortKeyGenerator::computeSortKeyFromIndexKey(
     const WorkingSetMember& member) const {
     invariant(member.getState() == WorkingSetMember::RID_AND_IDX);
     invariant(!_sortHasMeta);
@@ -109,7 +115,7 @@ StatusWith<BSONObj> SortKeyGenerator::computeSortKeyFromIndexKey(
         // non-simple collation.
         CollationIndexKey::collationAwareIndexKeyAppend(sortKeyElt, _collator, &objBuilder);
     }
-    return objBuilder.obj();
+    return DocumentMetadataFields::deserializeSortKey(isSingleElementKey(), objBuilder.obj());
 }
 
 StatusWith<BSONObj> SortKeyGenerator::computeSortKeyFromDocument(const BSONObj& obj,
diff --git a/src/mongo/db/index/sort_key_generator.h b/src/mongo/db/index/sort_key_generator.h
index fcdfe3309e5..b6ea016139a 100644
--- a/src/mongo/db/index/sort_key_generator.h
+++ b/src/mongo/db/index/sort_key_generator.h
@@ -65,17 +65,7 @@ public:
      * If the sort pattern contains a $meta sort (e.g. sort by "textScore" or "randVal"), then the
      * necessary metadata is obtained from the WorkingSetMember.
      */
-    StatusWith<BSONObj> computeSortKey(const WorkingSetMember&) const;
-
-    /**
-     * Returns the key which should be used to sort 'obj', or a non-OK status if no key could be
-     * generated.
-     *
-     * The caller must supply the appropriate 'metadata' in the case that the sort pattern includes
-     * a $meta sort (i.e. if sortHasMeta() is true). These values are filled in at the corresponding
-     * positions in the sort key.
-     */
-    StatusWith<BSONObj> computeSortKeyFromDocument(const BSONObj& obj, const Metadata*) const;
+    StatusWith<Value> computeSortKey(const WorkingSetMember&) const;
 
     /**
      * Returns the sort key for the input 'doc' as a Value. When the sort pattern has multiple
@@ -94,10 +84,20 @@ public:
     }
 
 private:
+    /**
+     * Returns the key which should be used to sort 'obj', or a non-OK status if no key could be
+     * generated.
+     *
+     * The caller must supply the appropriate 'metadata' in the case that the sort pattern includes
+     * a $meta sort (i.e. if sortHasMeta() is true). These values are filled in at the corresponding
+     * positions in the sort key.
+     */
+    StatusWith<BSONObj> computeSortKeyFromDocument(const BSONObj& obj, const Metadata*) const;
+
     // Extracts the sort key from a WorkingSetMember which represents an index key. It is illegal to
     // call this if the working set member is not in RID_AND_IDX state. It is also illegal to call
     // this if the sort pattern has any $meta components.
-    StatusWith<BSONObj> computeSortKeyFromIndexKey(const WorkingSetMember& member) const;
+    StatusWith<Value> computeSortKeyFromIndexKey(const WorkingSetMember& member) const;
 
     // Extracts the sort key from 'obj', using '_sortSpecWithoutMeta' and thus ignoring any $meta
     // sort components of the sort pattern. The caller is responsible for augmenting this key with
diff --git a/src/mongo/db/index/sort_key_generator_test.cpp b/src/mongo/db/index/sort_key_generator_test.cpp
index abbfef2f487..6383bdb9497 100644
--- a/src/mongo/db/index/sort_key_generator_test.cpp
+++ b/src/mongo/db/index/sort_key_generator_test.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/bson/json.h"
 #include "mongo/db/index/sort_key_generator.h"
+#include "mongo/db/pipeline/document_value_test_util.h"
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/db/query/collation/collator_interface_mock.h"
 #include "mongo/unittest/death_test.h"
@@ -52,98 +53,88 @@ std::unique_ptr<SortKeyGenerator> makeSortKeyGen(const BSONObj& sortSpec,
 
 TEST(SortKeyGeneratorTest, ExtractNumberKeyForNonCompoundSortNonNested) {
     auto sortKeyGen = makeSortKeyGen(BSON("a" << 1), nullptr);
-    auto sortKey = sortKeyGen->computeSortKeyFromDocument(fromjson("{_id: 0, a: 5}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 5));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(Document{{"_id", {5}}, {"a", {5}}});
+    ASSERT_VALUE_EQ(sortKey, Value{5});
 }
 
 TEST(SortKeyGeneratorTest, ExtractNumberKeyFromDocWithSeveralFields) {
     auto sortKeyGen = makeSortKeyGen(BSON("a" << 1), nullptr);
-    auto sortKey =
-        sortKeyGen->computeSortKeyFromDocument(fromjson("{_id: 0, z: 10, a: 6, b: 16}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 6));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(
+        Document{{"_id", {0}}, {"z", {10}}, {"a", {6}}, {"b", {16}}});
+    ASSERT_VALUE_EQ(sortKey, Value{6});
 }
 
 TEST(SortKeyGeneratorTest, ExtractStringKeyNonCompoundNonNested) {
     auto sortKeyGen = makeSortKeyGen(BSON("a" << 1), nullptr);
     auto sortKey = sortKeyGen->computeSortKeyFromDocument(
-        fromjson("{_id: 0, z: 'thing1', a: 'thing2', b: 16}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(),
-                      BSON(""
-                           << "thing2"));
+        Document{{"_id", {0}}, {"z", {"thing1"_sd}}, {"a", {"thing2"_sd}}, {"b", {16}}});
+    ASSERT_VALUE_EQ(sortKey, Value{"thing2"_sd});
 }
 
 TEST(SortKeyGeneratorTest, CompoundSortPattern) {
     auto sortKeyGen = makeSortKeyGen(BSON("a" << 1 << "b" << 1), nullptr);
-    auto sortKey = sortKeyGen->computeSortKeyFromDocument(
-        fromjson("{_id: 0, z: 'thing1', a: 99, c: {a: 4}, b: 16}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 99 << "" << 16));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(Document{
+        {"_id", {0}}, {"z", {"thing1"_sd}}, {"a", {99}}, {"c", Document{{"a", {4}}}}, {"b", {16}}});
+    ASSERT_VALUE_EQ(sortKey, (Value{std::vector<Value>{Value{99}, Value{16}}}));
 }
 
 TEST(SortKeyGeneratorTest, CompoundSortPatternWithDottedPath) {
     auto sortKeyGen = makeSortKeyGen(BSON("c.a" << 1 << "b" << 1), nullptr);
-    auto sortKey = sortKeyGen->computeSortKeyFromDocument(
-        fromjson("{_id: 0, z: 'thing1', a: 99, c: {a: 4}, b: 16}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 4 << "" << 16));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(Document{
+        {"_id", {0}}, {"z", {"thing1"_sd}}, {"a", {99}}, {"c", Document{{"a", {4}}}}, {"b", {16}}});
+    ASSERT_VALUE_EQ(sortKey, (Value{std::vector<Value>{Value{4}, Value{16}}}));
 }
 
 TEST(SortKeyGeneratorTest, CompoundPatternLeadingFieldIsArray) {
     auto sortKeyGen = makeSortKeyGen(BSON("c" << 1 << "b" << 1), nullptr);
     auto sortKey = sortKeyGen->computeSortKeyFromDocument(
-        fromjson("{_id: 0, z: 'thing1', a: 99, c: [2, 4, 1], b: 16}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 1 << "" << 16));
+        Document{{"_id", {0}},
+                 {"z", {"thing1"_sd}},
+                 {"a", {99}},
+                 {"c", Value{std::vector<Value>{Value{2}, Value{4}, Value{1}}}},
+                 {"b", {16}}});
+    ASSERT_VALUE_EQ(sortKey, (Value{std::vector<Value>{Value{1}, Value{16}}}));
 }
 
 TEST(SortKeyGeneratorTest, ExtractStringSortKeyWithCollatorUsesComparisonKey) {
     CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
     auto sortKeyGen = makeSortKeyGen(BSON("a" << 1), &collator);
     auto sortKey = sortKeyGen->computeSortKeyFromDocument(
-        fromjson("{_id: 0, z: 'thing1', a: 'thing2', b: 16}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(),
-                      BSON(""
-                           << "2gniht"));
+        Document{{"_id", {0}}, {"z", {"thing1"_sd}}, {"a", {"thing2"_sd}}, {"b", {16}}});
+    ASSERT_VALUE_EQ(sortKey, Value{"2gniht"_sd});
 }
 
 TEST(SortKeyGeneratorTest, CollatorHasNoEffectWhenExtractingNonStringSortKey) {
     CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
     auto sortKeyGen = makeSortKeyGen(BSON("a" << 1), &collator);
-    auto sortKey =
-        sortKeyGen->computeSortKeyFromDocument(fromjson("{_id: 0, z: 10, a: 6, b: 16}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 6));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(
+        Document{{"_id", {0}}, {"z", {10}}, {"a", {6}}, {"b", {16}}});
+    ASSERT_VALUE_EQ(sortKey, Value{6});
 }
 
 TEST(SortKeyGeneratorTest, SortKeyGenerationForArraysChoosesCorrectKey) {
     auto sortKeyGen = makeSortKeyGen(BSON("a" << -1), nullptr);
-    auto sortKey =
-        sortKeyGen->computeSortKeyFromDocument(fromjson("{_id: 0, a: [1, 2, 3, 4]}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 4));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(Document{
+        {"_id", {0}}, {"a", Value{std::vector<Value>{Value{1}, Value{2}, Value{3}, Value{4}}}}});
+    ASSERT_VALUE_EQ(sortKey, Value{4});
 }
 
 TEST(SortKeyGeneratorTest, EnsureSortKeyGenerationForArraysRespectsCollation) {
     CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
     auto sortKeyGen = makeSortKeyGen(BSON("a" << 1), &collator);
     auto sortKey = sortKeyGen->computeSortKeyFromDocument(
-        fromjson("{_id: 0, a: ['aaz', 'zza', 'yya', 'zzb']}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(),
-                      BSON(""
-                           << "ayy"));
+        Document{{"_id", {0}},
+                 {"a",
+                  Value{std::vector<Value>{
+                      Value{"aaz"_sd}, Value{"zza"_sd}, Value{"yya"_sd}, Value{"zzb"_sd}}}}});
+    ASSERT_VALUE_EQ(sortKey, Value{"ayy"_sd});
 }
 
 TEST(SortKeyGeneratorTest, SortKeyGenerationForArraysRespectsCompoundOrdering) {
     auto sortKeyGen = makeSortKeyGen(BSON("a.b" << 1 << "a.c" << -1), nullptr);
     auto sortKey = sortKeyGen->computeSortKeyFromDocument(
-        fromjson("{_id: 0, a: [{b: 1, c: 0}, {b: 0, c: 3}, {b: 0, c: 1}]}"), nullptr);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 0 << "" << 3));
+        Document{fromjson("{_id: 0, a: [{b: 1, c: 0}, {b: 0, c: 3}, {b: 0, c: 1}]}")});
+    ASSERT_VALUE_EQ(sortKey, (Value{std::vector<Value>{Value{0}, Value{3}}}));
 }
 
 TEST(SortKeyGeneratorTest, SortPatternComponentWithStringUasserts) {
@@ -208,57 +199,34 @@ TEST(SortKeyGeneratorTest, SortPatternComponentWithSearchHighlightsMetaKeywordUa
                                 "$meta sort by 'searchHighlights' metadata is not supported");
 }
 
-DEATH_TEST(SortKeyGeneratorTest,
-           NoMetadataWhenPatternHasMetaTextScoreIsFatal,
-           "Invariant failure metadata") {
-    auto sortKeyGen = makeSortKeyGen(BSON("a" << BSON("$meta"
-                                                      << "textScore")),
-                                     nullptr);
-    uassertStatusOK(sortKeyGen->computeSortKeyFromDocument(BSONObj{}, nullptr).getStatus());
-}
-
-DEATH_TEST(SortKeyGeneratorTest,
-           NoMetadataWhenPatternHasMetaRandValIsFatal,
-           "Invariant failure metadata") {
-    auto sortKeyGen = makeSortKeyGen(BSON("a" << BSON("$meta"
-                                                      << "randVal")),
-                                     nullptr);
-    uassertStatusOK(sortKeyGen->computeSortKeyFromDocument(BSONObj{}, nullptr).getStatus());
-}
-
 TEST(SortKeyGeneratorTest, CanGenerateKeysForTextScoreMetaSort) {
     auto sortKeyGen = makeSortKeyGen(BSON("a" << BSON("$meta"
                                                       << "textScore")),
                                      nullptr);
-    SortKeyGenerator::Metadata metadata;
-    metadata.textScore = 1.5;
-    auto sortKey = sortKeyGen->computeSortKeyFromDocument(BSONObj{}, &metadata);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 1.5));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(
+        Document::fromBsonWithMetaData(BSON(Document::metaFieldTextScore << 1.5)));
+    ASSERT_VALUE_EQ(sortKey, Value{1.5});
 }
 
 TEST(SortKeyGeneratorTest, CanGenerateKeysForRandValMetaSort) {
     auto sortKeyGen = makeSortKeyGen(BSON("a" << BSON("$meta"
                                                       << "randVal")),
                                      nullptr);
-    SortKeyGenerator::Metadata metadata;
-    metadata.randVal = 0.3;
-    auto sortKey = sortKeyGen->computeSortKeyFromDocument(BSONObj{}, &metadata);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(), BSON("" << 0.3));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(
+        Document::fromBsonWithMetaData(BSON(Document::metaFieldRandVal << 0.3)));
+    ASSERT_VALUE_EQ(sortKey, Value{0.3});
 }
 
 TEST(SortKeyGeneratorTest, CanGenerateKeysForCompoundMetaSort) {
     BSONObj pattern = fromjson(
         "{a: 1, b: {$meta: 'randVal'}, c: {$meta: 'textScore'}, d: -1, e: {$meta: 'textScore'}}");
     auto sortKeyGen = makeSortKeyGen(pattern, nullptr);
-    SortKeyGenerator::Metadata metadata;
-    metadata.randVal = 0.3;
-    metadata.textScore = 1.5;
-    auto sortKey = sortKeyGen->computeSortKeyFromDocument(BSON("a" << 4 << "d" << 5), &metadata);
-    ASSERT_OK(sortKey.getStatus());
-    ASSERT_BSONOBJ_EQ(sortKey.getValue(),
-                      BSON("" << 4 << "" << 0.3 << "" << 1.5 << "" << 5 << "" << 1.5));
+    auto sortKey = sortKeyGen->computeSortKeyFromDocument(
+        Document::fromBsonWithMetaData(BSON("a" << 4 << "d" << 5 << Document::metaFieldRandVal
+                                                << 0.3 << Document::metaFieldTextScore << 1.5)));
+    ASSERT_VALUE_EQ(
+        sortKey,
+        (Value{std::vector<Value>{Value{4}, Value{0.3}, Value{1.5}, Value{5}, Value{1.5}}}));
 }
 
 // A test fixture which creates a WorkingSet and allocates a WorkingSetMember inside of it. Used for
@@ -299,7 +267,7 @@ TEST_F(SortKeyGeneratorWorkingSetTest, CanGetSortKeyFromWorkingSetMemberWithObj)
     setRecordIdAndObj(BSON("x" << 1 << "a" << 2 << "y" << 3));
     auto sortKey = sortKeyGen->computeSortKey(member());
     ASSERT_OK(sortKey);
-    ASSERT_BSONOBJ_EQ(BSON("" << 2), sortKey.getValue());
+    ASSERT_VALUE_EQ(Value(2), sortKey.getValue());
 }
 
 TEST_F(SortKeyGeneratorWorkingSetTest, CanGetSortKeyFromWorkingSetMemberWithOwnedObj) {
@@ -307,7 +275,7 @@ TEST_F(SortKeyGeneratorWorkingSetTest, CanGetSortKeyFromWorkingSetMemberWithOwne
     setOwnedObj(BSON("x" << 1 << "a" << 2 << "y" << 3));
     auto sortKey = sortKeyGen->computeSortKey(member());
     ASSERT_OK(sortKey);
-    ASSERT_BSONOBJ_EQ(BSON("" << 2), sortKey.getValue());
+    ASSERT_VALUE_EQ(Value(2), sortKey.getValue());
 }
 
 TEST_F(SortKeyGeneratorWorkingSetTest, CanGenerateKeyFromWSMForTextScoreMetaSort) {
@@ -317,7 +285,7 @@ TEST_F(SortKeyGeneratorWorkingSetTest, CanGenerateKeyFromWSMForTextScoreMetaSort
     member().metadata().setTextScore(9.9);
     auto sortKey = sortKeyGen->computeSortKey(member());
     ASSERT_OK(sortKey);
-    ASSERT_BSONOBJ_EQ(BSON("" << 2 << "" << 9.9 << "" << 6), sortKey.getValue());
+    ASSERT_VALUE_EQ(Value({Value(2), Value(9.9), Value(6)}), sortKey.getValue());
 }
 
 TEST_F(SortKeyGeneratorWorkingSetTest, CanGenerateSortKeyFromWSMInIndexKeyState) {
@@ -325,7 +293,7 @@ TEST_F(SortKeyGeneratorWorkingSetTest, CanGenerateSortKeyFromWSMInIndexKeyState)
     setRecordIdAndIdx(BSON("a" << 1 << "b" << 1), BSON("" << 2 << "" << 3));
     auto sortKey = sortKeyGen->computeSortKey(member());
     ASSERT_OK(sortKey);
-    ASSERT_BSONOBJ_EQ(BSON("" << 2), sortKey.getValue());
+    ASSERT_VALUE_EQ(Value(2), sortKey.getValue());
 }
 
 TEST_F(SortKeyGeneratorWorkingSetTest, CanGenerateSortKeyFromWSMInIndexKeyStateWithCollator) {
@@ -338,9 +306,7 @@ TEST_F(SortKeyGeneratorWorkingSetTest, CanGenerateSortKeyFromWSMInIndexKeyStateW
                            << "string2"));
     auto sortKey = sortKeyGen->computeSortKey(member());
     ASSERT_OK(sortKey);
-    ASSERT_BSONOBJ_EQ(BSON(""
-                           << "1gnirts"),
-                      sortKey.getValue());
+    ASSERT_VALUE_EQ(Value("1gnirts"_sd), sortKey.getValue());
 }
 
 DEATH_TEST_F(SortKeyGeneratorWorkingSetTest,
diff --git a/src/mongo/db/pipeline/document.cpp b/src/mongo/db/pipeline/document.cpp
index cf7c58a819e..9ffab53c34c 100644
--- a/src/mongo/db/pipeline/document.cpp
+++ b/src/mongo/db/pipeline/document.cpp
@@ -36,6 +36,7 @@
 #include "mongo/bson/bson_depth.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/pipeline/field_path.h"
+#include "mongo/db/pipeline/resume_token.h"
 #include "mongo/util/str.h"
 
 namespace mongo {
@@ -380,7 +381,37 @@ void DocumentStorage::loadLazyMetadata() const {
             } else if (fieldName == Document::metaFieldRandVal) {
                 _metadataFields.setRandVal(elem.Double());
             } else if (fieldName == Document::metaFieldSortKey) {
-                _metadataFields.setSortKey(elem.Obj());
+                auto bsonSortKey = elem.Obj();
+
+                bool isSingleElementKey = false;
+
+                BSONObjIterator sortKeyIt(bsonSortKey);
+                uassert(31282, "Empty sort key in metadata", sortKeyIt.more());
+                auto firstElementName = sortKeyIt.next().fieldNameStringData();
+
+                // If the sort key has exactly one field, we say it is a "single element key."
+                boost::optional<StringData> secondElementName;
+                if (!sortKeyIt.more()) {
+                    isSingleElementKey = true;
+                } else {
+                    secondElementName = sortKeyIt.next().fieldNameStringData();
+                }
+
+                // If the sort key looks like {_data: ...} or {_data: ..., _typeBits: ...}, we know
+                // that it came from a change stream, and we also treat it as a "single element
+                // key."
+                if (!sortKeyIt.more() && (firstElementName == ResumeToken::kDataFieldName) &&
+                    (!secondElementName || secondElementName == ResumeToken::kTypeBitsFieldName)) {
+                    // TODO (SERVER-43361): In 4.2 and earlier, the "sort key" for a change stream
+                    // document gets serialized differently than sort keys for normal pipeline
+                    // documents.
+                    isSingleElementKey = true;
+                    _metadataFields.setSortKey(Value(bsonSortKey), isSingleElementKey);
+                } else {
+                    _metadataFields.setSortKey(
+                        DocumentMetadataFields::deserializeSortKey(isSingleElementKey, bsonSortKey),
+                        isSingleElementKey);
+                }
             } else if (fieldName == Document::metaFieldGeoNearDistance) {
                 _metadataFields.setGeoNearDistance(elem.Double());
             } else if (fieldName == Document::metaFieldGeoNearPoint) {
@@ -459,15 +490,27 @@ constexpr StringData Document::metaFieldGeoNearPoint;
 constexpr StringData Document::metaFieldSearchScore;
 constexpr StringData Document::metaFieldSearchHighlights;
 
-BSONObj Document::toBsonWithMetaData() const {
+BSONObj Document::toBsonWithMetaData(bool use42ChangeStreamSortKeys) const {
     BSONObjBuilder bb;
     toBson(&bb);
     if (metadata().hasTextScore())
         bb.append(metaFieldTextScore, metadata().getTextScore());
     if (metadata().hasRandVal())
         bb.append(metaFieldRandVal, metadata().getRandVal());
-    if (metadata().hasSortKey())
-        bb.append(metaFieldSortKey, metadata().getSortKey());
+    if (metadata().hasSortKey()) {
+        if (use42ChangeStreamSortKeys) {
+            // TODO (SERVER-43361): In 4.2 and earlier, the "sort key" for a change stream document
+            // gets serialized differently than sort keys for normal pipeline documents. Once we no
+            // longer need to support that format, we can remove the 'use42ChangeStreamSortKeys'
+            // flag and this special case along with it.
+            invariant(metadata().isSingleElementKey());
+            metadata().getSortKey().addToBsonObj(&bb, metaFieldSortKey);
+        } else {
+            bb.append(metaFieldSortKey,
+                      DocumentMetadataFields::serializeSortKey(metadata().isSingleElementKey(),
+                                                               metadata().getSortKey()));
+        }
+    }
     if (metadata().hasGeoNearDistance())
         bb.append(metaFieldGeoNearDistance, metadata().getGeoNearDistance());
     if (metadata().hasGeoNearPoint())
diff --git a/src/mongo/db/pipeline/document.h b/src/mongo/db/pipeline/document.h
index fefbcbcdd67..3be4444c95d 100644
--- a/src/mongo/db/pipeline/document.h
+++ b/src/mongo/db/pipeline/document.h
@@ -222,10 +222,14 @@ public:
     BSONObj toBson() const;
 
     /**
-     * Like toBson, but includes metadata at the top-level.
-     * Output is parseable by fromBsonWithMetaData
-     */
-    BSONObj toBsonWithMetaData() const;
+     * Like the 'toBson()' method, but includes metadata at the top-level. When
+     * 'use42ChangeStreamSortKeys' is true, we assume that any Value in the "sortKey" metadata
+     * represents the resume token, which gets assigned directly to the "$sortKey" field. Otherwise,
+     * the "$sortKey" field gets assigned using DocumentMetadataFields::serializeSortKey(). Output
+     * is parseable by the 'fromBsonWithMetaData()' method. Note that parsing is able to infer the
+     * value of 'use42ChangeStreamSortKeys' from the format of the '$sortKey' field.
+     */
+    BSONObj toBsonWithMetaData(bool use42ChangeStreamSortKeys = false) const;
 
     /**
      * Like Document(BSONObj) but treats top-level fields with special names as metadata.
diff --git a/src/mongo/db/pipeline/document_metadata_fields.cpp b/src/mongo/db/pipeline/document_metadata_fields.cpp
index 73120cd68ed..62617144b03 100644
--- a/src/mongo/db/pipeline/document_metadata_fields.cpp
+++ b/src/mongo/db/pipeline/document_metadata_fields.cpp
@@ -65,7 +65,7 @@ void DocumentMetadataFields::mergeWith(const DocumentMetadataFields& other) {
         setRandVal(other.getRandVal());
     }
     if (!hasSortKey() && other.hasSortKey()) {
-        setSortKey(other.getSortKey());
+        setSortKey(other.getSortKey(), other.isSingleElementKey());
     }
     if (!hasGeoNearDistance() && other.hasGeoNearDistance()) {
         setGeoNearDistance(other.getGeoNearDistance());
@@ -92,7 +92,7 @@ void DocumentMetadataFields::copyFrom(const DocumentMetadataFields& other) {
         setRandVal(other.getRandVal());
     }
     if (other.hasSortKey()) {
-        setSortKey(other.getSortKey());
+        setSortKey(other.getSortKey(), other.isSingleElementKey());
     }
     if (other.hasGeoNearDistance()) {
         setGeoNearDistance(other.getGeoNearDistance());
@@ -121,7 +121,7 @@ size_t DocumentMetadataFields::getApproximateSize() const {
     size_t size = sizeof(MetadataHolder);
 
     // Count the "deep" portion of the metadata values.
-    size += _holder->sortKey.objsize();
+    size += _holder->sortKey.getApproximateSize();
     size += _holder->geoNearPoint.getApproximateSize();
     // Size of Value is double counted - once in sizeof(MetadataFields) and once in
     // getApproximateSize()
@@ -150,7 +150,8 @@ void DocumentMetadataFields::serializeForSorter(BufBuilder& buf) const {
     }
     if (hasSortKey()) {
         buf.appendNum(static_cast<char>(MetaType::kSortKey + 1));
-        getSortKey().appendSelfToBufBuilder(buf);
+        buf.appendChar(isSingleElementKey() ? 1 : 0);
+        getSortKey().serializeForSorter(buf);
     }
     if (hasGeoNearDistance()) {
         buf.appendNum(static_cast<char>(MetaType::kGeoNearDist + 1));
@@ -184,8 +185,9 @@ void DocumentMetadataFields::deserializeForSorter(BufReader& buf, DocumentMetada
         } else if (marker == static_cast<char>(MetaType::kRandVal) + 1) {
             out->setRandVal(buf.read<LittleEndian<double>>());
         } else if (marker == static_cast<char>(MetaType::kSortKey) + 1) {
-            out->setSortKey(
-                BSONObj::deserializeForSorter(buf, BSONObj::SorterDeserializeSettings()));
+            char isSingleElementKey = buf.read<char>();
+            out->setSortKey(Value::deserializeForSorter(buf, Value::SorterDeserializeSettings()),
+                            isSingleElementKey);
         } else if (marker == static_cast<char>(MetaType::kGeoNearDist) + 1) {
             out->setGeoNearDistance(buf.read<LittleEndian<double>>());
         } else if (marker == static_cast<char>(MetaType::kGeoNearPoint) + 1) {
diff --git a/src/mongo/db/pipeline/document_metadata_fields.h b/src/mongo/db/pipeline/document_metadata_fields.h
index 609aa8c4c5c..59db8dc941f 100644
--- a/src/mongo/db/pipeline/document_metadata_fields.h
+++ b/src/mongo/db/pipeline/document_metadata_fields.h
@@ -170,18 +170,23 @@ public:
         return _holder && _holder->metaFields.test(MetaType::kSortKey);
     }
 
-    BSONObj getSortKey() const {
+    Value getSortKey() const {
         invariant(hasSortKey());
         return _holder->sortKey;
     }
 
-    void setSortKey(BSONObj sortKey) {
+    void setSortKey(Value sortKey, bool isSingleElementKey) {
         if (!_holder) {
             _holder = std::make_unique<MetadataHolder>();
         }
 
         _holder->metaFields.set(MetaType::kSortKey);
-        _holder->sortKey = sortKey.getOwned();
+        _holder->isSingleElementKey = isSingleElementKey;
+        _holder->sortKey = std::move(sortKey);
+    }
+
+    bool isSingleElementKey() const {
+        return _holder && _holder->isSingleElementKey;
     }
 
     bool hasGeoNearDistance() const {
@@ -298,9 +303,15 @@ private:
     // A simple data struct housing all possible metadata fields.
     struct MetadataHolder {
         std::bitset<MetaType::kNumFields> metaFields;
+
+        // True when the sort key corresponds to a single-element sort pattern, meaning that
+        // comparisons should treat the sort key value as a single element, even if it is an array.
+        // Only relevant when 'kSortKey' is set.
+        bool isSingleElementKey;
+
         double textScore{0.0};
         double randVal{0.0};
-        BSONObj sortKey;
+        Value sortKey;
         double geoNearDistance{0.0};
         Value geoNearPoint;
         double searchScore{0.0};
diff --git a/src/mongo/db/pipeline/document_metadata_fields_test.cpp b/src/mongo/db/pipeline/document_metadata_fields_test.cpp
index 5e46ea7de7f..fc9c8be8779 100644
--- a/src/mongo/db/pipeline/document_metadata_fields_test.cpp
+++ b/src/mongo/db/pipeline/document_metadata_fields_test.cpp
@@ -42,7 +42,7 @@ TEST(DocumentMetadataFieldsTest, AllMetadataRoundtripsThroughSerialization) {
     DocumentMetadataFields metadata;
     metadata.setTextScore(9.9);
     metadata.setRandVal(42.0);
-    metadata.setSortKey(BSON("a" << 1));
+    metadata.setSortKey(Value(1), /* isSingleElementKey = */ true);
     metadata.setGeoNearDistance(3.2);
     metadata.setGeoNearPoint(Value{BSON_ARRAY(1 << 2)});
     metadata.setSearchScore(5.4);
@@ -57,7 +57,8 @@ TEST(DocumentMetadataFieldsTest, AllMetadataRoundtripsThroughSerialization) {
 
     ASSERT_EQ(deserialized.getTextScore(), 9.9);
     ASSERT_EQ(deserialized.getRandVal(), 42.0);
-    ASSERT_BSONOBJ_EQ(deserialized.getSortKey(), BSON("a" << 1));
+    ASSERT_VALUE_EQ(deserialized.getSortKey(), Value(1));
+    ASSERT_TRUE(deserialized.isSingleElementKey());
     ASSERT_EQ(deserialized.getGeoNearDistance(), 3.2);
     ASSERT_VALUE_EQ(deserialized.getGeoNearPoint(), Value{BSON_ARRAY(1 << 2)});
     ASSERT_EQ(deserialized.getSearchScore(), 5.4);
@@ -90,8 +91,9 @@ TEST(DocumentMetadataFieldsTest, HasMethodsReturnTrueForInitializedMetadata) {
     ASSERT_TRUE(metadata.hasRandVal());
 
     ASSERT_FALSE(metadata.hasSortKey());
-    metadata.setSortKey(BSON("a" << 1));
+    metadata.setSortKey(Value(1), /* isSingleElementKey = */ true);
     ASSERT_TRUE(metadata.hasSortKey());
+    ASSERT_TRUE(metadata.isSingleElementKey());
 
     ASSERT_FALSE(metadata.hasGeoNearDistance());
     metadata.setGeoNearDistance(3.2);
@@ -118,7 +120,7 @@ TEST(DocumentMetadataFieldsTest, MoveConstructor) {
     DocumentMetadataFields metadata;
     metadata.setTextScore(9.9);
     metadata.setRandVal(42.0);
-    metadata.setSortKey(BSON("a" << 1));
+    metadata.setSortKey(Value(1), /* isSingleElementKey = */ true);
     metadata.setGeoNearDistance(3.2);
     metadata.setGeoNearPoint(Value{BSON_ARRAY(1 << 2)});
     metadata.setSearchScore(5.4);
@@ -129,7 +131,8 @@ TEST(DocumentMetadataFieldsTest, MoveConstructor) {
     ASSERT_TRUE(moveConstructed);
     ASSERT_EQ(moveConstructed.getTextScore(), 9.9);
     ASSERT_EQ(moveConstructed.getRandVal(), 42.0);
-    ASSERT_BSONOBJ_EQ(moveConstructed.getSortKey(), BSON("a" << 1));
+    ASSERT_VALUE_EQ(moveConstructed.getSortKey(), Value(1));
+    ASSERT_TRUE(moveConstructed.isSingleElementKey());
     ASSERT_EQ(moveConstructed.getGeoNearDistance(), 3.2);
     ASSERT_VALUE_EQ(moveConstructed.getGeoNearPoint(), Value{BSON_ARRAY(1 << 2)});
     ASSERT_EQ(moveConstructed.getSearchScore(), 5.4);
@@ -143,7 +146,7 @@ TEST(DocumentMetadataFieldsTest, MoveAssignmentOperator) {
     DocumentMetadataFields metadata;
     metadata.setTextScore(9.9);
     metadata.setRandVal(42.0);
-    metadata.setSortKey(BSON("a" << 1));
+    metadata.setSortKey(Value(1), /* isSingleElementKey = */ true);
     metadata.setGeoNearDistance(3.2);
     metadata.setGeoNearPoint(Value{BSON_ARRAY(1 << 2)});
     metadata.setSearchScore(5.4);
@@ -157,7 +160,8 @@ TEST(DocumentMetadataFieldsTest, MoveAssignmentOperator) {
 
     ASSERT_EQ(moveAssigned.getTextScore(), 9.9);
     ASSERT_EQ(moveAssigned.getRandVal(), 42.0);
-    ASSERT_BSONOBJ_EQ(moveAssigned.getSortKey(), BSON("a" << 1));
+    ASSERT_VALUE_EQ(moveAssigned.getSortKey(), Value(1));
+    ASSERT_TRUE(moveAssigned.isSingleElementKey());
     ASSERT_EQ(moveAssigned.getGeoNearDistance(), 3.2);
     ASSERT_VALUE_EQ(moveAssigned.getGeoNearPoint(), Value{BSON_ARRAY(1 << 2)});
     ASSERT_EQ(moveAssigned.getSearchScore(), 5.4);
@@ -197,7 +201,7 @@ TEST(DocumentMetadataFieldsTest, MergeWithOnlyCopiesMetadataThatDestinationDoesN
     DocumentMetadataFields source;
     source.setTextScore(9.9);
     source.setRandVal(42.0);
-    source.setSortKey(BSON("a" << 1));
+    source.setSortKey(Value(1), /* isSingleElementKey = */ true);
     source.setGeoNearDistance(3.2);
 
     DocumentMetadataFields destination;
@@ -208,7 +212,8 @@ TEST(DocumentMetadataFieldsTest, MergeWithOnlyCopiesMetadataThatDestinationDoesN
 
     ASSERT_EQ(destination.getTextScore(), 12.3);
     ASSERT_EQ(destination.getRandVal(), 84.0);
-    ASSERT_BSONOBJ_EQ(destination.getSortKey(), BSON("a" << 1));
+    ASSERT_VALUE_EQ(destination.getSortKey(), Value(1));
+    ASSERT_TRUE(destination.isSingleElementKey());
     ASSERT_EQ(destination.getGeoNearDistance(), 3.2);
     ASSERT_FALSE(destination.hasGeoNearPoint());
     ASSERT_FALSE(destination.hasSearchScore());
@@ -220,7 +225,7 @@ TEST(DocumentMetadataFieldsTest, CopyFromCopiesAllMetadataThatSourceHas) {
     DocumentMetadataFields source;
     source.setTextScore(9.9);
     source.setRandVal(42.0);
-    source.setSortKey(BSON("a" << 1));
+    source.setSortKey(Value(1), /* isSingleElementKey = */ true);
     source.setGeoNearDistance(3.2);
 
     DocumentMetadataFields destination;
@@ -231,7 +236,8 @@ TEST(DocumentMetadataFieldsTest, CopyFromCopiesAllMetadataThatSourceHas) {
 
     ASSERT_EQ(destination.getTextScore(), 9.9);
     ASSERT_EQ(destination.getRandVal(), 42.0);
-    ASSERT_BSONOBJ_EQ(destination.getSortKey(), BSON("a" << 1));
+    ASSERT_VALUE_EQ(destination.getSortKey(), Value(1));
+    ASSERT_TRUE(destination.isSingleElementKey());
     ASSERT_EQ(destination.getGeoNearDistance(), 3.2);
     ASSERT_FALSE(destination.hasGeoNearPoint());
     ASSERT_FALSE(destination.hasSearchScore());
diff --git a/src/mongo/db/pipeline/document_source_change_stream_test.cpp b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
index e9248465431..592e3190786 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
@@ -1687,11 +1687,10 @@ TEST_F(ChangeStreamStageTest, UsesResumeTokenAsSortKeyIfNeedsMergeIsFalse) {
 
     auto next = stages.back()->getNext();
 
-    auto expectedSortKey =
-        makeResumeToken(kDefaultTs, testUuid(), BSON("x" << 2 << "_id" << 1)).toBson();
+    auto expectedSortKey = makeResumeToken(kDefaultTs, testUuid(), BSON("x" << 2 << "_id" << 1));
 
     ASSERT_TRUE(next.isAdvanced());
-    ASSERT_BSONOBJ_EQ(next.releaseDocument().metadata().getSortKey(), expectedSortKey);
+    ASSERT_VALUE_EQ(next.releaseDocument().metadata().getSortKey(), Value(expectedSortKey));
 }
 
 //
diff --git a/src/mongo/db/pipeline/document_source_change_stream_transform.cpp b/src/mongo/db/pipeline/document_source_change_stream_transform.cpp
index 5a558a924b9..a147e5271c3 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_transform.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_transform.cpp
@@ -328,7 +328,8 @@ Document DocumentSourceChangeStreamTransform::applyTransformation(const Document
 
     // We set the resume token as the document's sort key in both the sharded and non-sharded cases,
     // since we will subsequently rely upon it to generate a correct postBatchResumeToken.
-    doc.metadata().setSortKey(resumeToken.toBson());
+    const bool isSingleElementKey = true;
+    doc.metadata().setSortKey(Value{resumeToken}, isSingleElementKey);
 
     // "invalidate" and "newShardDetected" entries have fewer fields.
     if (operationType == DocumentSourceChangeStream::kInvalidateOpType ||
diff --git a/src/mongo/db/pipeline/document_source_check_invalidate.cpp b/src/mongo/db/pipeline/document_source_check_invalidate.cpp
index 5b286b585b0..330ef4d02df 100644
--- a/src/mongo/db/pipeline/document_source_check_invalidate.cpp
+++ b/src/mongo/db/pipeline/document_source_check_invalidate.cpp
@@ -105,7 +105,8 @@ DocumentSource::GetNextResult DocumentSourceCheckInvalidate::doGetNext() {
         // We set the resume token as the document's sort key in both the sharded and non-sharded
         // cases, since we will later rely upon it to generate a correct postBatchResumeToken. We
         // must therefore update the sort key to match the new resume token that we generated above.
-        result.metadata().setSortKey(resumeTokenDoc.toBson());
+        const bool isSingleElementKey = true;
+        result.metadata().setSortKey(Value{resumeTokenDoc}, isSingleElementKey);
 
         _queuedInvalidate = result.freeze();
     }
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 24253846166..1fa3e8d673c 100644
--- a/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_geo_near_cursor.cpp
@@ -105,7 +105,8 @@ Document DocumentSourceGeoNearCursor::transformBSONObjToDocument(const BSONObj&
 
     // In a cluster, $geoNear will be merged via $sort, so add the sort key.
     if (pExpCtx->needsMerge) {
-        output.metadata().setSortKey(BSON("" << distance));
+        const bool isSingleElementKey = true;
+        output.metadata().setSortKey(Value(distance), isSingleElementKey);
     }
 
     return output.freeze();
diff --git a/src/mongo/db/pipeline/document_source_sample_from_random_cursor.cpp b/src/mongo/db/pipeline/document_source_sample_from_random_cursor.cpp
index ba2c99cbbfb..31389b9d44d 100644
--- a/src/mongo/db/pipeline/document_source_sample_from_random_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_sample_from_random_cursor.cpp
@@ -94,7 +94,8 @@ DocumentSource::GetNextResult DocumentSourceSampleFromRandomCursor::doGetNext()
     if (pExpCtx->needsMerge) {
         // This stage will be merged by sorting results according to this random metadata field, but
         // the merging logic expects to sort by the sort key metadata.
-        md.metadata().setSortKey(BSON("" << _randMetaFieldVal));
+        const bool isSingleElementKey = true;
+        md.metadata().setSortKey(Value(_randMetaFieldVal), isSingleElementKey);
     }
     return md.freeze();
 }
diff --git a/src/mongo/db/pipeline/document_source_sort.cpp b/src/mongo/db/pipeline/document_source_sort.cpp
index 027d88037f8..5bce2d6acd0 100644
--- a/src/mongo/db/pipeline/document_source_sort.cpp
+++ b/src/mongo/db/pipeline/document_source_sort.cpp
@@ -215,20 +215,17 @@ bool DocumentSourceSort::usedDisk() {
 }
 
 std::pair<Value, Document> DocumentSourceSort::extractSortKey(Document&& doc) const {
-    Value inMemorySortKey = _sortKeyGen->computeSortKeyFromDocument(doc);
+    Value sortKey = _sortKeyGen->computeSortKeyFromDocument(doc);
 
     if (pExpCtx->needsMerge) {
         // If this sort stage is part of a merged pipeline, make sure that each Document's sort key
         // gets saved with its metadata.
-        auto serializedSortKey = DocumentMetadataFields::serializeSortKey(
-            _sortKeyGen->isSingleElementKey(), inMemorySortKey);
-
         MutableDocument toBeSorted(std::move(doc));
-        toBeSorted.metadata().setSortKey(serializedSortKey);
+        toBeSorted.metadata().setSortKey(sortKey, _sortKeyGen->isSingleElementKey());
 
-        return std::make_pair(std::move(inMemorySortKey), toBeSorted.freeze());
+        return std::make_pair(std::move(sortKey), toBeSorted.freeze());
     } else {
-        return std::make_pair(std::move(inMemorySortKey), std::move(doc));
+        return std::make_pair(std::move(sortKey), std::move(doc));
     }
 }
 
diff --git a/src/mongo/db/pipeline/document_value_test.cpp b/src/mongo/db/pipeline/document_value_test.cpp
index 43ed44e5c9a..872743679a4 100644
--- a/src/mongo/db/pipeline/document_value_test.cpp
+++ b/src/mongo/db/pipeline/document_value_test.cpp
@@ -622,7 +622,7 @@ TEST(MetaFields, CopyMetadataFromCopiesAllMetadata) {
 
     ASSERT_EQ(result.metadata().getTextScore(), 9.9);
     ASSERT_EQ(result.metadata().getRandVal(), 42.0);
-    ASSERT_BSONOBJ_EQ(result.metadata().getSortKey(), BSON("x" << 1));
+    ASSERT_VALUE_EQ(result.metadata().getSortKey(), Value(1));
     ASSERT_EQ(result.metadata().getGeoNearDistance(), 3.2);
     ASSERT_VALUE_EQ(result.metadata().getGeoNearPoint(), Value{BSON_ARRAY(1 << 2)});
     ASSERT_EQ(result.metadata().getSearchScore(), 5.4);
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index 9d21614c1ed..2ac2bd0402f 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -2623,7 +2623,10 @@ Value ExpressionMeta::evaluate(const Document& root, Variables* variables) const
         case MetaType::kIndexKey:
             return metadata.hasIndexKey() ? Value(metadata.getIndexKey()) : Value();
         case MetaType::kSortKey:
-            return metadata.hasSortKey() ? Value(metadata.getSortKey()) : Value();
+            return metadata.hasSortKey()
+                ? Value(DocumentMetadataFields::serializeSortKey(metadata.isSingleElementKey(),
+                                                                 metadata.getSortKey()))
+                : Value();
         default:
             MONGO_UNREACHABLE;
     }
diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index dbac88c28e1..e7e4c7ac2a7 100644
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -277,6 +277,10 @@ public:
     boost::optional<ServerGlobalParams::FeatureCompatibility::Version>
         maxFeatureCompatibilityVersion;
 
+    // True if this ExpressionContext is associated with a Change Stream that should serialize its
+    // "$sortKey" using the 4.2 format.
+    bool use42ChangeStreamSortKeys = false;
+
 protected:
     static const int kInterruptCheckPeriod = 128;
 
diff --git a/src/mongo/db/pipeline/expression_test.cpp b/src/mongo/db/pipeline/expression_test.cpp
index 9ed9e3a03b8..3043399787a 100644
--- a/src/mongo/db/pipeline/expression_test.cpp
+++ b/src/mongo/db/pipeline/expression_test.cpp
@@ -6199,10 +6199,10 @@ TEST(ExpressionMetaTest, ExpressionMetaSortKey) {
         ExpressionMeta::parse(expCtx, expr.firstElement(), expCtx->variablesParseState);
 
     MutableDocument doc;
-    BSONObj sortKey = BSON("" << 1 << "" << 2);
-    doc.metadata().setSortKey(sortKey);
+    Value sortKey = Value({Value(1), Value(2)});
+    doc.metadata().setSortKey(sortKey, /* isSingleElementSortKey = */ false);
     Value val = expressionMeta->evaluate(doc.freeze(), &expCtx->variables);
-    ASSERT_DOCUMENT_EQ(val.getDocument(), Document(sortKey));
+    ASSERT_BSONOBJ_EQ(val.getDocument().toBson(), BSON("" << 1 << "" << 2));
 }
 
 TEST(ExpressionMetaTest, ExpressionMetaTextScore) {
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index 14584438e03..4b183916e35 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -631,6 +631,11 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
     if (pipeline->peekFront() && pipeline->peekFront()->constraints().isChangeStreamStage()) {
         invariant(expCtx->tailableMode == TailableModeEnum::kTailableAndAwaitData);
         plannerOpts |= QueryPlannerParams::TRACK_LATEST_OPLOG_TS;
+
+        // TODO (SERVER-42713): When we change the format of Change Stream sort keys for 4.4, this
+        // function will determine whether we use the new format, based on the AggregationRequest
+        // parameters. For now, we always use the old 4.2 format.
+        expCtx->use42ChangeStreamSortKeys = true;
     }
 
     if (rewrittenGroupStage) {
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index eb6bda6c7b4..e751b0d31de 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -578,13 +578,13 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     invariant(executionResult.getValue().root);
     // We must have a tree of stages in order to have a valid plan executor, but the query
     // solution may be null.
-    return PlanExecutor::make(opCtx,
+    return PlanExecutor::make(std::move(executionResult.getValue().canonicalQuery),
                               std::move(ws),
                               std::move(executionResult.getValue().root),
-                              std::move(executionResult.getValue().querySolution),
-                              std::move(executionResult.getValue().canonicalQuery),
                               collection,
-                              yieldPolicy);
+                              yieldPolicy,
+                              NamespaceString(),
+                              std::move(executionResult.getValue().querySolution));
 }
 
 //
@@ -723,7 +723,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDelete(
         LOG(2) << "Collection " << nss.ns() << " does not exist."
                << " Using EOF stage: " << redact(request->getQuery());
         return PlanExecutor::make(
-            opCtx, std::move(ws), std::make_unique<EOFStage>(opCtx), nss, policy);
+            opCtx, std::move(ws), std::make_unique<EOFStage>(opCtx), nullptr, policy, nss);
     }
 
     if (!parsedDelete->hasParsedQuery()) {
@@ -798,13 +798,13 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDelete(
 
     // We must have a tree of stages in order to have a valid plan executor, but the query
     // solution may be null.
-    return PlanExecutor::make(opCtx,
+    return PlanExecutor::make(std::move(cq),
                               std::move(ws),
                               std::move(root),
-                              std::move(querySolution),
-                              std::move(cq),
                               collection,
-                              policy);
+                              policy,
+                              NamespaceString(),
+                              std::move(querySolution));
 }
 
 //
@@ -862,7 +862,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorUpdate(
         LOG(2) << "Collection " << nss.ns() << " does not exist."
                << " Using EOF stage: " << redact(request->getQuery());
         return PlanExecutor::make(
-            opCtx, std::move(ws), std::make_unique<EOFStage>(opCtx), nss, policy);
+            opCtx, std::move(ws), std::make_unique<EOFStage>(opCtx), nullptr, policy, nss);
     }
 
     // Pass index information to the update driver, so that it can determine for us whether the
@@ -942,13 +942,13 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorUpdate(
 
     // We must have a tree of stages in order to have a valid plan executor, but the query
     // solution may be null. Takes ownership of all args other than 'collection' and 'opCtx'
-    return PlanExecutor::make(opCtx,
+    return PlanExecutor::make(std::move(cq),
                               std::move(ws),
                               std::move(root),
-                              std::move(querySolution),
-                              std::move(cq),
                               collection,
-                              policy);
+                              policy,
+                              NamespaceString(),
+                              std::move(querySolution));
 }
 
 //
@@ -1106,7 +1106,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCount(
         // this case we put a CountStage on top of an EOFStage.
         unique_ptr<PlanStage> root = std::make_unique<CountStage>(
             opCtx, collection, limit, skip, ws.get(), new EOFStage(opCtx));
-        return PlanExecutor::make(opCtx, std::move(ws), std::move(root), nss, yieldPolicy);
+        return PlanExecutor::make(opCtx, std::move(ws), std::move(root), nullptr, yieldPolicy, nss);
     }
 
     // If the query is empty, then we can determine the count by just asking the collection
@@ -1121,7 +1121,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCount(
     if (useRecordStoreCount) {
         unique_ptr<PlanStage> root =
             std::make_unique<RecordStoreFastCountStage>(opCtx, collection, skip, limit);
-        return PlanExecutor::make(opCtx, std::move(ws), std::move(root), nss, yieldPolicy);
+        return PlanExecutor::make(opCtx, std::move(ws), std::move(root), nullptr, yieldPolicy, nss);
     }
 
     size_t plannerOptions = QueryPlannerParams::IS_COUNT;
@@ -1144,13 +1144,13 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorCount(
     root = std::make_unique<CountStage>(opCtx, collection, limit, skip, ws.get(), root.release());
     // We must have a tree of stages in order to have a valid plan executor, but the query
     // solution may be NULL. Takes ownership of all args other than 'collection' and 'opCtx'
-    return PlanExecutor::make(opCtx,
+    return PlanExecutor::make(std::move(cq),
                               std::move(ws),
                               std::move(root),
-                              std::move(querySolution),
-                              std::move(cq),
                               collection,
-                              yieldPolicy);
+                              yieldPolicy,
+                              NamespaceString(),
+                              std::move(querySolution));
 }
 
 //
@@ -1454,13 +1454,13 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorForS
     LOG(2) << "Using fast distinct: " << redact(parsedDistinct->getQuery()->toStringShort())
            << ", planSummary: " << Explain::getPlanSummary(root.get());
 
-    return PlanExecutor::make(opCtx,
+    return PlanExecutor::make(parsedDistinct->releaseQuery(),
                               std::move(ws),
                               std::move(root),
-                              std::move(soln),
-                              parsedDistinct->releaseQuery(),
                               collection,
-                              yieldPolicy);
+                              yieldPolicy,
+                              NamespaceString(),
+                              std::move(soln));
 }
 
 // Checks each solution in the 'solutions' vector to see if one includes an IXSCAN that can be
@@ -1495,13 +1495,13 @@ getExecutorDistinctFromIndexSolutions(OperationContext* opCtx,
             LOG(2) << "Using fast distinct: " << redact(parsedDistinct->getQuery()->toStringShort())
                    << ", planSummary: " << Explain::getPlanSummary(root.get());
 
-            return PlanExecutor::make(opCtx,
+            return PlanExecutor::make(parsedDistinct->releaseQuery(),
                                       std::move(ws),
                                       std::move(root),
-                                      std::move(currentSolution),
-                                      parsedDistinct->releaseQuery(),
                                       collection,
-                                      yieldPolicy);
+                                      yieldPolicy,
+                                      NamespaceString(),
+                                      std::move(currentSolution));
         }
     }
 
@@ -1545,10 +1545,9 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorDistinct(
 
     if (!collection) {
         // Treat collections that do not exist as empty collections.
-        return PlanExecutor::make(opCtx,
+        return PlanExecutor::make(parsedDistinct->releaseQuery(),
                                   std::make_unique<WorkingSet>(),
                                   std::make_unique<EOFStage>(opCtx),
-                                  parsedDistinct->releaseQuery(),
                                   collection,
                                   yieldPolicy);
     }
diff --git a/src/mongo/db/query/internal_plans.cpp b/src/mongo/db/query/internal_plans.cpp
index 860cf5a4ebd..53328901c88 100644
--- a/src/mongo/db/query/internal_plans.cpp
+++ b/src/mongo/db/query/internal_plans.cpp
@@ -58,7 +58,7 @@ std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> InternalPlanner::collection
         auto eof = std::make_unique<EOFStage>(opCtx);
         // Takes ownership of 'ws' and 'eof'.
         auto statusWithPlanExecutor = PlanExecutor::make(
-            opCtx, std::move(ws), std::move(eof), NamespaceString(ns), yieldPolicy);
+            opCtx, std::move(ws), std::move(eof), nullptr, yieldPolicy, NamespaceString(ns));
         invariant(statusWithPlanExecutor.isOK());
         return std::move(statusWithPlanExecutor.getValue());
     }
diff --git a/src/mongo/db/query/plan_executor.h b/src/mongo/db/query/plan_executor.h
index 03f94017379..a0bf2344a6b 100644
--- a/src/mongo/db/query/plan_executor.h
+++ b/src/mongo/db/query/plan_executor.h
@@ -197,8 +197,8 @@ public:
     //
     // On success, return a new PlanExecutor, owned by the caller.
     //
-    // Passing YIELD_AUTO to any of these factories will construct a yielding executor which
-    // may yield in the following circumstances:
+    // Passing YIELD_AUTO to any of these factories will construct a yielding executor which may
+    // yield in the following circumstances:
     //   - During plan selection inside the call to make().
     //   - On any call to getNext().
     //   - On any call to restoreState().
@@ -207,55 +207,51 @@ public:
     // If auto-yielding is enabled, a yield during make() may result in the PlanExecutor being
     // killed, in which case this method will return a non-OK status.
     //
+    // All callers of these factory methods should provide either a non-null value for 'collection'
+    // or a non-empty 'nss' NamespaceString but not both.
+    //
 
     /**
-     * Used when there is no canonical query and no query solution.
-     *
-     * Right now this is only for idhack updates which neither canonicalize nor go through normal
-     * planning.
+     * Note that the PlanExecutor will use the ExpressionContext associated with 'cq' and the
+     * OperationContext associated with that ExpressionContext.
      */
     static StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
-        OperationContext* opCtx,
+        std::unique_ptr<CanonicalQuery> cq,
         std::unique_ptr<WorkingSet> ws,
         std::unique_ptr<PlanStage> rt,
         const Collection* collection,
-        YieldPolicy yieldPolicy);
+        YieldPolicy yieldPolicy,
+        NamespaceString nss = NamespaceString(),
+        std::unique_ptr<QuerySolution> qs = nullptr);
 
     /**
-     * Used when we have a NULL collection and no canonical query. In this case, we need to
-     * explicitly pass a namespace to the plan executor.
-     */
-    static StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
-        OperationContext* opCtx,
-        std::unique_ptr<WorkingSet> ws,
-        std::unique_ptr<PlanStage> rt,
-        NamespaceString nss,
-        YieldPolicy yieldPolicy);
-
-    /**
-     * Used when there is a canonical query but no query solution (e.g. idhack queries, queries
-     * against a NULL collection, queries using the subplan stage).
+     * This overload is provided for executors that do not need a CanonicalQuery. For example, the
+     * outer plan executor for an aggregate command does not have a CanonicalQuery.
+     *
+     * Note that the PlanExecutor will use the OperationContext associated with the 'expCtx'
+     * ExpressionContext.
      */
     static StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
-        OperationContext* opCtx,
+        const boost::intrusive_ptr<ExpressionContext>& expCtx,
         std::unique_ptr<WorkingSet> ws,
         std::unique_ptr<PlanStage> rt,
-        std::unique_ptr<CanonicalQuery> cq,
         const Collection* collection,
-        YieldPolicy yieldPolicy);
+        YieldPolicy yieldPolicy,
+        NamespaceString nss = NamespaceString(),
+        std::unique_ptr<QuerySolution> qs = nullptr);
 
     /**
-     * The constructor for the normal case, when you have a collection, a canonical query, and a
-     * query solution.
+     * This overload is provided for executors that do not have a CanonicalQuery or an
+     * ExpressionContext, such as an aggregation command with a $listCollections stage.
      */
     static StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
         OperationContext* opCtx,
         std::unique_ptr<WorkingSet> ws,
         std::unique_ptr<PlanStage> rt,
-        std::unique_ptr<QuerySolution> qs,
-        std::unique_ptr<CanonicalQuery> cq,
         const Collection* collection,
-        YieldPolicy yieldPolicy);
+        YieldPolicy yieldPolicy,
+        NamespaceString nss = NamespaceString(),
+        std::unique_ptr<QuerySolution> qs = nullptr);
 
     /**
      * A PlanExecutor must be disposed before destruction. In most cases, this will happen
diff --git a/src/mongo/db/query/plan_executor_impl.cpp b/src/mongo/db/query/plan_executor_impl.cpp
index 565a2e1df76..6281a2d9899 100644
--- a/src/mongo/db/query/plan_executor_impl.cpp
+++ b/src/mongo/db/query/plan_executor_impl.cpp
@@ -126,72 +126,71 @@ PlanStage* getStageByType(PlanStage* root, StageType type) {
 }
 }  // namespace
 
-// static
 StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PlanExecutor::make(
-    OperationContext* opCtx,
-    unique_ptr<WorkingSet> ws,
-    unique_ptr<PlanStage> rt,
+    std::unique_ptr<CanonicalQuery> cq,
+    std::unique_ptr<WorkingSet> ws,
+    std::unique_ptr<PlanStage> rt,
     const Collection* collection,
-    YieldPolicy yieldPolicy) {
-    return PlanExecutorImpl::make(
-        opCtx, std::move(ws), std::move(rt), nullptr, nullptr, collection, {}, yieldPolicy);
-}
-
-// static
-StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PlanExecutor::make(
-    OperationContext* opCtx,
-    unique_ptr<WorkingSet> ws,
-    unique_ptr<PlanStage> rt,
+    YieldPolicy yieldPolicy,
     NamespaceString nss,
-    YieldPolicy yieldPolicy) {
-    return PlanExecutorImpl::make(opCtx,
+    std::unique_ptr<QuerySolution> qs) {
+    auto expCtx = cq->getExpCtx();
+    return PlanExecutorImpl::make(expCtx->opCtx,
                                   std::move(ws),
                                   std::move(rt),
-                                  nullptr,
-                                  nullptr,
-                                  nullptr,
-                                  std::move(nss),
+                                  std::move(qs),
+                                  std::move(cq),
+                                  expCtx,
+                                  collection,
+                                  nss,
                                   yieldPolicy);
 }
 
-// static
 StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PlanExecutor::make(
-    OperationContext* opCtx,
-    unique_ptr<WorkingSet> ws,
-    unique_ptr<PlanStage> rt,
-    unique_ptr<CanonicalQuery> cq,
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    std::unique_ptr<WorkingSet> ws,
+    std::unique_ptr<PlanStage> rt,
     const Collection* collection,
-    YieldPolicy yieldPolicy) {
-    return PlanExecutorImpl::make(
-        opCtx, std::move(ws), std::move(rt), nullptr, std::move(cq), collection, {}, yieldPolicy);
+    YieldPolicy yieldPolicy,
+    NamespaceString nss,
+    std::unique_ptr<QuerySolution> qs) {
+    return PlanExecutorImpl::make(expCtx->opCtx,
+                                  std::move(ws),
+                                  std::move(rt),
+                                  std::move(qs),
+                                  nullptr,
+                                  expCtx,
+                                  collection,
+                                  nss,
+                                  yieldPolicy);
 }
 
-// static
 StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PlanExecutor::make(
     OperationContext* opCtx,
-    unique_ptr<WorkingSet> ws,
-    unique_ptr<PlanStage> rt,
-    unique_ptr<QuerySolution> qs,
-    unique_ptr<CanonicalQuery> cq,
+    std::unique_ptr<WorkingSet> ws,
+    std::unique_ptr<PlanStage> rt,
     const Collection* collection,
-    YieldPolicy yieldPolicy) {
+    YieldPolicy yieldPolicy,
+    NamespaceString nss,
+    std::unique_ptr<QuerySolution> qs) {
     return PlanExecutorImpl::make(opCtx,
                                   std::move(ws),
                                   std::move(rt),
                                   std::move(qs),
-                                  std::move(cq),
+                                  nullptr,
+                                  nullptr,
                                   collection,
-                                  {},
+                                  nss,
                                   yieldPolicy);
 }
 
-// static
 StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PlanExecutorImpl::make(
     OperationContext* opCtx,
     unique_ptr<WorkingSet> ws,
     unique_ptr<PlanStage> rt,
     unique_ptr<QuerySolution> qs,
     unique_ptr<CanonicalQuery> cq,
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
     const Collection* collection,
     NamespaceString nss,
     YieldPolicy yieldPolicy) {
@@ -201,6 +200,7 @@ StatusWith<unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PlanExecutorImpl::ma
                                          std::move(rt),
                                          std::move(qs),
                                          std::move(cq),
+                                         expCtx,
                                          collection,
                                          std::move(nss),
                                          yieldPolicy);
@@ -221,17 +221,22 @@ PlanExecutorImpl::PlanExecutorImpl(OperationContext* opCtx,
                                    unique_ptr<PlanStage> rt,
                                    unique_ptr<QuerySolution> qs,
                                    unique_ptr<CanonicalQuery> cq,
+                                   const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                    const Collection* collection,
                                    NamespaceString nss,
                                    YieldPolicy yieldPolicy)
     : _opCtx(opCtx),
       _cq(std::move(cq)),
+      _expCtx(_cq ? _cq->getExpCtx() : expCtx),
       _workingSet(std::move(ws)),
       _qs(std::move(qs)),
       _root(std::move(rt)),
       _nss(std::move(nss)),
       // There's no point in yielding if the collection doesn't exist.
       _yieldPolicy(makeYieldPolicy(this, collection ? yieldPolicy : NO_YIELD)) {
+    invariant(!_expCtx || _expCtx->opCtx == _opCtx);
+    invariant(!_cq || !_expCtx || _cq->getExpCtx() == _expCtx);
+
     // We may still need to initialize _nss from either collection or _cq.
     if (!_nss.isEmpty()) {
         return;  // We already have an _nss set, so there's nothing more to do.
@@ -288,7 +293,6 @@ PlanExecutorImpl::~PlanExecutorImpl() {
     invariant(_currentState == kDisposed);
 }
 
-// static
 string PlanExecutor::statestr(ExecState s) {
     if (PlanExecutor::ADVANCED == s) {
         return "ADVANCED";
@@ -541,11 +545,12 @@ PlanExecutor::ExecState PlanExecutorImpl::_getNextImpl(Snapshotted<BSONObj>* obj
                         *objOut = Snapshotted<BSONObj>(SnapshotId(), member->keyData[0].keyData);
                     }
                 } else if (member->hasObj()) {
-                    *objOut =
-                        Snapshotted<BSONObj>(member->doc.snapshotId(),
-                                             member->metadata() && member->doc.value().metadata()
-                                                 ? member->doc.value().toBsonWithMetaData()
-                                                 : member->doc.value().toBson());
+                    *objOut = Snapshotted<BSONObj>(
+                        member->doc.snapshotId(),
+                        member->metadata() && member->doc.value().metadata()
+                            ? member->doc.value().toBsonWithMetaData(
+                                  _expCtx ? _expCtx->use42ChangeStreamSortKeys : false)
+                            : member->doc.value().toBson());
                 } else {
                     _workingSet->free(id);
                     hasRequestedData = false;
diff --git a/src/mongo/db/query/plan_executor_impl.h b/src/mongo/db/query/plan_executor_impl.h
index 1b4dfda3dda..629f66c6474 100644
--- a/src/mongo/db/query/plan_executor_impl.h
+++ b/src/mongo/db/query/plan_executor_impl.h
@@ -50,6 +50,7 @@ public:
         std::unique_ptr<PlanStage> rt,
         std::unique_ptr<QuerySolution> qs,
         std::unique_ptr<CanonicalQuery> cq,
+        const boost::intrusive_ptr<ExpressionContext>& expCtx,
         const Collection* collection,
         NamespaceString nss,
         YieldPolicy yieldPolicy);
@@ -89,6 +90,7 @@ private:
                      std::unique_ptr<PlanStage> rt,
                      std::unique_ptr<QuerySolution> qs,
                      std::unique_ptr<CanonicalQuery> cq,
+                     const boost::intrusive_ptr<ExpressionContext>& expCtx,
                      const Collection* collection,
                      NamespaceString nss,
                      YieldPolicy yieldPolicy);
@@ -151,7 +153,15 @@ private:
     // detachFromOperationContext() and reattachToOperationContext().
     OperationContext* _opCtx;
 
+    // Note, this can be null. Some queries don't need a CanonicalQuery for planning. For example,
+    // aggregation queries create a PlanExecutor with no CanonicalQuery.
     std::unique_ptr<CanonicalQuery> _cq;
+
+    // When '_cq' is not null, this will point to the same ExpressionContext that is in the '_cq'
+    // object. Note that this pointer can also be null when '_cq' is null. For example a "list
+    // collections" query does not need a CanonicalQuery or ExpressionContext.
+    boost::intrusive_ptr<ExpressionContext> _expCtx;
+
     std::unique_ptr<WorkingSet> _workingSet;
     std::unique_ptr<QuerySolution> _qs;
     std::unique_ptr<PlanStage> _root;
diff --git a/src/mongo/dbtests/cursor_manager_test.cpp b/src/mongo/dbtests/cursor_manager_test.cpp
index 72934d8348a..90def084bba 100644
--- a/src/mongo/dbtests/cursor_manager_test.cpp
+++ b/src/mongo/dbtests/cursor_manager_test.cpp
@@ -76,8 +76,9 @@ public:
         return unittest::assertGet(PlanExecutor::make(opCtx,
                                                       std::move(workingSet),
                                                       std::move(queuedDataStage),
-                                                      kTestNss,
-                                                      PlanExecutor::YieldPolicy::NO_YIELD));
+                                                      nullptr,
+                                                      PlanExecutor::YieldPolicy::NO_YIELD,
+                                                      kTestNss));
     }
 
     ClientCursorParams makeParams(OperationContext* opCtx) {
diff --git a/src/mongo/dbtests/documentsourcetests.cpp b/src/mongo/dbtests/documentsourcetests.cpp
index 8963c988bfd..2165ed6695a 100644
--- a/src/mongo/dbtests/documentsourcetests.cpp
+++ b/src/mongo/dbtests/documentsourcetests.cpp
@@ -343,10 +343,9 @@ TEST_F(DocumentSourceCursorTest, TailableAwaitDataCursorShouldErrorAfterTimeout)
     auto canonicalQuery = unittest::assertGet(
         CanonicalQuery::canonicalize(opCtx(), std::move(queryRequest), nullptr));
     auto planExecutor =
-        uassertStatusOK(PlanExecutor::make(opCtx(),
+        uassertStatusOK(PlanExecutor::make(std::move(canonicalQuery),
                                            std::move(workingSet),
                                            std::move(collectionScan),
-                                           std::move(canonicalQuery),
                                            readLock.getCollection(),
                                            PlanExecutor::YieldPolicy::ALWAYS_TIME_OUT));
 
@@ -380,10 +379,9 @@ TEST_F(DocumentSourceCursorTest, NonAwaitDataCursorShouldErrorAfterTimeout) {
     auto canonicalQuery = unittest::assertGet(
         CanonicalQuery::canonicalize(opCtx(), std::move(queryRequest), nullptr));
     auto planExecutor =
-        uassertStatusOK(PlanExecutor::make(opCtx(),
+        uassertStatusOK(PlanExecutor::make(std::move(canonicalQuery),
                                            std::move(workingSet),
                                            std::move(collectionScan),
-                                           std::move(canonicalQuery),
                                            readLock.getCollection(),
                                            PlanExecutor::YieldPolicy::ALWAYS_TIME_OUT));
 
@@ -426,10 +424,9 @@ TEST_F(DocumentSourceCursorTest, TailableAwaitDataCursorShouldErrorAfterBeingKil
     auto canonicalQuery = unittest::assertGet(
         CanonicalQuery::canonicalize(opCtx(), std::move(queryRequest), nullptr));
     auto planExecutor =
-        uassertStatusOK(PlanExecutor::make(opCtx(),
+        uassertStatusOK(PlanExecutor::make(std::move(canonicalQuery),
                                            std::move(workingSet),
                                            std::move(collectionScan),
-                                           std::move(canonicalQuery),
                                            readLock.getCollection(),
                                            PlanExecutor::YieldPolicy::ALWAYS_MARK_KILLED));
 
@@ -462,10 +459,9 @@ TEST_F(DocumentSourceCursorTest, NormalCursorShouldErrorAfterBeingKilled) {
     auto canonicalQuery = unittest::assertGet(
         CanonicalQuery::canonicalize(opCtx(), std::move(queryRequest), nullptr));
     auto planExecutor =
-        uassertStatusOK(PlanExecutor::make(opCtx(),
+        uassertStatusOK(PlanExecutor::make(std::move(canonicalQuery),
                                            std::move(workingSet),
                                            std::move(collectionScan),
-                                           std::move(canonicalQuery),
                                            readLock.getCollection(),
                                            PlanExecutor::YieldPolicy::ALWAYS_MARK_KILLED));
 
diff --git a/src/mongo/dbtests/plan_executor_invalidation_test.cpp b/src/mongo/dbtests/plan_executor_invalidation_test.cpp
index 61eedeab72f..590d518feae 100644
--- a/src/mongo/dbtests/plan_executor_invalidation_test.cpp
+++ b/src/mongo/dbtests/plan_executor_invalidation_test.cpp
@@ -85,10 +85,9 @@ public:
         std::unique_ptr<CanonicalQuery> cq = std::move(statusWithCQ.getValue());
 
         // Takes ownership of 'ws', 'scan', and 'cq'.
-        auto statusWithPlanExecutor = PlanExecutor::make(&_opCtx,
+        auto statusWithPlanExecutor = PlanExecutor::make(std::move(cq),
                                                          std::move(ws),
                                                          std::move(scan),
-                                                         std::move(cq),
                                                          _ctx->db()->getCollection(&_opCtx, nss),
                                                          PlanExecutor::YIELD_MANUAL);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
diff --git a/src/mongo/dbtests/query_plan_executor.cpp b/src/mongo/dbtests/query_plan_executor.cpp
index 82f6251867a..be1b3448555 100644
--- a/src/mongo/dbtests/query_plan_executor.cpp
+++ b/src/mongo/dbtests/query_plan_executor.cpp
@@ -119,8 +119,8 @@ public:
             new CollectionScan(&_opCtx, coll, csparams, ws.get(), cq.get()->root()));
 
         // Hand the plan off to the executor.
-        auto statusWithPlanExecutor = PlanExecutor::make(
-            &_opCtx, std::move(ws), std::move(root), std::move(cq), coll, yieldPolicy);
+        auto statusWithPlanExecutor =
+            PlanExecutor::make(std::move(cq), std::move(ws), std::move(root), coll, yieldPolicy);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         return std::move(statusWithPlanExecutor.getValue());
     }
@@ -162,12 +162,8 @@ public:
         verify(nullptr != cq.get());
 
         // Hand the plan off to the executor.
-        auto statusWithPlanExecutor = PlanExecutor::make(&_opCtx,
-                                                         std::move(ws),
-                                                         std::move(root),
-                                                         std::move(cq),
-                                                         coll,
-                                                         PlanExecutor::YIELD_MANUAL);
+        auto statusWithPlanExecutor = PlanExecutor::make(
+            std::move(cq), std::move(ws), std::move(root), coll, PlanExecutor::YIELD_MANUAL);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
         return std::move(statusWithPlanExecutor.getValue());
     }
diff --git a/src/mongo/dbtests/query_stage_multiplan.cpp b/src/mongo/dbtests/query_stage_multiplan.cpp
index c08ca0abad0..877b4c2213c 100644
--- a/src/mongo/dbtests/query_stage_multiplan.cpp
+++ b/src/mongo/dbtests/query_stage_multiplan.cpp
@@ -253,12 +253,8 @@ TEST_F(QueryStageMultiPlanTest, MPSCollectionScanVsHighlySelectiveIXScan) {
     ASSERT_EQUALS(0, mps->bestPlanIdx());
 
     // Takes ownership of arguments other than 'collection'.
-    auto statusWithPlanExecutor = PlanExecutor::make(_opCtx.get(),
-                                                     std::move(sharedWs),
-                                                     std::move(mps),
-                                                     std::move(cq),
-                                                     coll,
-                                                     PlanExecutor::NO_YIELD);
+    auto statusWithPlanExecutor = PlanExecutor::make(
+        std::move(cq), std::move(sharedWs), std::move(mps), coll, PlanExecutor::NO_YIELD);
     ASSERT_OK(statusWithPlanExecutor.getStatus());
     auto exec = std::move(statusWithPlanExecutor.getValue());
 
diff --git a/src/mongo/dbtests/query_stage_sort_key_generator.cpp b/src/mongo/dbtests/query_stage_sort_key_generator.cpp
index 74aab27fc88..6ee9b1ba86c 100644
--- a/src/mongo/dbtests/query_stage_sort_key_generator.cpp
+++ b/src/mongo/dbtests/query_stage_sort_key_generator.cpp
@@ -34,6 +34,7 @@
 #include "mongo/db/exec/queued_data_stage.h"
 #include "mongo/db/exec/sort_key_generator.h"
 #include "mongo/db/json.h"
+#include "mongo/db/pipeline/document_value_test_util.h"
 #include "mongo/db/query/collation/collator_interface_mock.h"
 #include "mongo/db/query/query_test_service_context.h"
 #include "mongo/unittest/unittest.h"
@@ -42,7 +43,7 @@ namespace mongo {
 
 namespace {
 
-BSONObj extractKeyFromKeyGenStage(SortKeyGeneratorStage* sortKeyGen, WorkingSet* workingSet) {
+Value extractKeyFromKeyGenStage(SortKeyGeneratorStage* sortKeyGen, WorkingSet* workingSet) {
     WorkingSetID wsid;
     PlanStage::StageState state = PlanStage::NEED_TIME;
     while (state == PlanStage::NEED_TIME) {
@@ -62,7 +63,7 @@ BSONObj extractKeyFromKeyGenStage(SortKeyGeneratorStage* sortKeyGen, WorkingSet*
  * The 'collator' is used to specify the string comparison semantics that should be used when
  * generating the sort key.
  */
-BSONObj extractSortKey(const char* sortSpec, const char* doc, const CollatorInterface* collator) {
+Value extractSortKey(const char* sortSpec, const char* doc, const CollatorInterface* collator) {
     QueryTestServiceContext serviceContext;
     auto opCtx = serviceContext.makeOperationContext();
     boost::intrusive_ptr<ExpressionContext> pExpCtx(new ExpressionContext(opCtx.get(), collator));
@@ -89,9 +90,9 @@ BSONObj extractSortKey(const char* sortSpec, const char* doc, const CollatorInte
  * The 'collator' is used to specify the string comparison semantics that should be used when
  * generating the sort key.
  */
-BSONObj extractSortKeyCovered(const char* sortSpec,
-                              const IndexKeyDatum& ikd,
-                              const CollatorInterface* collator) {
+Value extractSortKeyCovered(const char* sortSpec,
+                            const IndexKeyDatum& ikd,
+                            const CollatorInterface* collator) {
     QueryTestServiceContext serviceContext;
     auto opCtx = serviceContext.makeOperationContext();
     boost::intrusive_ptr<ExpressionContext> pExpCtx(new ExpressionContext(opCtx.get(), collator));
@@ -112,150 +113,146 @@ BSONObj extractSortKeyCovered(const char* sortSpec,
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyNormal) {
-    BSONObj actualOut = extractSortKey("{a: 1}", "{_id: 0, a: 5}", nullptr);
-    BSONObj expectedOut = BSON("" << 5);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value actualOut = extractSortKey("{a: 1}", "{_id: 0, a: 5}", nullptr);
+    Value expectedOut(5);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyNormal2) {
-    BSONObj actualOut = extractSortKey("{a: 1}", "{_id: 0, z: 10, a: 6, b: 16}", nullptr);
-    BSONObj expectedOut = BSON("" << 6);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value actualOut = extractSortKey("{a: 1}", "{_id: 0, z: 10, a: 6, b: 16}", nullptr);
+    Value expectedOut(6);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyString) {
-    BSONObj actualOut =
+    Value actualOut =
         extractSortKey("{a: 1}", "{_id: 0, z: 'thing1', a: 'thing2', b: 16}", nullptr);
-    BSONObj expectedOut = BSON(""
-                               << "thing2");
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut("thing2"_sd);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyCompound) {
-    BSONObj actualOut =
+    Value actualOut =
         extractSortKey("{a: 1, b: 1}", "{_id: 0, z: 'thing1', a: 99, c: {a: 4}, b: 16}", nullptr);
-    BSONObj expectedOut = BSON("" << 99 << "" << 16);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut({Value(99), Value(16)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyEmbedded) {
-    BSONObj actualOut = extractSortKey(
+    Value actualOut = extractSortKey(
         "{'c.a': 1, b: 1}", "{_id: 0, z: 'thing1', a: 99, c: {a: 4}, b: 16}", nullptr);
-    BSONObj expectedOut = BSON("" << 4 << "" << 16);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut = Value({Value(4), Value(16)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyArray) {
-    BSONObj actualOut = extractSortKey(
+    Value actualOut = extractSortKey(
         "{'c': 1, b: 1}", "{_id: 0, z: 'thing1', a: 99, c: [2, 4, 1], b: 16}", nullptr);
-    BSONObj expectedOut = BSON("" << 1 << "" << 16);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut({Value(1), Value(16)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyCoveredNormal) {
     CollatorInterface* collator = nullptr;
-    BSONObj actualOut = extractSortKeyCovered(
+    Value actualOut = extractSortKeyCovered(
         "{a: 1}", IndexKeyDatum(BSON("a" << 1), BSON("" << 5), 0, SnapshotId{}), collator);
-    BSONObj expectedOut = BSON("" << 5);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut({Value(5)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyCoveredEmbedded) {
     CollatorInterface* collator = nullptr;
-    BSONObj actualOut = extractSortKeyCovered(
+    Value actualOut = extractSortKeyCovered(
         "{'a.c': 1}",
         IndexKeyDatum(BSON("a.c" << 1 << "c" << 1), BSON("" << 5 << "" << 6), 0, SnapshotId{}),
         collator);
-    BSONObj expectedOut = BSON("" << 5);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut(5);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyCoveredCompound) {
     CollatorInterface* collator = nullptr;
-    BSONObj actualOut = extractSortKeyCovered(
+    Value actualOut = extractSortKeyCovered(
         "{a: 1, c: 1}",
         IndexKeyDatum(BSON("a" << 1 << "c" << 1), BSON("" << 5 << "" << 6), 0, SnapshotId{}),
         collator);
-    BSONObj expectedOut = BSON("" << 5 << "" << 6);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut({Value(5), Value(6)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyCoveredCompound2) {
     CollatorInterface* collator = nullptr;
-    BSONObj actualOut = extractSortKeyCovered("{a: 1, b: 1}",
-                                              IndexKeyDatum(BSON("a" << 1 << "b" << 1 << "c" << 1),
-                                                            BSON("" << 5 << "" << 6 << "" << 4),
-                                                            0,
-                                                            SnapshotId{}),
-                                              collator);
-    BSONObj expectedOut = BSON("" << 5 << "" << 6);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value actualOut = extractSortKeyCovered("{a: 1, b: 1}",
+                                            IndexKeyDatum(BSON("a" << 1 << "b" << 1 << "c" << 1),
+                                                          BSON("" << 5 << "" << 6 << "" << 4),
+                                                          0,
+                                                          SnapshotId{}),
+                                            collator);
+    Value expectedOut({Value(5), Value(6)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyCoveredCompound3) {
     CollatorInterface* collator = nullptr;
-    BSONObj actualOut =
+    Value actualOut =
         extractSortKeyCovered("{b: 1, c: 1}",
                               IndexKeyDatum(BSON("a" << 1 << "b" << 1 << "c" << 1 << "d" << 1),
                                             BSON("" << 5 << "" << 6 << "" << 4 << "" << 9000),
                                             0,
                                             SnapshotId{}),
                               collator);
-    BSONObj expectedOut = BSON("" << 6 << "" << 4);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut({Value(6), Value(4)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, ExtractStringSortKeyWithCollatorUsesComparisonKey) {
     CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
-    BSONObj actualOut =
+    Value actualOut =
         extractSortKey("{a: 1}", "{_id: 0, z: 'thing1', a: 'thing2', b: 16}", &collator);
-    BSONObj expectedOut = BSON(""
-                               << "2gniht");
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut = Value("2gniht"_sd);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, CollatorHasNoEffectWhenExtractingNonStringSortKey) {
     CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
-    BSONObj actualOut = extractSortKey("{a: 1}", "{_id: 0, z: 10, a: 6, b: 16}", &collator);
-    BSONObj expectedOut = BSON("" << 6);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value actualOut = extractSortKey("{a: 1}", "{_id: 0, z: 10, a: 6, b: 16}", &collator);
+    Value expectedOut = Value(6);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, CollatorAppliesWhenExtractingCoveredSortKeyString) {
     CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
-    BSONObj actualOut = extractSortKeyCovered("{b: 1}",
-                                              IndexKeyDatum(BSON("a" << 1 << "b" << 1),
-                                                            BSON("" << 4 << ""
-                                                                    << "foo"),
-                                                            0,
-                                                            SnapshotId{}),
-                                              &collator);
-    BSONObj expectedOut = BSON(""
-                               << "oof");
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value actualOut = extractSortKeyCovered("{b: 1}",
+                                            IndexKeyDatum(BSON("a" << 1 << "b" << 1),
+                                                          BSON("" << 4 << ""
+                                                                  << "foo"),
+                                                          0,
+                                                          SnapshotId{}),
+                                            &collator);
+    Value expectedOut = Value("oof"_sd);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyGenerationForArraysChoosesCorrectKey) {
-    BSONObj actualOut = extractSortKey("{a: -1}", "{_id: 0, a: [1, 2, 3, 4]}", nullptr);
-    BSONObj expectedOut = BSON("" << 4);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value actualOut = extractSortKey("{a: -1}", "{_id: 0, a: [1, 2, 3, 4]}", nullptr);
+    Value expectedOut(4);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, EnsureSortKeyGenerationForArraysRespectsCollation) {
     CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
-    BSONObj actualOut =
+    Value actualOut =
         extractSortKey("{a: 1}", "{_id: 0, a: ['aaz', 'zza', 'yya', 'zzb']}", &collator);
-    BSONObj expectedOut = BSON(""
-                               << "ayy");
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value expectedOut("ayy"_sd);
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 TEST(SortKeyGeneratorStageTest, SortKeyGenerationForArraysRespectsCompoundOrdering) {
-    BSONObj actualOut = extractSortKey("{'a.b': 1, 'a.c': -1}",
-                                       "{_id: 0, a: [{b: 1, c: 0}, {b: 0, c: 3}, {b: 0, c: 1}]}",
-                                       nullptr);
-    BSONObj expectedOut = BSON("" << 0 << "" << 3);
-    ASSERT_BSONOBJ_EQ(actualOut, expectedOut);
+    Value actualOut = extractSortKey("{'a.b': 1, 'a.c': -1}",
+                                     "{_id: 0, a: [{b: 1, c: 0}, {b: 0, c: 3}, {b: 0, c: 1}]}",
+                                     nullptr);
+    Value expectedOut({Value(0), Value(3)});
+    ASSERT_VALUE_EQ(actualOut, expectedOut);
 }
 
 }  // namespace
diff --git a/src/mongo/s/query/router_stage_pipeline.cpp b/src/mongo/s/query/router_stage_pipeline.cpp
index bfdd71fd104..4208f8fb254 100644
--- a/src/mongo/s/query/router_stage_pipeline.cpp
+++ b/src/mongo/s/query/router_stage_pipeline.cpp
@@ -99,7 +99,7 @@ BSONObj RouterStagePipeline::_validateAndConvertToBSON(const Document& event) {
     auto eventBSON = event.toBson();
     auto resumeToken = event.metadata().getSortKey();
     auto idField = eventBSON.getObjectField("_id");
-    invariant(!resumeToken.isEmpty());
+    invariant(!resumeToken.missing());
     uassert(ErrorCodes::ChangeStreamFatalError,
             str::stream() << "Encountered an event whose _id field, which contains the resume "
                              "token, was modified by the pipeline. Modifying the _id field of an "
@@ -108,7 +108,8 @@ BSONObj RouterStagePipeline::_validateAndConvertToBSON(const Document& event) {
                              "Expected: "
                           << BSON("_id" << resumeToken) << " but found: "
                           << (eventBSON["_id"] ? BSON("_id" << eventBSON["_id"]) : BSONObj()),
-            idField.binaryEqual(resumeToken));
+            (resumeToken.getType() == BSONType::Object) &&
+                idField.binaryEqual(resumeToken.getDocument().toBson()));
 
     // Return the event in BSONObj form, minus the $sortKey metadata.
     return eventBSON;