SERVER-70905 oplog application handles large batched writes

1 files changed, 55 insertions, 22 deletions

diff --git a/src/mongo/db/repl/oplog_applier_impl.cpp b/src/mongo/db/repl/oplog_applier_impl.cpp
index 2b89323bb27..5ebe08bee80 100644
--- a/src/mongo/db/repl/oplog_applier_impl.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl.cpp
@@ -52,7 +52,7 @@
 #include "mongo/db/storage/control/journal_flusher.h"
 #include "mongo/db/storage/storage_util.h"
 #include "mongo/logv2/log.h"
-#include "mongo/platform/basic.h"
+#include "mongo/stdx/unordered_map.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/log_with_sampling.h"
 
@@ -695,7 +695,31 @@ void OplogApplierImpl::_deriveOpsAndFillWriterVectors(
     std::vector<std::vector<OplogEntry>>* derivedOps,
     SessionUpdateTracker* sessionUpdateTracker) noexcept {
 
-    LogicalSessionIdMap<std::vector<OplogEntry*>> partialTxnOps;
+    // Caches partial transaction operations. Each map entry
+    // contains a cumulative list of operations seen in this batch so
+    // far.
+    stdx::unordered_map<OpTime, std::vector<OplogEntry*>, OpTime::Hasher> partialTxnOps;
+
+    // Provided to _addOplogChainOpsToWriterVectors() when 'partialTxnOps' does not have any entries
+    // for 'prevOpTime'.
+    std::vector<OplogEntry*> emptyPartialTxnOps;
+
+    // Returns a mutable partial transaction list that is either an existing entry in
+    // 'partialTxnOps' or 'emptyPartialTxnOps'.
+    auto getPartialTxnList = [&](const auto& op) {
+        auto prevOpTime = op.getPrevWriteOpTimeInTransaction();
+        invariant(prevOpTime, op.toStringForLogging());
+        auto it = partialTxnOps.find(*prevOpTime);
+        if (it != partialTxnOps.end()) {
+            auto& partialTxnList = it->second;
+            invariant(!prevOpTime->isNull(), op.toStringForLogging());
+            invariant(!partialTxnList.empty(), op.toStringForLogging());
+            return &partialTxnList;
+        }
+
+        return &emptyPartialTxnOps;  // cleared in _addOplogChainOpsToWriterVectors().
+    };
+
     CachedCollectionProperties collPropertiesCache;
 
     for (auto&& op : *ops) {
@@ -722,17 +746,25 @@ void OplogApplierImpl::_deriveOpsAndFillWriterVectors(
         // or commit. We must save it here because we are not guaranteed it has been written to
         // the oplog yet.
         if (op.isPartialTransaction()) {
-            // applyOps entries generated by a transaction must have a sessionId and a
-            // transaction number.
-            invariant(op.getSessionId(), op.toStringForLogging());
-            invariant(op.getTxnNumber(), op.toStringForLogging());
-
-            auto& partialTxnList = partialTxnOps[*op.getSessionId()];
-            // If this operation belongs to an existing partial transaction, partialTxnList
-            // must contain the previous operations of the transaction.
-            invariant(partialTxnList.empty() ||
-                      partialTxnList.front()->getTxnNumber() == op.getTxnNumber());
-            partialTxnList.push_back(&op);
+            auto prevOpTime = op.getPrevWriteOpTimeInTransaction();
+            invariant(prevOpTime, op.toStringForLogging());
+            if (auto it = partialTxnOps.find(*prevOpTime); it != partialTxnOps.end()) {
+                auto& partialTxnList = it->second;
+                invariant(!prevOpTime->isNull(), op.toStringForLogging());
+                invariant(!partialTxnList.empty(), op.toStringForLogging());
+                // If this operation belongs to an existing partial transaction, partialTxnList
+                // must contain the previous operations of the transaction.
+                invariant(partialTxnList.front()->getTxnNumber() == op.getTxnNumber(),
+                          op.toStringForLogging());
+                partialTxnList.push_back(&op);
+                // Replace key with new optime corresponding to updated list.
+                auto nodeHandle = partialTxnOps.extract(it);
+                nodeHandle.key() = op.getOpTime();
+                invariant(partialTxnOps.insert(std::move(nodeHandle)).inserted,
+                          op.toStringForLogging());
+            } else {
+                partialTxnOps[op.getOpTime()].push_back(&op);
+            }
             continue;
         }
 
@@ -752,15 +784,15 @@ void OplogApplierImpl::_deriveOpsAndFillWriterVectors(
         // Extract applyOps operations and fill writers with extracted operations.
         if (op.isTerminalApplyOps()) {
             auto logicalSessionId = op.getSessionId();
-            // applyOps entries generated by a transaction must have a sessionId and a
-            // transaction number.
-            if (logicalSessionId && op.getTxnNumber()) {
+            // applyOps entries generated by a transaction must have a prevOpTime.
+            if (auto prevOpTime = op.getPrevWriteOpTimeInTransaction()) {
                 // On commit of unprepared transactions, get transactional operations from the
                 // oplog and fill writers with those operations.
                 // Flush partialTxnList operations for current transaction.
-                auto& partialTxnList = partialTxnOps[*logicalSessionId];
+                auto* partialTxnList = getPartialTxnList(op);
                 _addOplogChainOpsToWriterVectors(
-                    opCtx, &partialTxnList, derivedOps, &op, &collPropertiesCache, writerVectors);
+                    opCtx, partialTxnList, derivedOps, &op, &collPropertiesCache, writerVectors);
+                invariant(partialTxnList->empty(), op.toStringForLogging());
             } else {
                 // The applyOps entry was not generated as part of a transaction.
                 invariant(!op.getPrevWriteOpTimeInTransaction());
@@ -782,9 +814,9 @@ void OplogApplierImpl::_deriveOpsAndFillWriterVectors(
             // Prepare entries in secondary mode do not come in their own batch, we will extract
             // applyOps operations and fill writers with the extracted operations.
             if (op.shouldPrepare() && (getOptions().mode == OplogApplication::Mode::kSecondary)) {
-                auto& partialTxnList = partialTxnOps[*op.getSessionId()];
+                auto* partialTxnList = getPartialTxnList(op);
                 _addOplogChainOpsToWriterVectors(
-                    opCtx, &partialTxnList, derivedOps, &op, &collPropertiesCache, writerVectors);
+                    opCtx, partialTxnList, derivedOps, &op, &collPropertiesCache, writerVectors);
                 continue;
             }
             if (op.isPreparedCommit() &&
@@ -798,9 +830,10 @@ void OplogApplierImpl::_deriveOpsAndFillWriterVectors(
         // initial sync, find the prepare oplog entry, extract applyOps operations, and fill writers
         // with the extracted operations.
         if (op.isPreparedCommit() && (getOptions().mode == OplogApplication::Mode::kInitialSync)) {
-            auto& partialTxnList = partialTxnOps[*op.getSessionId()];
+            auto* partialTxnList = getPartialTxnList(op);
             _addOplogChainOpsToWriterVectors(
-                opCtx, &partialTxnList, derivedOps, &op, &collPropertiesCache, writerVectors);
+                opCtx, partialTxnList, derivedOps, &op, &collPropertiesCache, writerVectors);
+            invariant(partialTxnList->empty(), op.toStringForLogging());
             continue;
         }