jstests/sharding/query/merge_stale_on_fields.js
// the operation will fail rather than proceed with the old shard key.
the operation will fail rather than proceed with the old shard key.
the operation will fail rather than proceed with the old shard key .
function testEpochChangeDuringAgg({mergeSpec, failpoint, failpointData}) {
    function indexSpecFromOnFields(onFields) {
        let spec = {};
        if (typeof (onFields) == "string") {
            spec[onFields] = 1;
        } else {
            onFields.forEach((field) => {
                spec[field] = 1;
            });
        }
        return spec;
    }
    target.drop();
    if (mergeSpec.hasOwnProperty('on')) {
        assert.commandWorked(
            target.createIndex(indexSpecFromOnFields(mergeSpec.on), {unique: true}));
        assert.commandWorked(st.s0.adminCommand(
            {shardCollection: target.getFullName(), key: indexSpecFromOnFields(mergeSpec.on)}));
    } else {
        assert.commandWorked(
            st.s0.adminCommand({shardCollection: target.getFullName(), key: {sk: 1, _id: 1}}));
    }
    [st.rs0.getPrimary(), st.rs1.getPrimary()].forEach((mongod) => {
        assert.commandWorked(mongod.adminCommand(
            {configureFailPoint: failpoint, mode: "alwaysOn", data: failpointData || {}}));
    });
    let parallelShellJoiner;
    try {
        let parallelCode = `
                const source = db.getSiblingDB("${dbName}").${source.getName()};
                const error = assert.throws(() => source.aggregate([
                    {$addFields: {sk: "$_id"}},
                    {$merge: ${tojsononeline(mergeSpec)}}
                ]));
                assert.eq(error.code, ErrorCodes.StaleEpoch);
            `;
        if (mergeSpec.hasOwnProperty("on")) {
            parallelCode = `
                    const source = db.getSiblingDB("${dbName}").${source.getName()};
                    assert.doesNotThrow(() => source.aggregate([
                        {$addFields: {sk: "$_id"}},
                        {$merge: ${tojsononeline(mergeSpec)}}
                    ]));
                `;
        }
        parallelShellJoiner = startParallelShell(parallelCode, st.s0.port);
        const getAggOps = function() {
            return st.s0.getDB("admin")
                .aggregate([
                    {$currentOp: {}},
                    {$match: {"cursor.originatingCommand.pipeline": {$exists: true}}}
                ])
                .toArray();
        };
        const hasMergeRunning = function() {
            return getAggOps()
                       .filter((op) => {
                           const pipeline = op.cursor.originatingCommand.pipeline;
                           return pipeline.length > 0 &&
                               pipeline[pipeline.length - 1].hasOwnProperty("$merge");
                       })
                       .length >= 1;
        };
        assert.soon(hasMergeRunning, () => tojson(getAggOps()));
        target.drop();
    } finally {
        [st.rs0.getPrimary(), st.rs1.getPrimary()].forEach((mongod) => {
            assert.commandWorked(mongod.adminCommand({configureFailPoint: failpoint, mode: "off"}));
        });
    }
    parallelShellJoiner();
}
function testEpochChangeDuringAgg ( { mergeSpec , failpoint , failpointData } ) { function indexSpecFromOnFields ( onFields ) { let spec = { } ; if ( typeof ( onFields ) == "string" ) { spec [ onFields ] = 1 ; } else { onFields . forEach ( ( field ) => { spec [ field ] = 1 ; } ) ; } return spec ; } target . drop ( ) ; if ( mergeSpec . hasOwnProperty ( 'on' ) ) { assert . commandWorked ( target . createIndex ( indexSpecFromOnFields ( mergeSpec . on ) , { unique : true } ) ) ; assert . commandWorked ( st . s0 . adminCommand ( { shardCollection : target . getFullName ( ) , key : indexSpecFromOnFields ( mergeSpec . on ) } ) ) ; } else { assert . commandWorked ( st . s0 . adminCommand ( { shardCollection : target . getFullName ( ) , key : { sk : 1 , _id : 1 } } ) ) ; } [ st . rs0 . getPrimary ( ) , st . rs1 . getPrimary ( ) ] . forEach ( ( mongod ) => { assert . commandWorked ( mongod . adminCommand ( { configureFailPoint : failpoint , mode : "alwaysOn" , data : failpointData || { } } ) ) ; } ) ; let parallelShellJoiner ; try { let parallelCode = ` ${ dbName } ${ source . getName ( ) } ${ tojsononeline ( mergeSpec ) } ` ; if ( mergeSpec . hasOwnProperty ( "on" ) ) { parallelCode = ` ${ dbName } ${ source . getName ( ) } ${ tojsononeline ( mergeSpec ) } ` ; } parallelShellJoiner = startParallelShell ( parallelCode , st . s0 . port ) ; const getAggOps = function ( ) { return st . s0 . getDB ( "admin" ) . aggregate ( [ { $currentOp : { } } , { $match : { "cursor.originatingCommand.pipeline" : { $exists : true } } } ] ) . toArray ( ) ; } ; const hasMergeRunning = function ( ) { return getAggOps ( ) . filter ( ( op ) => { const pipeline = op . cursor . originatingCommand . pipeline ; return pipeline . length > 0 && pipeline [ pipeline . length - 1 ] . hasOwnProperty ( "$merge" ) ; } ) . length >= 1 ; } ; assert . soon ( hasMergeRunning , ( ) => tojson ( getAggOps ( ) ) ) ; target . drop ( ) ; } finally { [ st . rs0 . getPrimary ( ) , st . rs1 . getPrimary ( ) ] . forEach ( ( mongod ) => { assert . commandWorked ( mongod . adminCommand ( { configureFailPoint : failpoint , mode : "off" } ) ) ; } ) ; } parallelShellJoiner ( ) ; }
the operation will fail rather than proceed with the old shard key.
the operation will fail rather than proceed with the old shard key .
// Converts a single string or an array of strings into it's object spec form. For instance, for // input ["a", "b"] the returned object would be {a: 1, b: 1}. // Drop the collection and reshard it with a different shard key // Use a failpoint to make the query feeding into the aggregate hang while we drop the // collection. // If a user specifies their own "on" fields, we don't need to fail an aggregation if // the collection is dropped and recreated or the epoch otherwise changes. We are // allowed to fail such an operation should we choose to in the future, but for now we // don't expect to because we do not do anything special on mongos to ensure the catalog // cache is up to date, so do not want to attach mongos's believed epoch to the command // for the shards. // Wait for the merging $merge to appear in the currentOp output from the shards. We should // see that the $merge stage has an 'epoch' field serialized from the mongos. // Drop the collection so that the epoch changes while the merge operation is executing