mongodb/jstests/change_streams/lookup_post_image.js

// Tests the 'fullDocument' argument to the $changeStream stage.
//
// The $changeStream stage is not allowed within a $facet stage.
// @tags: [
//   do_not_wrap_aggregations_in_facets,
//   uses_multiple_connections,
// ]
(function() {
    "use strict";

    load("jstests/libs/change_stream_util.js");
    load("jstests/libs/collection_drop_recreate.js");  // For assert[Drop|Create]Collection.
    load("jstests/libs/fixture_helpers.js");           // For FixtureHelpers.
    load("jstests/replsets/libs/two_phase_drops.js");  // For 'TwoPhaseDropCollectionTest'.

    const coll = assertDropAndRecreateCollection(db, "change_post_image");
    const cst = new ChangeStreamTest(db);

    jsTestLog("Testing change streams without 'fullDocument' specified");
    // Test that not specifying 'fullDocument' does include a 'fullDocument' in the result for
    // an insert.
    let cursor = cst.startWatchingChanges({pipeline: [{$changeStream: {}}], collection: coll});
    assert.writeOK(coll.insert({_id: "fullDocument not specified"}));
    let latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "insert");
    assert.eq(latestChange.fullDocument, {_id: "fullDocument not specified"});

    // Test that not specifying 'fullDocument' does include a 'fullDocument' in the result for a
    // replacement-style update.
    assert.writeOK(coll.update({_id: "fullDocument not specified"},
                               {_id: "fullDocument not specified", replaced: true}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "replace");
    assert.eq(latestChange.fullDocument, {_id: "fullDocument not specified", replaced: true});

    // Test that not specifying 'fullDocument' does not include a 'fullDocument' in the result
    // for a non-replacement update.
    assert.writeOK(coll.update({_id: "fullDocument not specified"}, {$set: {updated: true}}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "update");
    assert(!latestChange.hasOwnProperty("fullDocument"));

    jsTestLog("Testing change streams with 'fullDocument' specified as 'default'");

    // Test that specifying 'fullDocument' as 'default' does include a 'fullDocument' in the
    // result for an insert.
    cursor = cst.startWatchingChanges(
        {collection: coll, pipeline: [{$changeStream: {fullDocument: "default"}}]});
    assert.writeOK(coll.insert({_id: "fullDocument is default"}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "insert");
    assert.eq(latestChange.fullDocument, {_id: "fullDocument is default"});

    // Test that specifying 'fullDocument' as 'default' does include a 'fullDocument' in the
    // result for a replacement-style update.
    assert.writeOK(coll.update({_id: "fullDocument is default"},
                               {_id: "fullDocument is default", replaced: true}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "replace");
    assert.eq(latestChange.fullDocument, {_id: "fullDocument is default", replaced: true});

    // Test that specifying 'fullDocument' as 'default' does not include a 'fullDocument' in the
    // result for a non-replacement update.
    assert.writeOK(coll.update({_id: "fullDocument is default"}, {$set: {updated: true}}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "update");
    assert(!latestChange.hasOwnProperty("fullDocument"));

    jsTestLog("Testing change streams with 'fullDocument' specified as 'updateLookup'");

    // Test that specifying 'fullDocument' as 'updateLookup' does include a 'fullDocument' in
    // the result for an insert.
    cursor = cst.startWatchingChanges(
        {collection: coll, pipeline: [{$changeStream: {fullDocument: "updateLookup"}}]});
    assert.writeOK(coll.insert({_id: "fullDocument is lookup"}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "insert");
    assert.eq(latestChange.fullDocument, {_id: "fullDocument is lookup"});

    // Test that specifying 'fullDocument' as 'updateLookup' does include a 'fullDocument' in
    // the result for a replacement-style update.
    assert.writeOK(coll.update({_id: "fullDocument is lookup"},
                               {_id: "fullDocument is lookup", replaced: true}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "replace");
    assert.eq(latestChange.fullDocument, {_id: "fullDocument is lookup", replaced: true});

    // Test that specifying 'fullDocument' as 'updateLookup' does include a 'fullDocument' in
    // the result for a non-replacement update.
    assert.writeOK(coll.update({_id: "fullDocument is lookup"}, {$set: {updated: true}}));
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "update");
    assert.eq(latestChange.fullDocument,
              {_id: "fullDocument is lookup", replaced: true, updated: true});

    // Test that looking up the post image of an update after deleting the document will result
    // in a 'fullDocument' with a value of null.
    cursor = cst.startWatchingChanges({
        collection: coll,
        pipeline: [
            {$changeStream: {fullDocument: "updateLookup"}},
            {$match: {operationType: "update"}}
        ]
    });
    assert.writeOK(coll.update({_id: "fullDocument is lookup"}, {$set: {updatedAgain: true}}));
    assert.writeOK(coll.remove({_id: "fullDocument is lookup"}));
    // If this test is running with secondary read preference, it's necessary for the remove
    // to propagate to all secondary nodes and be available for majority reads before we can
    // assume looking up the document will fail.
    FixtureHelpers.awaitLastOpCommitted(db);

    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "update");
    assert(latestChange.hasOwnProperty("fullDocument"));
    assert.eq(latestChange.fullDocument, null);
    const deleteDocResumePoint = latestChange._id;

    // Test that looking up the post image of an update after the collection has been dropped
    // will result in 'fullDocument' with a value of null.  This must be done using getMore
    // because new cursors cannot be established after a collection drop.
    assert.writeOK(coll.insert({_id: "fullDocument is lookup 2"}));
    assert.writeOK(coll.update({_id: "fullDocument is lookup 2"}, {$set: {updated: true}}));

    // Open a $changeStream cursor with batchSize 0, so that no oplog entries are retrieved yet.
    cursor = cst.startWatchingChanges({
        collection: coll,
        pipeline: [
            {$changeStream: {fullDocument: "updateLookup", resumeAfter: deleteDocResumePoint}},
            {$match: {operationType: {$ne: "delete"}}}
        ],
        aggregateOptions: {cursor: {batchSize: 0}}
    });

    // Save another stream to test post-image lookup after the collection is recreated.
    const cursorBeforeDrop = cst.startWatchingChanges({
        collection: coll,
        pipeline: [
            {$changeStream: {fullDocument: "updateLookup", resumeAfter: deleteDocResumePoint}},
            {$match: {operationType: {$ne: "delete"}}}
        ],
        aggregateOptions: {cursor: {batchSize: 0}}
    });

    // Retrieve the 'insert' operation from the latter stream. This is necessary on a sharded
    // collection so that the documentKey is retrieved before the collection is recreated;
    // otherwise, per SERVER-31691, a uassert will occur.
    latestChange = cst.getOneChange(cursorBeforeDrop);
    assert.eq(latestChange.operationType, "insert");
    assert(latestChange.hasOwnProperty("fullDocument"));
    assert.eq(latestChange.fullDocument, {_id: "fullDocument is lookup 2"});

    // Drop the collection and wait until two-phase drop finishes.
    assertDropCollection(db, coll.getName());
    assert.soon(function() {
        return !TwoPhaseDropCollectionTest.collectionIsPendingDropInDatabase(db, coll.getName());
    });
    // If this test is running with secondary read preference, it's necessary for the drop
    // to propagate to all secondary nodes and be available for majority reads before we can
    // assume looking up the document will fail.
    FixtureHelpers.awaitLastOpCommitted(db);

    // Check the next $changeStream entry; this is the test document inserted above.
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "insert");
    assert(latestChange.hasOwnProperty("fullDocument"));
    assert.eq(latestChange.fullDocument, {_id: "fullDocument is lookup 2"});

    // The next entry is the 'update' operation. Because the collection has been dropped, our
    // attempt to look up the post-image results in a null document.
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "update");
    assert(latestChange.hasOwnProperty("fullDocument"));
    assert.eq(latestChange.fullDocument, null);

    // Test that we can resume a change stream with 'fullDocument: updateLookup' after the
    // collection has been dropped. This is only allowed if an explicit collation is provided.
    cursor = cst.startWatchingChanges({
        collection: coll,
        pipeline: [
            {$changeStream: {resumeAfter: deleteDocResumePoint, fullDocument: "updateLookup"}},
            {$match: {operationType: {$ne: "delete"}}}
        ],
        aggregateOptions: {cursor: {batchSize: 0}, collation: {locale: "simple"}}
    });

    // Check the next $changeStream entry; this is the test document inserted above.
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "insert");
    assert(latestChange.hasOwnProperty("fullDocument"));
    assert.eq(latestChange.fullDocument, {_id: "fullDocument is lookup 2"});

    // The next entry is the 'update' operation. Because the collection has been dropped, our
    // attempt to look up the post-image results in a null document.
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "update");
    assert(latestChange.hasOwnProperty("fullDocument"));
    assert.eq(latestChange.fullDocument, null);

    // Test that looking up the post image of an update after the collection has been dropped
    // and created again will result in 'fullDocument' with a value of null. This must be done
    // using getMore because new cursors cannot be established after a collection drop.

    // Insert a document with the same _id, verify the change stream won't return it due to
    // different UUID.
    assertCreateCollection(db, coll.getName());
    assert.writeOK(coll.insert({_id: "fullDocument is lookup 2"}));

    // Confirm that the next entry's post-image is null since new collection has a different
    // UUID.
    latestChange = cst.getOneChange(cursorBeforeDrop);
    assert.eq(latestChange.operationType, "update");
    assert(latestChange.hasOwnProperty("fullDocument"));
    assert.eq(latestChange.fullDocument, null);

    jsTestLog("Testing full document lookup with a real getMore");
    assert.writeOK(coll.insert({_id: "getMoreEnabled"}));

    cursor = cst.startWatchingChanges({
        collection: coll,
        pipeline: [{$changeStream: {fullDocument: "updateLookup"}}],
    });
    assert.writeOK(coll.update({_id: "getMoreEnabled"}, {$set: {updated: true}}));

    const doc = cst.getOneChange(cursor);
    assert.docEq(doc["fullDocument"], {_id: "getMoreEnabled", updated: true});

    // Test that invalidate entries don't have 'fullDocument' even if 'updateLookup' is
    // specified.
    cursor = cst.startWatchingChanges({
        collection: coll,
        pipeline: [{$changeStream: {fullDocument: "updateLookup"}}],
        aggregateOptions: {cursor: {batchSize: 0}}
    });
    assert.writeOK(coll.insert({_id: "testing invalidate"}));
    assertDropCollection(db, coll.getName());
    // Wait until two-phase drop finishes.
    assert.soon(function() {
        return !TwoPhaseDropCollectionTest.collectionIsPendingDropInDatabase(db, coll.getName());
    });
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "insert");
    latestChange = cst.getOneChange(cursor);
    assert.eq(latestChange.operationType, "drop");
    // Only single-collection change streams will be invalidated by the drop.
    if (!isChangeStreamPassthrough()) {
        latestChange = cst.getOneChange(cursor, true);
        assert.eq(latestChange.operationType, "invalidate");
    }
}());