SERVER-60957 Support change stream pre and post images in a sharded cluster.

3 files changed, 166 insertions, 14 deletions

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
new file mode 100644
index 00000000000..c9b959abee1
--- /dev/null
+++ b/jstests/change_streams/change_stream_pre_image_lookup_whole_db_whole_cluster.js
@@ -0,0 +1,132 @@
+/**
+ * Tests that a whole-db or whole-cluster change stream can succeed when the
+ * "fullDocumentBeforeChange" option is set to "required", so long as the user
+ * specifies a pipeline that filters out changes to any collections which do not
+ * have pre-images enabled.
+ *
+ * @tags: [
+ *   uses_change_streams,
+ *   # TODO SERVER-58694: remove this tag.
+ *   change_stream_does_not_expect_txns,
+ *   # TODO SERVER-60238: remove this tag.
+ *   assumes_read_preference_unchanged
+ * ]
+ */
+(function() {
+"use strict";
+
+load("jstests/libs/change_stream_util.js");  // For canRecordPreImagesInConfigDatabase.
+load("jstests/libs/fixture_helpers.js");     // For FixtureHelpers.
+
+const testDB = db.getSiblingDB(jsTestName());
+const adminDB = db.getSiblingDB("admin");
+const collWithPreImageName = "coll_with_pre_images";
+const collWithNoPreImageName = "coll_with_no_pre_images";
+const canRecordPreImagesInConfigDb = canRecordPreImagesInConfigDatabase(testDB);
+
+if (!canRecordPreImagesInConfigDb && FixtureHelpers.isMongos(db)) {
+    jsTestLog("Skipping test as pre image lookup is not supported in sharded cluster with feature" +
+              "flag 'featureFlagChangeStreamPreAndPostImages' disabled.");
+    return;
+}
+
+assert.commandWorked(testDB.dropDatabase());
+
+// Create one collection that has pre-image recording enabled...
+if (!canRecordPreImagesInConfigDb) {
+    assert.commandWorked(testDB.createCollection(collWithPreImageName, {recordPreImages: true}));
+} else {
+    assert.commandWorked(testDB.createCollection(collWithPreImageName,
+                                                 {changeStreamPreAndPostImages: {enabled: true}}));
+}
+
+//... and one collection which has pre-images disabled.
+if (!canRecordPreImagesInConfigDb) {
+    assert.commandWorked(testDB.createCollection(collWithNoPreImageName, {recordPreImages: false}));
+} else {
+    assert.commandWorked(testDB.createCollection(collWithNoPreImageName,
+                                                 {changeStreamPreAndPostImages: {enabled: false}}));
+}
+
+const collWithPreImages = testDB.coll_with_pre_images;
+const collWithNoPreImages = testDB.coll_with_no_pre_images;
+
+//... and a collection that will hold the sentinal document that marks the end of changes
+const sentinelColl = testDB.sentinelColl;
+
+// Insert one document as a starting point and extract its resume token.
+const resumeToken = (() => {
+    const csCursor = collWithNoPreImages.watch();
+    assert.commandWorked(collWithNoPreImages.insert({_id: -1}));
+    assert.soon(() => csCursor.hasNext());
+    return csCursor.next()._id;
+})();
+
+// Write a series of interleaving operations to each collection.
+assert.commandWorked(collWithNoPreImages.insert({_id: 0}));
+assert.commandWorked(collWithPreImages.insert({_id: 0}));
+
+assert.commandWorked(collWithNoPreImages.update({_id: 0}, {foo: "bar"}));
+assert.commandWorked(collWithPreImages.update({_id: 0}, {foo: "bar"}));
+
+assert.commandWorked(collWithNoPreImages.update({_id: 0}, {$set: {foo: "baz"}}));
+assert.commandWorked(collWithPreImages.update({_id: 0}, {$set: {foo: "baz"}}));
+
+assert.commandWorked(collWithNoPreImages.remove({_id: 0}));
+assert.commandWorked(collWithPreImages.remove({_id: 0}));
+
+// This will generate an insert change event we can wait for on the change stream that indicates
+// we have reached the end of changes this test is interested in.
+assert.commandWorked(sentinelColl.insert({_id: "last_change_sentinel"}));
+
+// Confirm that attempting to open a whole-db stream on this database with mode "required" fails.
+assert.throwsWithCode(function() {
+    const wholeDBStream =
+        testDB.watch([], {fullDocumentBeforeChange: "required", resumeAfter: resumeToken});
+
+    return assert.soon(() => wholeDBStream.hasNext() &&
+                           wholeDBStream.next().documentKey._id === "last_change_sentinel");
+}, [ErrorCodes.NoMatchingDocument, 51770]);
+
+// Confirm that attempting to open a whole-cluster stream on with mode "required" fails.
+assert.throwsWithCode(function() {
+    const wholeClusterStream = adminDB.watch([], {
+        fullDocumentBeforeChange: "required",
+        resumeAfter: resumeToken,
+        allChangesForCluster: true,
+    });
+
+    return assert.soon(() => wholeClusterStream.hasNext() &&
+                           wholeClusterStream.next().documentKey._id == "last_change_sentinel");
+}, [ErrorCodes.NoMatchingDocument, 51770]);
+
+// 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
+// pre-image lookup stage, so no events from 'collWithNoPreImages' ever reach it, and therefore
+// don't trip the validation checks for the existence of the pre-image.
+for (let runOnDB of [testDB, adminDB]) {
+    // Open a whole-db or whole-cluster stream that filters for the 'collWithPreImages' namespace.
+    const csCursor = runOnDB.watch(
+        [{$match: {$or: [{_id: resumeToken}, {"ns.coll": collWithPreImages.getName()}]}}], {
+            fullDocumentBeforeChange: "required",
+            resumeAfter: resumeToken,
+            allChangesForCluster: (runOnDB === adminDB)
+        });
+
+    // The list of events and pre-images that we expect to see in the stream.
+    const expectedPreImageEvents = [
+        {opType: "insert", fullDocumentBeforeChange: null},
+        {opType: "replace", fullDocumentBeforeChange: {_id: 0}},
+        {opType: "update", fullDocumentBeforeChange: {_id: 0, foo: "bar"}},
+        {opType: "delete", fullDocumentBeforeChange: {_id: 0, foo: "baz"}}
+    ];
+
+    // Confirm that the expected events are all seen, and in the expected order.
+    for (let expectedEvent of expectedPreImageEvents) {
+        assert.soon(() => csCursor.hasNext());
+        const observedEvent = csCursor.next();
+        assert.eq(observedEvent.operationType, expectedEvent.opType);
+        assert.eq(observedEvent.fullDocumentBeforeChange, expectedEvent.fullDocumentBeforeChange);
+    }
+}
+})();
diff --git a/jstests/change_streams/lookup_pit_pre_and_post_image.js b/jstests/change_streams/lookup_pit_pre_and_post_image.js
index 69153720d61..b06dba63bb6 100644
--- a/jstests/change_streams/lookup_pit_pre_and_post_image.js
+++ b/jstests/change_streams/lookup_pit_pre_and_post_image.js
@@ -1,24 +1,22 @@
 // Tests that the point-in-time pre- and post-images are loaded correctly in $changeStream running
 // with different arguments for collections with 'changeStreamPreAndPostImages' being enabled.
 // @tags: [
-//  assumes_against_mongod_not_mongos,
-//  change_stream_does_not_expect_txns,
-//  multiversion_incompatible,
+//   # TODO SERVER-58694: remove this tag.
+//   change_stream_does_not_expect_txns,
+//   multiversion_incompatible,
+//   # TODO SERVER-60238: remove this tag.
+//   assumes_read_preference_unchanged
 // ]
 (function() {
 "use strict";
 
 load("jstests/libs/collection_drop_recreate.js");  // For assertDropAndRecreateCollection.
-load("jstests/libs/change_stream_util.js");        // For isChangeStreamPreAndPostImagesEnabled.
+load("jstests/libs/change_stream_util.js");        // For canRecordPreImagesInConfigDatabase.
 
 const testDB = db.getSiblingDB(jsTestName());
 const collName = "test";
 
-const clusteredIndexesEnabled =
-    assert.commandWorked(testDB.adminCommand({getParameter: 1, featureFlagClusteredIndexes: 1}))
-        .featureFlagClusteredIndexes.value;
-
-if (!(isChangeStreamPreAndPostImagesEnabled(db) && clusteredIndexesEnabled)) {
+if (!canRecordPreImagesInConfigDatabase(testDB)) {
     const coll = assertDropAndRecreateCollection(testDB, collName);
 
     // If feature flag is off, creating changeStream with new fullDocument arguments should throw.
@@ -27,7 +25,7 @@ if (!(isChangeStreamPreAndPostImagesEnabled(db) && clusteredIndexesEnabled)) {
     assert.throwsWithCode(() => coll.watch([], {fullDocument: 'required'}), ErrorCodes.BadValue);
 
     jsTestLog(
-        'Skipping test because featureFlagChangeStreamPreAndPostImages or featureFlagClusteredIndexes feature flag is not enabled');
+        "Skipping test because pre-image recording capability in 'system.preimages' is not enabled.");
     return;
 }
 
diff --git a/jstests/change_streams/lookup_pre_image.js b/jstests/change_streams/lookup_pre_image.js
index 120c6de5671..f85a46a9e16 100644
--- a/jstests/change_streams/lookup_pre_image.js
+++ b/jstests/change_streams/lookup_pre_image.js
@@ -2,25 +2,42 @@
  * Tests the behaviour of the 'fullDocumentBeforeChange' argument to the $changeStream stage.
  *
  * @tags: [
- *   assumes_against_mongod_not_mongos,
  *   assumes_unsharded_collection,
  *   do_not_wrap_aggregations_in_facets,
  *   uses_multiple_connections,
  *   multiversion_incompatible,
+ *   # TODO SERVER-58694: remove this tag.
+ *   change_stream_does_not_expect_txns,
+ *   # TODO SERVER-60238: remove this tag.
+ *   assumes_read_preference_unchanged
  * ]
  */
 (function() {
 "use strict";
 
-load("jstests/libs/change_stream_util.js");        // For ChangeStreamTest.
+load("jstests/libs/change_stream_util.js");        // For ChangeStreamTest and
+                                                   // canRecordPreImagesInConfigDatabase.
 load("jstests/libs/collection_drop_recreate.js");  // For assert[Drop|Create]Collection.
 load("jstests/libs/fixture_helpers.js");           // For FixtureHelpers.
 
+const canRecordPreImagesInConfigDb = canRecordPreImagesInConfigDatabase(db);
+
+if (!canRecordPreImagesInConfigDb && FixtureHelpers.isMongos(db)) {
+    jsTestLog("Skipping test as pre image lookup is not supported in sharded cluster with feature" +
+              "flag 'featureFlagChangeStreamPreAndPostImages' disabled.");
+    return;
+}
+
 const coll = assertDropAndRecreateCollection(db, "change_stream_pre_images");
 const cst = new ChangeStreamTest(db);
 
 // Enable pre-image recording on the test collection.
-assert.commandWorked(db.runCommand({collMod: coll.getName(), recordPreImages: true}));
+if (!canRecordPreImagesInConfigDb) {
+    assert.commandWorked(db.runCommand({collMod: coll.getName(), recordPreImages: true}));
+} else {
+    assert.commandWorked(
+        db.runCommand({collMod: coll.getName(), changeStreamPreAndPostImages: {enabled: true}}));
+}
 
 // Open three streams on the collection, one for each "fullDocumentBeforeChange" mode.
 const csNoPreImages = cst.startWatchingChanges({
@@ -97,7 +114,12 @@ assert.docEq(latestChange, cst.getOneChange(csPreImageWhenAvailableCursor));
 assert.docEq(latestChange, cst.getOneChange(csPreImageRequiredCursor));
 
 // Now disable pre-image generation on the test collection and re-test.
-assert.commandWorked(db.runCommand({collMod: coll.getName(), recordPreImages: false}));
+if (!canRecordPreImagesInConfigDb) {
+    assert.commandWorked(db.runCommand({collMod: coll.getName(), recordPreImages: false}));
+} else {
+    assert.commandWorked(
+        db.runCommand({collMod: coll.getName(), changeStreamPreAndPostImages: {enabled: false}}));
+}
 
 // Test pre-image lookup for an insertion. No pre-image exists on any cursor.
 assert.commandWorked(coll.insert({_id: "y"}));