SERVER-65858 Mark the applyOps entries when the operations involves orphaned documents

2 files changed, 223 insertions, 52 deletions

diff --git a/jstests/sharding/change_stream_no_orphans.js b/jstests/sharding/change_stream_no_orphans.js
index e835d75a72f..98d5fd1bcd6 100644
--- a/jstests/sharding/change_stream_no_orphans.js
+++ b/jstests/sharding/change_stream_no_orphans.js
@@ -3,10 +3,11 @@
  * streams and (2) have no effect on the persisted data.
  *
  * The behavior is tested in the following scenarios:
- *   - Direct update/delete (from shard) to orphaned document
- *   - Distributed update/delete (from router) on a single document
- *   - Distributed update/delete (from router) to multi-documents
- *   - Distributed update/delete (from router) via transaction on multi-documents
+ *   - Direct operations to shard on orphaned documents
+ *   - Broadcasted operations (from router) on orphaned documents
+ *   - Transaction from router updating both orphaned and non-orphaned documents
+ *   - Transaction to shard updating both orphaned and non-orphaned documents
+ *   - Batched deletes from router and to shard
  *
  * @tags: [
  *   requires_fcv_53,
@@ -41,12 +42,14 @@ assert.commandWorked(
     st.s.adminCommand({enableSharding: dbName, primaryShard: st.shard0.shardName}));
 assert.commandWorked(st.s.adminCommand({shardCollection: collNS, key: {_id: 1}}));
 const coll = st.s.getCollection(collNS);
-assert.commandWorked(coll.insert({_id: 0, name: 'matt', age: 20}));
-assert.commandWorked(coll.insert({_id: 1, name: 'john', age: 25}));
-assert.commandWorked(coll.insert({_id: 2, name: 'robert', age: 30}));
-assert.commandWorked(coll.insert({_id: 3, name: 'robert', age: 35}));
-assert.commandWorked(coll.insert({_id: 4, name: 'james', age: 40}));
-assert.commandWorked(coll.insert({_id: 5, name: 'james', age: 45}));
+assert.commandWorked(coll.insert({_id: -2, name: 'emma', age: 20}));
+assert.commandWorked(coll.insert({_id: -1, name: 'olivia', age: 25}));
+assert.commandWorked(coll.insert({_id: 0, name: 'matt', age: 30}));
+assert.commandWorked(coll.insert({_id: 1, name: 'john', age: 35}));
+assert.commandWorked(coll.insert({_id: 2, name: 'robert', age: 40}));
+assert.commandWorked(coll.insert({_id: 3, name: 'robert', age: 45}));
+assert.commandWorked(coll.insert({_id: 4, name: 'james', age: 50}));
+assert.commandWorked(coll.insert({_id: 5, name: 'liam', age: 55}));
 
 // Move the chunk to the second shard leaving orphaned documents on the first shard.
 assert.commandWorked(st.s.adminCommand({split: collNS, middle: {_id: 0}}));
@@ -56,10 +59,14 @@ assert.commandWorked(
 // Setup a change stream on the collection to receive real-time events on any data changes.
 const changeStream = coll.watch([]);
 
+////////////////////////////////////////////////////////////////////////////////////////////////////
+// Direct operations to shard on orphaned documents
+////////////////////////////////////////////////////////////////////////////////////////////////////
+
 jsTest.log('A direct insert to a shard of an orphaned document does not generate an insert event');
 {
     // Direct insert to first shard of an orphaned document.
-    assert.commandWorked(st.shard0.getCollection(collNS).insert({_id: 6, name: 'ken', age: 50}));
+    assert.commandWorked(st.shard0.getCollection(collNS).insert({_id: 6, name: 'ken', age: 60}));
 
     // No event is notified.
     assert(!changeStream.hasNext());
@@ -68,19 +75,19 @@ jsTest.log('A direct insert to a shard of an orphaned document does not generate
     assert.neq(null, st.shard0.getCollection(collNS).findOne({_id: 6}));
 }
 
-jsTest.log('A direct update to a shard on an orphaned document generates an update event');
+jsTest.log('A direct update to a shard of an orphaned document does not generate an update event');
 {
     // Send a direct update to first shard on an orphaned document.
-    assert.commandWorked(st.shard0.getCollection(collNS).update({name: 'matt'}, {$set: {age: 21}}));
+    assert.commandWorked(st.shard0.getCollection(collNS).update({name: 'matt'}, {$set: {age: 31}}));
 
     // No change stream event is generated.
     assert(!changeStream.hasNext());
 
     // The orphaned document on first shard has been updated.
-    assert.eq(21, st.shard0.getCollection(collNS).findOne({_id: 0}).age);
+    assert.eq(31, st.shard0.getCollection(collNS).findOne({_id: 0}).age);
 }
 
-jsTest.log('A direct delete to a shard on an orphaned document generates an update event');
+jsTest.log('A direct delete to a shard of an orphaned document does generate an update event');
 {
     // Send a direct delete to first shard on an orphaned document.
     assert.commandWorked(st.shard0.getCollection(collNS).remove({name: 'matt'}));
@@ -92,10 +99,14 @@ jsTest.log('A direct delete to a shard on an orphaned document generates an upda
     assert.eq(null, st.shard0.getCollection(collNS).findOne({_id: 0}));
 }
 
-jsTest.log('A distributed update on a single document generates an update event');
+////////////////////////////////////////////////////////////////////////////////////////////////////
+// Broadcasted operations (from router) on orphaned documents
+////////////////////////////////////////////////////////////////////////////////////////////////////
+
+jsTest.log('A broadcasted update of a single document generates an update event');
 {
-    // Send a distributed update (query on non-key field) on a single document to all the shards.
-    assert.commandWorked(coll.update({name: 'john'}, {$set: {age: 26}}, {multi: true}));
+    // Send a broadcasted update (query on non-key field) on a single document to all the shards.
+    assert.commandWorked(coll.update({name: 'john'}, {$set: {age: 36}}, {multi: true}));
 
     // The document is hosted by the second shard and the update event is notified. The first shard
     // still hosts the orphaned document so no additional event must be notified.
@@ -105,13 +116,13 @@ jsTest.log('A distributed update on a single document generates an update event'
 
     // The orphaned document on first shard has not been updated, unlike the non-orphaned one on the
     // second shard.
-    assert.eq(25, st.shard0.getCollection(collNS).findOne({_id: 1}).age);
-    assert.eq(26, st.shard1.getCollection(collNS).findOne({_id: 1}).age);
+    assert.eq(35, st.shard0.getCollection(collNS).findOne({_id: 1}).age);
+    assert.eq(36, st.shard1.getCollection(collNS).findOne({_id: 1}).age);
 }
 
-jsTest.log('A distributed delete on a single document generates a delete event');
+jsTest.log('A broadcasted delete of a single document generates a delete event');
 {
-    // Send a distributed delete (query on non-key field) on a single document to all the shards.
+    // Send a broadcasted delete (query on non-key field) on a single document to all the shards.
     assert.commandWorked(coll.remove({name: 'john'}));
 
     // The document is hosted by the second shard and the delete event is notified. The first shard
@@ -126,10 +137,10 @@ jsTest.log('A distributed delete on a single document generates a delete event')
     assert.eq(null, st.shard1.getCollection(collNS).findOne({_id: 1}));
 }
 
-jsTest.log('A distributed update on multi-documents generates more update events');
+jsTest.log('A broadcasted update of multi-documents generates more update events');
 {
-    // Send a distributed update (query on non-key field) on two documents to all the shards.
-    assert.commandWorked(coll.update({name: 'robert'}, {$set: {age: 36}}, {multi: true}));
+    // Send a broadcasted update (query on non-key field) on two documents to all the shards.
+    assert.commandWorked(coll.update({name: 'robert'}, {$set: {age: 41}}, {multi: true}));
 
     // The documents are hosted by the second shard and two delete events are notified. The first
     // shard still hosts the orphaned documents so no additional event must be notified.
@@ -141,15 +152,15 @@ jsTest.log('A distributed update on multi-documents generates more update events
 
     // The orphaned documents on first shard have not been updated, unlike the non-orphaned ones on
     // the second shard.
-    assert.eq(30, st.shard0.getCollection(collNS).findOne({_id: 2}).age);
-    assert.eq(35, st.shard0.getCollection(collNS).findOne({_id: 3}).age);
-    assert.eq(36, st.shard1.getCollection(collNS).findOne({_id: 2}).age);
-    assert.eq(36, st.shard1.getCollection(collNS).findOne({_id: 3}).age);
+    assert.eq(40, st.shard0.getCollection(collNS).findOne({_id: 2}).age);
+    assert.eq(45, st.shard0.getCollection(collNS).findOne({_id: 3}).age);
+    assert.eq(41, st.shard1.getCollection(collNS).findOne({_id: 2}).age);
+    assert.eq(41, st.shard1.getCollection(collNS).findOne({_id: 3}).age);
 }
 
-jsTest.log('A distributed delete on multi-documents generates more delete events');
+jsTest.log('A broadcasted delete of multi-documents generates more delete events');
 {
-    // Send a distributed delete (query on non-key field) on two documents to all the shards.
+    // Send a broadcasted delete (query on non-key field) on two documents to all the shards.
     assert.commandWorked(coll.remove({name: 'robert'}));
 
     // The documents are hosted by the second shard and two delete events are notified. The first
@@ -168,62 +179,132 @@ jsTest.log('A distributed delete on multi-documents generates more delete events
     assert.eq(null, st.shard1.getCollection(collNS).findOne({_id: 3}));
 }
 
-jsTest.log('A distributed update via transaction on multi-documents generates more update events');
+////////////////////////////////////////////////////////////////////////////////////////////////////
+// Transaction from router updating both orphaned and non-orphaned documents
+////////////////////////////////////////////////////////////////////////////////////////////////////
+
+jsTest.log('Broadcasted updates (via a transaction through the router) of both orphaned and ' +
+           'non-orphaned documents generate events only for operations on non-orphaned documents');
 {
-    // Send a distributed update (query on non-key field) via transaction on two documents to all
-    // the shards.
+    // Send a broadcasted transaction to the router updating both orphaned and non-orphaned
+    // documents.
     const session = st.s.startSession();
     const sessionDB = session.getDatabase(dbName);
     const sessionColl = sessionDB.getCollection(collName);
     session.startTransaction();
-    assert.commandWorked(sessionColl.update({name: 'james'}, {$set: {age: 46}}, {multi: true}));
+    assert.commandWorked(sessionColl.update({name: 'olivia'}, {$set: {age: 26}}, {multi: true}));
+    assert.commandWorked(sessionColl.update({name: 'james'}, {$set: {age: 51}}, {multi: true}));
     assert.commandWorked(session.commitTransaction_forTesting());
     session.endSession();
 
-    // The documents are hosted by the second shard and two delete events are notified. The first
-    // shard still hosts the orphaned documents so no additional event must be notified.
+    // The primary shard hosts orphaned (james) and non-orphaned (olivia) documents, whereas the
+    // second shard hosts a non-orphaned document (james). Consequently, two update events are
+    // notified.
     assert.soon(() => changeStream.hasNext(), 'A first update event is expected');
     assert.eq(changeStream.next().operationType, 'update');
     assert.soon(() => changeStream.hasNext(), 'A second update event is expected');
     assert.eq(changeStream.next().operationType, 'update');
     assert(!changeStream.hasNext());
 
-    // The orphaned documents on first shard have not been updated, unlike the non-orphaned ones on
-    // the second shard.
-    assert.eq(40, st.shard0.getCollection(collNS).findOne({_id: 4}).age);
-    assert.eq(45, st.shard0.getCollection(collNS).findOne({_id: 5}).age);
-    assert.eq(46, st.shard1.getCollection(collNS).findOne({_id: 4}).age);
-    assert.eq(46, st.shard1.getCollection(collNS).findOne({_id: 5}).age);
+    // The orphaned document on first shard (james) has not been updated, unlike the non-orphaned
+    // ones on both primary and second shards (olivia and james).
+    assert.eq(26, st.shard0.getCollection(collNS).findOne({_id: -1}).age);
+    assert.eq(50, st.shard0.getCollection(collNS).findOne({_id: 4}).age);
+    assert.eq(51, st.shard1.getCollection(collNS).findOne({_id: 4}).age);
 }
 
-jsTest.log('A distributed delete via transaction on multi-documents generates more delete events');
+jsTest.log('Broadcasted deletes (via a transaction through the router) of both orphaned and ' +
+           'non-orphaned documents generate events only for operations on non-orphaned documents');
 {
-    // Send a distributed delete (query on non-key field) via transaction on two documents to all
-    // the shards.
+    // Send a broadcasted transaction to the router deleting both orphaned and non-orphaned
+    // documents.
     const session = st.s.startSession();
     const sessionDB = session.getDatabase(dbName);
     const sessionColl = sessionDB.getCollection(collName);
     session.startTransaction();
+    assert.commandWorked(sessionColl.remove({name: 'olivia'}));
     assert.commandWorked(sessionColl.remove({name: 'james'}));
     assert.commandWorked(session.commitTransaction_forTesting());
     session.endSession();
 
-    // The documents are hosted by the second shard and two delete events are notified. The first
-    // shard still hosts the orphaned documents so no additional event must be notified.
+    // The primary shard hosts orphaned (james) and non-orphaned (olivia) documents, whereas the
+    // second shard hosts a non-orphaned document (james). Consequently, two delete events are
+    // notified.
     assert.soon(() => changeStream.hasNext(), 'A first delete event is expected');
     assert.eq(changeStream.next().operationType, 'delete');
     assert.soon(() => changeStream.hasNext(), 'A second delete event is expected');
     assert.eq(changeStream.next().operationType, 'delete');
     assert(!changeStream.hasNext());
 
-    // The orphaned documents on first shard have not been removed, unlike the non-orphaned ones on
-    // the second shard.
+    // The orphaned document on first shard (james) has not been removed, unlike the non-orphaned
+    // ones on both primary and second shards (olivia and james).
+    assert.eq(null, st.shard0.getCollection(collNS).findOne({_id: -1}));
     assert.neq(null, st.shard0.getCollection(collNS).findOne({_id: 4}));
-    assert.neq(null, st.shard0.getCollection(collNS).findOne({_id: 5}));
     assert.eq(null, st.shard1.getCollection(collNS).findOne({_id: 4}));
-    assert.eq(null, st.shard1.getCollection(collNS).findOne({_id: 5}));
 }
 
+////////////////////////////////////////////////////////////////////////////////////////////////////
+// Transaction to shard updating both orphaned and non-orphaned documents
+////////////////////////////////////////////////////////////////////////////////////////////////////
+
+jsTest.log('Direct updates (via a transaction to a shard) of both orphaned and non-orphaned' +
+           'documents generate events only for operations on non-orphaned documents');
+{
+    // Send a direct transaction to a shard updating both orphaned and non-orphaned documents.
+    const session = st.rs0.getPrimary().getDB(dbName).getMongo().startSession();
+    const sessionDB = session.getDatabase(dbName);
+    const sessionColl = sessionDB.getCollection(collName);
+    session.startTransaction();
+    assert.commandWorked(sessionColl.update({name: 'emma'}, {$set: {age: 21}}, {multi: true}));
+    assert.commandWorked(sessionColl.update({name: 'liam'}, {$set: {age: 56}}, {multi: true}));
+    assert.commandWorked(session.commitTransaction_forTesting());
+    session.endSession();
+
+    // The shard hosts both orphaned (liam) and non-orphaned (emma) documents. Consequently, only
+    // one update event is notified.
+    // TODO (SERVER-65859): The second update event will be filtered out when the ticket is
+    // completed.
+    assert.soon(() => changeStream.hasNext(), 'A first update event is expected');
+    assert.eq(changeStream.next().operationType, 'update');
+    assert.soon(() => changeStream.hasNext(), 'A second update event is expected');
+    assert.eq(changeStream.next().operationType, 'update');
+    assert(!changeStream.hasNext());
+
+    // Both orphaned (liam) and non-orphaned (emma) documents on the shard have been updated.
+    assert.eq(21, st.shard0.getCollection(collNS).findOne({_id: -2}).age);
+    assert.eq(56, st.shard0.getCollection(collNS).findOne({_id: 5}).age);
+}
+
+jsTest.log('Direct deletes (via a transaction to a shard) of both orphaned and non-orphaned' +
+           'documents generate events only for operations on non-orphaned documents');
+{
+    // Send a direct transaction to a shard deleting both orphaned and non-orphaned documents.
+    const session = st.rs0.getPrimary().getDB(dbName).getMongo().startSession();
+    const sessionDB = session.getDatabase(dbName);
+    const sessionColl = sessionDB.getCollection(collName);
+    session.startTransaction();
+    assert.commandWorked(sessionColl.remove({name: 'emma'}));
+    assert.commandWorked(sessionColl.remove({name: 'liam'}));
+    assert.commandWorked(session.commitTransaction_forTesting());
+    session.endSession();
+
+    // The shard hosts both orphaned (liam) and non-orphaned (emma) documents. Consequently, only
+    // one update event is notified.
+    // TODO (SERVER-65859): The second delete event will be filtered out when the ticket is
+    // completed.
+    assert.soon(() => changeStream.hasNext(), 'A first delete event is expected');
+    assert.eq(changeStream.next().operationType, 'delete');
+    assert.soon(() => changeStream.hasNext(), 'A second delete event is expected');
+    assert.eq(changeStream.next().operationType, 'delete');
+    assert(!changeStream.hasNext());
+
+    // Both orphaned (liam) and non-orphaned (emma) documents on the shard have been removed.
+    assert.eq(null, st.shard0.getCollection(collNS).findOne({_id: -2}));
+    assert.eq(null, st.shard0.getCollection(collNS).findOne({_id: 5}));
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////
+
 jsTest.log('The collection drop generates a drop event');
 {
     coll.drop();
@@ -234,6 +315,92 @@ jsTest.log('The collection drop generates a drop event');
     assert.eq(changeStream.next().operationType, 'drop');
 }
 
+////////////////////////////////////////////////////////////////////////////////////////////////////
+// Batched deletes from router and to shard
+////////////////////////////////////////////////////////////////////////////////////////////////////
+
+// Set the database to use batched deletes.
+const db2 = st.rs0.getPrimary().getDB(dbName);
+assert.commandWorked(db2.adminCommand({setParameter: 1, batchedDeletesTargetStagedDocBytes: 0}));
+assert.commandWorked(db2.adminCommand({setParameter: 1, batchedDeletesTargetBatchTimeMS: 0}));
+assert.commandWorked(db2.adminCommand({setParameter: 1, batchedDeletesTargetBatchDocs: 2}));
+
+// Create a non-sharded collection.
+const coll2 = db2.getCollection(collName);
+
+// Setup a change stream on the collection to receive real-time events on any data changes.
+const changeStream2 = coll2.watch([]);
+
+jsTest.log('A batched delete from router generates only one delete event');
+{
+    // Insert two documents in the collection (see 'batchedDeletesTargetBatchDocs') and skip the
+    // generated events.
+    assert.commandWorked(coll2.insert({_id: 0, name: 'volkswagen'}));
+    assert.commandWorked(coll2.insert({_id: 1, name: 'renault'}));
+    assert.soon(() => changeStream2.hasNext(), 'A first insert event is expected');
+    assert.eq(changeStream2.next().operationType, 'insert');
+    assert.soon(() => changeStream2.hasNext(), 'A second insert event is expected');
+    assert.eq(changeStream2.next().operationType, 'insert');
+    assert(!changeStream2.hasNext());
+
+    // Delete all documents in batch from the collection.
+    assert.commandWorked(coll2.deleteMany({_id: {$gte: 0}}));
+
+    // Actually only one delete operation is performed. Consequently, only one delete event is
+    // notified.
+    // TODO (SERVER-65859): The second delete event will be filtered out when the ticket is
+    // completed.
+    assert.soon(() => changeStream2.hasNext(), 'A first delete event is expected');
+    assert.eq(changeStream2.next().operationType, 'delete');
+    assert.soon(() => changeStream2.hasNext(), 'A second delete event is expected for now');
+    assert.eq(changeStream2.next().operationType, 'delete');
+    assert(!changeStream2.hasNext());
+
+    // All documents have been removed from the collection.
+    assert.eq(0, coll2.find().itcount());
+}
+
+jsTest.log('A batched delete to shard generates only one delete event');
+{
+    // Insert two documents in the collection (see 'batchedDeletesTargetBatchDocs') and skip the
+    // generated events.
+    assert.commandWorked(coll2.insert({_id: 0, name: 'volkswagen'}));
+    assert.commandWorked(coll2.insert({_id: 1, name: 'renault'}));
+    assert.soon(() => changeStream2.hasNext(), 'A first insert event is expected');
+    assert.eq(changeStream2.next().operationType, 'insert');
+    assert.soon(() => changeStream2.hasNext(), 'A second insert event is expected');
+    assert.eq(changeStream2.next().operationType, 'insert');
+    assert(!changeStream2.hasNext());
+
+    // Delete all documents in batch from the collection.
+    assert.commandWorked(st.shard0.getCollection(collNS).deleteMany({_id: {$gte: 0}}));
+
+    // Actually only one delete operation is performed. Consequently, only one delete event is
+    // notified.
+    // TODO (SERVER-65859): The second delete event will be filtered out when the ticket is
+    // completed.
+    assert.soon(() => changeStream2.hasNext(), 'A first delete event is expected');
+    assert.eq(changeStream2.next().operationType, 'delete');
+    assert.soon(() => changeStream2.hasNext(), 'A second delete event is expected for now');
+    assert.eq(changeStream2.next().operationType, 'delete');
+    assert(!changeStream2.hasNext());
+
+    // All documents have been removed from the collection.
+    assert.eq(0, coll2.find().itcount());
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////
+
+jsTest.log('The collection drop generates a drop event');
+{
+    coll2.drop();
+
+    // Essentially, this verifies that the operation before dropping the collection did not notify
+    // additional and unexpected events.
+    assert.soon(() => changeStream2.hasNext(), 'A drop event is expected');
+    assert.eq(changeStream2.next().operationType, 'drop');
+}
+
 suspendRangeDeletionShard0.off();
 
 st.stop();
diff --git a/jstests/sharding/internal_txns/retry_on_transient_error_basic.js b/jstests/sharding/internal_txns/retry_on_transient_error_basic.js
index b417d5a545b..2bc6e61425f 100644
--- a/jstests/sharding/internal_txns/retry_on_transient_error_basic.js
+++ b/jstests/sharding/internal_txns/retry_on_transient_error_basic.js
@@ -145,6 +145,10 @@ function testPersistence(shardRst, lsid, txnNumber, txnDocFilter, oplogEntryFilt
     const txnRetryCounter1 = NumberInt(1);
     let db = shardRst.getPrimary().getDB(kDbName);
 
+    // Preload the collection metadata to avoid repeating the insert command if it fails due to
+    // StaleConfig error.
+    assert.commandWorked(db.adminCommand({_flushRoutingTableCacheUpdates: kNs}));
+
     const insertCmdObj = {
         insert: kCollName,
         documents: [{x: 0}],