SERVER-58469 Add option "changeStreamsPreAndPostImages" to the "create" and "collMod" commands

24 files changed, 463 insertions, 21 deletions

diff --git a/buildscripts/idl/idl_check_compatibility.py b/buildscripts/idl/idl_check_compatibility.py
index 2da512358f0..d5b2d6c699b 100644
--- a/buildscripts/idl/idl_check_compatibility.py
+++ b/buildscripts/idl/idl_check_compatibility.py
@@ -97,6 +97,8 @@ ALLOW_ANY_TYPE_LIST: List[str] = [
     'saslStart-reply-payload',
     'saslContinue-param-payload',
     'saslContinue-reply-payload',
+    "create-param-changeStreamPreAndPostImages",
+    "collMod-param-changeStreamPreAndPostImages",
 
     # These commands (aggregate, find, update, delete, findAndModify, explain) might contain some
     # fields with type `any`. Currently, it's not possible to avoid the `any` type in those cases.
diff --git a/jstests/change_streams/lookup_pit_post_image.js b/jstests/change_streams/lookup_pit_post_image.js
index 28686df7810..8a7e426a4b1 100644
--- a/jstests/change_streams/lookup_pit_post_image.js
+++ b/jstests/change_streams/lookup_pit_post_image.js
@@ -17,7 +17,7 @@ if (!isChangeStreamPreAndPostImagesEnabled(db)) {
     assert.throwsWithCode(() => coll.watch([], {fullDocument: 'required'}), ErrorCodes.BadValue);
 
     jsTestLog(
-        'Skipping test because featureFlagChangeStreamsPreAndPostImages feature flag is not enabled');
+        'Skipping test because featureFlagChangeStreamPreAndPostImages feature flag is not enabled');
     return;
 }
 
diff --git a/jstests/libs/change_stream_util.js b/jstests/libs/change_stream_util.js
index e4c220743e0..e2d8f1280de 100644
--- a/jstests/libs/change_stream_util.js
+++ b/jstests/libs/change_stream_util.js
@@ -24,14 +24,13 @@ function isChangeStreamsOptimizationEnabled(db) {
 }
 
 /**
- * Returns true if feature flag 'featureFlagChangeStreamsPreAndPostImages' is enabled, false
+ * Returns true if feature flag 'featureFlagChangeStreamPreAndPostImages' is enabled, false
  * otherwise.
  */
 function isChangeStreamPreAndPostImagesEnabled(db) {
-    const getParam =
-        db.adminCommand({getParameter: 1, featureFlagChangeStreamsPreAndPostImages: 1});
-    return getParam.hasOwnProperty("featureFlagChangeStreamsPreAndPostImages") &&
-        getParam.featureFlagChangeStreamsPreAndPostImages.value;
+    const getParam = db.adminCommand({getParameter: 1, featureFlagChangeStreamPreAndPostImages: 1});
+    return getParam.hasOwnProperty("featureFlagChangeStreamPreAndPostImages") &&
+        getParam.featureFlagChangeStreamPreAndPostImages.value;
 }
 
 /**
diff --git a/jstests/libs/collection_drop_recreate.js b/jstests/libs/collection_drop_recreate.js
index 80044a5bc69..b6b3352b163 100644
--- a/jstests/libs/collection_drop_recreate.js
+++ b/jstests/libs/collection_drop_recreate.js
@@ -26,4 +26,4 @@ function assertCreateCollection(db, collName, collOpts) {
 function assertDropAndRecreateCollection(db, collName, collOpts) {
     assertDropCollection(db, collName);
     return assertCreateCollection(db, collName, collOpts);
-}
\ No newline at end of file
+}
diff --git a/jstests/libs/collection_options.js b/jstests/libs/collection_options.js
new file mode 100644
index 00000000000..356887a6c49
--- /dev/null
+++ b/jstests/libs/collection_options.js
@@ -0,0 +1,15 @@
+/**
+ * Asserts that the given collection option is set to true.
+ */
+function assertCollectionOptionIsEnabled(db, collName, option) {
+    const collectionInfos = db.getCollectionInfos({name: collName});
+    assert(collectionInfos[0].options[option] === true);
+}
+
+/**
+ * Asserts that the given collection option is absent.
+ */
+function assertCollectionOptionIsAbsent(db, collName, option) {
+    const collectionInfos = db.getCollectionInfos({name: collName});
+    assert(!collectionInfos[0].options.hasOwnProperty(option));
+}
diff --git a/jstests/multiVersion/change_streams_pre_and_post_images_upgrade_downgrade.js b/jstests/multiVersion/change_streams_pre_and_post_images_upgrade_downgrade.js
new file mode 100644
index 00000000000..bd8bb5b41f0
--- /dev/null
+++ b/jstests/multiVersion/change_streams_pre_and_post_images_upgrade_downgrade.js
@@ -0,0 +1,112 @@
+/**
+ * Verifies that it is possible to upgrade a replica set with collections with 'recordPreImages'
+ * option to use 'changeStreamPreAndPostImages' option, and to do a corresponding downgrade.
+ * @tags: [requires_fcv_51, featureFlagChangeStreamPreAndPostImages]
+ */
+(function() {
+'use strict';
+
+load("jstests/libs/collection_drop_recreate.js");  // For assertCreateCollection.
+load("jstests/libs/collection_options.js");        // For assertCollectionOptionIsEnabled,
+                                                   // assertCollectionOptionIsAbsent.
+load("jstests/multiVersion/libs/multi_rs.js");     // For upgradeSet.
+
+const collName = "test";
+
+function runTest(downgradeVersion) {
+    const downgradeFCV = binVersionToFCV(downgradeVersion);
+
+    const rst = new ReplSetTest({
+        nodes: 2,
+        nodeOptions: {binVersion: downgradeVersion},
+    });
+    rst.startSet();
+    rst.initiate();
+
+    // Create the collection with recorded pre-images enabled.
+    let testDB = rst.getPrimary().getDB(jsTestName());
+    assertCreateCollection(testDB, collName, {"recordPreImages": true});
+
+    // Upgrade the replica set.
+    rst.upgradeSet({binVersion: "latest"});
+    testDB = rst.getPrimary().getDB(jsTestName());
+
+    // Set the FCV to the latest.
+    testDB.adminCommand({setFeatureCompatibilityVersion: latestFCV});
+
+    // 'changeStreamPreAndPostImages' field must be absent and 'recordPreImages' field must be set
+    // to true.
+    assertCollectionOptionIsEnabled(testDB, collName, "recordPreImages");
+    assertCollectionOptionIsAbsent(testDB, collName, "changeStreamPreAndPostImages");
+
+    // Enable pre-/post-images for the collection with "recordPreImages" enabled.
+    assert.commandWorked(
+        testDB.runCommand({"collMod": collName, "changeStreamPreAndPostImages": true}));
+
+    // 'changeStreamPreAndPostImages' field must be set to true and 'recordPreImages' should be
+    // absent.
+    assertCollectionOptionIsAbsent(testDB, collName, "recordPreImages");
+    assertCollectionOptionIsEnabled(testDB, collName, "changeStreamPreAndPostImages");
+
+    // Set 'recordPreImages: true' to disable 'changeStreamPreAndPostImages' option.
+    assert.commandWorked(testDB.runCommand({"collMod": collName, "recordPreImages": true}));
+
+    // 'changeStreamPreAndPostImages' field must be absent and 'recordPreImages' should be set to
+    // true.
+    assertCollectionOptionIsEnabled(testDB, collName, "recordPreImages");
+    assertCollectionOptionIsAbsent(testDB, collName, "changeStreamPreAndPostImages");
+
+    // Downgrade the FCV.
+    testDB.adminCommand({setFeatureCompatibilityVersion: downgradeFCV});
+
+    // Verify that an attempt to set 'changeStreamPreAndPostImages' options fails for the downgrade
+    // version.
+    assert.commandFailedWithCode(
+        testDB.createCollection(collName, {"changeStreamPreAndPostImages": false}), 5846900);
+    assert.commandFailedWithCode(
+        testDB.runCommand({"collMod": collName, "changeStreamPreAndPostImages": false}), 5846901);
+
+    // Downgrade the cluster.
+    rst.upgradeSet({binVersion: downgradeVersion});
+
+    // Reset the db reference.
+    testDB = rst.getPrimary().getDB(jsTestName());
+
+    // 'changeStreamPreAndPostImages' field must be absent and 'recordPreImages' should be set to
+    // true.
+    assertCollectionOptionIsEnabled(testDB, collName, "recordPreImages");
+    assertCollectionOptionIsAbsent(testDB, collName, "changeStreamPreAndPostImages");
+
+    // Upgrade the replica set.
+    rst.upgradeSet({binVersion: "latest"});
+    testDB = rst.getPrimary().getDB(jsTestName());
+
+    // Set the FCV to the latest.
+    testDB.adminCommand({setFeatureCompatibilityVersion: latestFCV});
+
+    // Enable pre-/post-images for the collection with "recordPreImages" enabled.
+    assert.commandWorked(
+        testDB.runCommand({"collMod": collName, "changeStreamPreAndPostImages": true}));
+
+    // 'changeStreamPreAndPostImages' field must be set to true and 'recordPreImages' field must be
+    // absent.
+    assertCollectionOptionIsEnabled(testDB, collName, "changeStreamPreAndPostImages");
+    assertCollectionOptionIsAbsent(testDB, collName, "recordPreImages");
+
+    // Downgrade the FCV.
+    testDB.adminCommand({setFeatureCompatibilityVersion: downgradeFCV});
+
+    // Downgrading the cluster should fail, since unsupported field 'changeStreamPreAndPostImages'
+    // is set to true for the collection.
+    try {
+        rst.upgradeSet({binVersion: downgradeVersion});
+        assert(false);
+    } catch (exception) {
+        assert.eq(exception.returnCode, MongoRunner.EXIT_UNCAUGHT);
+    }
+
+    rst.stopSet();
+}
+
+runFeatureFlagMultiversionTest('featureFlagChangeStreamPreAndPostImages', runTest);
+})();
diff --git a/jstests/noPassthrough/change_streams_pre_and_post_images_in_create_and_collmod.js b/jstests/noPassthrough/change_streams_pre_and_post_images_in_create_and_collmod.js
new file mode 100644
index 00000000000..04c11b59bca
--- /dev/null
+++ b/jstests/noPassthrough/change_streams_pre_and_post_images_in_create_and_collmod.js
@@ -0,0 +1,129 @@
+/*
+ * Tests that the 'changeStreamPreAndPostImages' option is settable via the collMod and create
+ * commands. Also tests that this option cannot be set on collections in the 'local' or 'admin'
+ * databases as well as timeseries and view collections.
+ * @tags: [requires_fcv_51, featureFlagChangeStreamPreAndPostImages]
+ */
+(function() {
+'use strict';
+
+load("jstests/libs/collection_options.js");  // For assertCollectionOptionIsEnabled,
+                                             // assertCollectionOptionIsAbsent.
+
+const rsTest = new ReplSetTest({name: jsTestName(), nodes: 1});
+rsTest.startSet();
+rsTest.initiate();
+
+const dbName = 'testDB';
+const collName = 'changeStreamPreAndPostImages';
+const collName2 = 'changeStreamPreAndPostImages2';
+const collName3 = 'changeStreamPreAndPostImages3';
+const collName4 = 'changeStreamPreAndPostImages4';
+const viewName = "view";
+const createTimeseriesOptions = {
+    timeField: "a"
+};
+
+const primary = rsTest.getPrimary();
+const adminDB = primary.getDB("admin");
+const localDB = primary.getDB("local");
+const configDB = primary.getDB("config");
+const testDB = primary.getDB(dbName);
+
+// Check that we cannot set 'changeStreamPreAndPostImages' on the local or admin databases.
+for (const db of [localDB, adminDB, configDB]) {
+    assert.commandFailedWithCode(
+        db.runCommand({create: collName, changeStreamPreAndPostImages: true}),
+        ErrorCodes.InvalidOptions);
+
+    assert.commandWorked(db.runCommand({create: collName}));
+    assert.commandFailedWithCode(
+        db.runCommand({collMod: collName, changeStreamPreAndPostImages: true}),
+        ErrorCodes.InvalidOptions);
+}
+
+// Should be able to set the 'changeStreamPreAndPostImages' via create or collMod.
+assert.commandWorked(testDB.runCommand({create: collName, changeStreamPreAndPostImages: true}));
+assertCollectionOptionIsEnabled(testDB, collName, "changeStreamPreAndPostImages");
+
+assert.commandWorked(testDB.runCommand({create: collName2}));
+assert.commandWorked(testDB.runCommand({collMod: collName2, changeStreamPreAndPostImages: true}));
+assertCollectionOptionIsEnabled(testDB, collName2, "changeStreamPreAndPostImages");
+
+// Verify that setting collection options with 'collMod' command does not affect
+// 'changeStreamPreAndPostImages' option.
+assert.commandWorked(testDB.runCommand({"collMod": collName2, validationLevel: "off"}));
+assertCollectionOptionIsEnabled(testDB, collName2, "changeStreamPreAndPostImages");
+
+// Should successfully unset 'changeStreamPreAndPostImages' using the 'collMod' command.
+assert.commandWorked(testDB.runCommand({collMod: collName2, changeStreamPreAndPostImages: false}));
+assertCollectionOptionIsAbsent(testDB, collName2, "changeStreamPreAndPostImages");
+
+// Both 'recordPreImages' and 'changeStreamPreAndPostImages' may not be set to true at the same
+// time.
+assert.commandFailedWithCode(
+    testDB.runCommand(
+        {create: collName3, recordPreImages: true, changeStreamPreAndPostImages: true}),
+    ErrorCodes.InvalidOptions);
+
+assert.commandWorked(testDB.runCommand({create: collName3}));
+assert.commandFailedWithCode(
+    testDB.runCommand(
+        {collMod: collName3, recordPreImages: true, changeStreamPreAndPostImages: true}),
+    ErrorCodes.InvalidOptions);
+
+// Should set 'recordPreImages' to true and 'changeStreamPreAndPostImages' to false.
+assert.commandWorked(testDB.runCommand(
+    {collMod: collName3, recordPreImages: true, changeStreamPreAndPostImages: false}));
+assertCollectionOptionIsAbsent(testDB, collName3, "changeStreamPreAndPostImages");
+assertCollectionOptionIsEnabled(testDB, collName3, "recordPreImages");
+
+// Should set 'recordPreImages' to false and 'changeStreamPreAndPostImages' to true.
+assert.commandWorked(testDB.runCommand(
+    {collMod: collName3, recordPreImages: false, changeStreamPreAndPostImages: true}));
+assertCollectionOptionIsEnabled(testDB, collName3, "changeStreamPreAndPostImages");
+assertCollectionOptionIsAbsent(testDB, collName3, "recordPreImages");
+
+// Set 'recordPreImages: true' to disable 'changeStreamPreAndPostImages' option.
+assert.commandWorked(testDB.runCommand({"collMod": collName3, "recordPreImages": true}));
+
+// 'changeStreamPreAndPostImages' field must be absent and 'recordPreImages' should be set to
+// true.
+assertCollectionOptionIsEnabled(testDB, collName3, "recordPreImages");
+assertCollectionOptionIsAbsent(testDB, collName3, "changeStreamPreAndPostImages");
+
+// Enable pre-/post-images for the collection with 'changeStreamPreAndPostImages' enabled.
+// Set 'changeStreamPreAndPostImages: true' to disable 'recordPreImages' option.
+assert.commandWorked(
+    testDB.runCommand({"collMod": collName3, "changeStreamPreAndPostImages": true}));
+
+// 'changeStreamPreAndPostImages' field must be set to true and 'recordPreImages' should be
+// absent.
+assertCollectionOptionIsAbsent(testDB, collName3, "recordPreImages");
+assertCollectionOptionIsEnabled(testDB, collName3, "changeStreamPreAndPostImages");
+
+// Should fail to create a timeseries collection with 'changeStreamPreAndPostImages' set to true.
+assert.commandFailedWithCode(testDB.runCommand({
+    create: collName4,
+    timeseries: createTimeseriesOptions,
+    changeStreamPreAndPostImages: true
+}),
+                             ErrorCodes.InvalidOptions);
+
+assert.commandWorked(testDB.runCommand({create: collName4, timeseries: createTimeseriesOptions}));
+assert.commandFailedWithCode(
+    testDB.runCommand({collMod: collName4, changeStreamPreAndPostImages: true}),
+    ErrorCodes.InvalidOptions);
+assertCollectionOptionIsAbsent(testDB, collName4, "changeStreamPreAndPostImages");
+
+// Should fail to create a view with 'changeStreamPreAndPostImages' set to true.
+assert.commandFailedWithCode(
+    testDB.runCommand({create: viewName, viewOn: collName, changeStreamPreAndPostImages: true}),
+    ErrorCodes.InvalidOptions);
+assert.commandWorked(testDB.runCommand({create: viewName, viewOn: collName}));
+assert.commandFailedWithCode(
+    testDB.runCommand({collMod: viewName, changeStreamPreAndPostImages: true}),
+    ErrorCodes.InvalidOptions);
+
+rsTest.stopSet();
+}());
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 13d29084640..314e5f79e37 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -24,8 +24,10 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/query/collation/collator_interface',
+        '$BUILD_DIR/mongo/db/query/query_knobs',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
         '$BUILD_DIR/mongo/idl/basic_types',
+        '$BUILD_DIR/mongo/idl/feature_flag',
         '$BUILD_DIR/mongo/idl/idl_parser',
     ],
 )
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 593aa92b348..2a2a660dc38 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -112,6 +112,7 @@ struct CollModRequest {
     boost::optional<ValidationActionEnum> collValidationAction;
     boost::optional<ValidationLevelEnum> collValidationLevel;
     bool recordPreImages = false;
+    OptionalBool changeStreamPreAndPostImagesEnabled;
 };
 
 StatusWith<CollModRequest> parseCollModRequest(OperationContext* opCtx,
@@ -317,6 +318,24 @@ StatusWith<CollModRequest> parseCollModRequest(OperationContext* opCtx,
             }
 
             cmr.recordPreImages = e.trueValue();
+        } else if (fieldName == "changeStreamPreAndPostImages") {
+            if (nss.isTimeseriesBucketsCollection()) {
+                return {ErrorCodes::InvalidOptions,
+                        str::stream()
+                            << "option not supported on a timeseries collection: " << fieldName};
+            }
+
+            if (isView) {
+                return {ErrorCodes::InvalidOptions,
+                        str::stream() << "option not supported on a view: " << fieldName};
+            }
+
+            if (e.type() != mongo::Bool) {
+                return {ErrorCodes::InvalidOptions,
+                        "'changeStreamPreAndPostImages' option must be a boolean"};
+            }
+
+            cmr.changeStreamPreAndPostImagesEnabled = e.trueValue();
         } else if (fieldName == "expireAfterSeconds") {
             if (coll->getRecordStore()->keyFormat() != KeyFormat::String) {
                 return Status(ErrorCodes::InvalidOptions,
@@ -550,6 +569,20 @@ Status _collModInternal(OperationContext* opCtx,
 
         const CollectionOptions& oldCollOptions = coll->getCollectionOptions();
 
+        // TODO SERVER-58584: remove the feature flag.
+        if (feature_flags::gFeatureFlagChangeStreamPreAndPostImages.isEnabledAndIgnoreFCV()) {
+            // If 'changeStreamPreAndPostImagesEnabled' is set to true, 'recordPreImages' must be
+            // set to false. If 'recordPreImages' is set to true,
+            // 'changeStreamPreAndPostImagesEnabled' must be set to false.
+            if (cmrNew.changeStreamPreAndPostImagesEnabled) {
+                cmrNew.recordPreImages = false;
+            }
+
+            if (cmrNew.recordPreImages) {
+                cmrNew.changeStreamPreAndPostImagesEnabled = false;
+            }
+        }
+
         boost::optional<IndexCollModInfo> indexCollModInfo;
 
         // Handle collMod operation type appropriately.
@@ -638,6 +671,15 @@ Status _collModInternal(OperationContext* opCtx,
             coll.getWritableCollection()->setRecordPreImages(opCtx, cmrNew.recordPreImages);
         }
 
+        // TODO SERVER-58584: remove the feature flag.
+        if (feature_flags::gFeatureFlagChangeStreamPreAndPostImages.isEnabledAndIgnoreFCV() &&
+            cmrNew.changeStreamPreAndPostImagesEnabled.has_value() &&
+            cmrNew.changeStreamPreAndPostImagesEnabled !=
+                oldCollOptions.changeStreamPreAndPostImagesEnabled) {
+            coll.getWritableCollection()->setChangeStreamPreAndPostImages(
+                opCtx, cmrNew.changeStreamPreAndPostImagesEnabled);
+        }
+
         if (ts.isABSONObj()) {
             auto res = timeseries::applyTimeseriesOptionsModifications(*oldCollOptions.timeseries,
                                                                        ts.Obj());
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 511fa9d1e3c..66dc7be085d 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -515,6 +515,9 @@ public:
     virtual bool getRecordPreImages() const = 0;
     virtual void setRecordPreImages(OperationContext* opCtx, bool val) = 0;
 
+    virtual bool isChangeStreamPreAndPostImagesEnabled() const = 0;
+    virtual void setChangeStreamPreAndPostImages(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 1f5c892ff95..501fac64012 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -203,13 +203,29 @@ Status checkValidatorCanBeUsedOnNs(const BSONObj& validator,
     return Status::OK();
 }
 
-Status validatePreImageRecording(OperationContext* opCtx, const NamespaceString& ns) {
-    if (ns.db() == NamespaceString::kAdminDb || ns.db() == NamespaceString::kLocalDb) {
+Status validateIsNotInDbs(OperationContext* opCtx,
+                          const NamespaceString& ns,
+                          const std::vector<StringData>& disallowedDbs,
+                          StringData optionName) {
+    if (std::find(disallowedDbs.begin(), disallowedDbs.end(), ns.db()) != disallowedDbs.end()) {
         return {ErrorCodes::InvalidOptions,
-                str::stream() << "recordPreImages collection option is not supported on the "
+                str::stream() << optionName << " collection option is not supported on the "
                               << ns.db() << " database"};
     }
 
+    return Status::OK();
+}
+
+// Validates that the option is not used on admin or local db as well as not being used on shards
+// or config servers.
+Status validateRecordPreImagesOptionIsPermitted(OperationContext* opCtx,
+                                                const NamespaceString& ns) {
+    const auto validationStatus = validateIsNotInDbs(
+        opCtx, ns, {NamespaceString::kAdminDb, NamespaceString::kLocalDb}, "recordPreImages");
+    if (validationStatus != Status::OK()) {
+        return validationStatus;
+    }
+
     if (serverGlobalParams.clusterRole != ClusterRole::None) {
         return {
             ErrorCodes::InvalidOptions,
@@ -223,6 +239,28 @@ Status validatePreImageRecording(OperationContext* opCtx, const NamespaceString&
     return Status::OK();
 }
 
+// Validates that the option is not used on admin or local db as well as not being used on config
+// servers.
+Status validateChangeStreamPreAndPostImagesOptionIsPermitted(OperationContext* opCtx,
+                                                             const NamespaceString& ns) {
+    const auto validationStatus = validateIsNotInDbs(
+        opCtx,
+        ns,
+        {NamespaceString::kAdminDb, NamespaceString::kLocalDb, NamespaceString::kConfigDb},
+        "changeStreamPreAndPostImages");
+    if (validationStatus != Status::OK()) {
+        return validationStatus;
+    }
+
+    if (serverGlobalParams.clusterRole == ClusterRole::ConfigServer) {
+        return {
+            ErrorCodes::InvalidOptions,
+            "changeStreamPreAndPostImages collection option is not supported on config servers"};
+    }
+
+    return Status::OK();
+}
+
 bool isRetryableWrite(OperationContext* opCtx) {
     auto txnParticipant = TransactionParticipant::get(opCtx);
     const bool inMultiDocumentTransaction = txnParticipant && txnParticipant.transactionIsOpen();
@@ -397,7 +435,11 @@ void CollectionImpl::init(OperationContext* opCtx) {
     // Make sure to copy the action and level before parsing MatchExpression, since certain features
     // are not supported with certain combinations of action and level.
     if (collectionOptions.recordPreImages) {
-        uassertStatusOK(validatePreImageRecording(opCtx, _ns));
+        uassertStatusOK(validateRecordPreImagesOptionIsPermitted(opCtx, _ns));
+    }
+
+    if (collectionOptions.changeStreamPreAndPostImagesEnabled) {
+        uassertStatusOK(validateChangeStreamPreAndPostImagesOptionIsPermitted(opCtx, _ns));
     }
 
     // Store the result (OK / error) of parsing the validator, but do not enforce that the result is
@@ -1350,13 +1392,27 @@ bool CollectionImpl::getRecordPreImages() const {
 
 void CollectionImpl::setRecordPreImages(OperationContext* opCtx, bool val) {
     if (val) {
-        uassertStatusOK(validatePreImageRecording(opCtx, _ns));
+        uassertStatusOK(validateRecordPreImagesOptionIsPermitted(opCtx, _ns));
     }
 
     _writeMetadata(
         opCtx, [&](BSONCollectionCatalogEntry::MetaData& md) { md.options.recordPreImages = val; });
 }
 
+bool CollectionImpl::isChangeStreamPreAndPostImagesEnabled() const {
+    return _metadata->options.changeStreamPreAndPostImagesEnabled;
+}
+
+void CollectionImpl::setChangeStreamPreAndPostImages(OperationContext* opCtx, bool val) {
+    if (val) {
+        uassertStatusOK(validateChangeStreamPreAndPostImagesOptionIsPermitted(opCtx, _ns));
+    }
+
+    _writeMetadata(opCtx, [&](BSONCollectionCatalogEntry::MetaData& md) {
+        md.options.changeStreamPreAndPostImagesEnabled = val;
+    });
+}
+
 bool CollectionImpl::isCapped() const {
     return _shared->_isCapped;
 }
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index 0e926262417..a00eb19e212 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -315,6 +315,9 @@ public:
     bool getRecordPreImages() const final;
     void setRecordPreImages(OperationContext* opCtx, bool val) final;
 
+    bool isChangeStreamPreAndPostImagesEnabled() const final;
+    void setChangeStreamPreAndPostImages(OperationContext* opCtx, bool val) final;
+
     bool isTemporary() const final;
 
     bool isClustered() const final;
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index 7cb2b2b302a..d7091f3622e 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -258,6 +258,14 @@ public:
         std::abort();
     }
 
+    bool isChangeStreamPreAndPostImagesEnabled() const {
+        std::abort();
+    }
+
+    void setChangeStreamPreAndPostImages(OperationContext* opCtx, bool val) {
+        std::abort();
+    }
+
     bool isCapped() const {
         return false;
     }
diff --git a/src/mongo/db/catalog/collection_options.cpp b/src/mongo/db/catalog/collection_options.cpp
index 2eadc1b11f8..012f6f6c147 100644
--- a/src/mongo/db/catalog/collection_options.cpp
+++ b/src/mongo/db/catalog/collection_options.cpp
@@ -39,6 +39,7 @@
 #include "mongo/db/commands/create_gen.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/query/collation/collator_interface.h"
+#include "mongo/db/query/query_feature_flags_gen.h"
 #include "mongo/idl/command_generic_argument.h"
 #include "mongo/util/str.h"
 
@@ -133,6 +134,8 @@ StatusWith<CollectionOptions> CollectionOptions::parse(const BSONObj& options, P
             collectionOptions.temp = e.trueValue();
         } else if (fieldName == "recordPreImages") {
             collectionOptions.recordPreImages = e.trueValue();
+        } else if (fieldName == "changeStreamPreAndPostImages") {
+            collectionOptions.changeStreamPreAndPostImagesEnabled = e.trueValue();
         } else if (fieldName == "storageEngine") {
             if (e.type() != mongo::Object) {
                 return {ErrorCodes::TypeMismatch, "'storageEngine' must be a document"};
@@ -300,6 +303,9 @@ CollectionOptions CollectionOptions::fromCreateCommand(const CreateCommand& cmd)
     if (auto recordPreImages = cmd.getRecordPreImages()) {
         options.recordPreImages = *recordPreImages;
     }
+    if (cmd.getChangeStreamPreAndPostImages().has_value()) {
+        options.changeStreamPreAndPostImagesEnabled = cmd.getChangeStreamPreAndPostImages();
+    }
     if (auto timeseries = cmd.getTimeseries()) {
         options.timeseries = std::move(*timeseries);
     }
@@ -351,6 +357,13 @@ void CollectionOptions::appendBSON(BSONObjBuilder* builder,
         builder->appendBool(CreateCommand::kRecordPreImagesFieldName, true);
     }
 
+    // TODO SERVER-58584: remove the feature flag.
+    if (feature_flags::gFeatureFlagChangeStreamPreAndPostImages.isEnabledAndIgnoreFCV() &&
+        changeStreamPreAndPostImagesEnabled &&
+        shouldAppend(CreateCommand::kChangeStreamPreAndPostImagesFieldName)) {
+        builder->appendBool(CreateCommand::kChangeStreamPreAndPostImagesFieldName, true);
+    }
+
     if (!storageEngine.isEmpty() && shouldAppend(CreateCommand::kStorageEngineFieldName)) {
         builder->append(CreateCommand::kStorageEngineFieldName, storageEngine);
     }
@@ -425,6 +438,10 @@ bool CollectionOptions::matchesStorageOptions(const CollectionOptions& other,
         return false;
     }
 
+    if (changeStreamPreAndPostImagesEnabled != other.changeStreamPreAndPostImagesEnabled) {
+        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 7a901f79c28..ab7aa9e2eae 100644
--- a/src/mongo/db/catalog/collection_options.h
+++ b/src/mongo/db/catalog/collection_options.h
@@ -131,6 +131,11 @@ struct CollectionOptions {
     bool temp = false;
     bool recordPreImages = false;
 
+    // If set to true, stores the pre-images of the documents affected by update and delete
+    // operations in a dedicated collection, that will be used for reading data via changeStreams.
+    // Can not be set to true together with 'recordPreImages' (mutually exclusive).
+    bool changeStreamPreAndPostImagesEnabled = false;
+
     // Storage engine collection options. Always owned or empty.
     BSONObj storageEngine;
 
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index b4bec1e15d6..302271ec8e7 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -93,6 +93,11 @@ Status _createView(OperationContext* opCtx,
                           str::stream() << "Not primary while creating collection " << nss);
         }
 
+        if (collectionOptions.changeStreamPreAndPostImagesEnabled) {
+            return Status(ErrorCodes::InvalidOptions,
+                          "option not supported on a view: changeStreamPreAndPostImages");
+        }
+
         _createSystemDotViewsIfNecessary(opCtx, db);
 
         WriteUnitOfWork wunit(opCtx);
diff --git a/src/mongo/db/coll_mod.idl b/src/mongo/db/coll_mod.idl
index 54671dcde46..0a71b5522d1 100644
--- a/src/mongo/db/coll_mod.idl
+++ b/src/mongo/db/coll_mod.idl
@@ -136,6 +136,10 @@ commands:
                               document in the oplog"
                 optional: true
                 type: safeBool
+            changeStreamPreAndPostImages:
+                description: "Determines whether pre- and post-images of documents are available in the change stream events."
+                type: optionalBool
+                unstable: true
             expireAfterSeconds:
                 description: "The number of seconds after which old data should be deleted. This can
                               be disabled by passing in 'off' as a value"
diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index b23d232c07f..712dd0d573e 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -275,8 +275,11 @@ env.Library(
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/auth/authprivilege',
         '$BUILD_DIR/mongo/db/catalog/collection_options',
+        '$BUILD_DIR/mongo/db/query/query_knobs',
+        '$BUILD_DIR/mongo/db/server_options',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_options',
         '$BUILD_DIR/mongo/idl/basic_types',
+        '$BUILD_DIR/mongo/idl/feature_flag',
         '$BUILD_DIR/mongo/idl/idl_parser',
     ],
 )
@@ -730,6 +733,7 @@ env.CppUnitTest(
         "$BUILD_DIR/mongo/db/service_context_d_test_fixture",
         '$BUILD_DIR/mongo/idl/idl_parser',
         "core",
+        "create_command",
         "mongod",
         "servers",
         "standalone",
diff --git a/src/mongo/db/commands/create.idl b/src/mongo/db/commands/create.idl
index 5314b402650..81491a0ad76 100644
--- a/src/mongo/db/commands/create.idl
+++ b/src/mongo/db/commands/create.idl
@@ -159,6 +159,10 @@ commands:
                 type: safeBool
                 optional: true
                 unstable: true
+            changeStreamPreAndPostImages:
+                description: "Determines whether point-in-time pre- and post-images are available for change streams opened on this collection."
+                type: optionalBool
+                unstable: true
             timeseries:
                 description: "The options to create the time-series collection with."
                 type: TimeseriesOptions
diff --git a/src/mongo/db/commands/create_command.cpp b/src/mongo/db/commands/create_command.cpp
index d94c77f5268..038f7e607be 100644
--- a/src/mongo/db/commands/create_command.cpp
+++ b/src/mongo/db/commands/create_command.cpp
@@ -36,7 +36,9 @@
 #include "mongo/db/catalog/index_key_validate.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/create_gen.h"
+#include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/query/collation/collator_factory_interface.h"
+#include "mongo/db/query/query_feature_flags_gen.h"
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
 #include "mongo/db/timeseries/timeseries_constants.h"
@@ -62,6 +64,7 @@ constexpr auto kCreateCommandHelp =
     "  viewOn: <string: name of source collection or view>,\n"
     "  pipeline: <array<object>: aggregation pipeline stage>,\n"
     "  collation: <document: default collation for the collection or view>,\n"
+    "  changeStreamPreAndPostImages: <bool: pre- and post-images for change streams enabled>,\n"
     "  writeConcern: <document: write concern expression for the operation>]\n"
     "}"_sd;
 
@@ -258,6 +261,19 @@ public:
                 cmd.setIdIndex(idIndexSpec);
             }
 
+            if (feature_flags::gFeatureFlagChangeStreamPreAndPostImages.isEnabled(
+                    serverGlobalParams.featureCompatibility)) {
+                const auto isRecordPreImagesEnabled = cmd.getRecordPreImages().get_value_or(false);
+                uassert(ErrorCodes::InvalidOptions,
+                        "recordPreImages and changeStreamPreAndPostImages can not be set to true "
+                        "simultaneously",
+                        !(cmd.getChangeStreamPreAndPostImages() && isRecordPreImagesEnabled));
+            } else {
+                uassert(5846900,
+                        "BSON field 'changeStreamPreAndPostImages' is an unknown field.",
+                        !cmd.getChangeStreamPreAndPostImages().has_value());
+            }
+
             OperationShardingState::ScopedAllowImplicitCollectionCreate_UNSAFE
                 unsafeCreateCollection(opCtx);
             uassertStatusOK(createCollection(opCtx, cmd.getNamespace(), cmd));
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index dcff0d91c02..debb9c109ed 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -56,6 +56,7 @@
 #include "mongo/db/coll_mod_gen.h"
 #include "mongo/db/coll_mod_reply_validation.h"
 #include "mongo/db/commands.h"
+#include "mongo/db/commands/feature_compatibility_version.h"
 #include "mongo/db/commands/server_status.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
@@ -78,6 +79,7 @@
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/query/internal_plans.h"
+#include "mongo/db/query/query_feature_flags_gen.h"
 #include "mongo/db/query/query_planner.h"
 #include "mongo/db/read_concern.h"
 #include "mongo/db/repl/optime.h"
@@ -143,6 +145,7 @@ std::unique_ptr<CollMod> makeTimeseriesBucketsCollModCommand(OperationContext* o
     cmd->setViewOn(origCmd.getViewOn());
     cmd->setPipeline(origCmd.getPipeline());
     cmd->setRecordPreImages(origCmd.getRecordPreImages());
+    cmd->setChangeStreamPreAndPostImages(origCmd.getChangeStreamPreAndPostImages());
     cmd->setExpireAfterSeconds(origCmd.getExpireAfterSeconds());
     cmd->setTimeseries(origCmd.getTimeseries());
 
@@ -608,6 +611,19 @@ public:
                               BSONObjBuilder& result) final {
         const auto* cmd = &requestParser.request();
 
+        if (feature_flags::gFeatureFlagChangeStreamPreAndPostImages.isEnabled(
+                serverGlobalParams.featureCompatibility)) {
+            const auto isRecordPreImagesEnabled = cmd->getRecordPreImages().get_value_or(false);
+            uassert(ErrorCodes::InvalidOptions,
+                    "recordPreImages and changeStreamPreAndPostImages can not be set to true "
+                    "simultaneously",
+                    !(cmd->getChangeStreamPreAndPostImages() && isRecordPreImagesEnabled));
+        } else {
+            uassert(5846901,
+                    "BSON field 'changeStreamPreAndPostImages' is an unknown field.",
+                    !cmd->getChangeStreamPreAndPostImages().has_value());
+        }
+
         // If the target namespace refers to a time-series collection, we will redirect the
         // collection modification request to the underlying bucket collection.
         // Aliasing collMod on a time-series collection in this manner has a few advantages:
diff --git a/src/mongo/db/pipeline/document_source_change_stream.cpp b/src/mongo/db/pipeline/document_source_change_stream.cpp
index 496091eee1c..ff9a8451eec 100644
--- a/src/mongo/db/pipeline/document_source_change_stream.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream.cpp
@@ -360,7 +360,7 @@ void DocumentSourceChangeStream::assertIsLegalSpecification(
             !(shouldAddPreImage && (expCtx->inMongos || expCtx->needsMerge)));
 
     // TODO SERVER-58584: remove the feature flag.
-    if (!feature_flags::gFeatureFlagChangeStreamsPreAndPostImages.isEnabled(
+    if (!feature_flags::gFeatureFlagChangeStreamPreAndPostImages.isEnabled(
             serverGlobalParams.featureCompatibility)) {
         uassert(ErrorCodes::BadValue,
                 str::stream() << "Specified value '"
diff --git a/src/mongo/db/pipeline/document_source_change_stream_test.cpp b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
index e42be1ab250..67a832834c8 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
@@ -450,8 +450,8 @@ bool getCSRewriteFeatureFlagValue() {
     return feature_flags::gFeatureFlagChangeStreamsRewrite.isEnabledAndIgnoreFCV();
 }
 
-bool isChangeStreamsPreAndPostImagesEnabled() {
-    return feature_flags::gFeatureFlagChangeStreamsPreAndPostImages.isEnabledAndIgnoreFCV();
+bool isChangeStreamPreAndPostImagesEnabled() {
+    return feature_flags::gFeatureFlagChangeStreamPreAndPostImages.isEnabledAndIgnoreFCV();
 }
 
 /**
@@ -589,8 +589,8 @@ TEST_F(ChangeStreamStageTest, ShouldRejectUnsupportedFullDocumentOption) {
         // TODO SERVER-58584: remove the feature flag.
         {
             RAIIServerParameterControllerForTest controller(
-                "featureFlagChangeStreamsPreAndPostImages", false);
-            ASSERT_FALSE(isChangeStreamsPreAndPostImagesEnabled());
+                "featureFlagChangeStreamPreAndPostImages", false);
+            ASSERT_FALSE(isChangeStreamPreAndPostImagesEnabled());
 
             // 'DSChangeStream' is not allowed to be instantiated with new document modes when
             // pre-/post-images feature flag is disabled.
@@ -600,8 +600,8 @@ TEST_F(ChangeStreamStageTest, ShouldRejectUnsupportedFullDocumentOption) {
         }
         {
             RAIIServerParameterControllerForTest controller(
-                "featureFlagChangeStreamsPreAndPostImages", true);
-            ASSERT(isChangeStreamsPreAndPostImagesEnabled());
+                "featureFlagChangeStreamPreAndPostImages", true);
+            ASSERT(isChangeStreamPreAndPostImagesEnabled());
 
             // 'DSChangeStream' is allowed to be instantiated with new document modes when
             // pre-/post-images feature flag is enabled.
diff --git a/src/mongo/db/query/query_feature_flags.idl b/src/mongo/db/query/query_feature_flags.idl
index a8d2c7c2099..86e5e9f68ec 100644
--- a/src/mongo/db/query/query_feature_flags.idl
+++ b/src/mongo/db/query/query_feature_flags.idl
@@ -71,9 +71,9 @@ feature_flags:
       cpp_varname: gFeatureFlagShardedLookup 
       default: false
 
-    featureFlagChangeStreamsPreAndPostImages:
+    featureFlagChangeStreamPreAndPostImages:
       description: "Feature flag for allowing usage of point-in-time pre- and post-images of documents in change streams"
-      cpp_varname: gFeatureFlagChangeStreamsPreAndPostImages
+      cpp_varname: gFeatureFlagChangeStreamPreAndPostImages
       default: false
 
     featureFlagSearchMeta: