SERVER-45805 Add recordPreImages flag to collMod and create commands

 create mode 100644 jstests/noPassthrough/record_preimage_startup_validation.js

7 files changed, 104 insertions, 4 deletions

diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 49a666c9c67..5ad39981d44 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -74,6 +74,7 @@ struct CollModRequest {
     BSONElement collValidator = {};
     std::string collValidationAction = {};
     std::string collValidationLevel = {};
+    bool recordPreImages = false;
 };
 
 StatusWith<CollModRequest> parseCollModRequest(OperationContext* opCtx,
@@ -222,6 +223,13 @@ StatusWith<CollModRequest> parseCollModRequest(OperationContext* opCtx,
                 return Status(ErrorCodes::InvalidOptions, "'viewOn' option must be a string");
             }
             cmr.viewOn = e.str();
+        } else if (fieldName == "recordPreImages") {
+            if (isView) {
+                return {ErrorCodes::InvalidOptions,
+                        str::stream() << "option not supported on a view: " << fieldName};
+            }
+
+            cmr.recordPreImages = e.trueValue();
         } else {
             if (isView) {
                 return Status(ErrorCodes::InvalidOptions,
@@ -396,6 +404,10 @@ Status _collModInternal(OperationContext* opCtx,
         if (!cmrNew.collValidationLevel.empty())
             invariant(coll->setValidationLevel(opCtx, cmrNew.collValidationLevel));
 
+        if (cmrNew.recordPreImages != oldCollOptions.recordPreImages) {
+            coll->setRecordPreImages(opCtx, cmrNew.recordPreImages);
+        }
+
         // Only observe non-view collMods, as view operations are observed as operations on the
         // system.views collection.
         getGlobalServiceContext()->getOpObserver()->onCollMod(
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 5cac53c94b3..e11a790382c 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -93,6 +93,7 @@ struct CollectionUpdateArgs {
     bool fromMigrate = false;
 
     StoreDocOption storeDocOption = StoreDocOption::None;
+    bool preImageRecordingEnabledForCollection = false;
 };
 
 /**
@@ -383,6 +384,9 @@ public:
                                    StringData newLevel,
                                    StringData newAction) = 0;
 
+    virtual bool getRecordPreImages() const = 0;
+    virtual void setRecordPreImages(OperationContext* opCtx, bool val) = 0;
+
     /**
      * Returns true if this is a temporary collection.
      *
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 553af36eaf5..9d845cb28da 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -211,6 +211,35 @@ Status checkValidatorCanBeUsedOnNs(const BSONObj& validator,
     return Status::OK();
 }
 
+Status validatePreImageRecording(OperationContext* opCtx, const NamespaceString& ns) {
+    if (ns.db() == NamespaceString::kAdminDb || ns.db() == NamespaceString::kLocalDb) {
+        return {ErrorCodes::InvalidOptions,
+                str::stream() << "recordPreImages collection option is not supported on the "
+                              << ns.db() << " database"};
+    }
+
+    if (!serverGlobalParams.featureCompatibility.isVersionInitialized() ||
+        serverGlobalParams.featureCompatibility.getVersion() !=
+            ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44) {
+        return {ErrorCodes::InvalidOptions,
+                "recordPreImages collection option is only supported when the feature "
+                "compatibility version is set to 4.4 or above"};
+    }
+
+    if (serverGlobalParams.clusterRole != ClusterRole::None) {
+        return {ErrorCodes::InvalidOptions,
+                "recordPreImages collection option is not supported on shards or config servers"};
+    }
+
+    auto replCoord = repl::ReplicationCoordinator::get(opCtx);
+    if (!replCoord->isReplEnabled()) {
+        return {ErrorCodes::InvalidOptions,
+                "recordPreImages collection option depends on being in a replica set"};
+    }
+
+    return Status::OK();
+}
+
 }  // namespace
 
 CollectionImpl::CollectionImpl(OperationContext* opCtx,
@@ -273,6 +302,11 @@ void CollectionImpl::init(OperationContext* opCtx) {
     }
     _validationAction = uassertStatusOK(_parseValidationAction(collectionOptions.validationAction));
     _validationLevel = uassertStatusOK(_parseValidationLevel(collectionOptions.validationLevel));
+    if (collectionOptions.recordPreImages) {
+        uassertStatusOK(validatePreImageRecording(opCtx, _ns));
+        _recordPreImages = true;
+    }
+
     getIndexCatalog()->init(opCtx).transitional_ignore();
     _initialized = true;
 }
@@ -643,7 +677,8 @@ void CollectionImpl::deleteDocument(OperationContext* opCtx,
     getGlobalServiceContext()->getOpObserver()->aboutToDelete(opCtx, ns(), doc.value());
 
     boost::optional<BSONObj> deletedDoc;
-    if (storeDeletedDoc == Collection::StoreDeletedDoc::On) {
+    if ((storeDeletedDoc == Collection::StoreDeletedDoc::On && opCtx->getTxnNumber()) ||
+        getRecordPreImages()) {
         deletedDoc.emplace(doc.value().getOwned());
     }
 
@@ -716,7 +751,13 @@ RecordId CollectionImpl::updateDocument(OperationContext* opCtx,
                   str::stream() << "Cannot change the size of a document in a capped collection: "
                                 << oldSize << " != " << newDoc.objsize());
 
-    args->preImageDoc = oldDoc.value().getOwned();
+    // The preImageDoc may not be boost::none if this update was a retryable findAndModify or if
+    // the update may have changed the shard key. For non-in-place updates we always set the
+    // preImageDoc here to an owned copy of the pre-image.
+    if (!args->preImageDoc) {
+        args->preImageDoc = oldDoc.value().getOwned();
+    }
+    args->preImageRecordingEnabledForCollection = getRecordPreImages();
 
     uassertStatusOK(
         _recordStore->updateRecord(opCtx, oldLocation, newDoc.objdata(), newDoc.objsize()));
@@ -725,7 +766,7 @@ RecordId CollectionImpl::updateDocument(OperationContext* opCtx,
         int64_t keysInserted, keysDeleted;
 
         uassertStatusOK(_indexCatalog->updateRecord(
-            opCtx, args->preImageDoc.get(), newDoc, oldLocation, &keysInserted, &keysDeleted));
+            opCtx, *args->preImageDoc, newDoc, oldLocation, &keysInserted, &keysDeleted));
 
         if (opDebug) {
             opDebug->additiveMetrics.incrementKeysInserted(keysInserted);
@@ -760,12 +801,19 @@ StatusWith<RecordData> CollectionImpl::updateDocumentWithDamages(
     invariant(oldRec.snapshotId() == opCtx->recoveryUnit()->getSnapshotId());
     invariant(updateWithDamagesSupported());
 
+    // For in-place updates we need to grab an owned copy of the pre-image doc if pre-image
+    // recording is enabled and we haven't already set the pre-image due to this update being
+    // a retryable findAndModify or a possible update to the shard key.
+    if (!args->preImageDoc && getRecordPreImages()) {
+        args->preImageDoc = oldRec.value().toBson().getOwned();
+    }
+
     auto newRecStatus =
         _recordStore->updateWithDamages(opCtx, loc, oldRec.value(), damageSource, damages);
 
     if (newRecStatus.isOK()) {
         args->updatedDoc = newRecStatus.getValue().toBson();
-
+        args->preImageRecordingEnabledForCollection = getRecordPreImages();
         OplogUpdateEntryArgs entryArgs(*args, ns(), _uuid);
         getGlobalServiceContext()->getOpObserver()->onUpdate(opCtx, entryArgs);
     }
@@ -776,6 +824,18 @@ bool CollectionImpl::isTemporary(OperationContext* opCtx) const {
     return DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, getCatalogId()).temp;
 }
 
+bool CollectionImpl::getRecordPreImages() const {
+    return _recordPreImages;
+}
+
+void CollectionImpl::setRecordPreImages(OperationContext* opCtx, bool val) {
+    if (val) {
+        uassertStatusOK(validatePreImageRecording(opCtx, _ns));
+    }
+    DurableCatalog::get(opCtx)->setRecordPreImages(opCtx, getCatalogId(), val);
+    _recordPreImages = val;
+}
+
 bool CollectionImpl::isCapped() const {
     return _cappedNotifier.get();
 }
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index 42178e704b8..864bb6b9ea7 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -266,6 +266,9 @@ public:
                            StringData newLevel,
                            StringData newAction) final;
 
+    bool getRecordPreImages() const final;
+    void setRecordPreImages(OperationContext* opCtx, bool val) final;
+
     bool isTemporary(OperationContext* opCtx) const final;
 
     //
@@ -403,6 +406,8 @@ private:
     ValidationAction _validationAction;
     ValidationLevel _validationLevel;
 
+    bool _recordPreImages = false;
+
     // Notifier object for awaitData. Threads polling a capped collection for new data can wait
     // on this object until notified of the arrival of new data.
     //
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index 7a9579fe67b..001bee4db5d 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -208,6 +208,14 @@ public:
         std::abort();
     }
 
+    bool getRecordPreImages() const {
+        std::abort();
+    }
+
+    void setRecordPreImages(OperationContext* opCtx, bool val) {
+        std::abort();
+    }
+
     bool isCapped() const {
         std::abort();
     }
diff --git a/src/mongo/db/catalog/collection_options.cpp b/src/mongo/db/catalog/collection_options.cpp
index 9c1561e8192..f086b90547c 100644
--- a/src/mongo/db/catalog/collection_options.cpp
+++ b/src/mongo/db/catalog/collection_options.cpp
@@ -168,6 +168,8 @@ StatusWith<CollectionOptions> CollectionOptions::parse(const BSONObj& options, P
             continue;
         } else if (fieldName == "temp") {
             collectionOptions.temp = e.trueValue();
+        } else if (fieldName == "recordPreImages") {
+            collectionOptions.recordPreImages = e.trueValue();
         } else if (fieldName == "storageEngine") {
             Status status = checkStorageEngineOptions(e);
             if (!status.isOK()) {
@@ -286,6 +288,10 @@ void CollectionOptions::appendBSON(BSONObjBuilder* builder) const {
     if (temp)
         builder->appendBool("temp", true);
 
+    if (recordPreImages) {
+        builder->appendBool("recordPreImages", true);
+    }
+
     if (!storageEngine.isEmpty()) {
         builder->append("storageEngine", storageEngine);
     }
@@ -341,6 +347,10 @@ bool CollectionOptions::matchesStorageOptions(const CollectionOptions& other,
         return false;
     }
 
+    if (recordPreImages != other.recordPreImages) {
+        return false;
+    }
+
     if (temp != other.temp) {
         return false;
     }
diff --git a/src/mongo/db/catalog/collection_options.h b/src/mongo/db/catalog/collection_options.h
index f1644c69dd8..d5de2b5cb11 100644
--- a/src/mongo/db/catalog/collection_options.h
+++ b/src/mongo/db/catalog/collection_options.h
@@ -119,6 +119,7 @@ struct CollectionOptions {
     } autoIndexId = DEFAULT;
 
     bool temp = false;
+    bool recordPreImages = false;
 
     // Storage engine collection options. Always owned or empty.
     BSONObj storageEngine;