jstests/change_streams/oplog_rewrite/change_stream_match_pushdown_to_rewrite.js
// 2. the filtering is been done at oplog level
2. the filtering is been done at oplog level
2 . the filtering is been done at oplog level
function verifyOnWholeCluster(
    resumeAfterToken, userMatchExpr, expectedResult, expectedOplogRetDocsForEachShard) {
    const cursor = db.getSiblingDB("admin").aggregate([
        {$changeStream: {resumeAfter: resumeAfterToken, allChangesForCluster: true}},
        userMatchExpr
    ]);
    for (const [coll, opDict] of Object.entries(expectedResult)) {
        for (const [op, eventIdentifierList] of Object.entries(opDict)) {
            eventIdentifierList.forEach(eventIdentifier => {
                assert.soon(() => cursor.hasNext());
                const event = cursor.next();
                assert.eq(event.operationType, op, event);
                if (op == "insert") {
                    assert.eq(event.documentKey._id, eventIdentifier, event);
                } else if (op == "rename") {
                    assert.eq(event.to.coll, eventIdentifier, event);
                } else {
                    assert(false, event);
                }
                assert.eq(event.ns.coll, coll);
            });
        }
    }
    assert(!cursor.hasNext());
    const stats = db.getSiblingDB("admin").runCommand({
        explain: {
            aggregate: 1,
            pipeline: [
                {$changeStream: {resumeAfter: resumeAfterToken, allChangesForCluster: true}},
                userMatchExpr
            ],
            cursor: {batchSize: 0}
        },
        verbosity: "executionStats"
    });
    assertNumMatchingOplogEventsForShard(stats, st.rs0.name, expectedOplogRetDocsForEachShard);
    assertNumMatchingOplogEventsForShard(stats, st.rs1.name, expectedOplogRetDocsForEachShard);
}
function verifyOnWholeCluster ( resumeAfterToken , userMatchExpr , expectedResult , expectedOplogRetDocsForEachShard ) { const cursor = db . getSiblingDB ( "admin" ) . aggregate ( [ { $changeStream : { resumeAfter : resumeAfterToken , allChangesForCluster : true } } , userMatchExpr ] ) ; for ( const [ coll , opDict ] of Object . entries ( expectedResult ) ) { for ( const [ op , eventIdentifierList ] of Object . entries ( opDict ) ) { eventIdentifierList . forEach ( eventIdentifier => { assert . soon ( ( ) => cursor . hasNext ( ) ) ; const event = cursor . next ( ) ; assert . eq ( event . operationType , op , event ) ; if ( op == "insert" ) { assert . eq ( event . documentKey . _id , eventIdentifier , event ) ; } else if ( op == "rename" ) { assert . eq ( event . to . coll , eventIdentifier , event ) ; } else { assert ( false , event ) ; } assert . eq ( event . ns . coll , coll ) ; } ) ; } } assert ( ! cursor . hasNext ( ) ) ; const stats = db . getSiblingDB ( "admin" ) . runCommand ( { explain : { aggregate : 1 , pipeline : [ { $changeStream : { resumeAfter : resumeAfterToken , allChangesForCluster : true } } , userMatchExpr ] , cursor : { batchSize : 0 } } , verbosity : "executionStats" } ) ; assertNumMatchingOplogEventsForShard ( stats , st . rs0 . name , expectedOplogRetDocsForEachShard ) ; assertNumMatchingOplogEventsForShard ( stats , st . rs1 . name , expectedOplogRetDocsForEachShard ) ; }
2. the filtering is been done at oplog level
2 . the filtering is been done at oplog level