mongodb/jstests/sharding/change_streams_unsharded_becomes_sharded.js

// Tests the behavior of change streams on a collection that was initially unsharded but then
// becomes sharded. In particular, test that post-shardCollection inserts update their cached
// 'documentKey' to include the new shard key, and that a resume token obtained prior to the
// shardCollection command can be used to resume the stream even after the collection has been
// sharded.
// @tags: [uses_change_streams]
(function() {
    "use strict";

    load('jstests/libs/change_stream_util.js');  // For ChangeStreamTest.

    // For supportsMajorityReadConcern().
    load("jstests/multiVersion/libs/causal_consistency_helpers.js");

    if (!supportsMajorityReadConcern()) {
        jsTestLog("Skipping test since storage engine doesn't support majority read concern.");
        return;
    }

    const testName = "change_streams_unsharded_becomes_sharded";
    const st = new ShardingTest({
        shards: 2,
        mongos: 1,
        rs: {
            nodes: 1,
            enableMajorityReadConcern: '',
            // Use a higher frequency for periodic noops to speed up the test.
            setParameter: {periodicNoopIntervalSecs: 1, writePeriodicNoops: true}
        }
    });

    const mongosDB = st.s0.getDB("test");
    const mongosColl = mongosDB[testName];

    function testUnshardedBecomesSharded(collToWatch) {
        mongosColl.drop();
        mongosDB.createCollection(testName);
        mongosColl.createIndex({x: 1});

        st.ensurePrimaryShard(mongosDB.getName(), st.rs0.getURL());

        // Establish a change stream cursor on the unsharded collection.
        const cst = new ChangeStreamTest(mongosDB);

        // Create a different collection in the same database, and verify that it doesn't affect the
        // results of the change stream.
        const mongosCollOther = mongosDB[testName + "other"];
        mongosCollOther.drop();
        mongosDB.createCollection(testName + "other");
        mongosCollOther.createIndex({y: 1});

        let cursor = cst.startWatchingChanges({
            pipeline: [{$changeStream: {}}, {$match: {"ns.coll": mongosColl.getName()}}],
            collection: collToWatch
        });
        assert.eq(0, cursor.firstBatch.length, "Cursor had changes: " + tojson(cursor));

        // Verify that the cursor picks up documents inserted while the collection is unsharded. The
        // 'documentKey' at this point is simply the _id field.
        assert.writeOK(mongosColl.insert({_id: 0, x: 0}));
        assert.writeOK(mongosCollOther.insert({_id: 0, y: 0}));
        const[preShardCollectionChange] = cst.assertNextChangesEqual({
            cursor: cursor,
            expectedChanges: [{
                documentKey: {_id: 0},
                fullDocument: {_id: 0, x: 0},
                ns: {db: mongosDB.getName(), coll: mongosColl.getName()},
                operationType: "insert",
            }]
        });

        // Record the resume token for this change, before the collection is sharded.
        const preShardCollectionResumeToken = preShardCollectionChange._id;

        // Shard the test collection with shard key {x: 1} and split into 2 chunks.
        st.shardColl(mongosColl.getName(), {x: 1}, {x: 0}, false, mongosDB.getName());

        // Shard the other collection with shard key {y: 1} and split into 2 chunks.
        st.shardColl(mongosCollOther.getName(), {y: 1}, {y: 0}, false, mongosDB.getName());

        // List the changes we expect to see for the next two operations on the sharded collection.
        // Later, we will resume the stream using the token generated before the collection was
        // sharded, and will need to confirm that we can still see these two changes.
        const postShardCollectionChanges = [
            {
              documentKey: {x: 1, _id: 1},
              fullDocument: {_id: 1, x: 1},
              ns: {db: mongosDB.getName(), coll: mongosColl.getName()},
              operationType: "insert",
            },
            {
              documentKey: {x: -1, _id: -1},
              fullDocument: {_id: -1, x: -1},
              ns: {db: mongosDB.getName(), coll: mongosColl.getName()},
              operationType: "insert",
            }
        ];

        // Verify that the cursor on the original shard is still valid and sees new inserted
        // documents. The 'documentKey' field should now include the shard key, even before a
        // 'kNewShardDetected' operation has been generated by the migration of a chunk to a new
        // shard.
        assert.writeOK(mongosColl.insert({_id: 1, x: 1}));
        assert.writeOK(mongosCollOther.insert({_id: 1, y: 1}));
        cst.assertNextChangesEqual(
            {cursor: cursor, expectedChanges: [postShardCollectionChanges[0]]});

        // Move the [minKey, 0) chunk to shard1.
        assert.commandWorked(mongosDB.adminCommand({
            moveChunk: mongosColl.getFullName(),
            find: {x: -1},
            to: st.rs1.getURL(),
            _waitForDelete: true
        }));
        assert.commandWorked(mongosDB.adminCommand({
            moveChunk: mongosCollOther.getFullName(),
            find: {y: -1},
            to: st.rs1.getURL(),
            _waitForDelete: true
        }));

        // Make sure the change stream cursor sees a document inserted on the recipient shard.
        assert.writeOK(mongosColl.insert({_id: -1, x: -1}));
        assert.writeOK(mongosCollOther.insert({_id: -1, y: -1}));
        cst.assertNextChangesEqual(
            {cursor: cursor, expectedChanges: [postShardCollectionChanges[1]]});

        // Confirm that we can resume the stream on the sharded collection using the token generated
        // while the collection was unsharded, whose documentKey contains the _id field but not the
        // shard key.
        let resumedCursor = cst.startWatchingChanges({
            pipeline: [{$changeStream: {resumeAfter: preShardCollectionResumeToken}}],
            collection: mongosColl
        });

        // Verify that we see both of the insertions which occurred after the collection was
        // sharded.
        cst.assertNextChangesEqual(
            {cursor: resumedCursor, expectedChanges: postShardCollectionChanges});

        // Test the behavior of a change stream when a sharded collection is dropped and recreated.
        cursor = cst.startWatchingChanges({
            pipeline: [{$changeStream: {}}, {$match: {"ns.coll": mongosColl.getName()}}],
            collection: collToWatch
        });
        assert.eq(0, cursor.firstBatch.length, "Cursor had changes: " + tojson(cursor));

        // Insert a couple documents to shard1, creating a scenario where the getMore to shard0 will
        // indicate that the change stream is invalidated yet shard1 will still have data to return.
        assert.writeOK(mongosColl.insert({_id: -2, x: -2}));
        assert.writeOK(mongosColl.insert({_id: -3, x: -3}));

        // Drop and recreate the collection.
        mongosColl.drop();
        mongosDB.createCollection(mongosColl.getName());
        mongosColl.createIndex({z: 1});

        // Shard the collection on a different shard key and ensure that each shard has a chunk.
        st.shardColl(mongosColl.getName(), {z: 1}, {z: 0}, {z: -1}, mongosDB.getName());

        assert.writeOK(mongosColl.insert({_id: -1, z: -1}));
        assert.writeOK(mongosColl.insert({_id: 1, z: 1}));

        // Verify that the change stream picks up the inserts, however the shard key is missing
        // since the collection has since been dropped and recreated.
        cst.assertNextChangesEqual({
            cursor: cursor,
            expectedChanges: [
                {
                  documentKey: {_id: -2},
                  fullDocument: {_id: -2, x: -2},
                  ns: {db: mongosDB.getName(), coll: mongosColl.getName()},
                  operationType: "insert",
                },
                {
                  documentKey: {_id: -3},
                  fullDocument: {_id: -3, x: -3},
                  ns: {db: mongosDB.getName(), coll: mongosColl.getName()},
                  operationType: "insert",
                }
            ]
        });

        cst.cleanUp();
    }

    // First test against a change stream on a single collection.
    testUnshardedBecomesSharded(mongosColl.getName());

    // Test against a change stream on the entire database.
    testUnshardedBecomesSharded(1);

    st.stop();
})();