jstests/change_streams/oplog_rewrite/change_stream_match_pushdown_documentKey_rewrite.js
//     as specified in 'expectedOplogCursorReturnedDocs'.
as specified in 'expectedOplogCursorReturnedDocs'.
as specified in ' expectedOplogCursorReturnedDocs ' .
function verifyOps(resumeAfterToken, userMatchExpr, expectedOps, expectedOplogCursorReturnedDocs) {
    const cursor =
        coll.aggregate([{$changeStream: {resumeAfter: resumeAfterToken}}, userMatchExpr]);
    let expectedChangeStreamDocsReturned = [0, 0];
    for (const [op, id, shardId] of expectedOps) {
        assert.soon(() => cursor.hasNext());
        const event = cursor.next();
        assert.eq(event.operationType, op, event);
        assert.eq(event.documentKey._id, id, event);
        assert.eq(event.documentKey.shard, shardId, event);
        if (shardId == 0 || shardId == 1) {
            ++expectedChangeStreamDocsReturned[shardId];
        }
    }
    assert(!cursor.hasNext());
    const stats = coll.explain("executionStats")
                      .aggregate([{$changeStream: {resumeAfter: resumeAfterToken}}, userMatchExpr]);
    assertNumChangeStreamDocsReturnedFromShard(
        stats, st.rs0.name, expectedChangeStreamDocsReturned[0]);
    assertNumChangeStreamDocsReturnedFromShard(
        stats, st.rs1.name, expectedChangeStreamDocsReturned[1]);
    assertNumMatchingOplogEventsForShard(stats, st.rs0.name, expectedOplogCursorReturnedDocs[0]);
    assertNumMatchingOplogEventsForShard(stats, st.rs1.name, expectedOplogCursorReturnedDocs[1]);
}
function verifyOps ( resumeAfterToken , userMatchExpr , expectedOps , expectedOplogCursorReturnedDocs ) { const cursor = coll . aggregate ( [ { $changeStream : { resumeAfter : resumeAfterToken } } , userMatchExpr ] ) ; let expectedChangeStreamDocsReturned = [ 0 , 0 ] ; for ( const [ op , id , shardId ] of expectedOps ) { assert . soon ( ( ) => cursor . hasNext ( ) ) ; const event = cursor . next ( ) ; assert . eq ( event . operationType , op , event ) ; assert . eq ( event . documentKey . _id , id , event ) ; assert . eq ( event . documentKey . shard , shardId , event ) ; if ( shardId == 0 || shardId == 1 ) { ++ expectedChangeStreamDocsReturned [ shardId ] ; } } assert ( ! cursor . hasNext ( ) ) ; const stats = coll . explain ( "executionStats" ) . aggregate ( [ { $changeStream : { resumeAfter : resumeAfterToken } } , userMatchExpr ] ) ; assertNumChangeStreamDocsReturnedFromShard ( stats , st . rs0 . name , expectedChangeStreamDocsReturned [ 0 ] ) ; assertNumChangeStreamDocsReturnedFromShard ( stats , st . rs1 . name , expectedChangeStreamDocsReturned [ 1 ] ) ; assertNumMatchingOplogEventsForShard ( stats , st . rs0 . name , expectedOplogCursorReturnedDocs [ 0 ] ) ; assertNumMatchingOplogEventsForShard ( stats , st . rs1 . name , expectedOplogCursorReturnedDocs [ 1 ] ) ; }
as specified in 'expectedOplogCursorReturnedDocs'.
as specified in ' expectedOplogCursorReturnedDocs ' .
// An 'executionStats' could only be captured for a non-invalidating stream.