SERVER-29135 Add post-image lookup to $changeNotification

This patch only adds support on an unsharded collection.

23 files changed, 916 insertions, 158 deletions

diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index 0d67ccbb2df..093cbb5a5c9 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -135,7 +135,9 @@ env.CppUnitTest(
         'document_source_geo_near_test.cpp',
         'document_source_group_test.cpp',
         'document_source_limit_test.cpp',
+        'document_source_lookup_change_post_image_test.cpp',
         'document_source_lookup_test.cpp',
+        'document_source_graph_lookup_test.cpp',
         'document_source_match_test.cpp',
         'document_source_mock_test.cpp',
         'document_source_project_test.cpp',
@@ -231,7 +233,6 @@ docSourceEnv.Library(
         'document_source_add_fields.cpp',
         'document_source_bucket.cpp',
         'document_source_bucket_auto.cpp',
-        'document_source_change_notification.cpp',
         'document_source_coll_stats.cpp',
         'document_source_count.cpp',
         'document_source_current_op.cpp',
@@ -262,6 +263,7 @@ docSourceEnv.Library(
         '$BUILD_DIR/mongo/db/matcher/expression_algo',
         '$BUILD_DIR/mongo/db/matcher/expressions',
         '$BUILD_DIR/mongo/db/pipeline/lite_parsed_document_source',
+        '$BUILD_DIR/mongo/db/repl/oplog_entry',
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/stats/top',
         '$BUILD_DIR/mongo/db/storage/encryption_hooks',
@@ -309,8 +311,10 @@ env.Library(
 env.Library(
     target='document_source_lookup',
     source=[
+        'document_source_change_notification.cpp',
         'document_source_graph_lookup.cpp',
         'document_source_lookup.cpp',
+        'document_source_lookup_change_post_image.cpp',
     ],
     LIBDEPS=[
         'document_source',
@@ -318,20 +322,6 @@ env.Library(
     ],
 )
 
-env.CppUnitTest(
-    target='document_source_graph_lookup_test',
-    source='document_source_graph_lookup_test.cpp',
-    LIBDEPS=[
-        'document_source',
-        'document_source_lookup',
-        'document_value_test_util',
-        '$BUILD_DIR/mongo/db/auth/authorization_manager_mock_init',
-        '$BUILD_DIR/mongo/db/query/query_test_service_context',
-        '$BUILD_DIR/mongo/db/service_context_noop_init',
-        '$BUILD_DIR/mongo/s/is_mongos',
-    ],
-)
-
 env.Library(
     target='document_source_facet',
     source=[
diff --git a/src/mongo/db/pipeline/document_source.cpp b/src/mongo/db/pipeline/document_source.cpp
index 6e58adae2bf..a9d9f74181a 100644
--- a/src/mongo/db/pipeline/document_source.cpp
+++ b/src/mongo/db/pipeline/document_source.cpp
@@ -41,6 +41,7 @@ namespace mongo {
 
 using Parser = DocumentSource::Parser;
 using boost::intrusive_ptr;
+using std::list;
 using std::string;
 using std::vector;
 
@@ -60,7 +61,7 @@ void DocumentSource::registerParser(string name, Parser parser) {
     parserMap[name] = parser;
 }
 
-vector<intrusive_ptr<DocumentSource>> DocumentSource::parse(
+list<intrusive_ptr<DocumentSource>> DocumentSource::parse(
     const intrusive_ptr<ExpressionContext>& expCtx, BSONObj stageObj) {
     uassert(16435,
             "A pipeline stage specification object must contain exactly one field.",
diff --git a/src/mongo/db/pipeline/document_source.h b/src/mongo/db/pipeline/document_source.h
index 2545f02d8de..383cd56e35f 100644
--- a/src/mongo/db/pipeline/document_source.h
+++ b/src/mongo/db/pipeline/document_source.h
@@ -85,7 +85,7 @@ class Document;
     MONGO_INITIALIZER(addToDocSourceParserMap_##key)(InitializerContext*) {                  \
         auto fullParserWrapper = [](BSONElement stageSpec,                                   \
                                     const boost::intrusive_ptr<ExpressionContext>& expCtx) { \
-            return std::vector<boost::intrusive_ptr<DocumentSource>>{                        \
+            return std::list<boost::intrusive_ptr<DocumentSource>>{                          \
                 (fullParser)(stageSpec, expCtx)};                                            \
         };                                                                                   \
         LiteParsedDocumentSource::registerParser("$" #key, liteParser);                      \
@@ -115,7 +115,7 @@ class Document;
 
 class DocumentSource : public IntrusiveCounterUnsigned {
 public:
-    using Parser = stdx::function<std::vector<boost::intrusive_ptr<DocumentSource>>(
+    using Parser = stdx::function<std::list<boost::intrusive_ptr<DocumentSource>>(
         BSONElement, const boost::intrusive_ptr<ExpressionContext>&)>;
 
     /**
@@ -303,7 +303,7 @@ public:
     /**
      * Create a DocumentSource pipeline stage from 'stageObj'.
      */
-    static std::vector<boost::intrusive_ptr<DocumentSource>> parse(
+    static std::list<boost::intrusive_ptr<DocumentSource>> parse(
         const boost::intrusive_ptr<ExpressionContext>& expCtx, BSONObj stageObj);
 
     /**
diff --git a/src/mongo/db/pipeline/document_source_bucket.cpp b/src/mongo/db/pipeline/document_source_bucket.cpp
index 853ef0aaa42..e209834b14d 100644
--- a/src/mongo/db/pipeline/document_source_bucket.cpp
+++ b/src/mongo/db/pipeline/document_source_bucket.cpp
@@ -37,6 +37,7 @@ namespace mongo {
 
 using boost::intrusive_ptr;
 using std::vector;
+using std::list;
 
 REGISTER_MULTI_STAGE_ALIAS(bucket,
                            LiteParsedDocumentSourceDefault::parse,
@@ -52,7 +53,7 @@ intrusive_ptr<ExpressionConstant> getExpressionConstant(
 }
 }  // namespace
 
-vector<intrusive_ptr<DocumentSource>> DocumentSourceBucket::createFromBson(
+list<intrusive_ptr<DocumentSource>> DocumentSourceBucket::createFromBson(
     BSONElement elem, const intrusive_ptr<ExpressionContext>& pExpCtx) {
     uassert(40201,
             str::stream() << "Argument to $bucket stage must be an object, but found type: "
diff --git a/src/mongo/db/pipeline/document_source_bucket.h b/src/mongo/db/pipeline/document_source_bucket.h
index aa83ee07cc2..e059660fc99 100644
--- a/src/mongo/db/pipeline/document_source_bucket.h
+++ b/src/mongo/db/pipeline/document_source_bucket.h
@@ -40,7 +40,7 @@ public:
     /**
      * Returns a $group stage followed by a $sort stage.
      */
-    static std::vector<boost::intrusive_ptr<DocumentSource>> createFromBson(
+    static std::list<boost::intrusive_ptr<DocumentSource>> createFromBson(
         BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
 private:
diff --git a/src/mongo/db/pipeline/document_source_bucket_test.cpp b/src/mongo/db/pipeline/document_source_bucket_test.cpp
index 1a61934dcb5..26f18c4dca9 100644
--- a/src/mongo/db/pipeline/document_source_bucket_test.cpp
+++ b/src/mongo/db/pipeline/document_source_bucket_test.cpp
@@ -47,21 +47,22 @@
 namespace mongo {
 namespace {
 
-using std::vector;
 using boost::intrusive_ptr;
+using std::list;
+using std::vector;
 
 class BucketReturnsGroupAndSort : public AggregationContextFixture {
 public:
     void testCreateFromBsonResult(BSONObj bucketSpec, Value expectedGroupExplain) {
-        vector<intrusive_ptr<DocumentSource>> result =
+        list<intrusive_ptr<DocumentSource>> result =
             DocumentSourceBucket::createFromBson(bucketSpec.firstElement(), getExpCtx());
 
         ASSERT_EQUALS(result.size(), 2UL);
 
-        const auto* groupStage = dynamic_cast<DocumentSourceGroup*>(result[0].get());
+        const auto* groupStage = dynamic_cast<DocumentSourceGroup*>(result.front().get());
         ASSERT(groupStage);
 
-        const auto* sortStage = dynamic_cast<DocumentSourceSort*>(result[1].get());
+        const auto* sortStage = dynamic_cast<DocumentSourceSort*>(result.back().get());
         ASSERT(sortStage);
 
         // Serialize the DocumentSourceGroup and DocumentSourceSort from $bucket so that we can
@@ -154,7 +155,7 @@ TEST_F(BucketReturnsGroupAndSort, BucketSucceedsWithMultipleBoundaryValues) {
 
 class InvalidBucketSpec : public AggregationContextFixture {
 public:
-    vector<intrusive_ptr<DocumentSource>> createBucket(BSONObj bucketSpec) {
+    list<intrusive_ptr<DocumentSource>> createBucket(BSONObj bucketSpec) {
         auto sources = DocumentSourceBucket::createFromBson(bucketSpec.firstElement(), getExpCtx());
         return sources;
     }
@@ -267,14 +268,14 @@ TEST_F(InvalidBucketSpec, GroupFailsForBucketWithInvalidOutputField) {
 
 TEST_F(InvalidBucketSpec, SwitchFailsForBucketWhenNoDefaultSpecified) {
     const auto spec = fromjson("{$bucket : {groupBy : '$x', boundaries : [1, 2, 3]}}");
-    vector<intrusive_ptr<DocumentSource>> bucketStages = createBucket(spec);
+    list<intrusive_ptr<DocumentSource>> bucketStages = createBucket(spec);
 
     ASSERT_EQUALS(bucketStages.size(), 2UL);
 
-    auto* groupStage = dynamic_cast<DocumentSourceGroup*>(bucketStages[0].get());
+    auto* groupStage = dynamic_cast<DocumentSourceGroup*>(bucketStages.front().get());
     ASSERT(groupStage);
 
-    const auto* sortStage = dynamic_cast<DocumentSourceSort*>(bucketStages[1].get());
+    const auto* sortStage = dynamic_cast<DocumentSourceSort*>(bucketStages.back().get());
     ASSERT(sortStage);
 
     auto doc = Document{{"x", 4}};
diff --git a/src/mongo/db/pipeline/document_source_change_notification.cpp b/src/mongo/db/pipeline/document_source_change_notification.cpp
index dff06f05802..d173d1183d5 100644
--- a/src/mongo/db/pipeline/document_source_change_notification.cpp
+++ b/src/mongo/db/pipeline/document_source_change_notification.cpp
@@ -30,30 +30,46 @@
 
 #include "mongo/db/pipeline/document_source_change_notification.h"
 
+#include "mongo/bson/simple_bsonelement_comparator.h"
 #include "mongo/db/pipeline/document_source_limit.h"
+#include "mongo/db/pipeline/document_source_lookup_change_post_image.h"
 #include "mongo/db/pipeline/document_source_match.h"
 #include "mongo/db/pipeline/document_source_sort.h"
 #include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/lite_parsed_document_source.h"
+#include "mongo/db/repl/oplog_entry.h"
+#include "mongo/db/repl/oplog_entry_gen.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
 
 using boost::intrusive_ptr;
 using boost::optional;
-using std::vector;
+using std::list;
 using std::string;
+using std::vector;
 
-constexpr StringData kNamespaceField = "ns"_sd;
-constexpr StringData kTimestmapField = "ts"_sd;
-constexpr StringData kOpTypeField = "op"_sd;
-constexpr StringData kOField = "o"_sd;
-constexpr StringData kIdField = "_id"_sd;
-
+// The $changeNotification stage is an alias for many stages, but we need to be able to serialize
+// and re-parse the pipeline. To make this work, the 'transformation' stage will serialize itself
+// with the original specification, and all other stages that are created during the alias expansion
+// will not serialize themselves.
 REGISTER_MULTI_STAGE_ALIAS(changeNotification,
                            DocumentSourceChangeNotification::LiteParsed::parse,
                            DocumentSourceChangeNotification::createFromBson);
 
+constexpr StringData DocumentSourceChangeNotification::kDocumentKeyField;
+constexpr StringData DocumentSourceChangeNotification::kFullDocumentField;
+constexpr StringData DocumentSourceChangeNotification::kIdField;
+constexpr StringData DocumentSourceChangeNotification::kNamespaceField;
+constexpr StringData DocumentSourceChangeNotification::kOperationTypeField;
+constexpr StringData DocumentSourceChangeNotification::kStageName;
+constexpr StringData DocumentSourceChangeNotification::kTimestmapField;
+constexpr StringData DocumentSourceChangeNotification::kUpdateOpType;
+constexpr StringData DocumentSourceChangeNotification::kDeleteOpType;
+constexpr StringData DocumentSourceChangeNotification::kReplaceOpType;
+constexpr StringData DocumentSourceChangeNotification::kInsertOpType;
+constexpr StringData DocumentSourceChangeNotification::kInvalidateOpType;
+
 namespace {
 
 static constexpr StringData kOplogMatchExplainName = "$_internalOplogMatch"_sd;
@@ -127,7 +143,7 @@ BSONObj DocumentSourceChangeNotification::buildMatchFilter(const NamespaceString
     return BSON("ts" << GT << Timestamp() << "$or" << BSON_ARRAY(opMatch << commandMatch));
 }
 
-vector<intrusive_ptr<DocumentSource>> DocumentSourceChangeNotification::createFromBson(
+list<intrusive_ptr<DocumentSource>> DocumentSourceChangeNotification::createFromBson(
     BSONElement elem, const intrusive_ptr<ExpressionContext>& expCtx) {
     // TODO: Add sharding support here (SERVER-29141).
     uassert(40470,
@@ -137,17 +153,54 @@ vector<intrusive_ptr<DocumentSource>> DocumentSourceChangeNotification::createFr
             "Only default collation is allowed when using a $changeNotification stage.",
             !expCtx->getCollator());
 
-    BSONObj filter = buildMatchFilter(expCtx->ns);
+    uassert(40573,
+            str::stream() << "the $changeNotification stage must be specified as an object, got "
+                          << typeName(elem.type()),
+            elem.type() == BSONType::Object);
+
+    bool shouldLookupPostImage = false;
+    for (auto&& option : elem.embeddedObject()) {
+        auto optionName = option.fieldNameStringData();
+        if (optionName == "fullDocument"_sd) {
+            uassert(40574,
+                    str::stream() << "the 'fullDocument' option to the $changeNotification stage "
+                                     "must be a string, got "
+                                  << typeName(option.type()),
+                    option.type() == BSONType::String);
+            auto fullDocOption = option.valueStringData();
+            uassert(40575,
+                    str::stream() << "unrecognized value for the 'fullDocument' option to the "
+                                     "$changeNotification stage. Expected \"none\" or "
+                                     "\"fullDocument\", got \""
+                                  << option.String()
+                                  << "\"",
+                    fullDocOption == "lookup"_sd || fullDocOption == "none"_sd);
+            shouldLookupPostImage = (fullDocOption == "lookup"_sd);
+        } else if (optionName == "resumeAfter"_sd) {
+            uasserted(
+                40576,
+                "the 'resumeAfter' option to the $changeNotification stage is not yet supported");
+        } else {
+            uasserted(40577,
+                      str::stream() << "unrecognized option to $changeNotification stage: \""
+                                    << optionName
+                                    << "\"");
+        }
+    }
 
-    auto oplogMatch = DocumentSourceOplogMatch::create(filter, expCtx);
-    auto transformation = createTransformationStage(expCtx);
-    return {oplogMatch, transformation};
+    auto oplogMatch = DocumentSourceOplogMatch::create(buildMatchFilter(expCtx->ns), expCtx);
+    auto transformation = createTransformationStage(elem.embeddedObject(), expCtx);
+    list<intrusive_ptr<DocumentSource>> stages = {oplogMatch, transformation};
+    if (shouldLookupPostImage) {
+        stages.push_back(DocumentSourceLookupChangePostImage::create(expCtx));
+    }
+    return stages;
 }
 
 intrusive_ptr<DocumentSource> DocumentSourceChangeNotification::createTransformationStage(
-    const intrusive_ptr<ExpressionContext>& expCtx) {
+    BSONObj changeNotificationSpec, const intrusive_ptr<ExpressionContext>& expCtx) {
     return intrusive_ptr<DocumentSource>(new DocumentSourceSingleDocumentTransformation(
-        expCtx, stdx::make_unique<Transformation>(), "$changeNotification"));
+        expCtx, stdx::make_unique<Transformation>(changeNotificationSpec), kStageName.toString()));
 }
 
 namespace {
@@ -167,89 +220,103 @@ Document DocumentSourceChangeNotification::Transformation::applyTransformation(
     MutableDocument doc;
 
     // Extract the fields we need.
-    checkValueType(input[kOpTypeField], kOpTypeField, BSONType::String);
-    string op = input[kOpTypeField].getString();
-    Value ts = input[kTimestmapField];
-    Value ns = input[kNamespaceField];
-    checkValueType(ns, kNamespaceField, BSONType::String);
+    checkValueType(input[repl::OplogEntry::kOpTypeFieldName],
+                   repl::OplogEntry::kOpTypeFieldName,
+                   BSONType::String);
+    string op = input[repl::OplogEntry::kOpTypeFieldName].getString();
+    Value ts = input[repl::OplogEntry::kTimestampFieldName];
+    Value ns = input[repl::OplogEntry::kNamespaceFieldName];
+    checkValueType(ns, repl::OplogEntry::kNamespaceFieldName, BSONType::String);
     NamespaceString nss(ns.getString());
     Value id = input.getNestedField("o._id");
     // Non-replace updates have the _id in field "o2".
     Value documentId = id.missing() ? input.getNestedField("o2._id") : id;
-    string operationType;
-    Value newDocument;
+    StringData operationType;
+    Value fullDocument = Value(BSONNULL);
     Value updateDescription;
 
     // Deal with CRUD operations and commands.
-    if (op == "i") {
-        operationType = "insert";
-        newDocument = input[kOField];
-    } else if (op == "d") {
-        operationType = "delete";
-    } else if (op == "u") {
-        if (id.missing()) {
-            operationType = "update";
-            checkValueType(input[kOField], kOField, BSONType::Object);
-            Document o = input[kOField].getDocument();
-            Value updatedFields = o["$set"];
-            Value removedFields = o["$unset"];
-
-            // Extract the field names of $unset document.
-            vector<Value> removedFieldsVector;
-            if (removedFields.getType() == BSONType::Object) {
-                auto iter = removedFields.getDocument().fieldIterator();
-                while (iter.more()) {
-                    removedFieldsVector.push_back(Value(iter.next().first));
+    auto opType = repl::OpType_parse(IDLParserErrorContext("ChangeNotificationEntry.op"), op);
+    switch (opType) {
+        case repl::OpTypeEnum::kInsert: {
+            operationType = kInsertOpType;
+            fullDocument = input[repl::OplogEntry::kObjectFieldName];
+            break;
+        }
+        case repl::OpTypeEnum::kDelete: {
+            operationType = kDeleteOpType;
+            break;
+        }
+        case repl::OpTypeEnum::kUpdate: {
+            if (id.missing()) {
+                operationType = kUpdateOpType;
+                checkValueType(input[repl::OplogEntry::kObjectFieldName],
+                               repl::OplogEntry::kObjectFieldName,
+                               BSONType::Object);
+                Document opObject = input[repl::OplogEntry::kObjectFieldName].getDocument();
+                Value updatedFields = opObject["$set"];
+                Value removedFields = opObject["$unset"];
+
+                // Extract the field names of $unset document.
+                vector<Value> removedFieldsVector;
+                if (removedFields.getType() == BSONType::Object) {
+                    auto iter = removedFields.getDocument().fieldIterator();
+                    while (iter.more()) {
+                        removedFieldsVector.push_back(Value(iter.next().first));
+                    }
                 }
+                updateDescription = Value(Document{
+                    {"updatedFields", updatedFields.missing() ? Value(Document()) : updatedFields},
+                    {"removedFields", removedFieldsVector}});
+            } else {
+                operationType = kReplaceOpType;
+                fullDocument = input[repl::OplogEntry::kObjectFieldName];
             }
-            updateDescription = Value(Document{
-                {"updatedFields", updatedFields.missing() ? Value(Document()) : updatedFields},
-                {"removedFields", removedFieldsVector}});
-        } else {
-            operationType = "replace";
-            newDocument = input[kOField];
+            break;
+        }
+        case repl::OpTypeEnum::kCommand: {
+            operationType = kInvalidateOpType;
+            // Make sure the result doesn't have a document id.
+            documentId = Value();
+            break;
         }
-    } else if (op == "c") {
-        operationType = "invalidate";
-        // Make sure the result doesn't have a document id.
-        documentId = Value();
+        default: { MONGO_UNREACHABLE; }
     }
 
-    // Construct the result document. If document id is missing, it will not appear in the output.
+    // Construct the result document. Note that 'documentId' might be the missing value, in which
+    // case it will not appear in the output.
     doc.addField(
         kIdField,
         Value(Document{{kTimestmapField, ts}, {kNamespaceField, ns}, {kIdField, documentId}}));
-    doc.addField("operationType", Value(operationType));
+    doc.addField(kOperationTypeField, Value(operationType));
+    doc.addField(kFullDocumentField, fullDocument);
 
     // "invalidate" entry has fewer fields.
-    if (op == "c") {
+    if (opType == repl::OpTypeEnum::kCommand) {
         return doc.freeze();
     }
 
-    // Add fields for normal operations.
     doc.addField(kNamespaceField, Value(Document{{"db", nss.db()}, {"coll", nss.coll()}}));
-    doc.addField("documentKey", Value(Document{{kIdField, documentId}}));
+    doc.addField(kDocumentKeyField, Value(Document{{kIdField, documentId}}));
 
-    // If newDocument or updateDescription is missing, it will not be serialized.
-    doc.addField("newDocument", newDocument);
+    // Note that 'updateDescription' might be the 'missing' value, in which case it will not be
+    // serialized.
     doc.addField("updateDescription", updateDescription);
     return doc.freeze();
 }
 
 Document DocumentSourceChangeNotification::Transformation::serializeStageOptions(
     boost::optional<ExplainOptions::Verbosity> explain) const {
-    // TODO SERVER-29135 Be sure to re-serialize the 'postImage' argument.
-    // TODO SERVER-29131 Be sure to re-serialize the 'resumeAfter' argument.
-    return Document();
+    return Document(_changeNotificationSpec);
 }
 
 DocumentSource::GetDepsReturn DocumentSourceChangeNotification::Transformation::addDependencies(
     DepsTracker* deps) const {
-    deps->fields.insert(kOpTypeField.toString());
-    deps->fields.insert(kTimestmapField.toString());
-    deps->fields.insert(kNamespaceField.toString());
-    deps->fields.insert(kOField.toString());
-    deps->fields.insert("o2");
+    deps->fields.insert(repl::OplogEntry::kOpTypeFieldName.toString());
+    deps->fields.insert(repl::OplogEntry::kTimestampFieldName.toString());
+    deps->fields.insert(repl::OplogEntry::kNamespaceFieldName.toString());
+    deps->fields.insert(repl::OplogEntry::kObjectFieldName.toString());
+    deps->fields.insert(repl::OplogEntry::kObject2FieldName.toString());
     return DocumentSource::GetDepsReturn::EXHAUSTIVE_ALL;
 }
 
diff --git a/src/mongo/db/pipeline/document_source_change_notification.h b/src/mongo/db/pipeline/document_source_change_notification.h
index 3b38a908626..ad10a5ad210 100644
--- a/src/mongo/db/pipeline/document_source_change_notification.h
+++ b/src/mongo/db/pipeline/document_source_change_notification.h
@@ -51,6 +51,8 @@ public:
         }
 
         stdx::unordered_set<NamespaceString> getInvolvedNamespaces() const final {
+            // TODO SERVER-29138: we need to communicate that this stage will need to look up
+            // documents from different collections.
             return stdx::unordered_set<NamespaceString>();
         }
 
@@ -62,6 +64,8 @@ public:
 
     class Transformation : public DocumentSourceSingleDocumentTransformation::TransformerInterface {
     public:
+        Transformation(BSONObj changeNotificationSpec)
+            : _changeNotificationSpec(changeNotificationSpec.getOwned()) {}
         ~Transformation() = default;
         Document applyTransformation(const Document& input) final;
         TransformerType getType() const final {
@@ -72,10 +76,48 @@ public:
             boost::optional<ExplainOptions::Verbosity> explain) const final;
         DocumentSource::GetDepsReturn addDependencies(DepsTracker* deps) const final;
         DocumentSource::GetModPathsReturn getModifiedPaths() const final;
+
+    private:
+        BSONObj _changeNotificationSpec;
     };
 
+    // The name of the field where the document key (_id and shard key, if present) will be found
+    // after the transformation.
+    static constexpr StringData kDocumentKeyField = "documentKey"_sd;
+
+    // The name of the field where the full document will be found after the transformation. The
+    // full document is only present for certain types of operations, such as an insert.
+    static constexpr StringData kFullDocumentField = "fullDocument"_sd;
+
+    // The name of the field where the change identifier will be located after the transformation.
+    static constexpr StringData kIdField = "_id"_sd;
+
+    // The name of the field where the namespace of the change will be located after the
+    // transformation.
+    static constexpr StringData kNamespaceField = "ns"_sd;
+
+    // The name of the field where the type of the operation will be located after the
+    // transformation.
+    static constexpr StringData kOperationTypeField = "operationType"_sd;
+
+    // The name of this stage.
+    static constexpr StringData kStageName = "$changeNotification"_sd;
+
+    // The name of the field where the timestamp of the change will be located after the
+    // transformation. The timestamp will be located inside the change identifier, so the full path
+    // to the timestamp will be kIdField + "." + kTimestampField.
+    static constexpr StringData kTimestmapField = "ts"_sd;
+
+    // The different types of operations we can use for the operation type.
+    static constexpr StringData kUpdateOpType = "update"_sd;
+    static constexpr StringData kDeleteOpType = "delete"_sd;
+    static constexpr StringData kReplaceOpType = "replace"_sd;
+    static constexpr StringData kInsertOpType = "insert"_sd;
+    static constexpr StringData kInvalidateOpType = "invalidate"_sd;
+
     /**
-     * Produce the BSON for the $match stage based on a $changeNotification stage.
+     * Produce the BSON object representing the filter for the $match stage to filter oplog entries
+     * to only those relevant for this $changeNotification stage.
      */
     static BSONObj buildMatchFilter(const NamespaceString& nss);
 
@@ -83,11 +125,11 @@ public:
      * Parses a $changeNotification stage from 'elem' and produces the $match and transformation
      * stages required.
      */
-    static std::vector<boost::intrusive_ptr<DocumentSource>> createFromBson(
+    static std::list<boost::intrusive_ptr<DocumentSource>> createFromBson(
         BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& expCtx);
 
     static boost::intrusive_ptr<DocumentSource> createTransformationStage(
-        const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
+        BSONObj changeNotificationSpec, const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
 private:
     // It is illegal to construct a DocumentSourceChangeNotification directly, use createFromBson()
diff --git a/src/mongo/db/pipeline/document_source_change_notification_test.cpp b/src/mongo/db/pipeline/document_source_change_notification_test.cpp
index b3595789577..9c08bd91637 100644
--- a/src/mongo/db/pipeline/document_source_change_notification_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_notification_test.cpp
@@ -49,14 +49,18 @@
 namespace mongo {
 namespace {
 
-using std::vector;
-using std::string;
 using boost::intrusive_ptr;
-using repl::OplogEntry;
 using repl::OpTypeEnum;
+using repl::OplogEntry;
+using std::list;
+using std::string;
+using std::vector;
+
 using D = Document;
 using V = Value;
 
+using DSChangeNotification = DocumentSourceChangeNotification;
+
 static const Timestamp ts(100, 1);
 static const repl::OpTime optime(ts, 1);
 static const NamespaceString nss("unittests.change_notification");
@@ -67,10 +71,10 @@ public:
 
     void checkTransformation(const OplogEntry& entry, const boost::optional<Document> expectedDoc) {
         const auto spec = fromjson("{$changeNotification: {}}");
-        vector<intrusive_ptr<DocumentSource>> result =
-            DocumentSourceChangeNotification::createFromBson(spec.firstElement(), getExpCtx());
+        list<intrusive_ptr<DocumentSource>> result =
+            DSChangeNotification::createFromBson(spec.firstElement(), getExpCtx());
 
-        auto match = dynamic_cast<DocumentSourceMatch*>(result[0].get());
+        auto match = dynamic_cast<DocumentSourceMatch*>(result.front().get());
         ASSERT(match);
         auto mock = DocumentSourceMock::create(D(entry.toBSON()));
         match->setSource(mock.get());
@@ -78,7 +82,7 @@ public:
         // Check the oplog entry is transformed correctly.
         auto transform = result.back().get();
         ASSERT(transform);
-        ASSERT_EQ(string(transform->getSourceName()), "$changeNotification");
+        ASSERT_EQ(string(transform->getSourceName()), DSChangeNotification::kStageName);
         transform->setSource(match);
 
         auto next = transform->getNext();
@@ -94,16 +98,62 @@ public:
     }
 };
 
+TEST_F(ChangeNotificationStageTest, ShouldRejectUnrecognizedOption) {
+    auto expCtx = getExpCtx();
+
+    ASSERT_THROWS_CODE(
+        DSChangeNotification::createFromBson(
+            BSON(DSChangeNotification::kStageName << BSON("unexpected" << 4)).firstElement(),
+            expCtx),
+        UserException,
+        40577);
+}
+
+TEST_F(ChangeNotificationStageTest, ShouldRejectResumeAfterOption) {
+    // TODO SERVER-29131 change this test to accept the option.
+    auto expCtx = getExpCtx();
+
+    ASSERT_THROWS_CODE(
+        DSChangeNotification::createFromBson(
+            BSON(DSChangeNotification::kStageName << BSON("resumeAfter" << ts)).firstElement(),
+            expCtx),
+        UserException,
+        40576);
+}
+
+TEST_F(ChangeNotificationStageTest, ShouldRejectNonStringFullDocumentOption) {
+    auto expCtx = getExpCtx();
+
+    ASSERT_THROWS_CODE(
+        DSChangeNotification::createFromBson(
+            BSON(DSChangeNotification::kStageName << BSON("fullDocument" << true)).firstElement(),
+            expCtx),
+        UserException,
+        40574);
+}
+
+TEST_F(ChangeNotificationStageTest, ShouldRejectUnrecognizedFullDocumentOption) {
+    auto expCtx = getExpCtx();
+
+    ASSERT_THROWS_CODE(DSChangeNotification::createFromBson(
+                           BSON(DSChangeNotification::kStageName << BSON("fullDocument"
+                                                                         << "unrecognized"))
+                               .firstElement(),
+                           expCtx),
+                       UserException,
+                       40575);
+}
+
 TEST_F(ChangeNotificationStageTest, StagesGeneratedCorrectly) {
     const auto spec = fromjson("{$changeNotification: {}}");
 
-    vector<intrusive_ptr<DocumentSource>> result =
-        DocumentSourceChangeNotification::createFromBson(spec.firstElement(), getExpCtx());
+    list<intrusive_ptr<DocumentSource>> result =
+        DSChangeNotification::createFromBson(spec.firstElement(), getExpCtx());
 
     ASSERT_EQUALS(result.size(), 2UL);
-    ASSERT_TRUE(dynamic_cast<DocumentSourceMatch*>(result[0].get()));
-    ASSERT_EQUALS(string(result[0]->getSourceName()), "$changeNotification");
-    ASSERT_EQUALS(string(result[1]->getSourceName()), "$changeNotification");
+    ASSERT_TRUE(dynamic_cast<DocumentSourceMatch*>(result.front().get()));
+    ASSERT_EQUALS(string(result.front()->getSourceName()), DSChangeNotification::kStageName);
+    ASSERT_EQUALS(string(result.back()->getSourceName()), DSChangeNotification::kStageName);
 
     // TODO: Check explain result.
 }
@@ -112,11 +162,11 @@ TEST_F(ChangeNotificationStageTest, TransformInsert) {
     OplogEntry insert(optime, 1, OpTypeEnum::kInsert, nss, BSON("_id" << 1 << "x" << 1));
     // Insert
     Document expectedInsert{
-        {"_id", D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
-        {"operationType", "insert"_sd},
-        {"ns", D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {"documentKey", D{{"_id", 1}}},
-        {"newDocument", D{{"_id", 1}, {"x", 1}}},
+        {DSChangeNotification::kIdField, D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
+        {DSChangeNotification::kOperationTypeField, DSChangeNotification::kInsertOpType},
+        {DSChangeNotification::kFullDocumentField, D{{"_id", 1}, {"x", 1}}},
+        {DSChangeNotification::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
+        {DSChangeNotification::kDocumentKeyField, D{{"_id", 1}}},
     };
     checkTransformation(insert, expectedInsert);
 }
@@ -126,13 +176,15 @@ TEST_F(ChangeNotificationStageTest, TransformUpdateFields) {
         optime, 1, OpTypeEnum::kUpdate, nss, BSON("$set" << BSON("y" << 1)), BSON("_id" << 1));
     // Update fields
     Document expectedUpdateField{
-        {"_id", D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
-        {"operationType", "update"_sd},
-        {"ns", D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {"documentKey", D{{"_id", 1}}},
+        {DSChangeNotification::kIdField, D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
+        {DSChangeNotification::kOperationTypeField, DSChangeNotification::kUpdateOpType},
+        {DSChangeNotification::kFullDocumentField, BSONNULL},
+        {DSChangeNotification::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
+        {DSChangeNotification::kDocumentKeyField, D{{"_id", 1}}},
         {
             "updateDescription", D{{"updatedFields", D{{"y", 1}}}, {"removedFields", vector<V>()}},
-        }};
+        },
+    };
     checkTransformation(updateField, expectedUpdateField);
 }
 
@@ -141,10 +193,11 @@ TEST_F(ChangeNotificationStageTest, TransformRemoveFields) {
         optime, 1, OpTypeEnum::kUpdate, nss, BSON("$unset" << BSON("y" << 1)), BSON("_id" << 1));
     // Remove fields
     Document expectedRemoveField{
-        {"_id", D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
-        {"operationType", "update"_sd},
-        {"ns", D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {"documentKey", D{{"_id", 1}}},
+        {DSChangeNotification::kIdField, D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
+        {DSChangeNotification::kOperationTypeField, DSChangeNotification::kUpdateOpType},
+        {DSChangeNotification::kFullDocumentField, BSONNULL},
+        {DSChangeNotification::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
+        {DSChangeNotification::kDocumentKeyField, D{{"_id", 1}}},
         {
             "updateDescription", D{{"updatedFields", D{}}, {"removedFields", vector<V>{V("y"_sd)}}},
         }};
@@ -156,11 +209,11 @@ TEST_F(ChangeNotificationStageTest, TransformReplace) {
         optime, 1, OpTypeEnum::kUpdate, nss, BSON("_id" << 1 << "y" << 1), BSON("_id" << 1));
     // Replace
     Document expectedReplace{
-        {"_id", D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
-        {"operationType", "replace"_sd},
-        {"ns", D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {"documentKey", D{{"_id", 1}}},
-        {"newDocument", D{{"_id", 1}, {"y", 1}}},
+        {DSChangeNotification::kIdField, D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
+        {DSChangeNotification::kOperationTypeField, DSChangeNotification::kReplaceOpType},
+        {DSChangeNotification::kFullDocumentField, D{{"_id", 1}, {"y", 1}}},
+        {DSChangeNotification::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
+        {DSChangeNotification::kDocumentKeyField, D{{"_id", 1}}},
     };
     checkTransformation(replace, expectedReplace);
 }
@@ -169,10 +222,11 @@ TEST_F(ChangeNotificationStageTest, TransformDelete) {
     OplogEntry deleteEntry(optime, 1, OpTypeEnum::kDelete, nss, BSON("_id" << 1));
     // Delete
     Document expectedDelete{
-        {"_id", D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
-        {"operationType", "delete"_sd},
-        {"ns", D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {"documentKey", D{{"_id", 1}}},
+        {DSChangeNotification::kIdField, D{{"ts", ts}, {"ns", nss.ns()}, {"_id", 1}}},
+        {DSChangeNotification::kOperationTypeField, DSChangeNotification::kDeleteOpType},
+        {DSChangeNotification::kFullDocumentField, BSONNULL},
+        {DSChangeNotification::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
+        {DSChangeNotification::kDocumentKeyField, D{{"_id", 1}}},
     };
     checkTransformation(deleteEntry, expectedDelete);
 }
@@ -187,7 +241,9 @@ TEST_F(ChangeNotificationStageTest, TransformInvalidate) {
 
     // Invalidate entry includes $cmd namespace in _id and doesn't have a document id.
     Document expectedInvalidate{
-        {"_id", D{{"ts", ts}, {"ns", nss.getCommandNS().ns()}}}, {"operationType", "invalidate"_sd},
+        {DSChangeNotification::kIdField, D{{"ts", ts}, {"ns", nss.getCommandNS().ns()}}},
+        {DSChangeNotification::kOperationTypeField, DSChangeNotification::kInvalidateOpType},
+        {DSChangeNotification::kFullDocumentField, BSONNULL},
     };
     for (auto& entry : {dropColl, dropDB, rename}) {
         checkTransformation(entry, expectedInvalidate);
@@ -203,8 +259,9 @@ TEST_F(ChangeNotificationStageTest, TransformInvalidateRenameDropTarget) {
                       otherColl.getCommandNS(),
                       BSON("renameCollection" << otherColl.ns() << "to" << nss.ns()));
     Document expectedInvalidate{
-        {"_id", D{{"ts", ts}, {"ns", otherColl.getCommandNS().ns()}}},
-        {"operationType", "invalidate"_sd},
+        {DSChangeNotification::kIdField, D{{"ts", ts}, {"ns", otherColl.getCommandNS().ns()}}},
+        {DSChangeNotification::kOperationTypeField, DSChangeNotification::kInvalidateOpType},
+        {DSChangeNotification::kFullDocumentField, BSONNULL},
     };
     checkTransformation(rename, expectedInvalidate);
 }
@@ -230,5 +287,38 @@ TEST_F(ChangeNotificationStageTest, MatchFiltersCreateIndex) {
     checkTransformation(createIndex, boost::none);
 }
 
+TEST_F(ChangeNotificationStageTest, TransformationShouldBeAbleToReParseSerializedStage) {
+    auto expCtx = getExpCtx();
+
+    auto originalSpec = BSON(DSChangeNotification::kStageName << BSONObj());
+    auto allStages = DSChangeNotification::createFromBson(originalSpec.firstElement(), expCtx);
+    ASSERT_EQ(allStages.size(), 2UL);
+    auto stage = allStages.back();
+    ASSERT(dynamic_cast<DocumentSourceSingleDocumentTransformation*>(stage.get()));
+
+    //
+    // Serialize the stage and confirm contents.
+    //
+    vector<Value> serialization;
+    stage->serializeToArray(serialization);
+    ASSERT_EQ(serialization.size(), 1UL);
+    ASSERT_EQ(serialization[0].getType(), BSONType::Object);
+    auto serializedDoc = serialization[0].getDocument();
+    ASSERT_BSONOBJ_EQ(serializedDoc.toBson(), originalSpec);
+
+    //
+    // Create a new stage from the serialization. Serialize the new stage and confirm that it is
+    // equivalent to the original serialization.
+    //
+    auto serializedBson = serializedDoc.toBson();
+    auto roundTripped = uassertStatusOK(Pipeline::create(
+        DSChangeNotification::createFromBson(serializedBson.firstElement(), expCtx), expCtx));
+
+    auto newSerialization = roundTripped->serialize();
+
+    ASSERT_EQ(newSerialization.size(), 1UL);
+    ASSERT_VALUE_EQ(newSerialization[0], serialization[0]);
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_count.cpp b/src/mongo/db/pipeline/document_source_count.cpp
index 01c4044fece..0a2d299200b 100644
--- a/src/mongo/db/pipeline/document_source_count.cpp
+++ b/src/mongo/db/pipeline/document_source_count.cpp
@@ -39,14 +39,14 @@
 namespace mongo {
 
 using boost::intrusive_ptr;
-using std::vector;
+using std::list;
 using std::string;
 
 REGISTER_MULTI_STAGE_ALIAS(count,
                            LiteParsedDocumentSourceDefault::parse,
                            DocumentSourceCount::createFromBson);
 
-vector<intrusive_ptr<DocumentSource>> DocumentSourceCount::createFromBson(
+list<intrusive_ptr<DocumentSource>> DocumentSourceCount::createFromBson(
     BSONElement elem, const intrusive_ptr<ExpressionContext>& pExpCtx) {
     uassert(40156,
             str::stream() << "the count field must be a non-empty string",
diff --git a/src/mongo/db/pipeline/document_source_count.h b/src/mongo/db/pipeline/document_source_count.h
index 50fbccdf41b..e77ce3eb0ed 100644
--- a/src/mongo/db/pipeline/document_source_count.h
+++ b/src/mongo/db/pipeline/document_source_count.h
@@ -40,7 +40,7 @@ public:
     /**
      * Returns a $group stage followed by a $project stage.
      */
-    static std::vector<boost::intrusive_ptr<DocumentSource>> createFromBson(
+    static std::list<boost::intrusive_ptr<DocumentSource>> createFromBson(
         BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
 private:
diff --git a/src/mongo/db/pipeline/document_source_count_test.cpp b/src/mongo/db/pipeline/document_source_count_test.cpp
index e083399b0fa..254f011ff1d 100644
--- a/src/mongo/db/pipeline/document_source_count_test.cpp
+++ b/src/mongo/db/pipeline/document_source_count_test.cpp
@@ -45,22 +45,23 @@
 
 namespace mongo {
 namespace {
-using std::vector;
 using boost::intrusive_ptr;
+using std::list;
+using std::vector;
 
 class CountReturnsGroupAndProjectStages : public AggregationContextFixture {
 public:
     void testCreateFromBsonResult(BSONObj countSpec) {
-        vector<intrusive_ptr<DocumentSource>> result =
+        list<intrusive_ptr<DocumentSource>> result =
             DocumentSourceCount::createFromBson(countSpec.firstElement(), getExpCtx());
 
         ASSERT_EQUALS(result.size(), 2UL);
 
-        const auto* groupStage = dynamic_cast<DocumentSourceGroup*>(result[0].get());
+        const auto* groupStage = dynamic_cast<DocumentSourceGroup*>(result.front().get());
         ASSERT(groupStage);
 
         const auto* projectStage =
-            dynamic_cast<DocumentSourceSingleDocumentTransformation*>(result[1].get());
+            dynamic_cast<DocumentSourceSingleDocumentTransformation*>(result.back().get());
         ASSERT(projectStage);
 
         auto explain = ExplainOptions::Verbosity::kQueryPlanner;
@@ -94,7 +95,7 @@ TEST_F(CountReturnsGroupAndProjectStages, ValidStringSpec) {
 
 class InvalidCountSpec : public AggregationContextFixture {
 public:
-    vector<intrusive_ptr<DocumentSource>> createCount(BSONObj countSpec) {
+    list<intrusive_ptr<DocumentSource>> createCount(BSONObj countSpec) {
         auto specElem = countSpec.firstElement();
         return DocumentSourceCount::createFromBson(specElem, getExpCtx());
     }
diff --git a/src/mongo/db/pipeline/document_source_lookup_change_post_image.cpp b/src/mongo/db/pipeline/document_source_lookup_change_post_image.cpp
new file mode 100644
index 00000000000..ee605710a54
--- /dev/null
+++ b/src/mongo/db/pipeline/document_source_lookup_change_post_image.cpp
@@ -0,0 +1,125 @@
+/**
+ * Copyright (C) 2017 MongoDB Inc.
+ *
+ * This program is free software: you can redistribute it and/or  modify
+ * it under the terms of the GNU Affero General Public License, version 3,
+ * as published by the Free Software Foundation.
+ *
+ * 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
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * 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 GNU Affero General 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/platform/basic.h"
+
+#include "mongo/db/pipeline/document_source_lookup_change_post_image.h"
+
+#include "mongo/bson/simple_bsonelement_comparator.h"
+
+namespace mongo {
+
+constexpr StringData DocumentSourceLookupChangePostImage::kStageName;
+constexpr StringData DocumentSourceLookupChangePostImage::kFullDocumentFieldName;
+
+namespace {
+Value assertFieldHasType(const Document& fullDoc, StringData fieldName, BSONType expectedType) {
+    auto val = fullDoc[fieldName];
+    uassert(40578,
+            str::stream() << "failed to look up post image after change: expected \"" << fieldName
+                          << "\" field to have type "
+                          << typeName(expectedType)
+                          << ", instead found type "
+                          << typeName(val.getType())
+                          << ": "
+                          << val.toString()
+                          << ", full object: "
+                          << fullDoc.toString(),
+            val.getType() == expectedType);
+    return val;
+}
+}  // namespace
+
+DocumentSource::GetNextResult DocumentSourceLookupChangePostImage::getNext() {
+    pExpCtx->checkForInterrupt();
+
+    auto input = pSource->getNext();
+    if (!input.isAdvanced()) {
+        return input;
+    }
+    auto opTypeVal = assertFieldHasType(input.getDocument(),
+                                        DocumentSourceChangeNotification::kOperationTypeField,
+                                        BSONType::String);
+    if (opTypeVal.getString() != DocumentSourceChangeNotification::kUpdateOpType) {
+        return input;
+    }
+
+    MutableDocument output(input.releaseDocument());
+    output[kFullDocumentFieldName] = lookupPostImage(output.peek());
+    return output.freeze();
+}
+
+NamespaceString DocumentSourceLookupChangePostImage::assertNamespaceMatches(
+    const Document& inputDoc) const {
+    auto namespaceObject = assertFieldHasType(inputDoc,
+                                              DocumentSourceChangeNotification::kNamespaceField,
+                                              BSONType::Object)
+                               .getDocument();
+    auto dbName = assertFieldHasType(namespaceObject, "db"_sd, BSONType::String);
+    auto collectionName = assertFieldHasType(namespaceObject, "coll"_sd, BSONType::String);
+    NamespaceString nss(dbName.getString(), collectionName.getString());
+    uassert(40579,
+            str::stream() << "unexpected namespace during post image lookup: " << nss.ns()
+                          << ", expected "
+                          << pExpCtx->ns.ns(),
+            nss == pExpCtx->ns);
+    return nss;
+}
+
+Value DocumentSourceLookupChangePostImage::lookupPostImage(const Document& updateOp) const {
+    // Make sure we have a well-formed input.
+    auto nss = assertNamespaceMatches(updateOp);
+
+    auto documentKey = assertFieldHasType(
+        updateOp, DocumentSourceChangeNotification::kDocumentKeyField, BSONType::Object);
+    auto matchSpec = BSON("$match" << documentKey);
+
+    // TODO SERVER-29134 we need to extract the namespace from the document and set them on the new
+    // ExpressionContext if we're getting notifications from an entire database.
+    auto foreignExpCtx = pExpCtx->copyWith(nss);
+    auto pipeline = uassertStatusOK(_mongod->makePipeline({matchSpec}, foreignExpCtx));
+
+    if (auto first = pipeline->getNext()) {
+        auto lookedUpDocument = Value(*first);
+        if (auto next = pipeline->getNext()) {
+            uasserted(40580,
+                      str::stream() << "found more than document with documentKey "
+                                    << documentKey.toString()
+                                    << " while looking up post image after change: ["
+                                    << lookedUpDocument.toString()
+                                    << ", "
+                                    << next->toString()
+                                    << "]");
+        }
+        return lookedUpDocument;
+    }
+    // We couldn't find it with the documentKey, it may have been deleted.
+    return Value(BSONNULL);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_lookup_change_post_image.h b/src/mongo/db/pipeline/document_source_lookup_change_post_image.h
new file mode 100644
index 00000000000..ca14656a63c
--- /dev/null
+++ b/src/mongo/db/pipeline/document_source_lookup_change_post_image.h
@@ -0,0 +1,115 @@
+/**
+ * Copyright (C) 2017 MongoDB Inc.
+ *
+ * This program is free software: you can redistribute it and/or  modify
+ * it under the terms of the GNU Affero General Public License, version 3,
+ * as published by the Free Software Foundation.
+ *
+ * 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
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * 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 GNU Affero General 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 "mongo/db/pipeline/document_source.h"
+#include "mongo/db/pipeline/document_source_change_notification.h"
+
+namespace mongo {
+
+/**
+ * Part of the change notification API machinery used to look up the post-image of a document. Uses
+ * the "documentKey" field of the input to look up the new version of the document.
+ *
+ * Uses the ExpressionContext to determine what collection to look up into.
+ * TODO SERVER-29134 When we allow change streams on multiple collections, this will need to change.
+ */
+class DocumentSourceLookupChangePostImage final : public DocumentSourceNeedsMongod {
+public:
+    static constexpr StringData kStageName = "$_internalLookupChangePostImage"_sd;
+    static constexpr StringData kFullDocumentFieldName =
+        DocumentSourceChangeNotification::kFullDocumentField;
+
+    /**
+     * Creates a DocumentSourceLookupChangePostImage stage.
+     */
+    static boost::intrusive_ptr<DocumentSourceLookupChangePostImage> create(
+        const boost::intrusive_ptr<ExpressionContext>& expCtx) {
+        return new DocumentSourceLookupChangePostImage(expCtx);
+    }
+
+    /**
+     * Only modifies a single path: "fullDocument".
+     */
+    GetModPathsReturn getModifiedPaths() const final {
+        return {GetModPathsReturn::Type::kFiniteSet, {kFullDocumentFieldName.toString()}, {}};
+    }
+
+    StageConstraints constraints() const final {
+        StageConstraints constraints;
+        constraints.canSwapWithMatch = true;
+        constraints.isAllowedInsideFacetStage = false;
+        return constraints;
+    }
+
+    GetDepsReturn getDependencies(DepsTracker* deps) const {
+        // The namespace is not technically needed yet, but we will if there is more than one
+        // collection involved.
+        deps->fields.insert(DocumentSourceChangeNotification::kNamespaceField.toString());
+        deps->fields.insert(DocumentSourceChangeNotification::kDocumentKeyField.toString());
+        deps->fields.insert(DocumentSourceChangeNotification::kOperationTypeField.toString());
+        // This stage does not restrict the output fields to a finite set, and has no impact on
+        // whether metadata is available or needed.
+        return SEE_NEXT;
+    }
+
+    /**
+     * Performs the lookup to retrieve the full document.
+     */
+    GetNextResult getNext() final;
+
+    Value serialize(boost::optional<ExplainOptions::Verbosity> explain = boost::none) const final {
+        if (explain) {
+            return Value{Document{{kStageName, Document()}}};
+        }
+        return Value();  // Do not serialize this stage unless we're explaining.
+    }
+
+    const char* getSourceName() const final {
+        return kStageName.rawData();
+    }
+
+private:
+    DocumentSourceLookupChangePostImage(const boost::intrusive_ptr<ExpressionContext>& expCtx)
+        : DocumentSourceNeedsMongod(expCtx) {}
+
+    /**
+     * Uses the "documentKey" field from 'updateOp' to look up the current version of the document.
+     * Returns Value(BSONNULL) if the document couldn't be found.
+     */
+    Value lookupPostImage(const Document& updateOp) const;
+
+    /**
+     * Throws a UserException if the namespace found in 'inputDoc' doesn't match the one on the
+     * ExpressionContext.
+     */
+    NamespaceString assertNamespaceMatches(const Document& inputDoc) const;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_lookup_change_post_image_test.cpp b/src/mongo/db/pipeline/document_source_lookup_change_post_image_test.cpp
new file mode 100644
index 00000000000..728db433c55
--- /dev/null
+++ b/src/mongo/db/pipeline/document_source_lookup_change_post_image_test.cpp
@@ -0,0 +1,266 @@
+/**
+ * Copyright (C) 2017 MongoDB Inc.
+ *
+ * This program is free software: you can redistribute it and/or  modify
+ * it under the terms of the GNU Affero General Public License, version 3,
+ * as published by the Free Software Foundation.
+ *
+ * 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
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * 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 GNU Affero General 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/platform/basic.h"
+
+#include <boost/intrusive_ptr.hpp>
+#include <deque>
+#include <vector>
+
+#include "mongo/bson/bsonmisc.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/db/pipeline/aggregation_context_fixture.h"
+#include "mongo/db/pipeline/document.h"
+#include "mongo/db/pipeline/document_source_change_notification.h"
+#include "mongo/db/pipeline/document_source_lookup_change_post_image.h"
+#include "mongo/db/pipeline/document_source_mock.h"
+#include "mongo/db/pipeline/document_value_test_util.h"
+#include "mongo/db/pipeline/field_path.h"
+#include "mongo/db/pipeline/stub_mongod_interface.h"
+#include "mongo/db/pipeline/value.h"
+
+namespace mongo {
+namespace {
+using boost::intrusive_ptr;
+using std::deque;
+using std::vector;
+
+// This provides access to getExpCtx(), but we'll use a different name for this test suite.
+using DocumentSourceLookupChangePostImageTest = AggregationContextFixture;
+
+/**
+ * A mock MongodInterface which allows mocking a foreign pipeline.
+ */
+class MockMongodInterface final : public StubMongodInterface {
+public:
+    MockMongodInterface(deque<DocumentSource::GetNextResult> mockResults)
+        : _mockResults(std::move(mockResults)) {}
+
+    bool isSharded(const NamespaceString& ns) final {
+        return false;
+    }
+
+    StatusWith<std::unique_ptr<Pipeline, Pipeline::Deleter>> makePipeline(
+        const std::vector<BSONObj>& rawPipeline,
+        const boost::intrusive_ptr<ExpressionContext>& expCtx) final {
+        auto pipeline = Pipeline::parse(rawPipeline, expCtx);
+        if (!pipeline.isOK()) {
+            return pipeline.getStatus();
+        }
+
+        pipeline.getValue()->addInitialSource(DocumentSourceMock::create(_mockResults));
+        pipeline.getValue()->optimizePipeline();
+
+        return pipeline;
+    }
+
+private:
+    deque<DocumentSource::GetNextResult> _mockResults;
+};
+
+TEST_F(DocumentSourceLookupChangePostImageTest, ShouldErrorIfMissingDocumentKeyOnUpdate) {
+    auto expCtx = getExpCtx();
+
+    // Set up the $lookup stage.
+    auto lookupChangeStage = DocumentSourceLookupChangePostImage::create(expCtx);
+
+    // Mock its input with a document without a "documentKey" field.
+    auto mockLocalSource = DocumentSourceMock::create(
+        Document{{"operationType", "update"_sd},
+                 {"fullDocument", Document{{"_id", 0}}},
+                 {"ns", Document{{"db", expCtx->ns.db()}, {"coll", expCtx->ns.coll()}}}});
+
+    lookupChangeStage->setSource(mockLocalSource.get());
+
+    // Mock out the foreign collection.
+    lookupChangeStage->injectMongodInterface(
+        std::make_shared<MockMongodInterface>(deque<DocumentSource::GetNextResult>{}));
+
+    ASSERT_THROWS_CODE(lookupChangeStage->getNext(), UserException, 40578);
+}
+
+TEST_F(DocumentSourceLookupChangePostImageTest, ShouldErrorIfMissingOperationType) {
+    auto expCtx = getExpCtx();
+
+    // Set up the $lookup stage.
+    auto lookupChangeStage = DocumentSourceLookupChangePostImage::create(expCtx);
+
+    // Mock its input with a document without a "ns" field.
+    auto mockLocalSource = DocumentSourceMock::create(
+        Document{{"documentKey", Document{{"_id", 0}}},
+                 {"fullDocument", Document{{"_id", 0}}},
+                 {"ns", Document{{"db", expCtx->ns.db()}, {"coll", expCtx->ns.coll()}}}});
+
+    lookupChangeStage->setSource(mockLocalSource.get());
+
+    // Mock out the foreign collection.
+    lookupChangeStage->injectMongodInterface(
+        std::make_shared<MockMongodInterface>(deque<DocumentSource::GetNextResult>{}));
+
+    ASSERT_THROWS_CODE(lookupChangeStage->getNext(), UserException, 40578);
+}
+
+TEST_F(DocumentSourceLookupChangePostImageTest, ShouldErrorIfMissingNamespace) {
+    auto expCtx = getExpCtx();
+
+    // Set up the $lookup stage.
+    auto lookupChangeStage = DocumentSourceLookupChangePostImage::create(expCtx);
+
+    // Mock its input with a document without a "ns" field.
+    auto mockLocalSource = DocumentSourceMock::create(Document{
+        {"documentKey", Document{{"_id", 0}}}, {"operationType", "update"_sd},
+    });
+
+    lookupChangeStage->setSource(mockLocalSource.get());
+
+    // Mock out the foreign collection.
+    lookupChangeStage->injectMongodInterface(
+        std::make_shared<MockMongodInterface>(deque<DocumentSource::GetNextResult>{}));
+
+    ASSERT_THROWS_CODE(lookupChangeStage->getNext(), UserException, 40578);
+}
+
+TEST_F(DocumentSourceLookupChangePostImageTest, ShouldErrorIfNsFieldHasWrongType) {
+    auto expCtx = getExpCtx();
+
+    // Set up the $lookup stage.
+    auto lookupChangeStage = DocumentSourceLookupChangePostImage::create(expCtx);
+
+    // Mock its input with a document without a "ns" field.
+    auto mockLocalSource = DocumentSourceMock::create(
+        Document{{"documentKey", Document{{"_id", 0}}}, {"operationType", "update"_sd}, {"ns", 4}});
+
+    lookupChangeStage->setSource(mockLocalSource.get());
+
+    // Mock out the foreign collection.
+    lookupChangeStage->injectMongodInterface(
+        std::make_shared<MockMongodInterface>(deque<DocumentSource::GetNextResult>{}));
+
+    ASSERT_THROWS_CODE(lookupChangeStage->getNext(), UserException, 40578);
+}
+
+TEST_F(DocumentSourceLookupChangePostImageTest, ShouldErrorIfNsFieldDoesNotMatchPipeline) {
+    auto expCtx = getExpCtx();
+
+    // Set up the $lookup stage.
+    auto lookupChangeStage = DocumentSourceLookupChangePostImage::create(expCtx);
+
+    // Mock its input with a document without a "ns" field.
+    auto mockLocalSource = DocumentSourceMock::create(
+        Document{{"documentKey", Document{{"_id", 0}}},
+                 {"operationType", "update"_sd},
+                 {"ns", Document{{"db", "DIFFERENT"_sd}, {"coll", expCtx->ns.coll()}}}});
+
+    lookupChangeStage->setSource(mockLocalSource.get());
+
+    // Mock out the foreign collection.
+    lookupChangeStage->injectMongodInterface(
+        std::make_shared<MockMongodInterface>(deque<DocumentSource::GetNextResult>{}));
+
+    ASSERT_THROWS_CODE(lookupChangeStage->getNext(), UserException, 40579);
+}
+
+TEST_F(DocumentSourceLookupChangePostImageTest, ShouldErrorIfDocumentKeyIsNotUnique) {
+    auto expCtx = getExpCtx();
+
+    // Set up the $lookup stage.
+    auto lookupChangeStage = DocumentSourceLookupChangePostImage::create(expCtx);
+
+    // Mock its input with an update document.
+    auto mockLocalSource = DocumentSourceMock::create(
+        Document{{"documentKey", Document{{"_id", 0}}},
+                 {"operationType", "update"_sd},
+                 {"ns", Document{{"db", expCtx->ns.db()}, {"coll", expCtx->ns.coll()}}}});
+
+    lookupChangeStage->setSource(mockLocalSource.get());
+
+    // Mock out the foreign collection to have two documents with the same document key.
+    deque<DocumentSource::GetNextResult> foreignCollection = {Document{{"_id", 0}},
+                                                              Document{{"_id", 0}}};
+    lookupChangeStage->injectMongodInterface(
+        std::make_shared<MockMongodInterface>(std::move(foreignCollection)));
+
+    ASSERT_THROWS_CODE(lookupChangeStage->getNext(), UserException, 40580);
+}
+
+TEST_F(DocumentSourceLookupChangePostImageTest, ShouldPropagatePauses) {
+    auto expCtx = getExpCtx();
+
+    // Set up the $lookup stage.
+    auto lookupChangeStage = DocumentSourceLookupChangePostImage::create(expCtx);
+
+    // Mock its input, pausing every other result.
+    auto mockLocalSource = DocumentSourceMock::create(
+        {Document{{"documentKey", Document{{"_id", 0}}},
+                  {"operationType", "insert"_sd},
+                  {"fullDocument", Document{{"_id", 0}}},
+                  {"ns", Document{{"db", expCtx->ns.db()}, {"coll", expCtx->ns.coll()}}}},
+         DocumentSource::GetNextResult::makePauseExecution(),
+         Document{{"documentKey", Document{{"_id", 1}}},
+                  {"operationType", "update"_sd},
+                  {"fullDocument", BSONNULL},
+                  {"ns", Document{{"db", expCtx->ns.db()}, {"coll", expCtx->ns.coll()}}}},
+         DocumentSource::GetNextResult::makePauseExecution()});
+
+    lookupChangeStage->setSource(mockLocalSource.get());
+
+    // Mock out the foreign collection.
+    deque<DocumentSource::GetNextResult> mockForeignContents{Document{{"_id", 0}},
+                                                             Document{{"_id", 1}}};
+    lookupChangeStage->injectMongodInterface(
+        std::make_shared<MockMongodInterface>(std::move(mockForeignContents)));
+
+    auto next = lookupChangeStage->getNext();
+    ASSERT_TRUE(next.isAdvanced());
+    ASSERT_DOCUMENT_EQ(
+        next.releaseDocument(),
+        (Document{{"documentKey", Document{{"_id", 0}}},
+                  {"operationType", "insert"_sd},
+                  {"fullDocument", Document{{"_id", 0}}},
+                  {"ns", Document{{"db", expCtx->ns.db()}, {"coll", expCtx->ns.coll()}}}}));
+
+    ASSERT_TRUE(lookupChangeStage->getNext().isPaused());
+
+    next = lookupChangeStage->getNext();
+    ASSERT_TRUE(next.isAdvanced());
+    ASSERT_DOCUMENT_EQ(
+        next.releaseDocument(),
+        (Document{{"documentKey", Document{{"_id", 1}}},
+                  {"operationType", "update"_sd},
+                  {"fullDocument", Document{{"_id", 1}}},
+                  {"ns", Document{{"db", expCtx->ns.db()}, {"coll", expCtx->ns.coll()}}}}));
+
+    ASSERT_TRUE(lookupChangeStage->getNext().isPaused());
+
+    ASSERT_TRUE(lookupChangeStage->getNext().isEOF());
+    ASSERT_TRUE(lookupChangeStage->getNext().isEOF());
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_sort_by_count.cpp b/src/mongo/db/pipeline/document_source_sort_by_count.cpp
index 48cfaa9f559..97c5e0f0297 100644
--- a/src/mongo/db/pipeline/document_source_sort_by_count.cpp
+++ b/src/mongo/db/pipeline/document_source_sort_by_count.cpp
@@ -39,13 +39,13 @@
 namespace mongo {
 
 using boost::intrusive_ptr;
-using std::vector;
+using std::list;
 
 REGISTER_MULTI_STAGE_ALIAS(sortByCount,
                            LiteParsedDocumentSourceDefault::parse,
                            DocumentSourceSortByCount::createFromBson);
 
-vector<intrusive_ptr<DocumentSource>> DocumentSourceSortByCount::createFromBson(
+list<intrusive_ptr<DocumentSource>> DocumentSourceSortByCount::createFromBson(
     BSONElement elem, const intrusive_ptr<ExpressionContext>& pExpCtx) {
     if (elem.type() == Object) {
         // Make sure that the sortByCount field is an expression inside an object
diff --git a/src/mongo/db/pipeline/document_source_sort_by_count.h b/src/mongo/db/pipeline/document_source_sort_by_count.h
index 69bf8d2c5e0..c8196c9b498 100644
--- a/src/mongo/db/pipeline/document_source_sort_by_count.h
+++ b/src/mongo/db/pipeline/document_source_sort_by_count.h
@@ -40,7 +40,7 @@ public:
     /**
      * Returns a $group stage followed by a $sort stage.
      */
-    static std::vector<boost::intrusive_ptr<DocumentSource>> createFromBson(
+    static std::list<boost::intrusive_ptr<DocumentSource>> createFromBson(
         BSONElement elem, const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
 private:
diff --git a/src/mongo/db/pipeline/document_source_sort_by_count_test.cpp b/src/mongo/db/pipeline/document_source_sort_by_count_test.cpp
index 3e0a007c59b..12975e96149 100644
--- a/src/mongo/db/pipeline/document_source_sort_by_count_test.cpp
+++ b/src/mongo/db/pipeline/document_source_sort_by_count_test.cpp
@@ -46,8 +46,9 @@
 
 namespace mongo {
 namespace {
-using std::vector;
 using boost::intrusive_ptr;
+using std::list;
+using std::vector;
 
 /**
  * Fixture to test that $sortByCount returns a DocumentSourceGroup and DocumentSourceSort.
@@ -55,15 +56,15 @@ using boost::intrusive_ptr;
 class SortByCountReturnsGroupAndSort : public AggregationContextFixture {
 public:
     void testCreateFromBsonResult(BSONObj sortByCountSpec, Value expectedGroupExplain) {
-        vector<intrusive_ptr<DocumentSource>> result =
+        list<intrusive_ptr<DocumentSource>> result =
             DocumentSourceSortByCount::createFromBson(sortByCountSpec.firstElement(), getExpCtx());
 
         ASSERT_EQUALS(result.size(), 2UL);
 
-        const auto* groupStage = dynamic_cast<DocumentSourceGroup*>(result[0].get());
+        const auto* groupStage = dynamic_cast<DocumentSourceGroup*>(result.front().get());
         ASSERT(groupStage);
 
-        const auto* sortStage = dynamic_cast<DocumentSourceSort*>(result[1].get());
+        const auto* sortStage = dynamic_cast<DocumentSourceSort*>(result.back().get());
         ASSERT(sortStage);
 
         // Serialize the DocumentSourceGroup and DocumentSourceSort from $sortByCount so that we can
@@ -111,7 +112,7 @@ TEST_F(SortByCountReturnsGroupAndSort, ExpressionInObjectSpec) {
  */
 class InvalidSortByCountSpec : public AggregationContextFixture {
 public:
-    vector<intrusive_ptr<DocumentSource>> createSortByCount(BSONObj sortByCountSpec) {
+    list<intrusive_ptr<DocumentSource>> createSortByCount(BSONObj sortByCountSpec) {
         auto specElem = sortByCountSpec.firstElement();
         return DocumentSourceSortByCount::createFromBson(specElem, getExpCtx());
     }
diff --git a/src/mongo/db/pipeline/expression_context.cpp b/src/mongo/db/pipeline/expression_context.cpp
index d2a16a6ca87..f5f42d197c0 100644
--- a/src/mongo/db/pipeline/expression_context.cpp
+++ b/src/mongo/db/pipeline/expression_context.cpp
@@ -73,7 +73,7 @@ void ExpressionContext::setCollator(std::unique_ptr<CollatorInterface> coll) {
 }
 
 intrusive_ptr<ExpressionContext> ExpressionContext::copyWith(NamespaceString ns) const {
-    intrusive_ptr<ExpressionContext> expCtx = new ExpressionContext();
+    intrusive_ptr<ExpressionContext> expCtx = new ExpressionContext(std::move(ns));
 
     expCtx->explain = explain;
     expCtx->inShard = inShard;
@@ -81,7 +81,6 @@ intrusive_ptr<ExpressionContext> ExpressionContext::copyWith(NamespaceString ns)
     expCtx->extSortAllowed = extSortAllowed;
     expCtx->bypassDocumentValidation = bypassDocumentValidation;
 
-    expCtx->ns = std::move(ns);
     expCtx->tempDir = tempDir;
 
     expCtx->opCtx = opCtx;
diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index d04b095aa4b..ba2690796f6 100644
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -134,7 +134,9 @@ public:
 
 protected:
     static const int kInterruptCheckPeriod = 128;
-    ExpressionContext() : variablesParseState(variables.useIdGenerator()) {}
+
+    ExpressionContext(NamespaceString nss)
+        : ns(std::move(nss)), variablesParseState(variables.useIdGenerator()) {}
 
     /**
      * Sets '_collator' and resets '_documentComparator' and '_valueComparator'.
diff --git a/src/mongo/db/pipeline/expression_context_for_test.h b/src/mongo/db/pipeline/expression_context_for_test.h
index 234b54e1a10..542e5b91eea 100644
--- a/src/mongo/db/pipeline/expression_context_for_test.h
+++ b/src/mongo/db/pipeline/expression_context_for_test.h
@@ -40,7 +40,9 @@ namespace mongo {
  */
 class ExpressionContextForTest : public ExpressionContext {
 public:
-    ExpressionContextForTest() = default;
+    ExpressionContextForTest() : ExpressionContext(NamespaceString{"test"_sd, "namespace"_sd}) {}
+
+    ExpressionContextForTest(NamespaceString nss) : ExpressionContext(std::move(nss)) {}
 
     ExpressionContextForTest(OperationContext* opCtx, const AggregationRequest& request)
         : ExpressionContext(opCtx, request, nullptr, {}) {}
diff --git a/src/mongo/db/pipeline/pipeline_test.cpp b/src/mongo/db/pipeline/pipeline_test.cpp
index 1f37b30fc90..1402f6f504d 100644
--- a/src/mongo/db/pipeline/pipeline_test.cpp
+++ b/src/mongo/db/pipeline/pipeline_test.cpp
@@ -37,7 +37,11 @@
 #include "mongo/db/pipeline/dependencies.h"
 #include "mongo/db/pipeline/document.h"
 #include "mongo/db/pipeline/document_source.h"
+#include "mongo/db/pipeline/document_source_change_notification.h"
+#include "mongo/db/pipeline/document_source_lookup_change_post_image.h"
+#include "mongo/db/pipeline/document_source_match.h"
 #include "mongo/db/pipeline/document_source_mock.h"
+#include "mongo/db/pipeline/document_source_project.h"
 #include "mongo/db/pipeline/document_value_test_util.h"
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/db/pipeline/field_path.h"
@@ -52,6 +56,8 @@ using boost::intrusive_ptr;
 using std::string;
 using std::vector;
 
+const NamespaceString kTestNss = NamespaceString("a.collection");
+
 namespace Optimizations {
 using namespace mongo;
 
@@ -77,7 +83,7 @@ void assertPipelineOptimizesAndSerializesTo(std::string inputPipeJson,
         ASSERT_EQUALS(stageElem.type(), BSONType::Object);
         rawPipeline.push_back(stageElem.embeddedObject());
     }
-    AggregationRequest request(NamespaceString("a.collection"), rawPipeline);
+    AggregationRequest request(kTestNss, rawPipeline);
     intrusive_ptr<ExpressionContextForTest> ctx =
         new ExpressionContextForTest(opCtx.get(), request);
 
@@ -962,6 +968,52 @@ TEST(PipelineOptimizationTest, MatchOnMaxLengthShouldMoveAcrossRename) {
     assertPipelineOptimizesTo(inputPipe, outputPipe);
 }
 
+TEST(PipelineOptimizationTest, ChangeNotificationLookupSwapsWithIndependentMatch) {
+    QueryTestServiceContext testServiceContext;
+    auto opCtx = testServiceContext.makeOperationContext();
+
+    intrusive_ptr<ExpressionContext> expCtx(new ExpressionContextForTest(kTestNss));
+    expCtx->opCtx = opCtx.get();
+
+    auto spec = BSON("$changeNotification" << BSON("fullDocument"
+                                                   << "lookup"));
+    auto stages = DocumentSourceChangeNotification::createFromBson(spec.firstElement(), expCtx);
+    ASSERT_EQ(stages.size(), 3UL);
+    // Make sure the change lookup is at the end.
+    ASSERT(dynamic_cast<DocumentSourceLookupChangePostImage*>(stages.back().get()));
+
+    auto matchPredicate = BSON("extra"
+                               << "predicate");
+    stages.push_back(DocumentSourceMatch::create(matchPredicate, expCtx));
+    auto pipeline = uassertStatusOK(Pipeline::create(stages, expCtx));
+    pipeline->optimizePipeline();
+
+    // Make sure the $match stage has swapped before the change look up.
+    ASSERT(dynamic_cast<DocumentSourceLookupChangePostImage*>(pipeline->getSources().back().get()));
+}
+
+TEST(PipelineOptimizationTest, ChangeNotificationLookupDoesNotSwapWithMatchOnPostImage) {
+    QueryTestServiceContext testServiceContext;
+    auto opCtx = testServiceContext.makeOperationContext();
+
+    intrusive_ptr<ExpressionContext> expCtx(new ExpressionContextForTest(kTestNss));
+    expCtx->opCtx = opCtx.get();
+
+    auto spec = BSON("$changeNotification" << BSON("fullDocument"
+                                                   << "lookup"));
+    auto stages = DocumentSourceChangeNotification::createFromBson(spec.firstElement(), expCtx);
+    ASSERT_EQ(stages.size(), 3UL);
+    // Make sure the change lookup is at the end.
+    ASSERT(dynamic_cast<DocumentSourceLookupChangePostImage*>(stages.back().get()));
+
+    stages.push_back(DocumentSourceMatch::create(
+        BSON(DocumentSourceLookupChangePostImage::kFullDocumentFieldName << BSONNULL), expCtx));
+    auto pipeline = uassertStatusOK(Pipeline::create(stages, expCtx));
+    pipeline->optimizePipeline();
+
+    // Make sure the $match stage stays at the end.
+    ASSERT(dynamic_cast<DocumentSourceMatch*>(pipeline->getSources().back().get()));
+}
 }  // namespace Local
 
 namespace Sharded {
@@ -986,7 +1038,7 @@ public:
             ASSERT_EQUALS(stageElem.type(), BSONType::Object);
             rawPipeline.push_back(stageElem.embeddedObject());
         }
-        AggregationRequest request(NamespaceString("a.collection"), rawPipeline);
+        AggregationRequest request(kTestNss, rawPipeline);
         intrusive_ptr<ExpressionContextForTest> ctx =
             new ExpressionContextForTest(&_opCtx, request);
 
@@ -1391,7 +1443,7 @@ TEST_F(PipelineInitialSourceNSTest, CollectionNSNotValidIfInitialStageIsCollecti
     auto collectionlessSource = DocumentSourceCollectionlessMock::create();
     auto ctx = getExpCtx();
 
-    ctx->ns = NamespaceString("a.collection");
+    ctx->ns = kTestNss;
 
     ASSERT_NOT_OK(Pipeline::create({collectionlessSource}, ctx).getStatus());
 }
diff --git a/src/mongo/db/pipeline/value.h b/src/mongo/db/pipeline/value.h
index 9f7a357f677..04e396dc58d 100644
--- a/src/mongo/db/pipeline/value.h
+++ b/src/mongo/db/pipeline/value.h
@@ -117,6 +117,9 @@ public:
     explicit Value(const MaxKeyLabeler&) : _storage(MaxKey) {}        // MAXKEY
     explicit Value(const Date_t& date) : _storage(Date, date.toMillisSinceEpoch()) {}
 
+    explicit Value(const char*) = delete;  // Use StringData instead to prevent accidentally
+                                           // terminating the string at the first null byte.
+
     // TODO: add an unsafe version that can share storage with the BSONElement
     /// Deep-convert from BSONElement to Value
     explicit Value(const BSONElement& elem);