SERVER-63880 Make resharding handle applyOps oplog entries with WouldChangeOwningShard sentinel noop entry

2 files changed, 63 insertions, 13 deletions

diff --git a/src/mongo/db/s/resharding/resharding_oplog_batch_preparer.cpp b/src/mongo/db/s/resharding/resharding_oplog_batch_preparer.cpp
index bde8901926a..95240e60907 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_batch_preparer.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_batch_preparer.cpp
@@ -35,6 +35,7 @@
 
 #include "mongo/bson/bsonelement_comparator.h"
 #include "mongo/db/logical_session_id.h"
+#include "mongo/db/ops/write_ops_retryability.h"
 #include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/db/repl/apply_ops.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
@@ -128,8 +129,6 @@ WriterVectors ReshardingOplogBatchPreparer::makeCrudOpWriterVectors(
             }
 
             auto applyOpsInfo = repl::ApplyOpsCommandInfo::parse(op.getObject());
-            // TODO (SERVER-63880): Make resharding handle applyOps oplog entries with
-            // WouldChangeOwningShard sentinel noop entry.
             uassert(
                 ErrorCodes::OplogOperationUnsupported,
                 str::stream() << "Commands within applyOps are not supported during resharding: "
@@ -143,6 +142,10 @@ WriterVectors ReshardingOplogBatchPreparer::makeCrudOpWriterVectors(
                 unrolledOp.setDurableReplOperation(repl::DurableReplOperation::parse(
                     {"ReshardingOplogBatchPreparer::makeCrudOpWriterVectors innerOp"}, innerOp));
 
+                if (isWouldChangeOwningShardSentinelOplogEntry(unrolledOp)) {
+                    continue;
+                }
+
                 // There isn't a direct way to convert from a MutableOplogEntry to a
                 // DurableOplogEntry or OplogEntry. We serialize the unrolledOp to have it get
                 // re-parsed into an OplogEntry.
@@ -214,8 +217,6 @@ WriterVectors ReshardingOplogBatchPreparer::makeSessionOpWriterVectors(
                 // transaction applyOps oplog entry.
 
                 auto applyOpsInfo = repl::ApplyOpsCommandInfo::parse(op.getObject());
-                // TODO (SERVER-63880): Make resharding handle applyOps oplog entries with
-                // WouldChangeOwningShard sentinel noop entry.
                 uassert(ErrorCodes::OplogOperationUnsupported,
                         str::stream()
                             << "Commands within applyOps are not supported during resharding: "
@@ -241,7 +242,8 @@ WriterVectors ReshardingOplogBatchPreparer::makeSessionOpWriterVectors(
                     // DurableOplogEntry or OplogEntry. We serialize the unrolledOp to have it get
                     // re-parsed into an OplogEntry.
                     auto& derivedOp = derivedOps.emplace_back(unrolledOp.toBSON());
-                    invariant(derivedOp.isCrudOpType());
+                    invariant(derivedOp.isCrudOpType() ||
+                              isWouldChangeOwningShardSentinelOplogEntry(unrolledOp));
 
                     // `&derivedOp` is guaranteed to remain stable while we append more derived
                     // oplog entries because `derivedOps` is a std::list.
diff --git a/src/mongo/db/s/resharding/resharding_oplog_batch_preparer_test.cpp b/src/mongo/db/s/resharding/resharding_oplog_batch_preparer_test.cpp
index c98104fcf3a..31d073a4790 100644
--- a/src/mongo/db/s/resharding/resharding_oplog_batch_preparer_test.cpp
+++ b/src/mongo/db/s/resharding/resharding_oplog_batch_preparer_test.cpp
@@ -32,6 +32,7 @@
 #include <boost/optional/optional_io.hpp>
 
 #include "mongo/db/logical_session_id_helpers.h"
+#include "mongo/db/ops/write_ops_retryability.h"
 #include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/db/s/resharding/resharding_oplog_batch_preparer.h"
 #include "mongo/db/s/resharding/resharding_server_parameters_gen.h"
@@ -77,19 +78,31 @@ protected:
                                            boost::optional<TxnNumber> txnNumber = boost::none,
                                            boost::optional<bool> isPrepare = boost::none,
                                            boost::optional<bool> isPartial = boost::none) {
-        BSONObjBuilder applyOpsBuilder;
-
-        BSONArrayBuilder opsArrayBuilder = applyOpsBuilder.subarrayStart("applyOps");
+        std::vector<repl::DurableReplOperation> ops;
         for (const auto& document : documents) {
             auto insertOp = repl::DurableReplOperation(repl::OpTypeEnum::kInsert, {}, document);
             if (lsid && isInternalSessionForRetryableWrite(*lsid)) {
-                if (!document.hasField("_id")) {
-                    continue;
+                if (document.hasField("_id")) {
+                    auto id = document.getIntField("_id");
+                    insertOp.setStatementIds({{id}});
                 }
-                auto id = document.getIntField("_id");
-                insertOp.setStatementIds({{id}});
             }
-            opsArrayBuilder.append(insertOp.toBSON());
+            ops.emplace_back(insertOp);
+        }
+
+        return makeApplyOpsOplogEntry(ops, lsid, txnNumber, isPrepare, isPartial);
+    }
+
+    repl::OplogEntry makeApplyOpsOplogEntry(std::vector<repl::DurableReplOperation> ops,
+                                            boost::optional<LogicalSessionId> lsid = boost::none,
+                                            boost::optional<TxnNumber> txnNumber = boost::none,
+                                            boost::optional<bool> isPrepare = boost::none,
+                                            boost::optional<bool> isPartial = boost::none) {
+        BSONObjBuilder applyOpsBuilder;
+
+        BSONArrayBuilder opsArrayBuilder = applyOpsBuilder.subarrayStart("applyOps");
+        for (const auto& op : ops) {
+            opsArrayBuilder.append(op.toBSON());
         }
         opsArrayBuilder.done();
 
@@ -414,6 +427,41 @@ TEST_F(ReshardingOplogBatchPreparerTest, DiscardsNoops) {
     runTest(makeLogicalSessionIdWithTxnNumberAndUUIDForTest(), txnNumber);
 }
 
+TEST_F(ReshardingOplogBatchPreparerTest,
+       SessionWriterDoesNotDiscardWouldChangeOwningShardNoopForRetryableInternalTransaction) {
+
+    const auto lsid = makeLogicalSessionIdWithTxnNumberAndUUIDForTest();
+    const TxnNumber txnNumber{1};
+
+    OplogBatch batch;
+
+    auto op =
+        repl::DurableReplOperation(repl::OpTypeEnum::kNoop, {}, kWouldChangeOwningShardSentinel);
+    op.setObject2(BSONObj());
+    op.setStatementIds({{0}});
+    batch.emplace_back(makeApplyOpsOplogEntry(
+        {op}, lsid, txnNumber, false /* isPrepare */, false /* isPartial */));
+
+    std::list<repl::OplogEntry> derivedOpsForCrudWriters;
+    auto crudWriterVectors =
+        _batchPreparer.makeCrudOpWriterVectors(batch, derivedOpsForCrudWriters);
+    ASSERT_EQ(crudWriterVectors.size(), kNumWriterVectors);
+    ASSERT_EQ(derivedOpsForCrudWriters.size(), 0U);
+    ASSERT_EQ(crudWriterVectors[0].size(), 0U);
+    ASSERT_EQ(crudWriterVectors[1].size(), 0U);
+
+    std::list<repl::OplogEntry> derivedOpsForSessionWriters;
+    auto sessionWriterVectors =
+        _batchPreparer.makeSessionOpWriterVectors(batch, derivedOpsForSessionWriters);
+    ASSERT_EQ(sessionWriterVectors.size(), kNumWriterVectors);
+    auto writer = getNonEmptyWriterVector(sessionWriterVectors);
+    ASSERT_EQ(writer.size(), 1U);
+    ASSERT_EQ(derivedOpsForSessionWriters.size(), 1U);
+    ASSERT_EQ(writer[0]->getSessionId(), *getParentSessionId(lsid));
+    ASSERT_EQ(*writer[0]->getTxnNumber(), *lsid.getTxnNumber());
+    ASSERT(isWouldChangeOwningShardSentinelOplogEntry(*writer[0]));
+}
+
 TEST_F(ReshardingOplogBatchPreparerTest, SessionWriteVectorsForApplyOpsWithoutTxnNumber) {
     OplogBatch batch;
     batch.emplace_back(makeApplyOpsForInsert({BSON("_id" << 0)}));