SERVER-37779 Step up enforcement that $changeStream must be first

Adds an additional check which enforces a $changeStream stage must be
the first stage in the pipeline which is robust to the case where a
$changeStream is only ever sent to a mongos.

1 files changed, 20 insertions, 0 deletions

diff --git a/jstests/change_streams/required_as_first_stage.js b/jstests/change_streams/required_as_first_stage.js
index d84d7e4c127..9eb15db9e9d 100644
--- a/jstests/change_streams/required_as_first_stage.js
+++ b/jstests/change_streams/required_as_first_stage.js
@@ -7,12 +7,32 @@
 
     const coll = assertDropAndRecreateCollection(db, "change_stream_required_as_first_stage");
 
+    assertErrorCode(coll, [{$match: {z: 34}}, {$changeStream: {}}], 40602);
     assertErrorCode(coll, [{$indexStats: {}}, {$changeStream: {}}], 40602);
     assertErrorCode(
         coll,
         [{$indexStats: {}}, {$changeStream: {}}, {$match: {test: "this is an extra stage"}}],
         40602);
 
+    let error = assert.throws(() => coll.aggregate([{$sort: {x: 1}}, {$changeStream: {}}]));
+    assert.contains(error.code, [40602, 50988], "Unexpected error: " + tojson(error));
+
+    error = assert.throws(
+        () => coll.aggregate([{$sort: {x: 1}}, {$changeStream: {}}], {allowDiskUse: true}));
+    assert.contains(error.code, [40602, 50988], "Unexpected error: " + tojson(error));
+
+    error = assert.throws(() => coll.aggregate([{$group: {_id: "$x"}}, {$changeStream: {}}]));
+    assert.contains(error.code, [40602, 50988], "Unexpected error: " + tojson(error));
+
+    // This one has a different error code because of conflicting host type requirements: the $group
+    // needs to merge on a shard, but the $changeStream needs to merge on mongos. This doesn't
+    // happen for the $sort because the half of the $sort running on mongos is pre-sorted, and so
+    // won't need disk space.
+    error = assert.throws(
+        () => coll.aggregate([{$group: {_id: "$x"}}, {$changeStream: {}}], {allowDiskUse: true}));
+    assert.contains(
+        error.code, [40602, ErrorCodes.IllegalOperation], "Unexpected error: " + tojson(error));
+
     // Test that a $changeStream stage is not allowed within a $facet stage.
     assertErrorCode(coll, [{$facet: {testPipe: [{$changeStream: {}}]}}], 40600);
     assertErrorCode(coll,