jstests/change_streams/oplog_rewrite/change_stream_match_pushdown_operation_type_rewrite.js
// explain used to verify the oplog-level rewrites to fail.
explain used to verify the oplog-level rewrites to fail.
explain used to verify the oplog - level rewrites to fail .
function verifyNonInvalidatingOps(
    resumeAfterToken, userMatchExpr, expectedOps, expectedOplogRetDocsForEachShard) {
    const cursor = coll.aggregate([
        {$changeStream: {resumeAfter: resumeAfterToken, showExpandedEvents: true}},
        userMatchExpr
    ]);
    const docIds = [1, 2];
    for (const op of expectedOps) {
        docIds.forEach(docId => {
            assert.soon(() => cursor.hasNext(), op);
            const event = cursor.next();
            assert.eq(event.operationType, op, event);
            if (op == "insert" || op == "replace" || op == "update" || op == "delete") {
                assert.eq(event.documentKey._id, docId, event);
            }
        });
    }
    assert(!cursor.hasNext(), () => tojson(cursor.next()));
    const stats = coll.explain("executionStats").aggregate([
        {$changeStream: {resumeAfter: resumeAfterToken, showExpandedEvents: true}},
        userMatchExpr
    ]);
    assertNumChangeStreamDocsReturnedFromShard(stats, st.rs0.name, expectedOps.length);
    assertNumChangeStreamDocsReturnedFromShard(stats, st.rs1.name, expectedOps.length);
    assertNumMatchingOplogEventsForShard(stats, st.rs0.name, expectedOplogRetDocsForEachShard);
    assertNumMatchingOplogEventsForShard(stats, st.rs1.name, expectedOplogRetDocsForEachShard);
}
function verifyNonInvalidatingOps ( resumeAfterToken , userMatchExpr , expectedOps , expectedOplogRetDocsForEachShard ) { const cursor = coll . aggregate ( [ { $changeStream : { resumeAfter : resumeAfterToken , showExpandedEvents : true } } , userMatchExpr ] ) ; const docIds = [ 1 , 2 ] ; for ( const op of expectedOps ) { docIds . forEach ( docId => { assert . soon ( ( ) => cursor . hasNext ( ) , op ) ; const event = cursor . next ( ) ; assert . eq ( event . operationType , op , event ) ; if ( op == "insert" || op == "replace" || op == "update" || op == "delete" ) { assert . eq ( event . documentKey . _id , docId , event ) ; } } ) ; } assert ( ! cursor . hasNext ( ) , ( ) => tojson ( cursor . next ( ) ) ) ; const stats = coll . explain ( "executionStats" ) . aggregate ( [ { $changeStream : { resumeAfter : resumeAfterToken , showExpandedEvents : true } } , userMatchExpr ] ) ; assertNumChangeStreamDocsReturnedFromShard ( stats , st . rs0 . name , expectedOps . length ) ; assertNumChangeStreamDocsReturnedFromShard ( stats , st . rs1 . name , expectedOps . length ) ; assertNumMatchingOplogEventsForShard ( stats , st . rs0 . name , expectedOplogRetDocsForEachShard ) ; assertNumMatchingOplogEventsForShard ( stats , st . rs1 . name , expectedOplogRetDocsForEachShard ) ; }
explain used to verify the oplog-level rewrites to fail.
explain used to verify the oplog - level rewrites to fail .
// For shard1, document id is '1' and for shard2, document id is '2'. // An 'executionStats' could only be captured for a non-invalidating stream.