SERVER-56375: Add ability to write retryable findAndModify deletes to `config.image_collection`

4 files changed, 60 insertions, 8 deletions

diff --git a/src/mongo/db/op_observer_impl.cpp b/src/mongo/db/op_observer_impl.cpp
index dfee2e4f4e5..45c03ab79b4 100644
--- a/src/mongo/db/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer_impl.cpp
@@ -63,6 +63,7 @@
 
 namespace mongo {
 using repl::OplogEntry;
+
 namespace {
 
 MONGO_FAIL_POINT_DEFINE(failCollectionUpdates);
@@ -266,7 +267,8 @@ OpTimeBundle replLogDelete(OperationContext* opCtx,
                            Session* session,
                            StmtId stmtId,
                            bool fromMigrate,
-                           const boost::optional<BSONObj>& deletedDoc) {
+                           const boost::optional<BSONObj>& deletedDoc,
+                           const bool storeImagesInSideCollection) {
     OperationSessionInfo sessionInfo;
     repl::OplogLink oplogLink;
 
@@ -281,9 +283,7 @@ OpTimeBundle replLogDelete(OperationContext* opCtx,
 
     boost::optional<repl::RetryImageEnum> needsRetryImage;
     if (deletedDoc && opCtx->getTxnNumber()) {
-        if (storeFindAndModifyImagesInSideCollection.load() &&
-            serverGlobalParams.featureCompatibility.getVersion() >=
-                ServerGlobalParams::FeatureCompatibility::Version::kUpgradingTo40) {
+        if (storeImagesInSideCollection) {
             needsRetryImage = repl::RetryImageEnum::kPreImage;
         } else {
             auto noteOplog = logOperation(opCtx,
@@ -584,7 +584,23 @@ void OpObserverImpl::onDelete(OperationContext* opCtx,
         auto operation = OplogEntry::makeDeleteOperation(nss, uuid, documentKey);
         session->addTransactionOperation(opCtx, operation);
     } else {
-        opTime = replLogDelete(opCtx, nss, uuid, session, stmtId, fromMigrate, deletedDoc);
+        const auto storeImagesInSideCollection = storeFindAndModifyImagesInSideCollection.load() &&
+            serverGlobalParams.featureCompatibility.getVersion() >=
+                ServerGlobalParams::FeatureCompatibility::Version::kUpgradingTo40;
+        opTime = replLogDelete(opCtx,
+                               nss,
+                               uuid,
+                               session,
+                               stmtId,
+                               fromMigrate,
+                               deletedDoc,
+                               storeImagesInSideCollection);
+        if (storeImagesInSideCollection && deletedDoc && opCtx->getTxnNumber()) {
+            writeToImagesCollection(opCtx,
+                                    *deletedDoc,
+                                    repl::RetryImageEnum::kPreImage,
+                                    opTime.writeOpTime.getTimestamp());
+        }
         onWriteOpCompleted(opCtx,
                            nss,
                            session,
diff --git a/src/mongo/db/ops/delete.cpp b/src/mongo/db/ops/delete.cpp
index e7d18e7ef5e..1b2813bcf54 100644
--- a/src/mongo/db/ops/delete.cpp
+++ b/src/mongo/db/ops/delete.cpp
@@ -34,7 +34,6 @@
 
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/exec/delete.h"
-#include "mongo/db/ops/delete_request.h"
 #include "mongo/db/ops/parsed_delete.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/repl/repl_client_info.h"
@@ -65,4 +64,24 @@ long long deleteObjects(OperationContext* opCtx,
     return DeleteStage::getNumDeleted(*exec);
 }
 
+boost::optional<BSONObj> deleteObject(OperationContext* opCtx,
+                                      Collection* collection,
+                                      const DeleteRequest& request) {
+    ParsedDelete parsedDelete(opCtx, &request);
+    uassertStatusOK(parsedDelete.parseRequest());
+
+    // This method doesn't support multi-deletes.
+    invariant(!request.isMulti());
+
+    auto exec = uassertStatusOK(
+        getExecutorDelete(opCtx, &CurOp::get(opCtx)->debug(), collection, &parsedDelete));
+
+    BSONObj docImage;
+    if (exec->getNext(&docImage, nullptr) == PlanExecutor::IS_EOF) {
+        return boost::none;
+    }
+
+    return {request.shouldReturnDeleted() ? docImage.getOwned() : BSONObj()};
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/ops/delete.h b/src/mongo/db/ops/delete.h
index 2e2b30cfede..4c2da9ef84e 100644
--- a/src/mongo/db/ops/delete.h
+++ b/src/mongo/db/ops/delete.h
@@ -33,6 +33,7 @@
 #pragma once
 
 #include "mongo/db/jsobj.h"
+#include "mongo/db/ops/delete_request.h"
 #include "mongo/db/query/plan_executor.h"
 
 
@@ -53,4 +54,8 @@ long long deleteObjects(OperationContext* opCtx,
                         bool justOne,
                         bool god = false,
                         bool fromMigrate = false);
+
+boost::optional<BSONObj> deleteObject(OperationContext* opCtx,
+                                      Collection* collection,
+                                      const DeleteRequest& request);
 }
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 0857070ccaf..46e9e45def2 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -1581,6 +1581,7 @@ Status applyOperation_inlock(OperationContext* opCtx,
             timestamp = fieldTs.timestamp();
         }
 
+        bool upsertConfigImage = true;
         const StringData ns = fieldNs.valuestrsafe();
         writeConflictRetry(opCtx, "applyOps_delete", ns, [&] {
             WriteUnitOfWork wuow(opCtx);
@@ -1589,8 +1590,19 @@ Status applyOperation_inlock(OperationContext* opCtx,
             }
 
             if (opType[1] == 0) {
-                const auto justOne = true;
-                deleteObjects(opCtx, collection, requestNss, deleteCriteria, justOne);
+                DeleteRequest request(requestNss);
+                request.setQuery(deleteCriteria);
+                if (op.hasField(OplogEntryBase::kNeedsRetryImageFieldName)) {
+                    request.setReturnDeleted(true);
+                }
+                boost::optional<BSONObj> preImage = deleteObject(opCtx, collection, request);
+                if (op.hasField(OplogEntryBase::kNeedsRetryImageFieldName)) {
+                    writeToImageCollection(opCtx,
+                                           op,
+                                           preImage.get(),
+                                           repl::RetryImageEnum::kPreImage,
+                                           &upsertConfigImage);
+                }
             } else
                 verify(opType[1] == 'b');  // "db" advertisement
             wuow.commit();