SERVER-50769 Change streams no longer balk at empty applyOps

(cherry picked from commit e9122ba)

Author: jseyster

buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml

@@ -53,6 +53,7 @@ selector:
   - jstests/sharding/agg_explain_fmt.js
   - jstests/sharding/change_stream_metadata_notifications.js
   - jstests/sharding/change_stream_transaction_sharded.js
+  - jstests/sharding/change_stream_empty_apply_ops.js
   - jstests/sharding/change_streams.js
   - jstests/sharding/collation_lookup.js
   - jstests/sharding/collation_targeting.js

etc/backports_required_for_multiversion_tests.yml

@@ -88,6 +88,8 @@ all:
     test_file: jstests/replsets/read_operations_during_step_down.js
   - ticket: SERVER-50417
     test_file: jstests/replsets/read_operations_during_step_up.js
+  - ticket: SERVER-50769
+    test_file: jstests/sharding/change_stream_empty_apply_ops.js
 
 # Tests that should only be excluded from particular suites should be listed under that suite.
 suites:

jstests/sharding/change_stream_empty_apply_ops.js

@@ -0,0 +1,75 @@
+// Confirms that change streams correctly handle prepared transactions with an empty applyOps entry.
+// This test creats a multi-shard transaction in which one of the participating shards has only a
+// no-op write, resulting in the empty applyOps scenario we wish to test. Exercises the fix for
+// SERVER-50769.
+// @tags: [
+//   requires_sharding,
+//   uses_change_streams,
+//   uses_multi_shard_transaction,
+//   uses_transactions,
+// ]
+(function() {
+"use strict";
+
+const dbName = "test";
+const collName = "change_stream_empty_apply_ops";
+const namespace = dbName + "." + collName;
+
+const st = new ShardingTest({
+    shards: 2,
+    rs: {nodes: 1, setParameter: {writePeriodicNoops: true, periodicNoopIntervalSecs: 1}}
+});
+
+const mongosConn = st.s;
+const db = mongosConn.getDB(dbName);
+const coll = db.getCollection(collName);
+
+assert.commandWorked(coll.createIndex({shard: 1}));
+st.ensurePrimaryShard(dbName, st.shard0.shardName);
+// Shard the test collection and split it into two chunks: one that contains all {shard: 1}
+// documents and one that contains all {shard: 2} documents.
+st.shardColl(collName,
+             {shard: 1} /* shard key */,
+             {shard: 2} /* split at */,
+             {shard: 2} /* move the chunk containing {shard: 2} to its own shard */,
+             dbName,
+             true);
+// Seed each chunk with an initial document.
+assert.commandWorked(coll.insert({shard: 1}, {writeConcern: {w: "majority"}}));
+assert.commandWorked(coll.insert({shard: 2}, {writeConcern: {w: "majority"}}));
+
+// Open up change streams.
+const changeStreamCursorColl = coll.watch();
+const changeStreamCursorDB = db.watch();
+const changeStreamCursorCluster = mongosConn.watch();
+
+// Start a transaction, which will include both shards.
+const sesion = db.getMongo().startSession({causalConsistency: true});
+const sessionDb = sesion.getDatabase(dbName);
+const sessionColl = sessionDb[collName];
+
+sesion.startTransaction({readConcern: {level: "majority"}});
+
+// This no-op will make one of the shards a transaction participant without generating an actual
+// write. The transaction will send an empty prepared transaction to the shard, in the form of an
+// applyOps command with no operations.
+sessionColl.findAndModify({query: {shard: 1}, update: {$setOnInsert: {a: 1}}});
+
+// This write, which is not a no-op, occurs on the other shard.
+sessionColl.findAndModify({query: {shard: 2}, update: {$set: {a: 1}}});
+
+assert.commandWorked(sesion.commitTransaction_forTesting());
+
+// Each change stream should see exactly one update, resulting from the valid write on shard 2.
+[changeStreamCursorColl, changeStreamCursorDB, changeStreamCursorCluster].forEach(function(
+    changeStreamCursor) {
+    assert.soon(() => changeStreamCursor.hasNext());
+    const changeDoc = changeStreamCursor.next();
+    assert.eq(changeDoc.documentKey.shard, 2);
+    assert.eq(changeDoc.operationType, "update");
+
+    assert(!changeStreamCursor.hasNext());
+});
+
+st.stop();
+})();

src/mongo/db/pipeline/document_source_change_stream.cpp

@@ -177,8 +177,10 @@ BSONObj getTxnApplyOpsFilter(BSONElement nsMatch, const NamespaceString& nss) {
     BSONObjBuilder applyOpsBuilder;
     applyOpsBuilder.append("op", "c");
 
-    // "o.applyOps" must be an array with at least one element
-    applyOpsBuilder.append("o.applyOps.0", BSON("$exists" << true));
+    // "o.applyOps" stores the list of operations, so it must be an array.
+    applyOpsBuilder.append("o.applyOps",
+                           BSON("$type"
+                                << "array"));
     applyOpsBuilder.append("lsid", BSON("$exists" << true));
     applyOpsBuilder.append("txnNumber", BSON("$exists" << true));
     applyOpsBuilder.append("o.prepare", BSON("$not" << BSON("$eq" << true)));