SERVER-60919 Remove oplog based change stream pre-image storage

37 files changed, 121 insertions, 1330 deletions

diff --git a/buildscripts/idl/idl_check_compatibility.py b/buildscripts/idl/idl_check_compatibility.py
index dc6e44422a8..2ea550f8dc6 100644
--- a/buildscripts/idl/idl_check_compatibility.py
+++ b/buildscripts/idl/idl_check_compatibility.py
@@ -203,8 +203,6 @@ IGNORE_STABLE_TO_UNSTABLE_LIST: List[str] = [
     # can return one or more cursors. Multiple cursors are covered under the 'cursors' field.
     'find-reply-cursor',
     'aggregate-reply-cursor',
-    # The 'recordPreImages' field is only used by Realm and is not documented to users.
-    'collMod-param-recordPreImages',
     # The 'ignoreUnknownIndexOptions' field is for internal use only and is not documented to users.
     'createIndexes-param-ignoreUnknownIndexOptions',
     # The 'runtimeConstants' field is a legacy field for internal use only and is not documented to
@@ -1059,6 +1057,9 @@ def check_command_params_or_type_struct_fields(
     # We allow collMod isTimeseriesNamespace parameter to be removed because it's implicitly
     # added from mongos and not documented in the API.
     allow_list += ["collMod-param-isTimeseriesNamespace"]
+    # We allow collMod "recordPreImages" parameter to be removed because it was incorrectly marked as stable
+    # in 5.0.x versions.
+    allow_list += ["collMod-param-recordPreImages"]
 
     for old_field in old_struct_fields or []:
         new_field_exists = False
diff --git a/jstests/change_streams/change_stream_pre_image_lookup_whole_db_whole_cluster.js b/jstests/change_streams/change_stream_pre_image_lookup_whole_db_whole_cluster.js
index 4087ab65f73..f2ca07b1c54 100644
--- a/jstests/change_streams/change_stream_pre_image_lookup_whole_db_whole_cluster.js
+++ b/jstests/change_streams/change_stream_pre_image_lookup_whole_db_whole_cluster.js
@@ -66,7 +66,7 @@ assert.throwsWithCode(function() {
 
     return assert.soon(() => wholeDBStream.hasNext() &&
                            wholeDBStream.next().documentKey._id === "last_change_sentinel");
-}, [ErrorCodes.NoMatchingDocument, 51770]);
+}, ErrorCodes.NoMatchingDocument);
 
 // Confirm that attempting to open a whole-cluster stream on with mode "required" fails.
 assert.throwsWithCode(function() {
@@ -78,7 +78,7 @@ assert.throwsWithCode(function() {
 
     return assert.soon(() => wholeClusterStream.hasNext() &&
                            wholeClusterStream.next().documentKey._id == "last_change_sentinel");
-}, [ErrorCodes.NoMatchingDocument, 51770]);
+}, ErrorCodes.NoMatchingDocument);
 
 // However, if we open a whole-db or whole-cluster stream that filters for only the namespace with
 // pre-images, then the cursor can proceed. This is because the $match gets moved ahead of the
diff --git a/jstests/change_streams/delete_in_txn_produces_correct_doc_key.js b/jstests/change_streams/delete_in_txn_produces_correct_doc_key.js
index 32cf184faf1..0225bf4d0f5 100644
--- a/jstests/change_streams/delete_in_txn_produces_correct_doc_key.js
+++ b/jstests/change_streams/delete_in_txn_produces_correct_doc_key.js
@@ -24,11 +24,6 @@ function testDeleteInMultiDocTxn({collName, deleteCommand, expectedChanges}) {
     // Initialize the collection.
     const coll = assertDropAndRecreateCollection(db, collName);
 
-    // Enable the 'recordPreImages' flag on the collection. This allows us to verify that the full
-    // document is not written to the 'documentKey' field even when we know it is available to the
-    // oplog writer during the delete operation.
-    assert.commandWorked(db.runCommand({collMod: collName, recordPreImages: true}));
-
     assert.commandWorked(coll.insertMany([
         {_id: 1, a: 0, fullDoc: "It's a full document!"},
         {_id: 2, a: 0},
diff --git a/jstests/change_streams/lookup_pre_image.js b/jstests/change_streams/lookup_pre_image.js
index a3357a3a994..8337052d9a9 100644
--- a/jstests/change_streams/lookup_pre_image.js
+++ b/jstests/change_streams/lookup_pre_image.js
@@ -122,7 +122,7 @@ latestChange.fullDocumentBeforeChange = null;
 assert.docEq(latestChange, cst.getOneChange(csPreImageWhenAvailableCursor));
 // ... but the "required" cursor throws an exception.
 assert.throwsWithCode(() => cst.getOneChange(csPreImageRequiredCursor),
-                      [ErrorCodes.NoMatchingDocument, 51770]);
+                      ErrorCodes.NoMatchingDocument);
 
 // Test pre-image lookup for an op-style update operation.
 assert.commandWorked(coll.update({_id: "y"}, {$set: {foo: "baz"}}));
diff --git a/jstests/core/timeseries/timeseries_collmod.js b/jstests/core/timeseries/timeseries_collmod.js
index 7c919878bad..b7e04f7e43b 100644
--- a/jstests/core/timeseries/timeseries_collmod.js
+++ b/jstests/core/timeseries/timeseries_collmod.js
@@ -47,10 +47,6 @@ assert.commandFailedWithCode(db.runCommand({"collMod": collName, "validationActi
 assert.commandFailedWithCode(db.runCommand({"collMod": collName, "viewOn": "foo", "pipeline": []}),
                              ErrorCodes.InvalidOptions);
 
-// Tries to set 'recordPreImages' for a time-series collection.
-assert.commandFailedWithCode(db.runCommand({"collMod": collName, "recordPreImages": true}),
-                             ErrorCodes.InvalidOptions);
-
 // Successfully sets 'expireAfterSeconds' for a time-series collection.
 assert.commandWorked(db.runCommand({"collMod": collName, "expireAfterSeconds": 60}));
 
diff --git a/jstests/noPassthrough/change_stream_preimages_standalone_mode.js b/jstests/noPassthrough/change_stream_preimages_standalone_mode.js
index c36d03433e6..e4fe15f43ba 100644
--- a/jstests/noPassthrough/change_stream_preimages_standalone_mode.js
+++ b/jstests/noPassthrough/change_stream_preimages_standalone_mode.js
@@ -1,6 +1,6 @@
 /**
- * Test that nodes are able to startup with 'recordPreImages' and 'changeStreamPreAndPostImages'
- * options set in collection metadata and no pre-images are recorded while being in standalone mode.
+ * Test that nodes are able to startup with 'changeStreamPreAndPostImages' options set in collection
+ * metadata and no pre-images are recorded while being in standalone mode.
  *
  * @tags: [
  *   # Servers are restarted in this test and the data must be retained.
@@ -138,22 +138,6 @@ function testStandaloneMode({
     rst.stopSet();
 }
 
-// Run the test for 'recordPreImages' option.
-testStandaloneMode({
-    collectionOptions: {recordPreImages: true},
-    assertPreImagesRecordingEnabledFunc: (db, collName) => {
-        assert.eq(findCollectionInfo(db, collName).options.recordPreImages, true);
-    },
-    assertPreImagesRecordedFunc: (db, writerOps) => {
-        const writtenOplogEntries = oplogEntriesForOps(db, writerOps);
-        assert.gt(writtenOplogEntries.length, 0, writtenOplogEntries);
-    },
-    assertNoPreImagesRecordedFunc: (db, writerOps) => {
-        const writtenOplogEntries = oplogEntriesForOps(db, writerOps);
-        assert.eq(writtenOplogEntries.length, 0, writtenOplogEntries);
-    }
-});
-
 // Run the test for 'changeStreamPreAndPostImages' option.
 testStandaloneMode({
     collectionOptions: {changeStreamPreAndPostImages: {enabled: true}},
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
index cd7b5e16bd7..5c57db9df73 100644
--- 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
@@ -26,7 +26,6 @@ const collName = 'changeStreamPreAndPostImages';
 const collName2 = 'changeStreamPreAndPostImages2';
 const collName3 = 'changeStreamPreAndPostImages3';
 const collName4 = 'changeStreamPreAndPostImages4';
-const collName5 = 'changeStreamPreAndPostImages5';
 const viewName = "view";
 
 const primary = rsTest.getPrimary();
@@ -67,55 +66,7 @@ assert.commandWorked(
     testDB.runCommand({collMod: collName2, changeStreamPreAndPostImages: {enabled: false}}));
 assertChangeStreamPreAndPostImagesCollectionOptionIsAbsent(testDB, collName2);
 
-// Both 'recordPreImages' and 'changeStreamPreAndPostImages' may not be enabled at the same
-// time.
-assert.commandFailedWithCode(
-    testDB.runCommand(
-        {collMod: collName3, recordPreImages: true, changeStreamPreAndPostImages: {enabled: true}}),
-    ErrorCodes.InvalidOptions);
-
 assert.commandWorked(testDB.runCommand({create: collName3}));
-assert.commandFailedWithCode(
-    testDB.runCommand(
-        {collMod: collName3, recordPreImages: true, changeStreamPreAndPostImages: {enabled: true}}),
-    ErrorCodes.InvalidOptions);
-
-// Should set 'recordPreImages' to true and disable 'changeStreamPreAndPostImages' option.
-assert.commandWorked(testDB.runCommand(
-    {collMod: collName3, recordPreImages: true, changeStreamPreAndPostImages: {enabled: false}}));
-assertChangeStreamPreAndPostImagesCollectionOptionIsAbsent(testDB, collName3);
-assertCollectionOptionIsEnabled(testDB, collName3, "recordPreImages");
-
-// Should set 'recordPreImages' to false and enable 'changeStreamPreAndPostImages'.
-assert.commandWorked(testDB.runCommand(
-    {collMod: collName3, recordPreImages: false, changeStreamPreAndPostImages: {enabled: true}}));
-assertChangeStreamPreAndPostImagesCollectionOptionIsEnabled(testDB, collName3);
-assertCollectionOptionIsAbsent(testDB, collName3, "recordPreImages");
-
-// Set 'recordPreImages: true' to disable 'changeStreamPreAndPostImages' option.
-assert.commandWorked(testDB.runCommand({"collMod": collName3, "recordPreImages": true}));
-
-// 'changeStreamPreAndPostImages' option must be absent and 'recordPreImages' should be set to
-// true.
-assertCollectionOptionIsEnabled(testDB, collName3, "recordPreImages");
-assertChangeStreamPreAndPostImagesCollectionOptionIsAbsent(testDB, collName3);
-
-// Enable pre-/post-images for the collection with 'changeStreamPreAndPostImages' enabled.
-// Set 'changeStreamPreAndPostImages: {enabled: true}' to disable 'recordPreImages' option.
-assert.commandWorked(
-    testDB.runCommand({collMod: collName3, changeStreamPreAndPostImages: {enabled: true}}));
-
-// 'changeStreamPreAndPostImages' option must be enabled and 'recordPreImages' should be
-// absent.
-assertCollectionOptionIsAbsent(testDB, collName3, "recordPreImages");
-assertChangeStreamPreAndPostImagesCollectionOptionIsEnabled(testDB, collName3);
-
-// Should set 'recordPreImages' to true and disable 'changeStreamPreAndPostImages' option.
-assert.commandWorked(testDB.runCommand(
-    {create: collName4, recordPreImages: true, changeStreamPreAndPostImages: {enabled: false}}));
-assert.commandWorked(testDB.runCommand(
-    {collMod: collName4, recordPreImages: true, changeStreamPreAndPostImages: {enabled: false}}));
-assertChangeStreamPreAndPostImagesCollectionOptionIsAbsent(testDB, collName4);
 
 // Should fail to create a view with enabled 'changeStreamPreAndPostImages' option.
 assert.commandFailedWithCode(
@@ -130,16 +81,16 @@ assert.commandFailedWithCode(
 // Should fail to create a timeseries collection with enabled 'changeStreamPreAndPostImages'
 // option.
 assert.commandFailedWithCode(testDB.runCommand({
-    create: collName5,
+    create: collName4,
     timeseries: {timeField: 'time'},
     changeStreamPreAndPostImages: {enabled: true}
 }),
                              ErrorCodes.InvalidOptions);
 
 // Should fail to enable 'changeStreamPreAndPostImages' option on a timeseries collection.
-assert.commandWorked(testDB.runCommand({create: collName5, timeseries: {timeField: 'time'}}));
+assert.commandWorked(testDB.runCommand({create: collName4, timeseries: {timeField: 'time'}}));
 assert.commandFailedWithCode(
-    testDB.runCommand({collMod: collName5, changeStreamPreAndPostImages: {enabled: true}}),
+    testDB.runCommand({collMod: collName4, changeStreamPreAndPostImages: {enabled: true}}),
     ErrorCodes.InvalidOptions);
 
 rsTest.stopSet();
diff --git a/jstests/noPassthrough/recordPreImages_in_create_and_collmod.js b/jstests/noPassthrough/recordPreImages_in_create_and_collmod.js
deleted file mode 100644
index 4807da95268..00000000000
--- a/jstests/noPassthrough/recordPreImages_in_create_and_collmod.js
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Tests the 'recordPreImage' flag is settable via the collMode and create commands. Also tests that
- * this flag cannot be set on collections in the 'local' or 'admin' databases.
- *
- * @tags: [requires_replication]
- */
-(function() {
-'use strict';
-
-const findCollectionInfo = function(collName) {
-    var all = testDB.getCollectionInfos();
-    if (all.length == 0) {
-        return {};
-    }
-    all = all.filter(function(z) {
-        return z.name == collName;
-    });
-    assert.eq(all.length, 1);
-    return all[0];
-};
-
-let rsTest = new ReplSetTest({nodes: 1});
-rsTest.startSet();
-rsTest.initiate();
-
-const dbName = 'testDB';
-const collName = 'recordPreImageColl';
-const collName2 = 'recordPreImageColl2';
-
-let primary = rsTest.getPrimary();
-let adminDB = primary.getDB("admin");
-let localDB = primary.getDB("local");
-let testDB = primary.getDB(dbName);
-
-// Check that we cannot set recordPreImages on the local or admin databases.
-assert.commandFailedWithCode(adminDB.runCommand({create: collName, recordPreImages: true}),
-                             ErrorCodes.InvalidOptions);
-assert.commandFailedWithCode(localDB.runCommand({create: collName, recordPreImages: true}),
-                             ErrorCodes.InvalidOptions);
-
-// We should be able to set the recordPreImages flag via create or collMod.
-assert.commandWorked(testDB.runCommand({create: collName, recordPreImages: true}));
-assert.eq(findCollectionInfo(collName).options.recordPreImages, true);
-
-assert.commandWorked(testDB.runCommand({create: collName2}));
-assert.commandWorked(testDB.runCommand({collMod: collName2, recordPreImages: true}));
-assert.eq(findCollectionInfo(collName2).options.recordPreImages, true);
-
-// Test that the recordPreImages flag can be unset successfully using the 'collMod' command.
-assert.commandWorked(testDB.runCommand({collMod: collName, recordPreImages: false}));
-assert.eq(findCollectionInfo(collName).options, {});
-
-rsTest.stopSet();
-}());
diff --git a/jstests/noPassthrough/store_retryable_find_and_modify_images_in_side_collection.js b/jstests/noPassthrough/store_retryable_find_and_modify_images_in_side_collection.js
index 62ea7884911..fde47c8956d 100644
--- a/jstests/noPassthrough/store_retryable_find_and_modify_images_in_side_collection.js
+++ b/jstests/noPassthrough/store_retryable_find_and_modify_images_in_side_collection.js
@@ -20,21 +20,9 @@ function checkOplogEntry(entry, lsid, txnNum, stmtId, prevTs, retryImageArgs) {
     assert.eq(prevTs.getTime(), oplogPrevTs.getTime(), entry);
 
     if (retryImageArgs.needsRetryImage) {
-        if (retryImageArgs.imageKind === "preImage" && retryImageArgs.preImageRecordingEnabled) {
-            assert(!entry.hasOwnProperty("needsRetryImage"));
-            assert(entry.hasOwnProperty("preImageOpTime"));
-            assert(!entry.hasOwnProperty("postImageOpTime"));
-        } else {
-            assert.eq(retryImageArgs.imageKind, entry.needsRetryImage, entry);
-            if (retryImageArgs.preImageRecordingEnabled) {
-                assert(entry.hasOwnProperty("preImageOpTime"), entry);
-            }
-        }
+        assert.eq(retryImageArgs.imageKind, entry.needsRetryImage, entry);
     } else {
         assert(!entry.hasOwnProperty("needsRetryImage"));
-        if (retryImageArgs.preImageRecordingEnabled) {
-            assert(entry.hasOwnProperty("preImageOpTime"));
-        }
     }
 }
 
@@ -109,7 +97,7 @@ function checkProfilingLogs(primary) {
     assert.eq(0, configProfileDocs.length);
 }
 
-function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEnabled) {
+function runTests(lsid, mainConn, primary, secondary, docId) {
     const setParam = {setParameter: 1, storeFindAndModifyImagesInSideCollection: true};
     primary.adminCommand(setParam);
 
@@ -120,11 +108,6 @@ function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEn
 
     const oplog = primary.getDB('local').oplog.rs;
 
-    if (preImageRecordingEnabled) {
-        assert.commandWorked(
-            mainConn.getDB('test').runCommand({create: "user", recordPreImages: true}));
-    }
-
     // ////////////////////////////////////////////////////////////////////////
     // // Test findAndModify command (upsert)
 
@@ -165,23 +148,15 @@ function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEn
     // and values.
     const expectedWriteTs = Timestamp(0, 0);
     const expectedStmtId = 0;
-    let retryArgs = {
-        needsRetryImage: true,
-        imageKind: "preImage",
-        preImageRecordingEnabled: preImageRecordingEnabled
-    };
+    let retryArgs = {needsRetryImage: true, imageKind: "preImage"};
     checkOplogEntry(updateOp, lsid, txnNumber, expectedStmtId, expectedWriteTs, retryArgs);
     checkSessionCatalog(primary, lsid, txnNumber, updateOp.ts);
     checkSessionCatalog(secondary, lsid, txnNumber, updateOp.ts);
-    if (!preImageRecordingEnabled) {
-        const sessionInfo = {sessionId: lsid, txnNum: txnNumber};
-        checkImageCollection(primary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
-        checkImageCollection(secondary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
-    } else {
-        // The preImage should be stored in the oplog.
-        const preImage = oplog.findOne({ns: 'test.user', op: 'n', ts: updateOp.preImageOpTime.ts});
-        assert.eq(expectedPreImage, preImage.o);
-    }
+
+    var sessionInfo = {sessionId: lsid, txnNum: txnNumber};
+    checkImageCollection(primary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
+    checkImageCollection(secondary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
+
     // Assert that retrying the command will produce the same response.
     let retryRes = assert.commandWorked(mainConn.getDB('test').runCommand(cmd));
     assertRetryCommand(res, retryRes);
@@ -207,22 +182,15 @@ function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEn
     updateOp = oplog.findOne({ns: 'test.user', op: 'u', txnNumber: txnNumber});
     // Check that the findAndModify oplog entry and sessions record has the appropriate fields
     // and values.
-    retryArgs = {
-        needsRetryImage: true,
-        imageKind: "postImage",
-        preImageRecordingEnabled: preImageRecordingEnabled
-    };
+    retryArgs = {needsRetryImage: true, imageKind: "postImage"};
     checkOplogEntry(updateOp, lsid, txnNumber, expectedStmtId, expectedWriteTs, retryArgs);
     checkSessionCatalog(primary, lsid, txnNumber, updateOp.ts);
     checkSessionCatalog(secondary, lsid, txnNumber, updateOp.ts);
 
-    var sessionInfo = {sessionId: lsid, txnNum: txnNumber};
+    sessionInfo = {sessionId: lsid, txnNum: txnNumber};
     checkImageCollection(primary, sessionInfo, updateOp.ts, expectedPostImage, "postImage");
     checkImageCollection(secondary, sessionInfo, updateOp.ts, expectedPostImage, "postImage");
-    if (preImageRecordingEnabled) {
-        const preImage = oplog.findOne({ns: 'test.user', op: 'n', ts: updateOp.preImageOpTime.ts});
-        assert.eq(expectedPreImage, preImage.o);
-    }
+
     // Assert that retrying the command will produce the same response.
     retryRes = assert.commandWorked(mainConn.getDB('test').runCommand(cmd));
     assertRetryCommand(res, retryRes);
@@ -245,25 +213,15 @@ function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEn
     res = assert.commandWorked(mainConn.getDB('test').runCommand(cmd));
     // Get update entry.
     updateOp = oplog.findOne({ns: 'test.user', op: 'u', txnNumber: txnNumber});
-    retryArgs = {
-        needsRetryImage: true,
-        imageKind: "preImage",
-        preImageRecordingEnabled: preImageRecordingEnabled
-    };
+    retryArgs = {needsRetryImage: true, imageKind: "preImage"};
     // Check that the findAndModify oplog entry and sessions record has the appropriate fields
     // and values.
     checkOplogEntry(updateOp, lsid, txnNumber, expectedStmtId, expectedWriteTs, retryArgs);
     checkSessionCatalog(primary, lsid, txnNumber, updateOp.ts);
     checkSessionCatalog(secondary, lsid, txnNumber, updateOp.ts);
-    if (!preImageRecordingEnabled) {
-        const sessionInfo = {sessionId: lsid, txnNum: txnNumber};
-        checkImageCollection(primary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
-        checkImageCollection(secondary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
-    } else {
-        // The preImage should be stored in the oplog.
-        const preImage = oplog.findOne({ns: 'test.user', op: 'n', ts: updateOp.preImageOpTime.ts});
-        assert.eq(expectedPreImage, preImage.o);
-    }
+    sessionInfo = {sessionId: lsid, txnNum: txnNumber};
+    checkImageCollection(primary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
+    checkImageCollection(secondary, sessionInfo, updateOp.ts, expectedPreImage, "preImage");
 
     // Assert that retrying the command will produce the same response.
     retryRes = assert.commandWorked(mainConn.getDB('test').runCommand(cmd));
@@ -289,11 +247,8 @@ function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEn
 
     // Get update entry.
     updateOp = oplog.findOne({ns: 'test.user', op: 'u', txnNumber: txnNumber});
-    retryArgs = {
-        needsRetryImage: true,
-        imageKind: "postImage",
-        preImageRecordingEnabled: preImageRecordingEnabled
-    };
+    retryArgs = {needsRetryImage: true, imageKind: "postImage"};
+
     // Check that the findAndModify oplog entry and sessions record has the appropriate fields
     // and values.
     checkOplogEntry(updateOp, lsid, txnNumber, expectedStmtId, expectedWriteTs, retryArgs);
@@ -303,10 +258,7 @@ function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEn
     sessionInfo = {sessionId: lsid, txnNum: txnNumber};
     checkImageCollection(primary, sessionInfo, updateOp.ts, expectedPostImage, "postImage");
     checkImageCollection(secondary, sessionInfo, updateOp.ts, expectedPostImage, "postImage");
-    if (preImageRecordingEnabled) {
-        const preImage = oplog.findOne({ns: 'test.user', op: 'n', ts: updateOp.preImageOpTime.ts});
-        assert.eq(expectedPreImage, preImage.o);
-    }
+
     // Assert that retrying the command will produce the same response.
     retryRes = assert.commandWorked(mainConn.getDB('test').runCommand(cmd));
     assertRetryCommand(res, retryRes);
@@ -329,23 +281,14 @@ function runTests(lsid, mainConn, primary, secondary, docId, preImageRecordingEn
 
     // Get delete entry from top of oplog.
     const deleteOp = oplog.findOne({ns: 'test.user', op: 'd', txnNumber: txnNumber});
-    retryArgs = {
-        needsRetryImage: true,
-        imageKind: "preImage",
-        preImageRecordingEnabled: preImageRecordingEnabled
-    };
+    retryArgs = {needsRetryImage: true, imageKind: "preImage"};
     checkOplogEntry(deleteOp, lsid, txnNumber, expectedStmtId, expectedWriteTs, retryArgs);
     checkSessionCatalog(primary, lsid, txnNumber, deleteOp.ts);
     checkSessionCatalog(secondary, lsid, txnNumber, deleteOp.ts);
-    if (!preImageRecordingEnabled) {
-        const sessionInfo = {sessionId: lsid, txnNum: txnNumber};
-        checkImageCollection(primary, sessionInfo, deleteOp.ts, expectedPreImage, "preImage");
-        checkImageCollection(secondary, sessionInfo, deleteOp.ts, expectedPreImage, "preImage");
-    } else {
-        // The preImage should be stored in the oplog.
-        const preImage = oplog.findOne({ns: 'test.user', op: 'n', ts: deleteOp.preImageOpTime.ts});
-        assert.eq(expectedPreImage, preImage.o);
-    }
+    sessionInfo = {sessionId: lsid, txnNum: txnNumber};
+    checkImageCollection(primary, sessionInfo, deleteOp.ts, expectedPreImage, "preImage");
+    checkImageCollection(secondary, sessionInfo, deleteOp.ts, expectedPreImage, "preImage");
+
     // Assert that retrying the command will produce the same response.
     retryRes = assert.commandWorked(mainConn.getDB('test').runCommand(cmd));
     assertRetryCommand(res, retryRes);
@@ -362,34 +305,13 @@ const rst = new ReplSetTest({nodes: numNodes});
 rst.startSet();
 rst.initiate();
 checkProfilingLogs(rst.getPrimary());
-runTests(lsid,
-         rst.getPrimary(),
-         rst.getPrimary(),
-         rst.getSecondary(),
-         40,
-         /*preImageRecordingEnabled=*/false);
-runTests(lsid,
-         rst.getPrimary(),
-         rst.getPrimary(),
-         rst.getSecondary(),
-         60,
-         /*preImageRecordingEnabled=*/true);
+runTests(lsid, rst.getPrimary(), rst.getPrimary(), rst.getSecondary(), 40);
 rst.stopSet();
+
 // Test that retryable findAndModifys will store pre- and post- images in the
-// 'config.image_collection' table. We do not support collection preImage recording on sharded
-// clusters.
+// 'config.image_collection' table.
 const st = new ShardingTest({shards: {rs0: {nodes: numNodes}}});
-runTests(lsid,
-         st.s,
-         st.rs0.getPrimary(),
-         st.rs0.getSecondary(),
-         70,
-         /*preImageRecordingEnabled=*/false);
-runTests(lsid,
-         st.s,
-         st.rs0.getPrimary(),
-         st.rs0.getSecondary(),
-         80,
-         /*preImageRecordingEnabled=*/false);
+runTests(lsid, st.s, st.rs0.getPrimary(), st.rs0.getSecondary(), 70);
+runTests(lsid, st.s, st.rs0.getPrimary(), st.rs0.getSecondary(), 80);
 st.stop();
 })();
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index ef743884654..a0e073bb098 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -102,7 +102,6 @@ struct ParsedCollModRequest {
     boost::optional<Collection::Validator> collValidator;
     boost::optional<ValidationActionEnum> collValidationAction;
     boost::optional<ValidationLevelEnum> collValidationLevel;
-    bool recordPreImages = false;
     boost::optional<ChangeStreamPreAndPostImagesOptions> changeStreamPreAndPostImagesOptions;
     int numModifications = 0;
     bool dryRun = false;
@@ -468,18 +467,6 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
         oplogEntryBuilder.append(CollMod::kViewOnFieldName, *viewOn);
     }
 
-    if (const auto& recordPreImages = cmr.getRecordPreImages()) {
-        if (isView) {
-            return getNotSupportedOnViewError(CollMod::kRecordPreImagesFieldName);
-        }
-        if (isTimeseries) {
-            return getNotSupportedOnTimeseriesError(CollMod::kRecordPreImagesFieldName);
-        }
-        parsed.numModifications++;
-        parsed.recordPreImages = *recordPreImages;
-        oplogEntryBuilder.append(CollMod::kRecordPreImagesFieldName, *recordPreImages);
-    }
-
     if (auto& changeStreamPreAndPostImages = cmr.getChangeStreamPreAndPostImages()) {
         if (isView) {
             return getNotSupportedOnViewError(CollMod::kChangeStreamPreAndPostImagesFieldName);
@@ -823,17 +810,6 @@ Status _collModInternal(OperationContext* opCtx,
 
         const CollectionOptions& oldCollOptions = coll->getCollectionOptions();
 
-        // If 'changeStreamPreAndPostImagesOptions' are enabled, 'recordPreImages' must be set
-        // to false. If 'recordPreImages' is set to true, 'changeStreamPreAndPostImagesOptions'
-        // must be disabled.
-        if (cmrNew.changeStreamPreAndPostImagesOptions &&
-            cmrNew.changeStreamPreAndPostImagesOptions->getEnabled()) {
-            cmrNew.recordPreImages = false;
-        }
-
-        if (cmrNew.recordPreImages) {
-            cmrNew.changeStreamPreAndPostImagesOptions = ChangeStreamPreAndPostImagesOptions(false);
-        }
         if (cmrNew.cappedSize || cmrNew.cappedMax) {
             // If the current capped collection size exceeds the newly set limits, future document
             // inserts will prompt document deletion.
@@ -868,10 +844,6 @@ Status _collModInternal(OperationContext* opCtx,
                                        "Failed to set validationLevel");
         }
 
-        if (cmrNew.recordPreImages != oldCollOptions.recordPreImages) {
-            coll.getWritableCollection(opCtx)->setRecordPreImages(opCtx, cmrNew.recordPreImages);
-        }
-
         if (cmrNew.changeStreamPreAndPostImagesOptions.has_value() &&
             *cmrNew.changeStreamPreAndPostImagesOptions !=
                 oldCollOptions.changeStreamPreAndPostImagesOptions) {
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index ad129695d7c..ac63bf5be86 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -83,7 +83,6 @@ struct CollectionUpdateArgs {
     OperationSource source = OperationSource::kStandard;
 
     StoreDocOption storeDocOption = StoreDocOption::None;
-    bool preImageRecordingEnabledForCollection = false;
     bool changeStreamPreAndPostImagesEnabledForCollection = false;
 
     // Set if OpTimes were reserved for the update ahead of time.
@@ -417,9 +416,6 @@ public:
 
     virtual Status checkValidatorAPIVersionCompatability(OperationContext* opCtx) const = 0;
 
-    virtual bool getRecordPreImages() const = 0;
-    virtual void setRecordPreImages(OperationContext* opCtx, bool val) = 0;
-
     virtual bool isChangeStreamPreAndPostImagesEnabled() const = 0;
     virtual void setChangeStreamPreAndPostImages(OperationContext* opCtx,
                                                  ChangeStreamPreAndPostImagesOptions val) = 0;
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index d61a693869f..5b50f768e51 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -171,28 +171,6 @@ Status validateIsNotInDbs(const NamespaceString& ns,
     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(const NamespaceString& ns) {
-    const auto validationStatus = validateIsNotInDbs(
-        ns, {NamespaceString::kAdminDb, NamespaceString::kLocalDb}, "recordPreImages");
-    if (validationStatus != Status::OK()) {
-        return validationStatus;
-    }
-
-    if (serverGlobalParams.clusterRole != ClusterRole::None) {
-        return {
-            ErrorCodes::InvalidOptions,
-            str::stream()
-                << "namespace " << ns.ns()
-                << " has the recordPreImages option set, this is not supported on a "
-                   "sharded cluster. Consider restarting without --shardsvr and --configsvr and "
-                   "disabling recordPreImages via collMod"};
-    }
-
-    return Status::OK();
-}
-
 // Validates that the option is not used on admin, local or config db as well as not being used on
 // config servers.
 Status validateChangeStreamPreAndPostImagesOptionIsPermitted(const NamespaceString& ns) {
@@ -225,8 +203,7 @@ bool isRetryableWrite(OperationContext* opCtx) {
         (!opCtx->inMultiDocumentTransaction() || txnParticipant.transactionIsOpen());
 }
 
-std::vector<OplogSlot> reserveOplogSlotsForRetryableFindAndModify(OperationContext* opCtx,
-                                                                  const int numSlots) {
+std::vector<OplogSlot> reserveOplogSlotsForRetryableFindAndModify(OperationContext* opCtx) {
     invariant(isRetryableWrite(opCtx));
 
     // For retryable findAndModify running in a multi-document transaction, we will reserve the
@@ -239,7 +216,7 @@ std::vector<OplogSlot> reserveOplogSlotsForRetryableFindAndModify(OperationConte
     // used as the oplog timestamp. Tenant migrations and resharding will forge no-op image oplog
     // entries and set the timestamp for these synthetic entries to be TS - 1.
     auto oplogInfo = LocalOplogInfo::get(opCtx);
-    auto slots = oplogInfo->getNextOpTimes(opCtx, numSlots);
+    auto slots = oplogInfo->getNextOpTimes(opCtx, 2);
     uassertStatusOK(opCtx->recoveryUnit()->setTimestamp(slots.back().getTimestamp()));
     return slots;
 }
@@ -474,10 +451,6 @@ void CollectionImpl::_initCommon(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(validateRecordPreImagesOptionIsPermitted(_ns));
-    }
-
     if (collectionOptions.changeStreamPreAndPostImagesOptions.getEnabled()) {
         uassertStatusOK(validateChangeStreamPreAndPostImagesOptionIsPermitted(_ns));
     }
@@ -821,14 +794,12 @@ void CollectionImpl::deleteDocument(OperationContext* opCtx,
 
     std::vector<OplogSlot> oplogSlots;
     auto retryableFindAndModifyLocation = RetryableFindAndModifyLocation::kNone;
-    if (storeDeletedDoc == Collection::StoreDeletedDoc::On && !getRecordPreImages() &&
-        isRetryableWrite(opCtx)) {
+    if (storeDeletedDoc == Collection::StoreDeletedDoc::On && isRetryableWrite(opCtx)) {
         retryableFindAndModifyLocation = RetryableFindAndModifyLocation::kSideCollection;
-        oplogSlots = reserveOplogSlotsForRetryableFindAndModify(opCtx, 2);
+        oplogSlots = reserveOplogSlotsForRetryableFindAndModify(opCtx);
     }
     OplogDeleteEntryArgs deleteArgs{nullptr /* deletedDoc */,
                                     fromMigrate,
-                                    getRecordPreImages(),
                                     isChangeStreamPreAndPostImagesEnabled(),
                                     retryableFindAndModifyLocation,
                                     oplogSlots};
@@ -838,8 +809,7 @@ void CollectionImpl::deleteDocument(OperationContext* opCtx,
     boost::optional<BSONObj> deletedDoc;
     const bool isRecordingPreImageForRetryableWrite =
         retryableFindAndModifyLocation != RetryableFindAndModifyLocation::kNone;
-    if (isRecordingPreImageForRetryableWrite || getRecordPreImages() ||
-        isChangeStreamPreAndPostImagesEnabled()) {
+    if (isRecordingPreImageForRetryableWrite || isChangeStreamPreAndPostImagesEnabled()) {
         deletedDoc.emplace(doc.value().getOwned());
     }
     int64_t keysDeleted = 0;
@@ -935,7 +905,6 @@ RecordId CollectionImpl::updateDocument(OperationContext* opCtx,
     if (!args->preImageDoc) {
         args->preImageDoc = oldDoc.value().getOwned();
     }
-    args->preImageRecordingEnabledForCollection = getRecordPreImages();
     args->changeStreamPreAndPostImagesEnabledForCollection =
         isChangeStreamPreAndPostImagesEnabled();
 
@@ -949,14 +918,10 @@ RecordId CollectionImpl::updateDocument(OperationContext* opCtx,
         // post-image in a side collection, then we must reserve oplog slots in advance. We
         // expect to use the reserved oplog slots as follows, where TS is the greatest
         // timestamp of 'oplogSlots':
-        // TS - 2: If 'getRecordPreImages()' is true, we reserve an extra oplog slot in case we
-        //         must account for storing a pre-image in the oplog and an eventual synthetic
-        //         no-op image oplog used by tenant migrations/resharding.
         // TS - 1: Tenant migrations and resharding will forge no-op image oplog entries and set
         //         the entry timestamps to TS - 1.
         // TS:     The timestamp given to the update oplog entry.
-        const auto numSlotsToReserve = getRecordPreImages() ? 3 : 2;
-        args->oplogSlots = reserveOplogSlotsForRetryableFindAndModify(opCtx, numSlotsToReserve);
+        args->oplogSlots = reserveOplogSlotsForRetryableFindAndModify(opCtx);
     } else {
         // Retryable findAndModify commands should not reserve oplog slots before entering this
         // function since tenant migrations and resharding rely on always being able to set
@@ -1021,7 +986,7 @@ StatusWith<RecordData> CollectionImpl::updateDocumentWithDamages(
     // 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() || isChangeStreamPreAndPostImagesEnabled())) {
+    if (!args->preImageDoc && isChangeStreamPreAndPostImagesEnabled()) {
         args->preImageDoc = oldRec.value().toBson().getOwned();
     }
     OplogUpdateEntryArgs onUpdateArgs(args, ns(), _uuid);
@@ -1034,14 +999,10 @@ StatusWith<RecordData> CollectionImpl::updateDocumentWithDamages(
         // post-image in a side collection, then we must reserve oplog slots in advance. We
         // expect to use the reserved oplog slots as follows, where TS is the greatest
         // timestamp of 'oplogSlots':
-        // TS - 2: If 'getRecordPreImages()' is true, we reserve an extra oplog slot in case we
-        //         must account for storing a pre-image in the oplog and an eventual synthetic
-        //         no-op image oplog used by tenant migrations/resharding.
         // TS - 1: Tenant migrations and resharding will forge no-op image oplog entries and set
         //         the entry timestamps to TS - 1.
         // TS:     The timestamp given to the update oplog entry.
-        const auto numSlotsToReserve = getRecordPreImages() ? 3 : 2;
-        args->oplogSlots = reserveOplogSlotsForRetryableFindAndModify(opCtx, numSlotsToReserve);
+        args->oplogSlots = reserveOplogSlotsForRetryableFindAndModify(opCtx);
     } else {
         // Retryable findAndModify commands should not reserve oplog slots before entering this
         // function since tenant migrations and resharding rely on always being able to set
@@ -1054,7 +1015,6 @@ StatusWith<RecordData> CollectionImpl::updateDocumentWithDamages(
 
     if (newRecStatus.isOK()) {
         args->updatedDoc = newRecStatus.getValue().toBson();
-        args->preImageRecordingEnabledForCollection = getRecordPreImages();
         args->changeStreamPreAndPostImagesEnabledForCollection =
             isChangeStreamPreAndPostImagesEnabled();
 
@@ -1170,19 +1130,6 @@ Status CollectionImpl::updateCappedSize(OperationContext* opCtx,
     return Status::OK();
 }
 
-bool CollectionImpl::getRecordPreImages() const {
-    return _metadata->options.recordPreImages;
-}
-
-void CollectionImpl::setRecordPreImages(OperationContext* opCtx, bool val) {
-    if (val) {
-        uassertStatusOK(validateRecordPreImagesOptionIsPermitted(_ns));
-    }
-
-    _writeMetadata(
-        opCtx, [&](BSONCollectionCatalogEntry::MetaData& md) { md.options.recordPreImages = val; });
-}
-
 bool CollectionImpl::isChangeStreamPreAndPostImagesEnabled() const {
     return _metadata->options.changeStreamPreAndPostImagesOptions.getEnabled();
 }
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index 4300a207635..95a8fd20249 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -265,9 +265,6 @@ public:
      */
     Status checkValidatorAPIVersionCompatability(OperationContext* opCtx) const final;
 
-    bool getRecordPreImages() const final;
-    void setRecordPreImages(OperationContext* opCtx, bool val) final;
-
     bool isChangeStreamPreAndPostImagesEnabled() const final;
     void setChangeStreamPreAndPostImages(OperationContext* opCtx,
                                          ChangeStreamPreAndPostImagesOptions val) final;
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index c5e1e138ec1..d910677df43 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -263,14 +263,6 @@ public:
         MONGO_UNREACHABLE;
     }
 
-    bool getRecordPreImages() const {
-        MONGO_UNREACHABLE;
-    }
-
-    void setRecordPreImages(OperationContext* opCtx, bool val) {
-        MONGO_UNREACHABLE;
-    }
-
     bool isChangeStreamPreAndPostImagesEnabled() const {
         MONGO_UNREACHABLE;
     }
diff --git a/src/mongo/db/catalog/collection_options.cpp b/src/mongo/db/catalog/collection_options.cpp
index 800e08172cb..d65b30d0101 100644
--- a/src/mongo/db/catalog/collection_options.cpp
+++ b/src/mongo/db/catalog/collection_options.cpp
@@ -176,8 +176,6 @@ 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 == "changeStreamPreAndPostImages") {
             if (e.type() != mongo::Object) {
                 return {ErrorCodes::InvalidOptions,
@@ -368,9 +366,6 @@ CollectionOptions CollectionOptions::fromCreateCommand(const CreateCommand& cmd)
     if (auto collation = cmd.getCollation()) {
         options.collation = collation->toBSON();
     }
-    if (auto recordPreImages = cmd.getRecordPreImages()) {
-        options.recordPreImages = *recordPreImages;
-    }
     if (auto changeStreamPreAndPostImagesOptions = cmd.getChangeStreamPreAndPostImages()) {
         options.changeStreamPreAndPostImagesOptions = *changeStreamPreAndPostImagesOptions;
     }
@@ -440,10 +435,6 @@ void CollectionOptions::appendBSON(BSONObjBuilder* builder,
     if (temp && shouldAppend(CreateCommand::kTempFieldName))
         builder->appendBool(CreateCommand::kTempFieldName, true);
 
-    if (recordPreImages && shouldAppend(CreateCommand::kRecordPreImagesFieldName)) {
-        builder->appendBool(CreateCommand::kRecordPreImagesFieldName, true);
-    }
-
     if (changeStreamPreAndPostImagesOptions.getEnabled() &&
         shouldAppend(CreateCommand::kChangeStreamPreAndPostImagesFieldName)) {
         builder->append(CreateCommand::kChangeStreamPreAndPostImagesFieldName,
@@ -530,10 +521,6 @@ bool CollectionOptions::matchesStorageOptions(const CollectionOptions& other,
         return false;
     }
 
-    if (recordPreImages != other.recordPreImages) {
-        return false;
-    }
-
     if (changeStreamPreAndPostImagesOptions != other.changeStreamPreAndPostImagesOptions) {
         return false;
     }
diff --git a/src/mongo/db/catalog/collection_options.h b/src/mongo/db/catalog/collection_options.h
index 41b0c8f5376..2e1d06aa190 100644
--- a/src/mongo/db/catalog/collection_options.h
+++ b/src/mongo/db/catalog/collection_options.h
@@ -123,11 +123,10 @@ struct CollectionOptions {
     } autoIndexId = DEFAULT;
 
     bool temp = false;
-    bool recordPreImages = false;
 
     // Change stream options define whether or not to store 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 enabled together with 'recordPreImages' (mutually exclusive).
+    // via changeStreams.
     ChangeStreamPreAndPostImagesOptions changeStreamPreAndPostImagesOptions{false};
 
     // Storage engine collection options. Always owned or empty.
diff --git a/src/mongo/db/coll_mod.idl b/src/mongo/db/coll_mod.idl
index 5e6fa56af42..83ecc09d218 100644
--- a/src/mongo/db/coll_mod.idl
+++ b/src/mongo/db/coll_mod.idl
@@ -153,12 +153,6 @@ structs:
                 type: array<object>
                 optional: true
                 stability: stable
-            recordPreImages:
-                description: "Sets whether updates/deletes should store the pre-image of the
-                              document in the oplog"
-                optional: true
-                type: safeBool
-                stability: unstable
             changeStreamPreAndPostImages:
                 description: "The options for point-in-time pre- and post-images in change streams opened on this collection."
                 type: ChangeStreamPreAndPostImagesOptions
diff --git a/src/mongo/db/commands/create.idl b/src/mongo/db/commands/create.idl
index b03cb1196d1..8ab95c40f1f 100644
--- a/src/mongo/db/commands/create.idl
+++ b/src/mongo/db/commands/create.idl
@@ -168,12 +168,6 @@ commands:
                 type: Collation
                 optional: true
                 stability: stable
-            recordPreImages:
-                description: "Sets whether updates/deletes should store the pre-image of the
-                              document in the oplog"
-                type: safeBool
-                optional: true
-                stability: unstable
             changeStreamPreAndPostImages:
                 description: "The options for point-in-time pre- and post-images in change streams opened on this collection."
                 type: ChangeStreamPreAndPostImagesOptions
diff --git a/src/mongo/db/commands/create_command.cpp b/src/mongo/db/commands/create_command.cpp
index e4191a75e19..f480980fa0b 100644
--- a/src/mongo/db/commands/create_command.cpp
+++ b/src/mongo/db/commands/create_command.cpp
@@ -323,15 +323,6 @@ public:
                 cmd.setIdIndex(idIndexSpec);
             }
 
-            const auto isChangeStreamPreAndPostImagesEnabled =
-                (cmd.getChangeStreamPreAndPostImages() &&
-                 cmd.getChangeStreamPreAndPostImages()->getEnabled());
-            const auto isRecordPreImagesEnabled = cmd.getRecordPreImages().get_value_or(false);
-            uassert(ErrorCodes::InvalidOptions,
-                    "'recordPreImages' and 'changeStreamPreAndPostImages.enabled' can not be "
-                    "set to true simultaneously",
-                    !(isChangeStreamPreAndPostImagesEnabled && isRecordPreImagesEnabled));
-
             OperationShardingState::ScopedAllowImplicitCollectionCreate_UNSAFE
                 unsafeCreateCollection(opCtx);
             uassertStatusOK(createCollection(opCtx, cmd));
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index d18fa21101f..eb3c3567077 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -533,15 +533,6 @@ public:
             "collMod on a time-series collection's underlying buckets collection is not supported.",
             !cmd->getNamespace().isTimeseriesBucketsCollection());
 
-        const auto isChangeStreamPreAndPostImagesEnabled =
-            (cmd->getChangeStreamPreAndPostImages() &&
-             cmd->getChangeStreamPreAndPostImages()->getEnabled());
-        const auto isRecordPreImagesEnabled = cmd->getRecordPreImages().get_value_or(false);
-        uassert(ErrorCodes::InvalidOptions,
-                "'recordPreImages' and 'changeStreamPreAndPostImages.enabled' can not be set "
-                "to true simultaneously",
-                !(isChangeStreamPreAndPostImagesEnabled && isRecordPreImagesEnabled));
-
         // Updating granularity on sharded time-series collections is not allowed.
         if (Grid::get(opCtx)->catalogClient() && cmd->getTimeseries() &&
             cmd->getTimeseries()->getGranularity()) {
diff --git a/src/mongo/db/op_observer/op_observer.h b/src/mongo/db/op_observer/op_observer.h
index 5756ad47af7..3dac0e56eea 100644
--- a/src/mongo/db/op_observer/op_observer.h
+++ b/src/mongo/db/op_observer/op_observer.h
@@ -78,7 +78,6 @@ struct OplogDeleteEntryArgs {
     // "fromMigrate" indicates whether the delete was induced by a chunk migration, and so
     // should be ignored by the user as an internal maintenance operation and not a real delete.
     bool fromMigrate = false;
-    bool preImageRecordingEnabledForCollection = false;
     bool changeStreamPreAndPostImagesEnabledForCollection = false;
 
     // Specifies the pre-image recording option for retryable "findAndModify" commands.
@@ -446,11 +445,10 @@ public:
     virtual void onBatchedWriteAbort(OperationContext* opCtx) = 0;
 
     /**
-     * Contains "applyOps" oplog entries and oplog slots to be used for writing pre- and post- image
-     * oplog entries for a transaction. "applyOps" entries are not actual "applyOps" entries to be
-     * written to the oplog, but comprise certain parts of those entries - BSON serialized
-     * operations, and the assigned oplog slot. The operations in field 'ApplyOpsEntry::operations'
-     * should be considered opaque outside the OpObserver.
+     * Contains "applyOps" oplog entries for a transaction. "applyOps" entries are not actual
+     * "applyOps" entries to be written to the oplog, but comprise certain parts of those entries -
+     * BSON serialized operations, and the assigned oplog slot. The operations in field
+     * 'ApplyOpsEntry::operations' should be considered opaque outside the OpObserver.
      */
     struct ApplyOpsOplogSlotAndOperationAssignment {
         struct ApplyOpsEntry {
@@ -458,9 +456,6 @@ public:
             std::vector<BSONObj> operations;
         };
 
-        // Oplog slots to be used for writing pre- and post- image oplog entries.
-        std::vector<OplogSlot> prePostImageOplogEntryOplogSlots;
-
         // Representation of "applyOps" oplog entries.
         std::vector<ApplyOpsEntry> applyOpsEntries;
 
diff --git a/src/mongo/db/op_observer/op_observer_impl.cpp b/src/mongo/db/op_observer/op_observer_impl.cpp
index da1330a81bc..d578e89382f 100644
--- a/src/mongo/db/op_observer/op_observer_impl.cpp
+++ b/src/mongo/db/op_observer/op_observer_impl.cpp
@@ -202,58 +202,10 @@ OpTimeBundle replLogUpdate(OperationContext* opCtx,
     oplogWriter->appendOplogEntryChainInfo(opCtx, oplogEntry, &oplogLink, args.updateArgs->stmtIds);
 
     OpTimeBundle opTimes;
-    // We never want to store pre- or post- images when we're migrating oplog entries from another
-    // replica set.
-    const auto& migrationRecipientInfo = repl::tenantMigrationInfo(opCtx);
-    const auto storePreImageInOplogForRetryableWrite =
-        (args.updateArgs->storeDocOption == CollectionUpdateArgs::StoreDocOption::PreImage &&
-         opCtx->getTxnNumber() && !oplogEntry->getNeedsRetryImage());
-    if ((storePreImageInOplogForRetryableWrite ||
-         args.updateArgs->preImageRecordingEnabledForCollection) &&
-        !migrationRecipientInfo) {
-        MutableOplogEntry noopEntry = *oplogEntry;
-        invariant(args.updateArgs->preImageDoc);
-        noopEntry.setOpType(repl::OpTypeEnum::kNoop);
-        noopEntry.setObject(*args.updateArgs->preImageDoc);
-        if (args.updateArgs->preImageRecordingEnabledForCollection &&
-            args.retryableFindAndModifyLocation ==
-                RetryableFindAndModifyLocation::kSideCollection) {
-            // We are writing a no-op pre-image oplog entry and storing a post-image into a side
-            // collection. In this case, we expect to have already reserved 3 oplog slots:
-            // TS - 2: Oplog slot for the current no-op preimage oplog entry
-            // TS - 1: Oplog slot for the forged no-op oplog entry that may eventually get used by
-            //         tenant migrations or resharding.
-            // TS:     Oplog slot for the actual update oplog entry.
-            const auto reservedOplogSlots = args.updateArgs->oplogSlots;
-            invariant(reservedOplogSlots.size() == 3);
-            noopEntry.setOpTime(repl::OpTime(reservedOplogSlots.front().getTimestamp(),
-                                             reservedOplogSlots.front().getTerm()));
-        }
-        oplogLink.preImageOpTime =
-            logOperation(opCtx, &noopEntry, true /*assignWallClockTime*/, oplogWriter);
-        if (storePreImageInOplogForRetryableWrite) {
-            opTimes.prePostImageOpTime = oplogLink.preImageOpTime;
-        }
-    }
-
-    // This case handles storing the post image for retryable findAndModify's.
-    if (args.updateArgs->storeDocOption == CollectionUpdateArgs::StoreDocOption::PostImage &&
-        opCtx->getTxnNumber() && !migrationRecipientInfo && !oplogEntry->getNeedsRetryImage()) {
-        MutableOplogEntry noopEntry = *oplogEntry;
-        noopEntry.setOpType(repl::OpTypeEnum::kNoop);
-        noopEntry.setObject(args.updateArgs->updatedDoc);
-        oplogLink.postImageOpTime =
-            logOperation(opCtx, &noopEntry, true /*assignWallClockTime*/, oplogWriter);
-        invariant(opTimes.prePostImageOpTime.isNull());
-        opTimes.prePostImageOpTime = oplogLink.postImageOpTime;
-    }
-
     oplogEntry->setOpType(repl::OpTypeEnum::kUpdate);
     oplogEntry->setObject(args.updateArgs->update);
     oplogEntry->setObject2(args.updateArgs->criteria);
     oplogEntry->setFromMigrateIfTrue(args.updateArgs->source == OperationSource::kFromMigrate);
-    // oplogLink could have been changed to include pre/postImageOpTime by the previous no-op write.
-    oplogWriter->appendOplogEntryChainInfo(opCtx, oplogEntry, &oplogLink, args.updateArgs->stmtIds);
     if (!args.updateArgs->oplogSlots.empty()) {
         oplogEntry->setOpTime(args.updateArgs->oplogSlots.back());
     }
@@ -272,7 +224,6 @@ OpTimeBundle replLogDelete(OperationContext* opCtx,
                            const boost::optional<UUID>& uuid,
                            StmtId stmtId,
                            bool fromMigrate,
-                           const boost::optional<BSONObj>& deletedDoc,
                            OplogWriter* oplogWriter) {
     oplogEntry->setTid(nss.tenantId());
     oplogEntry->setNss(nss);
@@ -283,23 +234,9 @@ OpTimeBundle replLogDelete(OperationContext* opCtx,
     oplogWriter->appendOplogEntryChainInfo(opCtx, oplogEntry, &oplogLink, {stmtId});
 
     OpTimeBundle opTimes;
-    // We never want to store pre-images when we're migrating oplog entries from another
-    // replica set.
-    const auto& migrationRecipientInfo = repl::tenantMigrationInfo(opCtx);
-    if (deletedDoc && !migrationRecipientInfo) {
-        MutableOplogEntry noopEntry = *oplogEntry;
-        noopEntry.setOpType(repl::OpTypeEnum::kNoop);
-        noopEntry.setObject(*deletedDoc);
-        auto noteOplog = logOperation(opCtx, &noopEntry, true /*assignWallClockTime*/, oplogWriter);
-        opTimes.prePostImageOpTime = noteOplog;
-        oplogLink.preImageOpTime = noteOplog;
-    }
-
     oplogEntry->setOpType(repl::OpTypeEnum::kDelete);
     oplogEntry->setObject(repl::documentKeyDecoration(opCtx).value().getShardKeyAndId());
     oplogEntry->setFromMigrateIfTrue(fromMigrate);
-    // oplogLink could have been changed to include preImageOpTime by the previous no-op write.
-    oplogWriter->appendOplogEntryChainInfo(opCtx, oplogEntry, &oplogLink, {stmtId});
     opTimes.writeOpTime =
         logOperation(opCtx, oplogEntry, true /*assignWallClockTime*/, oplogWriter);
     opTimes.wallClockTime = oplogEntry->getWallClockTime();
@@ -833,21 +770,13 @@ void OpObserverImpl::onUpdate(OperationContext* opCtx, const OplogUpdateEntryArg
             args.nss, args.uuid, args.updateArgs->update, args.updateArgs->criteria);
 
         if (inRetryableInternalTransaction) {
-            uassert(6462400,
-                    str::stream() << "Found a retryable internal transaction on a sharded cluster "
-                                  << "executing an update against the collection '" << args.nss
-                                  << "' with the 'recordPreImages' option enabled",
-                    !args.updateArgs->preImageRecordingEnabledForCollection ||
-                        serverGlobalParams.clusterRole == ClusterRole::None);
-
             operation.setInitializedStatementIds(args.updateArgs->stmtIds);
             if (args.updateArgs->storeDocOption == CollectionUpdateArgs::StoreDocOption::PreImage) {
                 invariant(args.updateArgs->preImageDoc);
                 operation.setPreImage(args.updateArgs->preImageDoc->getOwned());
                 operation.setPreImageRecordedForRetryableInternalTransaction();
                 if (args.retryableFindAndModifyLocation ==
-                        RetryableFindAndModifyLocation::kSideCollection &&
-                    !args.updateArgs->preImageRecordingEnabledForCollection) {
+                    RetryableFindAndModifyLocation::kSideCollection) {
                     operation.setNeedsRetryImage(repl::RetryImageEnum::kPreImage);
                 }
             }
@@ -862,25 +791,8 @@ void OpObserverImpl::onUpdate(OperationContext* opCtx, const OplogUpdateEntryArg
             }
         }
 
-        if (args.updateArgs->preImageRecordingEnabledForCollection) {
-            invariant(args.updateArgs->preImageDoc);
-            tassert(
-                5869402,
-                "Change stream pre-image recording to the oplog and to the pre-image collection "
-                "requested at the same time",
-                !args.updateArgs->changeStreamPreAndPostImagesEnabledForCollection);
-            operation.setPreImage(args.updateArgs->preImageDoc->getOwned());
-            operation.setChangeStreamPreImageRecordingMode(
-                ChangeStreamPreImageRecordingMode::kOplog);
-        }
-
         if (args.updateArgs->changeStreamPreAndPostImagesEnabledForCollection) {
             invariant(args.updateArgs->preImageDoc);
-            tassert(
-                5869403,
-                "Change stream pre-image recording to the oplog and to the pre-image collection "
-                "requested at the same time",
-                !args.updateArgs->preImageRecordingEnabledForCollection);
             operation.setPreImage(args.updateArgs->preImageDoc->getOwned());
             operation.setChangeStreamPreImageRecordingMode(
                 ChangeStreamPreImageRecordingMode::kPreImagesCollection);
@@ -897,10 +809,7 @@ void OpObserverImpl::onUpdate(OperationContext* opCtx, const OplogUpdateEntryArg
         if (args.retryableFindAndModifyLocation ==
             RetryableFindAndModifyLocation::kSideCollection) {
             // If we've stored a preImage:
-            if (args.updateArgs->storeDocOption == CollectionUpdateArgs::StoreDocOption::PreImage &&
-                // And we're not writing to a noop entry anyways for
-                // `preImageRecordingEnabledForCollection`:
-                !args.updateArgs->preImageRecordingEnabledForCollection) {
+            if (args.updateArgs->storeDocOption == CollectionUpdateArgs::StoreDocOption::PreImage) {
                 oplogEntry.setNeedsRetryImage({repl::RetryImageEnum::kPreImage});
             } else if (args.updateArgs->storeDocOption ==
                        CollectionUpdateArgs::StoreDocOption::PostImage) {
@@ -1055,13 +964,6 @@ void OpObserverImpl::onDelete(OperationContext* opCtx,
             MutableOplogEntry::makeDeleteOperation(nss, uuid, documentKey.getShardKeyAndId());
 
         if (inRetryableInternalTransaction) {
-            uassert(6462401,
-                    str::stream() << "Found a retryable internal transaction on a sharded cluster "
-                                  << "executing an delete against the collection '" << nss
-                                  << "' with the 'recordPreImages' option enabled",
-                    !args.preImageRecordingEnabledForCollection ||
-                        serverGlobalParams.clusterRole == ClusterRole::None);
-
             operation.setInitializedStatementIds({stmtId});
             if (args.retryableFindAndModifyLocation != RetryableFindAndModifyLocation::kNone) {
                 tassert(6054000,
@@ -1070,8 +972,7 @@ void OpObserverImpl::onDelete(OperationContext* opCtx,
                 operation.setPreImage(args.deletedDoc->getOwned());
                 operation.setPreImageRecordedForRetryableInternalTransaction();
                 if (args.retryableFindAndModifyLocation ==
-                        RetryableFindAndModifyLocation::kSideCollection &&
-                    !args.preImageRecordingEnabledForCollection) {
+                    RetryableFindAndModifyLocation::kSideCollection) {
                     operation.setNeedsRetryImage(repl::RetryImageEnum::kPreImage);
                 }
             }
@@ -1081,21 +982,9 @@ void OpObserverImpl::onDelete(OperationContext* opCtx,
             tassert(5869400,
                     "Deleted document must be present for pre-image recording",
                     args.deletedDoc);
-            tassert(
-                5869401,
-                "Change stream pre-image recording to the oplog and to the pre-image collection "
-                "requested at the same time",
-                !args.preImageRecordingEnabledForCollection);
             operation.setPreImage(args.deletedDoc->getOwned());
             operation.setChangeStreamPreImageRecordingMode(
                 ChangeStreamPreImageRecordingMode::kPreImagesCollection);
-        } else if (args.preImageRecordingEnabledForCollection) {
-            tassert(5868701,
-                    "Deleted document must be present for pre-image recording",
-                    args.deletedDoc);
-            operation.setPreImage(args.deletedDoc->getOwned());
-            operation.setChangeStreamPreImageRecordingMode(
-                ChangeStreamPreImageRecordingMode::kOplog);
         }
 
         operation.setDestinedRecipient(destinedRecipientDecoration(opCtx));
@@ -1105,13 +994,8 @@ void OpObserverImpl::onDelete(OperationContext* opCtx,
         MutableOplogEntry oplogEntry;
         boost::optional<BSONObj> deletedDocForOplog = boost::none;
 
-        if (args.preImageRecordingEnabledForCollection) {
-            tassert(5868702,
-                    "Deleted document must be present for pre-image recording",
-                    args.deletedDoc);
-            deletedDocForOplog = {*(args.deletedDoc)};
-        } else if (args.retryableFindAndModifyLocation ==
-                   RetryableFindAndModifyLocation::kSideCollection) {
+        if (args.retryableFindAndModifyLocation ==
+            RetryableFindAndModifyLocation::kSideCollection) {
             tassert(5868703,
                     "Deleted document must be present for pre-image recording",
                     args.deletedDoc);
@@ -1122,14 +1006,8 @@ void OpObserverImpl::onDelete(OperationContext* opCtx,
                 oplogEntry.setOpTime(args.oplogSlots.back());
             }
         }
-        opTime = replLogDelete(opCtx,
-                               nss,
-                               &oplogEntry,
-                               uuid,
-                               stmtId,
-                               args.fromMigrate,
-                               deletedDocForOplog,
-                               _oplogWriter.get());
+        opTime = replLogDelete(
+            opCtx, nss, &oplogEntry, uuid, stmtId, args.fromMigrate, _oplogWriter.get());
 
         if (oplogEntry.getNeedsRetryImage()) {
             writeToImageCollection(opCtx,
@@ -1671,57 +1549,26 @@ std::vector<BSONObj> packOperationsIntoApplyOps(
  * Returns oplog slots to be used for "applyOps" oplog entries, BSON serialized operations, their
  * assignments to "applyOps" entries, and oplog slots to be used for writing pre- and post- image
  * oplog entries for the transaction consisting of 'operations'. Allocates oplog slots from
- * 'oplogSlots'. The 'numberOfPrePostImagesToWrite' is the number of CRUD operations that have a
- * pre-image to write as a noop oplog entry. The 'prepare' indicates if the function is called when
- * preparing a transaction.
+ * 'oplogSlots'. The 'prepare' indicates if the function is called when preparing a transaction.
  */
 OpObserver::ApplyOpsOplogSlotAndOperationAssignment
 getApplyOpsOplogSlotAndOperationAssignmentForTransaction(
     OperationContext* opCtx,
     const std::vector<OplogSlot>& oplogSlots,
-    size_t numberOfPrePostImagesToWrite,
     bool prepare,
     std::vector<repl::ReplOperation>& operations) {
     if (operations.empty()) {
-        return {{}, {}, 0 /*numberOfOplogSlotsUsed*/};
+        return {{}, 0 /*numberOfOplogSlotsUsed*/};
     }
     tassert(6278504, "Insufficient number of oplogSlots", operations.size() <= oplogSlots.size());
 
-    std::vector<OplogSlot> prePostImageOplogEntryOplogSlots;
     std::vector<OpObserver::ApplyOpsOplogSlotAndOperationAssignment::ApplyOpsEntry> applyOpsEntries;
-    const auto operationCount = operations.size();
     auto oplogSlotIter = oplogSlots.begin();
     auto getNextOplogSlot = [&]() {
         tassert(6278505, "Unexpected end of oplog slot vector", oplogSlotIter != oplogSlots.end());
         return *oplogSlotIter++;
     };
 
-    auto isMigratingTenant = [&opCtx]() {
-        return static_cast<bool>(repl::tenantMigrationInfo(opCtx));
-    };
-
-    // We never want to store pre-images or post-images when we're migrating oplog entries from
-    // another replica set.
-    if (numberOfPrePostImagesToWrite > 0 && !isMigratingTenant()) {
-        for (size_t operationIdx = 0; operationIdx < operationCount; ++operationIdx) {
-            auto& statement = operations[operationIdx];
-            if (statement.isChangeStreamPreImageRecordedInOplog() ||
-                (statement.isPreImageRecordedForRetryableInternalTransaction() &&
-                 statement.getNeedsRetryImage() != repl::RetryImageEnum::kPreImage)) {
-                tassert(6278506, "Expected a pre-image", !statement.getPreImage().isEmpty());
-                auto oplogSlot = getNextOplogSlot();
-                prePostImageOplogEntryOplogSlots.push_back(oplogSlot);
-                statement.setPreImageOpTime(oplogSlot);
-            }
-            if (!statement.getPostImage().isEmpty() &&
-                statement.getNeedsRetryImage() != repl::RetryImageEnum::kPostImage) {
-                auto oplogSlot = getNextOplogSlot();
-                prePostImageOplogEntryOplogSlots.push_back(oplogSlot);
-                statement.setPostImageOpTime(oplogSlot);
-            }
-        }
-    }
-
     auto hasNeedsRetryImage = [](const repl::ReplOperation& operation) {
         return static_cast<bool>(operation.getNeedsRetryImage());
     };
@@ -1751,9 +1598,7 @@ getApplyOpsOplogSlotAndOperationAssignmentForTransaction(
     if (prepare) {
         applyOpsEntries.back().oplogSlot = oplogSlots.back();
     }
-    return {std::move(prePostImageOplogEntryOplogSlots),
-            std::move(applyOpsEntries),
-            static_cast<size_t>(oplogSlotIter - oplogSlots.begin())};
+    return {std::move(applyOpsEntries), static_cast<size_t>(oplogSlotIter - oplogSlots.begin())};
 }
 
 /**
@@ -1931,8 +1776,7 @@ OpTimeBundle logApplyOps(OperationContext* opCtx,
 // non-empty.
 //
 // The 'applyOpsOperationAssignment' contains BSON serialized transaction statements, their
-// assigment to "applyOps" oplog entries, and oplog slots to be used for writing pre- and post-
-// image oplog entries for a transaction.
+// assigment to "applyOps" oplog entries for a transaction.
 //
 // In the case of writing entries for a prepared transaction, the last oplog entry (i.e. the
 // implicit prepare) will always be written using the last oplog slot given, even if this means
@@ -1945,7 +1789,6 @@ int logOplogEntries(
     const std::vector<OplogSlot>& oplogSlots,
     const OpObserver::ApplyOpsOplogSlotAndOperationAssignment& applyOpsOperationAssignment,
     boost::optional<ImageBundle>* prePostImageToWriteToImageCollection,
-    size_t numberOfPrePostImagesToWrite,
     bool prepare,
     Date_t wallClockTime,
     OplogWriter* oplogWriter) {
@@ -1967,55 +1810,6 @@ int logOplogEntries(
     if (txnParticipant) {
         prevWriteOpTime.writeOpTime = txnParticipant.getLastWriteOpTime();
     }
-    auto currPrePostImageOplogEntryOplogSlot =
-        applyOpsOperationAssignment.prePostImageOplogEntryOplogSlots.begin();
-
-    // We never want to store pre-images or post-images when we're migrating oplog entries from
-    // another replica set.
-    const auto& migrationRecipientInfo = repl::tenantMigrationInfo(opCtx);
-
-    auto logPrePostImageNoopEntry = [&](const repl::ReplOperation& statement,
-                                        const BSONObj& imageDoc) {
-        auto slot = *currPrePostImageOplogEntryOplogSlot;
-        ++currPrePostImageOplogEntryOplogSlot;
-
-        MutableOplogEntry imageEntry;
-        imageEntry.setSessionId(*opCtx->getLogicalSessionId());
-        imageEntry.setTxnNumber(*opCtx->getTxnNumber());
-        imageEntry.setStatementIds(statement.getStatementIds());
-        imageEntry.setOpType(repl::OpTypeEnum::kNoop);
-        imageEntry.setObject(imageDoc);
-        imageEntry.setTid(statement.getTid());
-        imageEntry.setNss(statement.getNss());
-        imageEntry.setUuid(statement.getUuid());
-        imageEntry.setOpTime(slot);
-        imageEntry.setDestinedRecipient(statement.getDestinedRecipient());
-
-        logOperation(opCtx, &imageEntry, true /*assignWallClockTime*/, oplogWriter);
-    };
-
-    if (numberOfPrePostImagesToWrite > 0 && !migrationRecipientInfo) {
-        for (auto& statement : *stmts) {
-            if (statement.isChangeStreamPreImageRecordedInOplog() ||
-                (statement.isPreImageRecordedForRetryableInternalTransaction() &&
-                 statement.getNeedsRetryImage() != repl::RetryImageEnum::kPreImage)) {
-                invariant(!statement.getPreImage().isEmpty());
-
-                // Note that 'needsRetryImage' stores the image kind that needs to stored in the
-                // image collection. Therefore, when 'needsRetryImage' is equal to kPreImage, the
-                // pre-image will be written to the image collection (after all the applyOps oplog
-                // entries are written).
-                logPrePostImageNoopEntry(statement, statement.getPreImage());
-            }
-            if (!statement.getPostImage().isEmpty() &&
-                statement.getNeedsRetryImage() != repl::RetryImageEnum::kPostImage) {
-                // Likewise, when 'needsRetryImage' is equal to kPostImage, the post-image will be
-                // written to the image collection (after all the applyOps oplog entries are
-                // written).
-                logPrePostImageNoopEntry(statement, statement.getPostImage());
-            }
-        }
-    }
 
     // Stores the statement ids of all write statements in the transaction.
     std::vector<StmtId> stmtIdsWritten;
@@ -2225,7 +2019,7 @@ void OpObserverImpl::onUnpreparedTransactionCommit(OperationContext* opCtx,
     // entries.
     const auto applyOpsOplogSlotAndOperationAssignment =
         getApplyOpsOplogSlotAndOperationAssignmentForTransaction(
-            opCtx, oplogSlots, numberOfPrePostImagesToWrite, false /*prepare*/, *statements);
+            opCtx, oplogSlots, false /*prepare*/, *statements);
     const auto wallClockTime = getWallClockTimeForOpLog(opCtx);
 
     // Log in-progress entries for the transaction along with the implicit commit.
@@ -2235,7 +2029,6 @@ void OpObserverImpl::onUnpreparedTransactionCommit(OperationContext* opCtx,
                                           oplogSlots,
                                           applyOpsOplogSlotAndOperationAssignment,
                                           &imageToWrite,
-                                          numberOfPrePostImagesToWrite,
                                           false /* prepare*/,
                                           wallClockTime,
                                           _oplogWriter.get());
@@ -2294,14 +2087,13 @@ void OpObserverImpl::onBatchedWriteCommit(OperationContext* opCtx) {
     // entries.
     const auto applyOpsOplogSlotAndOperationAssignment =
         getApplyOpsOplogSlotAndOperationAssignmentForTransaction(
-            opCtx, oplogSlots, 0 /*numberOfPrePostImagesToWrite*/, false /*prepare*/, batchedOps);
+            opCtx, oplogSlots, false /*prepare*/, batchedOps);
     const auto wallClockTime = getWallClockTimeForOpLog(opCtx);
     logOplogEntries(opCtx,
                     &batchedOps,
                     oplogSlots,
                     applyOpsOplogSlotAndOperationAssignment,
                     &noPrePostImage,
-                    0 /* numberOfPrePostImagesToWrite */,
                     false,
                     wallClockTime,
                     _oplogWriter.get());
@@ -2345,7 +2137,7 @@ OpObserverImpl::preTransactionPrepare(OperationContext* opCtx,
                                       std::vector<repl::ReplOperation>* statements) {
     auto applyOpsOplogSlotAndOperationAssignment =
         getApplyOpsOplogSlotAndOperationAssignmentForTransaction(
-            opCtx, reservedSlots, numberOfPrePostImagesToWrite, true /*prepare*/, *statements);
+            opCtx, reservedSlots, true /*prepare*/, *statements);
     writeChangeStreamPreImagesForTransaction(
         opCtx, *statements, applyOpsOplogSlotAndOperationAssignment, wallClockTime);
     return std::make_unique<OpObserver::ApplyOpsOplogSlotAndOperationAssignment>(
@@ -2398,7 +2190,6 @@ void OpObserverImpl::onTransactionPrepare(
                                     reservedSlots,
                                     *applyOpsOperationAssignment,
                                     &imageToWrite,
-                                    numberOfPrePostImagesToWrite,
                                     true /* prepare */,
                                     wallClockTime,
                                     _oplogWriter.get());
diff --git a/src/mongo/db/op_observer/op_observer_impl_test.cpp b/src/mongo/db/op_observer/op_observer_impl_test.cpp
index 72c81a1f64d..6e120317ca8 100644
--- a/src/mongo/db/op_observer/op_observer_impl_test.cpp
+++ b/src/mongo/db/op_observer/op_observer_impl_test.cpp
@@ -2250,9 +2250,6 @@ const auto kNonFaM = StoreDocOption::None;
 const auto kFaMPre = StoreDocOption::PreImage;
 const auto kFaMPost = StoreDocOption::PostImage;
 
-const bool kRecordPreImages = true;
-const bool kDoNotRecordPreImages = false;
-
 const bool kChangeStreamImagesEnabled = true;
 const bool kChangeStreamImagesDisabled = false;
 
@@ -2264,7 +2261,6 @@ const std::vector<bool> kInMultiDocumentTransactionCases{false, true};
 
 struct UpdateTestCase {
     StoreDocOption imageType;
-    bool alwaysRecordPreImages;
     bool changeStreamImagesEnabled;
     RetryableFindAndModifyLocation retryableOptions;
 
@@ -2307,7 +2303,6 @@ protected:
         LOGV2(5739902,
               "UpdateTestCase",
               "ImageType"_attr = testCase.getImageTypeStr(),
-              "PreImageRecording"_attr = testCase.alwaysRecordPreImages,
               "ChangeStreamPreAndPostImagesEnabled"_attr = testCase.changeStreamImagesEnabled,
               "RetryableFindAndModifyLocation"_attr =
                   testCase.getRetryableFindAndModifyLocationStr(),
@@ -2317,7 +2312,6 @@ protected:
     void initializeOplogUpdateEntryArgs(OperationContext* opCtx,
                                         const UpdateTestCase& testCase,
                                         OplogUpdateEntryArgs* update) {
-        update->updateArgs->preImageRecordingEnabledForCollection = testCase.alwaysRecordPreImages;
         update->updateArgs->changeStreamPreAndPostImagesEnabledForCollection =
             testCase.changeStreamImagesEnabled;
 
@@ -2338,8 +2332,7 @@ protected:
                 break;
         }
         update->updateArgs->preImageDoc = boost::none;
-        if (testCase.imageType == StoreDocOption::PreImage || testCase.alwaysRecordPreImages ||
-            testCase.changeStreamImagesEnabled) {
+        if (testCase.imageType == StoreDocOption::PreImage || testCase.changeStreamImagesEnabled) {
             update->updateArgs->preImageDoc = BSON("_id" << 0 << "preImage" << true);
         }
         update->updateArgs->updatedDoc = BSON("_id" << 0 << "postImage" << true);
@@ -2358,13 +2351,6 @@ protected:
         const boost::optional<OplogEntry>& applyOpsOplogEntry = boost::none) {
         bool checkSideCollection =
             testCase.isFindAndModify() && testCase.retryableOptions == kRecordInSideCollection;
-        if (checkSideCollection && testCase.alwaysRecordPreImages &&
-            testCase.imageType == StoreDocOption::PreImage) {
-            // When `alwaysRecordPreImages` is enabled for a collection, we always store an
-            // image in the oplog. To avoid unnecessary writes, we won't also store an image
-            // in the side collection.
-            checkSideCollection = false;
-        }
         if (checkSideCollection) {
             repl::ImageEntry imageEntry =
                 getImageEntryFromSideCollection(opCtx, *updateOplogEntry.getSessionId());
@@ -2415,25 +2401,19 @@ protected:
 
     std::vector<UpdateTestCase> _cases = {
         // Regular updates.
-        {kNonFaM, kDoNotRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 1},
-        {kNonFaM, kDoNotRecordPreImages, kChangeStreamImagesEnabled, kNotRetryable, 1},
-        {kNonFaM, kDoNotRecordPreImages, kChangeStreamImagesEnabled, kRecordInSideCollection, 1},
-        {kNonFaM, kRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 2},
-        {kNonFaM, kRecordPreImages, kChangeStreamImagesDisabled, kRecordInSideCollection, 2},
+        {kNonFaM, kChangeStreamImagesDisabled, kNotRetryable, 1},
+        {kNonFaM, kChangeStreamImagesEnabled, kNotRetryable, 1},
+        {kNonFaM, kChangeStreamImagesEnabled, kRecordInSideCollection, 1},
         // FindAndModify asking for a preImage.
-        {kFaMPre, kDoNotRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 1},
-        {kFaMPre, kDoNotRecordPreImages, kChangeStreamImagesDisabled, kRecordInSideCollection, 1},
-        {kFaMPre, kDoNotRecordPreImages, kChangeStreamImagesEnabled, kNotRetryable, 1},
-        {kFaMPre, kDoNotRecordPreImages, kChangeStreamImagesEnabled, kRecordInSideCollection, 1},
-        {kFaMPre, kRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 2},
-        {kFaMPre, kRecordPreImages, kChangeStreamImagesDisabled, kRecordInSideCollection, 2},
+        {kFaMPre, kChangeStreamImagesDisabled, kNotRetryable, 1},
+        {kFaMPre, kChangeStreamImagesDisabled, kRecordInSideCollection, 1},
+        {kFaMPre, kChangeStreamImagesEnabled, kNotRetryable, 1},
+        {kFaMPre, kChangeStreamImagesEnabled, kRecordInSideCollection, 1},
         // FindAndModify asking for a postImage.
-        {kFaMPost, kDoNotRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 1},
-        {kFaMPost, kDoNotRecordPreImages, kChangeStreamImagesDisabled, kRecordInSideCollection, 1},
-        {kFaMPost, kDoNotRecordPreImages, kChangeStreamImagesEnabled, kNotRetryable, 1},
-        {kFaMPost, kDoNotRecordPreImages, kChangeStreamImagesEnabled, kRecordInSideCollection, 1},
-        {kFaMPost, kRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 2},
-        {kFaMPost, kRecordPreImages, kChangeStreamImagesDisabled, kRecordInSideCollection, 2}};
+        {kFaMPost, kChangeStreamImagesDisabled, kNotRetryable, 1},
+        {kFaMPost, kChangeStreamImagesDisabled, kRecordInSideCollection, 1},
+        {kFaMPost, kChangeStreamImagesEnabled, kNotRetryable, 1},
+        {kFaMPost, kChangeStreamImagesEnabled, kRecordInSideCollection, 1}};
 
     const NamespaceString _nss{boost::none, "test", "coll"};
     const UUID _uuid = UUID::gen();
@@ -2530,39 +2510,6 @@ TEST_F(OnUpdateOutputsTest, TestFundamentalTransactionOnUpdateOutputs) {
     }
 }
 
-TEST_F(OnUpdateOutputsTest,
-       RetryableInternalTransactionUpdateWithPreImageRecordingEnabledOnShardServerThrows) {
-    // Create a registry that only registers the Impl. It can be challenging to call methods on
-    // the Impl directly. It falls into cases where `ReservedTimes` is expected to be
-    // instantiated. Due to strong encapsulation, we use the registry that managers the
-    // `ReservedTimes` on our behalf.
-    OpObserverRegistry opObserver;
-    opObserver.addObserver(std::make_unique<OpObserverImpl>(std::make_unique<OplogWriterImpl>()));
-
-    auto opCtxRaii = cc().makeOperationContext();
-    OperationContext* opCtx = opCtxRaii.get();
-
-    resetOplogAndTransactions(opCtx);
-
-    std::unique_ptr<MongoDSessionCatalog::Session> contextSession;
-    beginRetryableInternalTransactionWithTxnNumber(opCtx, 0, contextSession);
-
-    CollectionUpdateArgs updateArgs;
-    updateArgs.preImageRecordingEnabledForCollection = true;
-    updateArgs.preImageDoc = BSON("_id" << 0 << "preImage" << true);
-    updateArgs.updatedDoc = BSON("_id" << 0 << "postImage" << true);
-    updateArgs.update =
-        BSON("$set" << BSON("postImage" << true) << "$unset" << BSON("preImage" << 1));
-    updateArgs.criteria = BSON("_id" << 0);
-    OplogUpdateEntryArgs updateEntryArgs(&updateArgs, _nss, _uuid);
-    serverGlobalParams.clusterRole = ClusterRole::ShardServer;
-    ON_BLOCK_EXIT([] { serverGlobalParams.clusterRole = ClusterRole::None; });
-
-    WriteUnitOfWork wuow(opCtx);
-    AutoGetCollection locks(opCtx, _nss, MODE_IX);
-    ASSERT_THROWS_CODE(opObserver.onUpdate(opCtx, updateEntryArgs), DBException, 6462400);
-}
-
 struct InsertTestCase {
     bool isRetryableWrite;
     int numDocsToInsert;
@@ -2660,7 +2607,6 @@ TEST_F(OpObserverTest, TestFundamentalOnInsertsOutputs) {
 }
 
 struct DeleteTestCase {
-    bool alwaysRecordPreImages;
     bool changeStreamImagesEnabled;
     RetryableFindAndModifyLocation retryableOptions;
 
@@ -2730,22 +2676,6 @@ DEATH_TEST_REGEX_F(BatchedWriteOutputsTest,
 }
 
 DEATH_TEST_REGEX_F(BatchedWriteOutputsTest,
-                   TestDoesNotSupportPreImagesInOplog,
-                   "Invariant "
-                   "failure.*getChangeStreamPreImageRecordingMode.*repl::ReplOperation::"
-                   "ChangeStreamPreImageRecordingMode::kOff") {
-    auto opCtxRaii = cc().makeOperationContext();
-    OperationContext* opCtx = opCtxRaii.get();
-    WriteUnitOfWork wuow(opCtx, true /* groupOplogEntries */);
-
-    auto& bwc = BatchedWriteContext::get(opCtx);
-    auto entry = repl::MutableOplogEntry::makeDeleteOperation(_nss, UUID::gen(), BSON("_id" << 0));
-    entry.setChangeStreamPreImageRecordingMode(
-        repl::ReplOperation::ChangeStreamPreImageRecordingMode::kOplog);
-    bwc.addBatchedOperation(opCtx, entry);
-}
-
-DEATH_TEST_REGEX_F(BatchedWriteOutputsTest,
                    TestDoesNotSupportMultiDocTxn,
                    "Invariant failure.*!opCtx->inMultiDocumentTransaction()") {
     auto opCtxRaii = cc().makeOperationContext();
@@ -3284,7 +3214,6 @@ protected:
     void logTestCase(const DeleteTestCase& testCase) {
         LOGV2(5739905,
               "DeleteTestCase",
-              "PreImageRecording"_attr = testCase.alwaysRecordPreImages,
               "ChangeStreamPreAndPostImagesEnabled"_attr = testCase.changeStreamImagesEnabled,
               "RetryableFindAndModifyLocation"_attr =
                   testCase.getRetryableFindAndModifyLocationStr(),
@@ -3294,7 +3223,6 @@ protected:
     void initializeOplogDeleteEntryArgs(OperationContext* opCtx,
                                         const DeleteTestCase& testCase,
                                         OplogDeleteEntryArgs* deleteArgs) {
-        deleteArgs->preImageRecordingEnabledForCollection = testCase.alwaysRecordPreImages;
         deleteArgs->changeStreamPreAndPostImagesEnabledForCollection =
             testCase.changeStreamImagesEnabled;
 
@@ -3306,8 +3234,7 @@ protected:
                 deleteArgs->retryableFindAndModifyLocation = kRecordInSideCollection;
                 break;
         }
-        if (testCase.isRetryable() || testCase.alwaysRecordPreImages ||
-            testCase.changeStreamImagesEnabled) {
+        if (testCase.isRetryable() || testCase.changeStreamImagesEnabled) {
             deleteArgs->deletedDoc = &_deletedDoc;
         }
     }
@@ -3320,8 +3247,7 @@ protected:
         const OplogEntry& deleteOplogEntry,
         const boost::optional<OplogEntry> applyOpsOplogEntry = boost::none) {
         bool didWriteInSideCollection =
-            deleteArgs.retryableFindAndModifyLocation == kRecordInSideCollection &&
-            !deleteArgs.preImageRecordingEnabledForCollection;
+            deleteArgs.retryableFindAndModifyLocation == kRecordInSideCollection;
         if (didWriteInSideCollection) {
             repl::ImageEntry imageEntry =
                 getImageEntryFromSideCollection(opCtx, *deleteOplogEntry.getSessionId());
@@ -3364,13 +3290,10 @@ protected:
         }
     }
 
-    std::vector<DeleteTestCase> _cases{
-        {kDoNotRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 1},
-        {kDoNotRecordPreImages, kChangeStreamImagesDisabled, kRecordInSideCollection, 1},
-        {kDoNotRecordPreImages, kChangeStreamImagesEnabled, kNotRetryable, 1},
-        {kDoNotRecordPreImages, kChangeStreamImagesEnabled, kRecordInSideCollection, 1},
-        {kRecordPreImages, kChangeStreamImagesDisabled, kNotRetryable, 2},
-        {kRecordPreImages, kChangeStreamImagesDisabled, kRecordInSideCollection, 2}};
+    std::vector<DeleteTestCase> _cases{{kChangeStreamImagesDisabled, kNotRetryable, 1},
+                                       {kChangeStreamImagesDisabled, kRecordInSideCollection, 1},
+                                       {kChangeStreamImagesEnabled, kNotRetryable, 1},
+                                       {kChangeStreamImagesEnabled, kRecordInSideCollection, 1}};
 
     const NamespaceString _nss{boost::none, "test", "coll"};
     const UUID _uuid = UUID::gen();
@@ -3477,39 +3400,6 @@ TEST_F(OnDeleteOutputsTest, TestTransactionFundamentalOnDeleteOutputs) {
     }
 }
 
-TEST_F(OnDeleteOutputsTest,
-       RetryableInternalTransactionDeleteWithPreImageRecordingEnabledOnShardServerThrows) {
-    // Create a registry that only registers the Impl. It can be challenging to call methods on
-    // the Impl directly. It falls into cases where `ReservedTimes` is expected to be
-    // instantiated. Due to strong encapsulation, we use the registry that managers the
-    // `ReservedTimes` on our behalf.
-    OpObserverRegistry opObserver;
-    opObserver.addObserver(std::make_unique<OpObserverImpl>(std::make_unique<OplogWriterImpl>()));
-
-    auto opCtxRaii = cc().makeOperationContext();
-    OperationContext* opCtx = opCtxRaii.get();
-
-    resetOplogAndTransactions(opCtx);
-
-    std::unique_ptr<MongoDSessionCatalog::Session> contextSession;
-    beginRetryableInternalTransactionWithTxnNumber(opCtx, 0, contextSession);
-
-    OplogDeleteEntryArgs deleteEntryArgs;
-    deleteEntryArgs.preImageRecordingEnabledForCollection = true;
-    serverGlobalParams.clusterRole = ClusterRole::ShardServer;
-    ON_BLOCK_EXIT([] { serverGlobalParams.clusterRole = ClusterRole::None; });
-
-    WriteUnitOfWork wuow(opCtx);
-    AutoGetCollection locks(opCtx, _nss, MODE_IX);
-    // This test does not call `OpObserver::aboutToDelete`. That method has the side-effect
-    // of setting of `documentKey` on the delete for sharding purposes.
-    // `OpObserverImpl::onDelete` asserts its existence.
-    repl::documentKeyDecoration(opCtx).emplace(_deletedDoc["_id"].wrap(), boost::none);
-    ASSERT_THROWS_CODE(opObserver.onDelete(opCtx, _nss, _uuid, 1 /* stmtId */, deleteEntryArgs),
-                       DBException,
-                       6462401);
-}
-
 class OpObserverMultiEntryTransactionTest : public OpObserverTransactionTest {
     void setUp() override {
         _prevPackingLimit = gMaxNumberOfTransactionOperationsInSingleOplogEntry;
@@ -3684,159 +3574,6 @@ TEST_F(OpObserverMultiEntryTransactionTest, TransactionalUpdateTest) {
     ASSERT_BSONOBJ_EQ(oExpected, oplogEntries[1].getObject());
 }
 
-TEST_F(OpObserverMultiEntryTransactionTest, TransactionPreImageTest) {
-    auto txnParticipant = TransactionParticipant::get(opCtx());
-    txnParticipant.unstashTransactionResources(opCtx(), "txntest");
-
-    CollectionUpdateArgs updateArgs1;
-    const auto updateSpec = BSON("$set" << BSON("data"
-                                                << "x"));
-    const auto updatePreImage = BSON("_id" << 0 << "data"
-                                           << "y");
-    const auto updatePostImage = BSON("_id" << 0 << "data"
-                                            << "x");
-    const auto updateFilter = BSON("_id" << 0);
-
-    updateArgs1.stmtIds = {0};
-    updateArgs1.updatedDoc = updatePostImage;
-    updateArgs1.update = updateSpec;
-    updateArgs1.preImageDoc = updatePreImage;
-    updateArgs1.preImageRecordingEnabledForCollection = true;
-    updateArgs1.criteria = updateFilter;
-    OplogUpdateEntryArgs update1(&updateArgs1, nss1, uuid1);
-
-    WriteUnitOfWork wuow(opCtx());
-    AutoGetCollection autoColl1(opCtx(), nss1, MODE_IX);
-    opObserver().onUpdate(opCtx(), update1);
-
-    const auto deletedDoc = BSON("_id" << 1 << "data"
-                                       << "z");
-    OplogDeleteEntryArgs args;
-    args.deletedDoc = &deletedDoc;
-    args.preImageRecordingEnabledForCollection = true;
-    opObserver().aboutToDelete(opCtx(), nss1, uuid1, deletedDoc);
-    opObserver().onDelete(opCtx(), nss1, uuid1, 0, args);
-
-    auto txnOps = txnParticipant.retrieveCompletedTransactionOperations(opCtx());
-    opObserver().onUnpreparedTransactionCommit(opCtx(), txnOps, 2);
-
-    auto oplogEntryObjs = getNOplogEntries(opCtx(), 4);
-    std::vector<OplogEntry> oplogEntries;
-    mongo::repl::OpTime expectedPrevWriteOpTime;
-    for (const auto& oplogEntryObj : oplogEntryObjs) {
-        oplogEntries.push_back(assertGet(OplogEntry::parse(oplogEntryObj)));
-        const auto& oplogEntry = oplogEntries.back();
-        if (oplogEntry.getOpType() == repl::OpTypeEnum::kNoop) {
-            continue;
-        }
-        checkSessionAndTransactionFields(oplogEntryObj);
-        ASSERT(!oplogEntry.shouldPrepare());
-        ASSERT_TRUE(oplogEntry.getPrevWriteOpTimeInTransaction());
-        ASSERT_EQ(expectedPrevWriteOpTime, *oplogEntry.getPrevWriteOpTimeInTransaction());
-        ASSERT_LT(expectedPrevWriteOpTime.getTimestamp(), oplogEntry.getTimestamp());
-        expectedPrevWriteOpTime = repl::OpTime{oplogEntry.getTimestamp(), *oplogEntry.getTerm()};
-    }
-
-    ASSERT(oplogEntries[0].getOpType() == repl::OpTypeEnum::kNoop);
-    ASSERT_BSONOBJ_EQ(updatePreImage, oplogEntries[0].getObject());
-    ASSERT(oplogEntries[1].getOpType() == repl::OpTypeEnum::kNoop);
-    ASSERT_BSONOBJ_EQ(deletedDoc, oplogEntries[1].getObject());
-    ASSERT_BSONOBJ_EQ(BSON("applyOps"
-                           << BSON_ARRAY(BSON("op"
-                                              << "u"
-                                              << "ns" << nss1.toString() << "ui" << uuid1 << "o"
-                                              << updateSpec << "o2" << BSON("_id" << 0)
-                                              << "preImageOpTime" << oplogEntries[0].getOpTime()))
-                           << "partialTxn" << true),
-                      oplogEntries[2].getObject());
-    ASSERT_BSONOBJ_EQ(BSON("applyOps"
-                           << BSON_ARRAY(BSON("op"
-                                              << "d"
-                                              << "ns" << nss1.toString() << "ui" << uuid1 << "o"
-                                              << BSON("_id" << 1) << "preImageOpTime"
-                                              << oplogEntries[1].getOpTime()))
-                           << "count" << 2),
-                      oplogEntries[3].getObject());
-}
-
-TEST_F(OpObserverMultiEntryTransactionTest, PreparedTransactionPreImageTest) {
-    auto txnParticipant = TransactionParticipant::get(opCtx());
-    txnParticipant.unstashTransactionResources(opCtx(), "txntest");
-
-    CollectionUpdateArgs updateArgs1;
-    const auto updateSpec = BSON("$set" << BSON("data"
-                                                << "x"));
-    const auto updatePreImage = BSON("_id" << 0 << "data"
-                                           << "y");
-    const auto updatePostImage = BSON("_id" << 0 << "data"
-                                            << "x");
-    const auto updateFilter = BSON("_id" << 0);
-
-    updateArgs1.stmtIds = {0};
-    updateArgs1.updatedDoc = updatePostImage;
-    updateArgs1.update = updateSpec;
-    updateArgs1.preImageDoc = updatePreImage;
-    updateArgs1.preImageRecordingEnabledForCollection = true;
-    updateArgs1.criteria = updateFilter;
-    OplogUpdateEntryArgs update1(&updateArgs1, nss1, uuid1);
-
-    AutoGetCollection autoColl1(opCtx(), nss1, MODE_IX);
-    opObserver().onUpdate(opCtx(), update1);
-
-    const auto deletedDoc = BSON("_id" << 1 << "data"
-                                       << "z");
-    OplogDeleteEntryArgs args;
-    args.deletedDoc = &deletedDoc;
-    args.preImageRecordingEnabledForCollection = true;
-    opObserver().aboutToDelete(opCtx(), nss1, uuid1, deletedDoc);
-    opObserver().onDelete(opCtx(), nss1, uuid1, 0, args);
-
-    auto reservedSlots = reserveOpTimesInSideTransaction(opCtx(), 4);
-    auto prepareOpTime = reservedSlots.back();
-    txnParticipant.transitionToPreparedforTest(opCtx(), prepareOpTime);
-    prepareTransaction(reservedSlots, prepareOpTime, 2);
-
-    txnParticipant.stashTransactionResources(opCtx());
-    auto oplogEntryObjs = getNOplogEntries(opCtx(), 4);
-    std::vector<OplogEntry> oplogEntries;
-    mongo::repl::OpTime expectedPrevWriteOpTime;
-    for (const auto& oplogEntryObj : oplogEntryObjs) {
-        oplogEntries.push_back(assertGet(OplogEntry::parse(oplogEntryObj)));
-        const auto& oplogEntry = oplogEntries.back();
-        if (oplogEntry.getOpType() == repl::OpTypeEnum::kNoop) {
-            continue;
-        }
-        checkSessionAndTransactionFields(oplogEntryObj);
-        ASSERT_TRUE(oplogEntry.getPrevWriteOpTimeInTransaction());
-        ASSERT_EQ(expectedPrevWriteOpTime, *oplogEntry.getPrevWriteOpTimeInTransaction());
-        ASSERT_LT(expectedPrevWriteOpTime.getTimestamp(), oplogEntry.getTimestamp());
-        expectedPrevWriteOpTime = repl::OpTime{oplogEntry.getTimestamp(), *oplogEntry.getTerm()};
-    }
-
-    ASSERT(oplogEntries[0].getOpType() == repl::OpTypeEnum::kNoop);
-    ASSERT_BSONOBJ_EQ(updatePreImage, oplogEntries[0].getObject());
-    ASSERT(oplogEntries[1].getOpType() == repl::OpTypeEnum::kNoop);
-    ASSERT_BSONOBJ_EQ(deletedDoc, oplogEntries[1].getObject());
-    ASSERT_BSONOBJ_EQ(BSON("applyOps"
-                           << BSON_ARRAY(BSON("op"
-                                              << "u"
-                                              << "ns" << nss1.toString() << "ui" << uuid1 << "o"
-                                              << updateSpec << "o2" << BSON("_id" << 0)
-                                              << "preImageOpTime" << oplogEntries[0].getOpTime()))
-                           << "partialTxn" << true),
-                      oplogEntries[2].getObject());
-    ASSERT_BSONOBJ_EQ(BSON("applyOps"
-                           << BSON_ARRAY(BSON("op"
-                                              << "d"
-                                              << "ns" << nss1.toString() << "ui" << uuid1 << "o"
-                                              << BSON("_id" << 1) << "preImageOpTime"
-                                              << oplogEntries[1].getOpTime()))
-                           << "prepare" << true << "count" << 2),
-                      oplogEntries[3].getObject());
-
-    txnParticipant.unstashTransactionResources(opCtx(), "abortTransaction");
-}
-
 TEST_F(OpObserverMultiEntryTransactionTest, TransactionalDeleteTest) {
     auto txnParticipant = TransactionParticipant::get(opCtx());
     txnParticipant.unstashTransactionResources(opCtx(), "delete");
diff --git a/src/mongo/db/pipeline/change_stream_event_transform.cpp b/src/mongo/db/pipeline/change_stream_event_transform.cpp
index ef2aad694c5..e04f1f2763e 100644
--- a/src/mongo/db/pipeline/change_stream_event_transform.cpp
+++ b/src/mongo/db/pipeline/change_stream_event_transform.cpp
@@ -144,7 +144,6 @@ std::set<std::string> ChangeStreamDefaultEventTransformation::getFieldNameDepend
                                             repl::OplogEntry::kWallClockTimeFieldName.toString()};
 
     if (_preImageRequested || _postImageRequested) {
-        accessedFields.insert(repl::OplogEntry::kPreImageOpTimeFieldName.toString());
         accessedFields.insert(DocumentSourceChangeStream::kApplyOpsIndexField.toString());
         accessedFields.insert(DocumentSourceChangeStream::kApplyOpsTsField.toString());
     }
@@ -458,20 +457,13 @@ Document ChangeStreamDefaultEventTransformation::applyTransformation(const Docum
     static const std::set<StringData> postImageOps = {DocumentSourceChangeStream::kUpdateOpType};
     if ((_preImageRequested && preImageOps.count(operationType)) ||
         (_postImageRequested && postImageOps.count(operationType))) {
-        auto preImageOpTime = input[repl::OplogEntry::kPreImageOpTimeFieldName];
-        if (!preImageOpTime.missing()) {
-            // Set 'kPreImageIdField' to the pre-image optime. The DSCSAddPreImage stage will use
-            // this optime in order to fetch the pre-image from the oplog.
-            doc.addField(DocumentSourceChangeStream::kPreImageIdField, std::move(preImageOpTime));
-        } else {
-            // Set 'kPreImageIdField' to the 'ChangeStreamPreImageId'. The DSCSAddPreImage stage
-            // will use the id in order to fetch the pre-image from the pre-images collection.
-            const auto preImageId = ChangeStreamPreImageId(
-                uuid.getUuid(),
-                applyOpsEntryTs.missing() ? ts.getTimestamp() : applyOpsEntryTs.getTimestamp(),
-                applyOpsIndex.missing() ? 0 : applyOpsIndex.getLong());
-            doc.addField(DocumentSourceChangeStream::kPreImageIdField, Value(preImageId.toBSON()));
-        }
+        // Set 'kPreImageIdField' to the 'ChangeStreamPreImageId'. The DSCSAddPreImage stage
+        // will use the id in order to fetch the pre-image from the pre-images collection.
+        const auto preImageId = ChangeStreamPreImageId(
+            uuid.getUuid(),
+            applyOpsEntryTs.missing() ? ts.getTimestamp() : applyOpsEntryTs.getTimestamp(),
+            applyOpsIndex.missing() ? 0 : applyOpsIndex.getLong());
+        doc.addField(DocumentSourceChangeStream::kPreImageIdField, Value(preImageId.toBSON()));
     }
 
     // Add the 'ns' field to the change stream document, based on the final value of 'nss'.
diff --git a/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp b/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp
index 24f1e506991..4bc23d7920d 100644
--- a/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp
+++ b/src/mongo/db/pipeline/change_stream_helpers_legacy.cpp
@@ -49,43 +49,6 @@
 namespace mongo {
 namespace change_stream_legacy {
 
-// TODO SERVER-60919 remove 'legacyLookupPreImage' function.
-boost::optional<Document> legacyLookupPreImage(boost::intrusive_ptr<ExpressionContext> pExpCtx,
-                                               const Document& preImageId) {
-    // We need the oplog's UUID for lookup, so obtain the collection info via MongoProcessInterface.
-    auto localOplogInfo = pExpCtx->mongoProcessInterface->getCollectionOptions(
-        pExpCtx->opCtx, NamespaceString::kRsOplogNamespace);
-
-    // Extract the UUID from the collection information. We should always have a valid uuid here.
-    auto oplogUUID = invariantStatusOK(UUID::parse(localOplogInfo["uuid"]));
-
-    // Look up the pre-image oplog entry using the opTime as the query filter.
-    const auto opTime = repl::OpTime::parse(preImageId.toBson());
-    auto lookedUpDoc =
-        pExpCtx->mongoProcessInterface->lookupSingleDocument(pExpCtx,
-                                                             NamespaceString::kRsOplogNamespace,
-                                                             oplogUUID,
-                                                             Document{opTime.asQuery()},
-                                                             boost::none);
-
-    // Return boost::none to signify that we failed to find the pre-image.
-    if (!lookedUpDoc) {
-        return boost::none;
-    }
-
-    // If we had an optime to look up, and we found an oplog entry with that timestamp, then we
-    // should always have a valid no-op entry containing a valid, non-empty pre-image document.
-    auto opLogEntry = uassertStatusOK(repl::OplogEntry::parse(lookedUpDoc->toBson()));
-    tassert(5868901,
-            "Oplog entry type must be a no-op",
-            opLogEntry.getOpType() == repl::OpTypeEnum::kNoop);
-    tassert(5868902,
-            "Oplog entry must contait a non-empty pre-image document",
-            !opLogEntry.getObject().isEmpty());
-
-    return Document{opLogEntry.getObject().getOwned()};
-}
-
 // TODO SERVER-66138: This function can be removed after we branch for 7.0.
 void populateInternalOperationFilter(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                      BSONArrayBuilder* orBuilder) {
diff --git a/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.cpp b/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.cpp
index 934d353ce9f..583e3198a42 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.cpp
@@ -90,19 +90,8 @@ DocumentSource::GetNextResult DocumentSourceChangeStreamAddPreImage::doGetNext()
         return input;
     }
 
-    // If a pre-image is available, the transform stage will have populated it in the event's
-    // 'fullDocumentBeforeChange' field. If this field is missing and the pre-imaging mode is
-    // 'required', we throw an exception. Otherwise, we pass along the document unmodified.
     auto preImageId = input.getDocument()[kPreImageIdFieldName];
-    if (preImageId.missing()) {
-        uassert(51770,
-                str::stream()
-                    << "Change stream was configured to require a pre-image for all update, delete "
-                       "and replace events, but pre-image id was not available for event: "
-                    << makePreImageNotFoundErrorMsg(input.getDocument()),
-                _fullDocumentBeforeChangeMode != FullDocumentBeforeChangeModeEnum::kRequired);
-        return input;
-    }
+    tassert(6091900, "Pre-image id field is missing", !preImageId.missing());
     tassert(5868900,
             "Expected pre-image id field to be a document",
             preImageId.getType() == BSONType::Object);
@@ -130,12 +119,6 @@ DocumentSource::GetNextResult DocumentSourceChangeStreamAddPreImage::doGetNext()
 
 boost::optional<Document> DocumentSourceChangeStreamAddPreImage::lookupPreImage(
     boost::intrusive_ptr<ExpressionContext> pExpCtx, const Document& preImageId) {
-    // If the pre-image id does not contain the nsUUID field, then it is in legacy format. Look
-    // up the pre-image in the oplog.
-    if (preImageId[ChangeStreamPreImageId::kNsUUIDFieldName].missing()) {
-        return change_stream_legacy::legacyLookupPreImage(pExpCtx, preImageId);
-    }
-
     // Look up the pre-image document on the local node by id.
     auto lookedUpDoc = pExpCtx->mongoProcessInterface->lookupSingleDocumentLocally(
         pExpCtx,
diff --git a/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.h b/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.h
index 3ae0bf03763..ba2214ff702 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.h
+++ b/src/mongo/db/pipeline/document_source_change_stream_add_pre_image.h
@@ -37,9 +37,8 @@ namespace mongo {
 /**
  * Part of the change stream API machinery used to look up the pre-image of a document.
  *
- * After a document that should have its pre-image included is transformed from the oplog,
- * its "fullDocumentBeforeChange" field shall be the optime of the noop oplog entry containing the
- * pre-image. This stage replaces that field with the actual pre-image document.
+ * The identifier of pre-image is in "preImageId" field of the incoming document. The pre-image is
+ * set to "fullDocumentBeforeChange" field of the returned document.
  */
 class DocumentSourceChangeStreamAddPreImage final : public DocumentSource {
 public:
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 3b98a53104f..96f0715cb11 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_test.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream_test.cpp
@@ -3242,303 +3242,6 @@ TEST_F(ChangeStreamStageDBTest, TransformDropDatabaseShowExpandedEvents) {
     checkTransformation(dropDB, expectedDropDatabase, kShowExpandedEventsSpec, expectedInvalidate);
 }
 
-TEST_F(ChangeStreamStageTest, TransformPreImageForDelete) {
-    // Set the pre-image opTime to 1 second prior to the default event optime.
-    repl::OpTime preImageOpTime{Timestamp(kDefaultTs.getSecs() - 1, 1), 1};
-    const auto preImageObj = BSON("_id" << 1 << "x" << 2);
-
-    // The documentKey for the main change stream event.
-    const auto documentKey = BSON("_id" << 1);
-
-    // The mock oplog UUID used by MockMongoInterface.
-    auto oplogUUID = MockMongoInterface::oplogUuid();
-
-    // Create an oplog entry for the pre-image no-op event.
-    auto preImageEntry = makeOplogEntry(OpTypeEnum::kNoop,
-                                        NamespaceString::kRsOplogNamespace,
-                                        preImageObj,    // o
-                                        oplogUUID,      // uuid
-                                        boost::none,    // fromMigrate
-                                        boost::none,    // o2
-                                        preImageOpTime  // opTime
-    );
-
-    // Create an oplog entry for the delete event that will look up the pre-image.
-    auto deleteEntry = makeOplogEntry(OpTypeEnum::kDelete,
-                                      nss,
-                                      documentKey,     // o
-                                      testUuid(),      // uuid
-                                      boost::none,     // fromMigrate
-                                      boost::none,     // o2
-                                      kDefaultOpTime,  // opTime
-                                      {},              // sessionInfo
-                                      {},              // prevOpTime
-                                      preImageOpTime   // preImageOpTime
-    );
-
-    // Add the preImage oplog entry into a vector of documents that will be looked up. Add a dummy
-    // entry before it so that we know we are finding the pre-image based on the given timestamp.
-    repl::OpTime dummyOpTime{preImageOpTime.getTimestamp(), repl::OpTime::kInitialTerm};
-    std::vector<Document> documentsForLookup = {Document{dummyOpTime.toBSON()},
-                                                Document{preImageEntry.getEntry().toBSON()}};
-
-    // When run with {fullDocumentBeforeChange: "off"}, we do not see a pre-image even if available.
-    auto spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                             << "off"));
-    Document expectedDeleteNoPreImage{
-        {DSChangeStream::kIdField,
-         makeResumeToken(kDefaultTs, testUuid(), documentKey, DSChangeStream::kDeleteOpType)},
-        {DSChangeStream::kOperationTypeField, DSChangeStream::kDeleteOpType},
-        {DSChangeStream::kClusterTimeField, kDefaultTs},
-        {DSChangeStream::kWallTimeField, Date_t()},
-        {DSChangeStream::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {DSChangeStream::kDocumentKeyField, documentKey},
-    };
-    checkTransformation(
-        deleteEntry, expectedDeleteNoPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "whenAvailable"}, we see the pre-image.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "whenAvailable"));
-    Document expectedDeleteWithPreImage{
-        {DSChangeStream::kIdField,
-         makeResumeToken(kDefaultTs, testUuid(), documentKey, DSChangeStream::kDeleteOpType)},
-        {DSChangeStream::kOperationTypeField, DSChangeStream::kDeleteOpType},
-        {DSChangeStream::kClusterTimeField, kDefaultTs},
-        {DSChangeStream::kWallTimeField, Date_t()},
-        {DSChangeStream::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {DSChangeStream::kDocumentKeyField, documentKey},
-        {DSChangeStream::kFullDocumentBeforeChangeField, preImageObj},
-    };
-    checkTransformation(
-        deleteEntry, expectedDeleteWithPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "required"}, we see the pre-image.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "required"));
-    checkTransformation(
-        deleteEntry, expectedDeleteWithPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "whenAvailable"} but no pre-image is available, the
-    // output 'fullDocumentBeforeChange' field is explicitly set to 'null'.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "whenAvailable"));
-    MutableDocument expectedDeleteWithNullPreImage(expectedDeleteNoPreImage);
-    expectedDeleteWithNullPreImage.addField(DSChangeStream::kFullDocumentBeforeChangeField,
-                                            Value(BSONNULL));
-    checkTransformation(deleteEntry, expectedDeleteWithNullPreImage.freeze(), spec);
-
-    // When run with {fullDocumentBeforeChange: "required"} but we cannot find the pre-image, we
-    // throw NoMatchingDocument.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "required"));
-    ASSERT_THROWS_CODE(checkTransformation(deleteEntry, boost::none, spec),
-                       AssertionException,
-                       ErrorCodes::NoMatchingDocument);
-}
-
-TEST_F(ChangeStreamStageTest, TransformPreImageForUpdate) {
-    // Set the pre-image opTime to 1 second prior to the default event optime.
-    repl::OpTime preImageOpTime{Timestamp(kDefaultTs.getSecs() - 1, 1), 1};
-
-    // Define the pre-image object, the update operation spec, and the document key.
-    const auto updateSpec = BSON("$unset" << BSON("x" << 1));
-    const auto preImageObj = BSON("_id" << 1 << "x" << 2);
-    const auto documentKey = BSON("_id" << 1);
-
-    // The mock oplog UUID used by MockMongoInterface.
-    auto oplogUUID = MockMongoInterface::oplogUuid();
-
-    // Create an oplog entry for the pre-image no-op event.
-    auto preImageEntry = makeOplogEntry(OpTypeEnum::kNoop,
-                                        NamespaceString::kRsOplogNamespace,
-                                        preImageObj,    // o
-                                        oplogUUID,      // uuid
-                                        boost::none,    // fromMigrate
-                                        boost::none,    // o2
-                                        preImageOpTime  // opTime
-    );
-
-    // Create an oplog entry for the update event that will look up the pre-image.
-    auto updateEntry = makeOplogEntry(OpTypeEnum::kUpdate,
-                                      nss,
-                                      updateSpec,      // o
-                                      testUuid(),      // uuid
-                                      boost::none,     // fromMigrate
-                                      documentKey,     // o2
-                                      kDefaultOpTime,  // opTime
-                                      {},              // sessionInfo
-                                      {},              // prevOpTime
-                                      preImageOpTime   // preImageOpTime
-    );
-
-    // Add the preImage oplog entry into a vector of documents that will be looked up. Add a dummy
-    // entry before it so that we know we are finding the pre-image based on the given timestamp.
-    repl::OpTime dummyOpTime{preImageOpTime.getTimestamp(), repl::OpTime::kInitialTerm};
-    std::vector<Document> documentsForLookup = {Document{dummyOpTime.toBSON()},
-                                                Document{preImageEntry.getEntry().toBSON()}};
-
-    // When run with {fullDocumentBeforeChange: "off"}, we do not see a pre-image even if available.
-    auto spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                             << "off"));
-    Document expectedUpdateNoPreImage{
-        {DSChangeStream::kIdField,
-         makeResumeToken(kDefaultTs, testUuid(), documentKey, DSChangeStream::kUpdateOpType)},
-        {DSChangeStream::kOperationTypeField, DSChangeStream::kUpdateOpType},
-        {DSChangeStream::kClusterTimeField, kDefaultTs},
-        {DSChangeStream::kWallTimeField, Date_t()},
-        {DSChangeStream::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {DSChangeStream::kDocumentKeyField, documentKey},
-        {
-            "updateDescription",
-            D{{"updatedFields", D{}}, {"removedFields", vector<V>{V("x"_sd)}}},
-        },
-    };
-    checkTransformation(
-        updateEntry, expectedUpdateNoPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "whenAvailable"}, we see the pre-image.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "whenAvailable"));
-    Document expectedUpdateWithPreImage{
-        {DSChangeStream::kIdField,
-         makeResumeToken(kDefaultTs, testUuid(), documentKey, DSChangeStream::kUpdateOpType)},
-        {DSChangeStream::kOperationTypeField, DSChangeStream::kUpdateOpType},
-        {DSChangeStream::kClusterTimeField, kDefaultTs},
-        {DSChangeStream::kWallTimeField, Date_t()},
-        {DSChangeStream::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {DSChangeStream::kDocumentKeyField, documentKey},
-        {
-            "updateDescription",
-            D{{"updatedFields", D{}}, {"removedFields", vector<V>{V("x"_sd)}}},
-        },
-        {DSChangeStream::kFullDocumentBeforeChangeField, preImageObj},
-    };
-    checkTransformation(
-        updateEntry, expectedUpdateWithPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "required"}, we see the pre-image.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "required"));
-    checkTransformation(
-        updateEntry, expectedUpdateWithPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "whenAvailable"} but no pre-image is available, the
-    // output 'fullDocumentBeforeChange' field is explicitly set to 'null'.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "whenAvailable"));
-    MutableDocument expectedUpdateWithNullPreImage(expectedUpdateNoPreImage);
-    expectedUpdateWithNullPreImage.addField(DSChangeStream::kFullDocumentBeforeChangeField,
-                                            Value(BSONNULL));
-    checkTransformation(updateEntry, expectedUpdateWithNullPreImage.freeze(), spec);
-
-    // When run with {fullDocumentBeforeChange: "required"} but we cannot find the pre-image, we
-    // throw NoMatchingDocument.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "required"));
-    ASSERT_THROWS_CODE(checkTransformation(updateEntry, boost::none, spec),
-                       AssertionException,
-                       ErrorCodes::NoMatchingDocument);
-}
-
-TEST_F(ChangeStreamStageTest, TransformPreImageForReplace) {
-    // Set the pre-image opTime to 1 second prior to the default event optime.
-    repl::OpTime preImageOpTime{Timestamp(kDefaultTs.getSecs() - 1, 1), 1};
-
-    // Define the pre-image object, the replacement document, and the document key.
-    const auto replacementDoc = BSON("_id" << 1 << "y" << 3);
-    const auto preImageObj = BSON("_id" << 1 << "x" << 2);
-    const auto documentKey = BSON("_id" << 1);
-
-    // The mock oplog UUID used by MockMongoInterface.
-    auto oplogUUID = MockMongoInterface::oplogUuid();
-
-    // Create an oplog entry for the pre-image no-op event.
-    auto preImageEntry = makeOplogEntry(OpTypeEnum::kNoop,
-                                        NamespaceString::kRsOplogNamespace,
-                                        preImageObj,    // o
-                                        oplogUUID,      // uuid
-                                        boost::none,    // fromMigrate
-                                        boost::none,    // o2
-                                        preImageOpTime  // opTime
-    );
-
-    // Create an oplog entry for the replacement event that will look up the pre-image.
-    auto replaceEntry = makeOplogEntry(OpTypeEnum::kUpdate,
-                                       nss,
-                                       replacementDoc,  // o
-                                       testUuid(),      // uuid
-                                       boost::none,     // fromMigrate
-                                       documentKey,     // o2
-                                       kDefaultOpTime,  // opTime
-                                       {},              // sessionInfo
-                                       {},              // prevOpTime
-                                       preImageOpTime   // preImageOpTime
-    );
-
-    // Add the preImage oplog entry into a vector of documents that will be looked up. Add a dummy
-    // entry before it so that we know we are finding the pre-image based on the given timestamp.
-    repl::OpTime dummyOpTime{preImageOpTime.getTimestamp(), repl::OpTime::kInitialTerm};
-    std::vector<Document> documentsForLookup = {Document{dummyOpTime.toBSON()},
-                                                Document{preImageEntry.getEntry().toBSON()}};
-
-    // When run with {fullDocumentBeforeChange: "off"}, we do not see a pre-image even if available.
-    auto spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                             << "off"));
-    Document expectedReplaceNoPreImage{
-        {DSChangeStream::kIdField,
-         makeResumeToken(kDefaultTs, testUuid(), documentKey, DSChangeStream::kReplaceOpType)},
-        {DSChangeStream::kOperationTypeField, DSChangeStream::kReplaceOpType},
-        {DSChangeStream::kClusterTimeField, kDefaultTs},
-        {DSChangeStream::kWallTimeField, Date_t()},
-        {DSChangeStream::kFullDocumentField, replacementDoc},
-        {DSChangeStream::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {DSChangeStream::kDocumentKeyField, documentKey},
-    };
-    checkTransformation(
-        replaceEntry, expectedReplaceNoPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "whenAvailable"}, we see the pre-image.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "whenAvailable"));
-    Document expectedReplaceWithPreImage{
-        {DSChangeStream::kIdField,
-         makeResumeToken(kDefaultTs, testUuid(), documentKey, DSChangeStream::kReplaceOpType)},
-        {DSChangeStream::kOperationTypeField, DSChangeStream::kReplaceOpType},
-        {DSChangeStream::kClusterTimeField, kDefaultTs},
-        {DSChangeStream::kWallTimeField, Date_t()},
-        {DSChangeStream::kFullDocumentField, replacementDoc},
-        {DSChangeStream::kNamespaceField, D{{"db", nss.db()}, {"coll", nss.coll()}}},
-        {DSChangeStream::kDocumentKeyField, documentKey},
-        {DSChangeStream::kFullDocumentBeforeChangeField, preImageObj},
-    };
-    checkTransformation(
-        replaceEntry, expectedReplaceWithPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "required"}, we see the pre-image.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "required"));
-    checkTransformation(
-        replaceEntry, expectedReplaceWithPreImage, spec, boost::none, {}, documentsForLookup);
-
-    // When run with {fullDocumentBeforeChange: "whenAvailable"} but no pre-image is available, the
-    // output 'fullDocumentBeforeChange' field is explicitly set to 'null'.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "whenAvailable"));
-    MutableDocument expectedReplaceWithNullPreImage(expectedReplaceNoPreImage);
-    expectedReplaceWithNullPreImage.addField(DSChangeStream::kFullDocumentBeforeChangeField,
-                                             Value(BSONNULL));
-    checkTransformation(replaceEntry, expectedReplaceWithNullPreImage.freeze(), spec);
-
-    // When run with {fullDocumentBeforeChange: "required"} but we cannot find the pre-image, we
-    // throw NoMatchingDocument.
-    spec = BSON("$changeStream" << BSON("fullDocumentBeforeChange"
-                                        << "required"));
-    ASSERT_THROWS_CODE(checkTransformation(replaceEntry, boost::none, spec),
-                       AssertionException,
-                       ErrorCodes::NoMatchingDocument);
-}
-
 TEST_F(ChangeStreamStageDBTest, MatchFiltersOperationsOnSystemCollections) {
     NamespaceString systemColl(nss.db() + ".system.users");
     OplogEntry insert = makeOplogEntry(OpTypeEnum::kInsert, systemColl, BSON("_id" << 1));
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index 03365f0db74..b17e5f108ec 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -622,12 +622,6 @@ void appendOplogEntryChainInfo(OperationContext* opCtx,
                                const std::vector<StmtId>& stmtIds) {
     invariant(!stmtIds.empty());
 
-    // We sometimes have a pre-image no-op entry even for normal non-retryable writes
-    // if recordPreImages is enabled on the collection.
-    if (!oplogLink->preImageOpTime.isNull()) {
-        oplogEntry->setPreImageOpTime(oplogLink->preImageOpTime);
-    }
-
     // Not a retryable write.
     if (stmtIds.front() == kUninitializedStmtId) {
         // If the statement id is uninitialized, it must be the only one. There cannot also be
@@ -645,9 +639,6 @@ void appendOplogEntryChainInfo(OperationContext* opCtx,
         oplogLink->prevOpTime = txnParticipant.getLastWriteOpTime();
     }
     oplogEntry->setPrevWriteOpTimeInTransaction(oplogLink->prevOpTime);
-    if (!oplogLink->postImageOpTime.isNull()) {
-        oplogEntry->setPostImageOpTime(oplogLink->postImageOpTime);
-    }
 }
 
 namespace {
diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h
index f3af3ff146f..b67364b9653 100644
--- a/src/mongo/db/repl/oplog.h
+++ b/src/mongo/db/repl/oplog.h
@@ -84,20 +84,14 @@ struct OplogLink {
     OplogLink() = default;
 
     OpTime prevOpTime;
-    OpTime preImageOpTime;
-    OpTime postImageOpTime;
 };
 
 /**
- * Set the "lsid", "txnNumber", "stmtId", "prevOpTime", "preImageOpTime" and "postImageOpTime"
- * fields of the oplogEntry based on the given oplogLink for retryable writes (i.e. when
- * stmtIds.front() != kUninitializedStmtId).
+ * Set the "lsid", "txnNumber", "stmtId", "prevOpTime" fields of the oplogEntry based on the given
+ * oplogLink for retryable writes (i.e. when stmtIds.front() != kUninitializedStmtId).
  *
  * If the given oplogLink.prevOpTime is a null OpTime, both the oplogLink.prevOpTime and the
  * "prevOpTime" field of the oplogEntry will be set to the TransactionParticipant's lastWriteOpTime.
- * The "preImageOpTime" field will only be set if the given oplogLink.preImageOpTime is not null.
- * Similarly, the "postImageOpTime" field will only be set if the given oplogLink.postImageOpTime is
- * not null.
  */
 void appendOplogEntryChainInfo(OperationContext* opCtx,
                                MutableOplogEntry* oplogEntry,
diff --git a/src/mongo/db/repl/oplog_applier_impl_test.cpp b/src/mongo/db/repl/oplog_applier_impl_test.cpp
index ff4441515d4..d644d5ad734 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test.cpp
@@ -275,8 +275,12 @@ TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionAndDocExist) {
+    // Setup the pre-images collection.
+    ChangeStreamPreImagesCollectionManager::createPreImagesCollection(_opCtx.get(),
+                                                                      boost::none /* tenantId */);
     const NamespaceString nss("test.t");
-    createCollection(_opCtx.get(), nss, createRecordPreImageCollectionOptions());
+    createCollection(
+        _opCtx.get(), nss, createRecordChangeStreamPreAndPostImagesCollectionOptions());
     ASSERT_OK(getStorageInterface()->insertDocument(_opCtx.get(), nss, {BSON("_id" << 0)}, 0));
     auto op = makeOplogEntry(OpTypeEnum::kDelete, nss, {});
     _testApplyOplogEntryOrGroupedInsertsCrudOperation(ErrorCodes::OK, op, nss, true);
@@ -391,8 +395,11 @@ TEST_F(OplogApplierImplTestEnableSteadyStateConstraints,
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentCollectionLockedByUUID) {
+    // Setup the pre-images collection.
+    ChangeStreamPreImagesCollectionManager::createPreImagesCollection(_opCtx.get(),
+                                                                      boost::none /* tenantId */);
     const NamespaceString nss("test.t");
-    CollectionOptions options = createRecordPreImageCollectionOptions();
+    CollectionOptions options = createRecordChangeStreamPreAndPostImagesCollectionOptions();
     options.uuid = kUuid;
     createCollection(_opCtx.get(), nss, options);
 
@@ -742,6 +749,9 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsInsertDocumentIncorr
 }
 
 TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentIncludesTenantId) {
+    // Setup the pre-images collection.
+    ChangeStreamPreImagesCollectionManager::createPreImagesCollection(_opCtx.get(),
+                                                                      boost::none /* tenantId */);
     RAIIServerParameterControllerForTest multitenancyController("multitenancySupport", true);
     RAIIServerParameterControllerForTest featureFlagController("featureFlagRequireTenantID", true);
     const TenantId tid(OID::gen());
@@ -750,7 +760,7 @@ TEST_F(OplogApplierImplTest, applyOplogEntryOrGroupedInsertsDeleteDocumentInclud
 
     // this allows us to set deleteArgs.deletedDoc needed by the onDeleteFn validation function in
     // _testApplyOplogEntryOrGroupedInsertsCrudOperation below
-    CollectionOptions options = createRecordPreImageCollectionOptions();
+    CollectionOptions options = createRecordChangeStreamPreAndPostImagesCollectionOptions();
 
     repl::createCollection(_opCtx.get(), nss, options);
     ASSERT_OK(getStorageInterface()->insertDocument(_opCtx.get(), nss, {doc}, 0));
diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
index 007d9da4e0e..a24e438feb0 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.cpp
@@ -454,9 +454,9 @@ CollectionOptions createOplogCollectionOptions() {
     return options;
 }
 
-CollectionOptions createRecordPreImageCollectionOptions() {
+CollectionOptions createRecordChangeStreamPreAndPostImagesCollectionOptions() {
     CollectionOptions options;
-    options.recordPreImages = true;
+    options.changeStreamPreAndPostImagesOptions.setEnabled(true);
     return options;
 }
 
diff --git a/src/mongo/db/repl/oplog_applier_impl_test_fixture.h b/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
index e2c374e24b4..1792cb54dba 100644
--- a/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
+++ b/src/mongo/db/repl/oplog_applier_impl_test_fixture.h
@@ -299,10 +299,10 @@ OplogEntry makeOplogEntry(OpTypeEnum opType, NamespaceString nss, boost::optiona
  */
 CollectionOptions createOplogCollectionOptions();
 
-/*
- * Creates collection options for recording pre-images for testing deletes.
+/**
+ * Creates collection options for recording change stream pre-images for testing deletes.
  */
-CollectionOptions createRecordPreImageCollectionOptions();
+CollectionOptions createRecordChangeStreamPreAndPostImagesCollectionOptions();
 
 /**
  * Create test collection.
diff --git a/src/mongo/db/repl/oplog_entry.h b/src/mongo/db/repl/oplog_entry.h
index 07c63bb58b7..8be2baff48b 100644
--- a/src/mongo/db/repl/oplog_entry.h
+++ b/src/mongo/db/repl/oplog_entry.h
@@ -78,9 +78,6 @@ public:
 
         // The pre-image is recorded in the change stream pre-images collection.
         kPreImagesCollection,
-
-        // The pre-image is recorded in the oplog as a separate entry.
-        kOplog,
     };
 
     static ReplOperation parse(const IDLParserContext& ctxt, const BSONObj& bsonObject) {
@@ -138,15 +135,6 @@ public:
     }
 
     /**
-     * Returns true if the change stream pre-image is recorded in a dedicated oplog entry for this
-     * operation.
-     */
-    bool isChangeStreamPreImageRecordedInOplog() const {
-        return ReplOperation::ChangeStreamPreImageRecordingMode::kOplog ==
-            getChangeStreamPreImageRecordingMode();
-    }
-
-    /**
      * Returns true if the change stream pre-image is recorded in the change stream pre-images
      * collection for this operation.
      */
diff --git a/src/mongo/db/s/README.md b/src/mongo/db/s/README.md
index ec15a74f525..8e64b59b78f 100644
--- a/src/mongo/db/s/README.md
+++ b/src/mongo/db/s/README.md
@@ -964,24 +964,16 @@ to assign to the forged no-op oplog entry as result of the "downconvert", we mus
 extra oplog slot when writing the original retryable findAndModify oplog entry with
 `needsRetryImage: true`.
 
-The server also supports [collection-level pre-images](https://docs.mongodb.com/realm/mongodb/trigger-preimages/#overview), a feature used by MongoDB Realm. This feature continues to store document pre-images in
-the oplog, and is expected to work with the new retryable findAndModify behavior described above.
-With this, it's possible that for a particular update operation, we must store a pre-image
-(for collection-level pre-images) in the oplog while storing the post-image
-(for retryable findAndModify) in `config.image_collection`. In order to avoid certain WiredTiger
-constraints surrounding setting multiple timestamps in a single storage transaction, we must reserve
+In order to avoid certain WiredTiger constraints surrounding setting multiple timestamps in a single storage transaction, we must reserve
 oplog slots before entering the OpObserver, which is where we would normally create an oplog entry
 and assign it the next available timestamp. Here, we have a table that describes the different
 scenarios, along with the timestamps that are reserved and the oplog entries assigned to each of
 those timestamps:
-| Parameters | NumSlotsReserved | TS - 2 | TS - 1 | TS | Oplog fields for entry with timestamp: TS |
-| --- | --- | --- | --- | --- | --- |
-| Update, NeedsRetryImage=postImage, preImageRecordingEnabled = True | 3 | No-op oplog entry storing the pre-image|Reserved for forged no-op entry eventually used by tenant migrations/resharding | Update oplog entry | NeedsRetryImage: postImage, preImageOpTime: \{TS - 2} |
-| Update, NeedsRetryImage=preImage, preImageRecordingEnabled=True | 3 |No-op oplog entry storing the pre-image | Reserved but will not be used|Update Oplog entry | preImageOpTime: \{TS - 2} |
-| Update, NeedsRetryImage=preImage, preImageRecordingEnabled=False | 2 | N/A | Reserved for forged no-op entry eventually used by tenant migrations/resharding|Update oplog entry|NeedsRetryImage: preImage |
-| Update, NeedsRetryImage=postImage, preImageRecordingEnabled=False | 2 | N/A | Reserved for forged no-op entry eventually used by tenant migrations/resharding|Update oplog entry | NeedsRetryImage: postImage |
-| Delete, NeedsRetryImage=preImage, preImageRecordingEnabled = True | 0. Note that the OpObserver will still create a no-op entry along with the delete oplog entry, assigning them the next two available timestamps (TS - 1 and TS respectively). | N/A | No-op oplog entry storing the pre-image|Delete oplog entry | preImageOpTime: \{TS - 1} |
-|Delete, NeedsRetryImage=preImage, preImageRecordingEnabled = False|2|N/A|Reserved for forged no-op entry eventually used by tenant migrations/resharding|Delete oplog entry|NeedsRetryImage: preImage|
+| Parameters | NumSlotsReserved | TS - 1 | TS | Oplog fields for entry with timestamp: TS |
+| --- | --- | --- | --- | --- |
+| Update, NeedsRetryImage=preImage | 2 | Reserved for forged no-op entry eventually used by tenant migrations/resharding|Update oplog entry|NeedsRetryImage: preImage |
+| Update, NeedsRetryImage=postImage | 2 | Reserved for forged no-op entry eventually used by tenant migrations/resharding|Update oplog entry | NeedsRetryImage: postImage |
+|Delete, NeedsRetryImage=preImage |2|Reserved for forged no-op entry eventually used by tenant migrations/resharding|Delete oplog entry|NeedsRetryImage: preImage|
 
 #### Code references
 * [**TransactionParticipant class**](https://github.com/mongodb/mongo/blob/r4.3.4/src/mongo/db/transaction_participant.h)
diff --git a/src/mongo/db/timeseries/timeseries_collmod.cpp b/src/mongo/db/timeseries/timeseries_collmod.cpp
index a58e1462348..c90240a5695 100644
--- a/src/mongo/db/timeseries/timeseries_collmod.cpp
+++ b/src/mongo/db/timeseries/timeseries_collmod.cpp
@@ -74,7 +74,6 @@ std::unique_ptr<CollMod> makeTimeseriesBucketsCollModCommand(OperationContext* o
     request.setValidationAction(origCmd.getValidationAction());
     request.setViewOn(origCmd.getViewOn());
     request.setPipeline(origCmd.getPipeline());
-    request.setRecordPreImages(origCmd.getRecordPreImages());
     request.setChangeStreamPreAndPostImages(origCmd.getChangeStreamPreAndPostImages());
     request.setExpireAfterSeconds(origCmd.getExpireAfterSeconds());
     request.setTimeseries(origCmd.getTimeseries());
diff --git a/src/mongo/db/transaction/transaction_participant.cpp b/src/mongo/db/transaction/transaction_participant.cpp
index 110f76ed126..1a811e6f68b 100644
--- a/src/mongo/db/transaction/transaction_participant.cpp
+++ b/src/mongo/db/transaction/transaction_participant.cpp
@@ -1759,8 +1759,7 @@ void TransactionParticipant::Participant::addTransactionOperation(
         repl::DurableOplogEntry::getDurableReplOperationSize(operation);
     if (!operation.getPreImage().isEmpty()) {
         p().transactionOperationBytes += operation.getPreImage().objsize();
-        if (operation.isChangeStreamPreImageRecordedInOplog() ||
-            operation.isPreImageRecordedForRetryableInternalTransaction()) {
+        if (operation.isPreImageRecordedForRetryableInternalTransaction()) {
             ++p().numberOfPrePostImagesToWrite;
         }
     }