SERVER-29134: Support change streams on an entire database in a sharded cluster

1 files changed, 29 insertions, 22 deletions

diff --git a/jstests/sharding/lookup_change_stream_post_image_compound_shard_key.js b/jstests/sharding/lookup_change_stream_post_image_compound_shard_key.js
index bc69d734a78..29689156c87 100644
--- a/jstests/sharding/lookup_change_stream_post_image_compound_shard_key.js
+++ b/jstests/sharding/lookup_change_stream_post_image_compound_shard_key.js
@@ -49,7 +49,8 @@
         to: st.rs1.getURL()
     }));
 
-    const changeStream = mongosColl.aggregate([{$changeStream: {fullDocument: "updateLookup"}}]);
+    const changeStreamSingleColl = mongosColl.watch([], {fullDocument: "updateLookup"});
+    const changeStreamWholeDb = mongosDB.watch([], {fullDocument: "updateLookup"});
 
     const nDocs = 6;
     const bValues = ["one", "two", "three", "four", "five", "six"];
@@ -66,19 +67,22 @@
         assert.writeOK(mongosColl.update(documentKey, {$set: {updatedCount: 1}}));
     }
 
-    for (let id = 0; id < nDocs; ++id) {
-        assert.soon(() => changeStream.hasNext());
-        let next = changeStream.next();
-        assert.eq(next.operationType, "insert");
-        assert.eq(next.documentKey, Object.merge(shardKeyFromId(id), {_id: id}));
-
-        assert.soon(() => changeStream.hasNext());
-        next = changeStream.next();
-        assert.eq(next.operationType, "update");
-        assert.eq(next.documentKey, Object.merge(shardKeyFromId(id), {_id: id}));
-        assert.docEq(next.fullDocument,
-                     Object.merge(shardKeyFromId(id), {_id: id, updatedCount: 1}));
-    }
+    [changeStreamSingleColl, changeStreamWholeDb].forEach(function(changeStream) {
+        jsTestLog(`Testing updateLookup on namespace ${changeStream._ns}`);
+        for (let id = 0; id < nDocs; ++id) {
+            assert.soon(() => changeStream.hasNext());
+            let next = changeStream.next();
+            assert.eq(next.operationType, "insert");
+            assert.eq(next.documentKey, Object.merge(shardKeyFromId(id), {_id: id}));
+
+            assert.soon(() => changeStream.hasNext());
+            next = changeStream.next();
+            assert.eq(next.operationType, "update");
+            assert.eq(next.documentKey, Object.merge(shardKeyFromId(id), {_id: id}));
+            assert.docEq(next.fullDocument,
+                         Object.merge(shardKeyFromId(id), {_id: id, updatedCount: 1}));
+        }
+    });
 
     // Test that the change stream can still see the updated post image, even if a chunk is
     // migrated.
@@ -94,14 +98,17 @@
         to: st.rs0.getURL()
     }));
 
-    for (let id = 0; id < nDocs; ++id) {
-        assert.soon(() => changeStream.hasNext());
-        let next = changeStream.next();
-        assert.eq(next.operationType, "update");
-        assert.eq(next.documentKey, Object.merge(shardKeyFromId(id), {_id: id}));
-        assert.docEq(next.fullDocument,
-                     Object.merge(shardKeyFromId(id), {_id: id, updatedCount: 2}));
-    }
+    [changeStreamSingleColl, changeStreamWholeDb].forEach(function(changeStream) {
+        jsTestLog(`Testing updateLookup after moveChunk on namespace ${changeStream._ns}`);
+        for (let id = 0; id < nDocs; ++id) {
+            assert.soon(() => changeStream.hasNext());
+            let next = changeStream.next();
+            assert.eq(next.operationType, "update");
+            assert.eq(next.documentKey, Object.merge(shardKeyFromId(id), {_id: id}));
+            assert.docEq(next.fullDocument,
+                         Object.merge(shardKeyFromId(id), {_id: id, updatedCount: 2}));
+        }
+    });
 
     st.stop();
 })();