jstests/libs/change_stream_rewrite_util.js
// a change stream may consume. Assumes that the specified collection does not already exist.
a change stream may consume. Assumes that the specified collection does not already exist.
a change stream may consume . Assumes that the specified collection does not already exist .
function generateChangeStreamWriteWorkload(db, collName, numDocs, includInvalidatingEvents = true) {
    if (FixtureHelpers.isMongos(db)) {
        assert.commandWorked(db.adminCommand({enableSharding: db.getName()}));
        assert.commandWorked(db.adminCommand(
            {shardCollection: `${db.getName()}.${collName}`, key: {shardKey: "hashed"}}));
    }
    let testColl = assertCreateCollection(db, collName);
    assert.commandWorked(testColl.createIndex({a: 1}));
    assert.commandWorked(db.runCommand({
        collMod: testColl.getName(),
        index: {keyPattern: {a: 1}, hidden: true, expireAfterSeconds: 500}
    }));
    assert.commandWorked(testColl.dropIndex({a: 1}));
    assert.commandWorked(testColl.runCommand({
        collMod: collName,
        validator: {},
        validationLevel: "off",
        validationAction: "warn",
    }));
    assert.commandWorked(testColl.runCommand({
        collMod: collName,
        validator: {},
        validationLevel: "strict",
        validationAction: "error",
    }));
    for (let i = 0; i < numDocs; ++i) {
        assert.commandWorked(testColl.insert(
            {_id: i, shardKey: i, a: [1, [2], {b: 3}], f1: {subField: true}, f2: false}));
    }
    const updateSpecs = [
        [{$set: {f2: true}}],                                
        [{$unset: ["f1"]}],                                  
        [{$set: {a: [1, [2]]}}],                             
        [{$set: {a: [1, [2]], f2: true}}, {$unset: ["f1"]}]  
    ];
    for (let i = 0; i < numDocs / 2; ++i) {
        assert.commandWorked(
            testColl.update({_id: i, shardKey: i}, updateSpecs[(i % updateSpecs.length)]));
    }
    for (let i = numDocs / 2; i < numDocs; ++i) {
        assert.commandWorked(testColl.replaceOne({_id: i, shardKey: i}, {_id: i, shardKey: i}));
    }
    for (let i = 0; i < numDocs / 4; ++i) {
        assert.commandWorked(testColl.remove({_id: i, shardKey: i}));
    }
    assert.commandWorked(db.createView("view", collName, []));
    assert.commandWorked(db.runCommand({collMod: "view", viewOn: "viewOnView", pipeline: []}));
    assertDropCollection(db, "view");
    if (includInvalidatingEvents) {
        const collNameAfterRename = `${testColl.getName()}_renamed`;
        assert.commandWorked(testColl.renameCollection(collNameAfterRename));
        testColl = db[collNameAfterRename];
        assert.commandWorked(testColl.renameCollection(collName));
        testColl = db[collName];
        assert(testColl.drop());
        assert.commandWorked(db.dropDatabase());
    }
    return testColl;
}
function generateChangeStreamWriteWorkload ( db , collName , numDocs , includInvalidatingEvents = true ) { if ( FixtureHelpers . isMongos ( db ) ) { assert . commandWorked ( db . adminCommand ( { enableSharding : db . getName ( ) } ) ) ; assert . commandWorked ( db . adminCommand ( { shardCollection : ` ${ db . getName ( ) } ${ collName } ` , key : { shardKey : "hashed" } } ) ) ; } let testColl = assertCreateCollection ( db , collName ) ; assert . commandWorked ( testColl . createIndex ( { a : 1 } ) ) ; assert . commandWorked ( db . runCommand ( { collMod : testColl . getName ( ) , index : { keyPattern : { a : 1 } , hidden : true , expireAfterSeconds : 500 } } ) ) ; assert . commandWorked ( testColl . dropIndex ( { a : 1 } ) ) ; assert . commandWorked ( testColl . runCommand ( { collMod : collName , validator : { } , validationLevel : "off" , validationAction : "warn" , } ) ) ; assert . commandWorked ( testColl . runCommand ( { collMod : collName , validator : { } , validationLevel : "strict" , validationAction : "error" , } ) ) ; for ( let i = 0 ; i < numDocs ; ++ i ) { assert . commandWorked ( testColl . insert ( { _id : i , shardKey : i , a : [ 1 , [ 2 ] , { b : 3 } ] , f1 : { subField : true } , f2 : false } ) ) ; } const updateSpecs = [ [ { $set : { f2 : true } } ] , [ { $unset : [ "f1" ] } ] , [ { $set : { a : [ 1 , [ 2 ] ] } } ] , [ { $set : { a : [ 1 , [ 2 ] ] , f2 : true } } , { $unset : [ "f1" ] } ] ] ; for ( let i = 0 ; i < numDocs / 2 ; ++ i ) { assert . commandWorked ( testColl . update ( { _id : i , shardKey : i } , updateSpecs [ ( i % updateSpecs . length ) ] ) ) ; } for ( let i = numDocs / 2 ; i < numDocs ; ++ i ) { assert . commandWorked ( testColl . replaceOne ( { _id : i , shardKey : i } , { _id : i , shardKey : i } ) ) ; } for ( let i = 0 ; i < numDocs / 4 ; ++ i ) { assert . commandWorked ( testColl . remove ( { _id : i , shardKey : i } ) ) ; } assert . commandWorked ( db . createView ( "view" , collName , [ ] ) ) ; assert . commandWorked ( db . runCommand ( { collMod : "view" , viewOn : "viewOnView" , pipeline : [ ] } ) ) ; assertDropCollection ( db , "view" ) ; if ( includInvalidatingEvents ) { const collNameAfterRename = ` ${ testColl . getName ( ) } ` ; assert . commandWorked ( testColl . renameCollection ( collNameAfterRename ) ) ; testColl = db [ collNameAfterRename ] ; assert . commandWorked ( testColl . renameCollection ( collName ) ) ; testColl = db [ collName ] ; assert ( testColl . drop ( ) ) ; assert . commandWorked ( db . dropDatabase ( ) ) ; } return testColl ; }
a change stream may consume.
a change stream may consume .
// If this is a sharded passthrough, make sure we shard on something other than _id so that a // non-id field appears in the documentKey. This will generate 'create' and 'shardCollection'. // If the collection hasn't already been created, do so here. // Build an index, collMod it, then drop it. // Modify the collection's validation options. // Change the validation options back. // Insert some documents. // Update half of them. We generate these updates individually so that they generate different // values for the 'updatedFields', 'removedFields' and 'truncatedArrays' subfields. // only populates 'updatedFields' // only populates 'removedFields' // only populates 'truncatedArrays' // populates all fields // Replace the other half. // Delete half of the updated documents. // Create, modify, and drop a view on the collection. // If the caller is prepared to handle potential invalidations, include the following events. // Rename the collection. // Rename it back. // Drop the collection. // Drop the database.