SERVER-32308: Add the ability for a $lookup stage to execute on mongos against a sharded foreign collection

4 files changed, 1361 insertions, 0 deletions

diff --git a/jstests/sharding/collation_lookup.js b/jstests/sharding/collation_lookup.js
new file mode 100644
index 00000000000..f06e92ab3fc
--- /dev/null
+++ b/jstests/sharding/collation_lookup.js
@@ -0,0 +1,454 @@
+/**
+ * Tests that the $lookup stage respects the collation when the local and/or foreign collections
+ * are sharded.
+ *
+ * The comparison of string values between the 'localField' and 'foreignField' should use the
+ * collation either explicitly set on the aggregation operation, or the collation inherited from the
+ * collection the "aggregate" command was performed on.
+ */
+(function() {
+    "use strict";
+
+    load("jstests/aggregation/extras/utils.js");  // for arrayEq
+
+    function runTests(withDefaultCollationColl, withoutDefaultCollationColl, collation) {
+        // Test that the $lookup stage respects the inherited collation.
+        let res = withDefaultCollationColl
+                      .aggregate([{
+                          $lookup: {
+                              from: withoutDefaultCollationColl.getName(),
+                              localField: "str",
+                              foreignField: "str",
+                              as: "matched",
+                          },
+                      }])
+                      .toArray();
+        assert.eq(1, res.length, tojson(res));
+
+        let expected = [{_id: "lowercase", str: "abc"}, {_id: "uppercase", str: "ABC"}];
+        assert(arrayEq(expected, res[0].matched),
+               "Expected " + tojson(expected) + " to equal " + tojson(res[0].matched) +
+                   " up to ordering");
+
+        res = withDefaultCollationColl
+                  .aggregate([{
+                      $lookup: {
+                          from: withoutDefaultCollationColl.getName(),
+                          let : {str1: "$str"},
+                          pipeline: [
+                              {$match: {$expr: {$eq: ["$str", "$$str1"]}}},
+                              {
+                                $lookup: {
+                                    from: withoutDefaultCollationColl.getName(),
+                                    let : {str2: "$str"},
+                                    pipeline: [{$match: {$expr: {$eq: ["$str", "$$str1"]}}}],
+                                    as: "matched2"
+                                }
+                              }
+                          ],
+                          as: "matched1",
+                      },
+                  }])
+                  .toArray();
+        assert.eq(1, res.length, tojson(res));
+
+        expected = [
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched2": [{"_id": "lowercase", "str": "abc"}, {"_id": "uppercase", "str": "ABC"}]
+            },
+            {
+              "_id": "uppercase",
+              "str": "ABC",
+              "matched2":
+                  [{"_id": "lowercase", "str": "abc"}, {"_id": "uppercase", "str": "ABC"}]
+            }
+        ];
+        assert(arrayEq(expected, res[0].matched1),
+               "Expected " + tojson(expected) + " to equal " + tojson(res[0].matched1) +
+                   " up to ordering. " + tojson(res));
+
+        // Test that the $lookup stage respects the inherited collation when it optimizes with an
+        // $unwind stage.
+        res = withDefaultCollationColl
+                  .aggregate([
+                      {
+                        $lookup: {
+                            from: withoutDefaultCollationColl.getName(),
+                            localField: "str",
+                            foreignField: "str",
+                            as: "matched",
+                        },
+                      },
+                      {$unwind: "$matched"},
+                  ])
+                  .toArray();
+        assert.eq(2, res.length, tojson(res));
+
+        expected = [
+            {_id: "lowercase", str: "abc", matched: {_id: "lowercase", str: "abc"}},
+            {_id: "lowercase", str: "abc", matched: {_id: "uppercase", str: "ABC"}}
+        ];
+        assert(arrayEq(expected, res),
+               "Expected " + tojson(expected) + " to equal " + tojson(res) + " up to ordering");
+
+        res = withDefaultCollationColl
+                  .aggregate([
+                      {
+                        $lookup: {
+                            from: withoutDefaultCollationColl.getName(),
+                            let : {str1: "$str"},
+                            pipeline: [
+                                {$match: {$expr: {$eq: ["$str", "$$str1"]}}},
+                                {
+                                  $lookup: {
+                                      from: withoutDefaultCollationColl.getName(),
+                                      let : {str2: "$str"},
+                                      pipeline: [{$match: {$expr: {$eq: ["$str", "$$str1"]}}}],
+                                      as: "matched2"
+                                  }
+                                },
+                                {$unwind: "$matched2"},
+                            ],
+                            as: "matched1",
+                        },
+                      },
+                      {$unwind: "$matched1"},
+                  ])
+                  .toArray();
+        assert.eq(4, res.length, tojson(res));
+
+        expected = [
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "lowercase",
+                  "str": "abc",
+                  "matched2": {"_id": "lowercase", "str": "abc"}
+              }
+            },
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "lowercase",
+                  "str": "abc",
+                  "matched2": {"_id": "uppercase", "str": "ABC"}
+              }
+            },
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "uppercase",
+                  "str": "ABC",
+                  "matched2": {"_id": "lowercase", "str": "abc"}
+              }
+            },
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "uppercase",
+                  "str": "ABC",
+                  "matched2": {"_id": "uppercase", "str": "ABC"}
+              }
+            }
+        ];
+        assert(arrayEq(expected, res),
+               "Expected " + tojson(expected) + " to equal " + tojson(res) + " up to ordering");
+
+        // Test that the $lookup stage respects an explicit collation on the aggregation operation.
+        res = withoutDefaultCollationColl
+                  .aggregate(
+                      [
+                        {$match: {_id: "lowercase"}},
+                        {
+                          $lookup: {
+                              from: withoutDefaultCollationColl.getName(),
+                              localField: "str",
+                              foreignField: "str",
+                              as: "matched",
+                          },
+                        },
+                      ],
+                      collation)
+                  .toArray();
+        assert.eq(1, res.length, tojson(res));
+
+        expected = [{_id: "lowercase", str: "abc"}, {_id: "uppercase", str: "ABC"}];
+        assert(arrayEq(expected, res[0].matched),
+               "Expected " + tojson(expected) + " to equal " + tojson(res[0].matched) +
+                   " up to ordering");
+
+        res = withoutDefaultCollationColl
+                  .aggregate(
+                      [
+                        {$match: {_id: "lowercase"}},
+                        {
+                          $lookup: {
+                              from: withoutDefaultCollationColl.getName(),
+                              let : {str1: "$str"},
+                              pipeline: [
+                                  {$match: {$expr: {$eq: ["$str", "$$str1"]}}},
+                                  {
+                                    $lookup: {
+                                        from: withoutDefaultCollationColl.getName(),
+                                        let : {str2: "$str"},
+                                        pipeline: [{$match: {$expr: {$eq: ["$str", "$$str1"]}}}],
+                                        as: "matched2"
+                                    }
+                                  }
+                              ],
+                              as: "matched1",
+                          },
+                        }
+                      ],
+                      collation)
+                  .toArray();
+        assert.eq(1, res.length, tojson(res));
+
+        expected = [
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched2": [{"_id": "lowercase", "str": "abc"}, {"_id": "uppercase", "str": "ABC"}]
+            },
+            {
+              "_id": "uppercase",
+              "str": "ABC",
+              "matched2":
+                  [{"_id": "lowercase", "str": "abc"}, {"_id": "uppercase", "str": "ABC"}]
+            }
+        ];
+        assert(arrayEq(expected, res[0].matched1),
+               "Expected " + tojson(expected) + " to equal " + tojson(res[0].matched1) +
+                   " up to ordering");
+
+        // Test that the $lookup stage respects an explicit collation on the aggregation operation
+        // when
+        // it optimizes with an $unwind stage.
+        res = withoutDefaultCollationColl
+                  .aggregate(
+                      [
+                        {$match: {_id: "lowercase"}},
+                        {
+                          $lookup: {
+                              from: withoutDefaultCollationColl.getName(),
+                              localField: "str",
+                              foreignField: "str",
+                              as: "matched",
+                          },
+                        },
+                        {$unwind: "$matched"},
+                      ],
+                      collation)
+                  .toArray();
+        assert.eq(2, res.length, tojson(res));
+
+        expected = [
+            {_id: "lowercase", str: "abc", matched: {_id: "lowercase", str: "abc"}},
+            {_id: "lowercase", str: "abc", matched: {_id: "uppercase", str: "ABC"}}
+        ];
+        assert(arrayEq(expected, res),
+               "Expected " + tojson(expected) + " to equal " + tojson(res) + " up to ordering");
+
+        res = withoutDefaultCollationColl
+                  .aggregate(
+                      [
+                        {$match: {_id: "lowercase"}},
+                        {
+                          $lookup: {
+                              from: withoutDefaultCollationColl.getName(),
+                              let : {str1: "$str"},
+                              pipeline: [
+                                  {$match: {$expr: {$eq: ["$str", "$$str1"]}}},
+                                  {
+                                    $lookup: {
+                                        from: withoutDefaultCollationColl.getName(),
+                                        let : {str2: "$str"},
+                                        pipeline: [{$match: {$expr: {$eq: ["$str", "$$str1"]}}}],
+                                        as: "matched2"
+                                    }
+                                  },
+                                  {$unwind: "$matched2"},
+                              ],
+                              as: "matched1",
+                          },
+                        },
+                        {$unwind: "$matched1"},
+                      ],
+                      collation)
+                  .toArray();
+        assert.eq(4, res.length, tojson(res));
+
+        expected = [
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "lowercase",
+                  "str": "abc",
+                  "matched2": {"_id": "lowercase", "str": "abc"}
+              }
+            },
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "lowercase",
+                  "str": "abc",
+                  "matched2": {"_id": "uppercase", "str": "ABC"}
+              }
+            },
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "uppercase",
+                  "str": "ABC",
+                  "matched2": {"_id": "lowercase", "str": "abc"}
+              }
+            },
+            {
+              "_id": "lowercase",
+              "str": "abc",
+              "matched1": {
+                  "_id": "uppercase",
+                  "str": "ABC",
+                  "matched2": {"_id": "uppercase", "str": "ABC"}
+              }
+            }
+        ];
+        assert(arrayEq(expected, res),
+               "Expected " + tojson(expected) + " to equal " + tojson(res) + " up to ordering");
+
+        // Test that the $lookup stage uses the "simple" collation if a collation isn't set on the
+        // collection or the aggregation operation.
+        res = withoutDefaultCollationColl
+                  .aggregate([
+                      {$match: {_id: "lowercase"}},
+                      {
+                        $lookup: {
+                            from: withDefaultCollationColl.getName(),
+                            localField: "str",
+                            foreignField: "str",
+                            as: "matched",
+                        },
+                      },
+                  ])
+                  .toArray();
+        assert.eq([{_id: "lowercase", str: "abc", matched: [{_id: "lowercase", str: "abc"}]}], res);
+
+        res = withoutDefaultCollationColl
+                  .aggregate([
+                      {$match: {_id: "lowercase"}},
+                      {
+                        $lookup: {
+                            from: withoutDefaultCollationColl.getName(),
+                            let : {str1: "$str"},
+                            pipeline: [
+                                {$match: {$expr: {$eq: ["$str", "$$str1"]}}},
+                                {
+                                  $lookup: {
+                                      from: withoutDefaultCollationColl.getName(),
+                                      let : {str2: "$str"},
+                                      pipeline: [{$match: {$expr: {$eq: ["$str", "$$str1"]}}}],
+                                      as: "matched2"
+                                  }
+                                },
+                                {$unwind: "$matched2"},
+                            ],
+                            as: "matched1",
+                        },
+                      },
+                  ])
+                  .toArray();
+        assert.eq([{
+                     "_id": "lowercase",
+                     "str": "abc",
+                     "matched1": [{
+                         "_id": "lowercase",
+                         "str": "abc",
+                         "matched2": {"_id": "lowercase", "str": "abc"}
+                     }]
+                  }],
+                  res);
+    }
+
+    const st = new ShardingTest({shards: 2, config: 1});
+    const testName = "collation_lookup";
+    const caseInsensitive = {collation: {locale: "en_US", strength: 2}};
+
+    const mongosDB = st.s0.getDB(testName);
+    const withDefaultCollationColl = mongosDB[testName + "_with_default"];
+    const withoutDefaultCollationColl = mongosDB[testName + "_without_default"];
+
+    assert.commandWorked(
+        mongosDB.createCollection(withDefaultCollationColl.getName(), caseInsensitive));
+    assert.writeOK(withDefaultCollationColl.insert({_id: "lowercase", str: "abc"}));
+
+    assert.writeOK(withoutDefaultCollationColl.insert({_id: "lowercase", str: "abc"}));
+    assert.writeOK(withoutDefaultCollationColl.insert({_id: "uppercase", str: "ABC"}));
+    assert.writeOK(withoutDefaultCollationColl.insert({_id: "unmatched", str: "def"}));
+
+    //
+    // Sharded collection with default collation and unsharded collection without a default
+    // collation.
+    //
+    assert.commandWorked(
+        withDefaultCollationColl.createIndex({str: 1}, {collation: {locale: "simple"}}));
+
+    // Enable sharding on the test DB and ensure its primary is shard0000.
+    assert.commandWorked(mongosDB.adminCommand({enableSharding: mongosDB.getName()}));
+    st.ensurePrimaryShard(mongosDB.getName(), st.shard0.shardName);
+
+    // Shard the collection with a default collation.
+    assert.commandWorked(mongosDB.adminCommand({
+        shardCollection: withDefaultCollationColl.getFullName(),
+        key: {str: 1},
+        collation: {locale: "simple"}
+    }));
+
+    // Split the collection into 2 chunks.
+    assert.commandWorked(mongosDB.adminCommand(
+        {split: withDefaultCollationColl.getFullName(), middle: {str: "abc"}}));
+
+    // Move the chunk containing {str: "abc"} to shard0001.
+    assert.commandWorked(mongosDB.adminCommand({
+        moveChunk: withDefaultCollationColl.getFullName(),
+        find: {str: "abc"},
+        to: st.shard1.shardName
+    }));
+
+    runTests(withDefaultCollationColl, withoutDefaultCollationColl, caseInsensitive);
+
+    // TODO: Enable the following tests once SERVER-32536 is fixed.
+    //
+    // Sharded collection with default collation and sharded collection without a default
+    // collation.
+    //
+
+    // Shard the collection without a default collation.
+    // assert.commandWorked(mongosDB.adminCommand({
+    //     shardCollection: withoutDefaultCollationColl.getFullName(),
+    //     key: {_id: 1},
+    // }));
+
+    // // Split the collection into 2 chunks.
+    // assert.commandWorked(mongosDB.adminCommand(
+    //     {split: withoutDefaultCollationColl.getFullName(), middle: {_id: "unmatched"}}));
+
+    // // Move the chunk containing {_id: "lowercase"} to shard0001.
+    // assert.commandWorked(mongosDB.adminCommand({
+    //     moveChunk: withoutDefaultCollationColl.getFullName(),
+    //     find: {_id: "lowercase"},
+    //     to: st.shard1.shardName
+    // }));
+
+    // runTests(withDefaultCollationColl, withoutDefaultCollationColl, caseInsensitive);
+
+    st.stop();
+})();
diff --git a/jstests/sharding/lookup.js b/jstests/sharding/lookup.js
new file mode 100644
index 00000000000..cc41cbff319
--- /dev/null
+++ b/jstests/sharding/lookup.js
@@ -0,0 +1,609 @@
+// Basic $lookup regression tests.
+(function() {
+    "use strict";
+
+    load("jstests/aggregation/extras/utils.js");  // For assertErrorCode.
+
+    const st = new ShardingTest({shards: 2, config: 1, mongos: 1});
+    const testName = "lookup_sharded";
+
+    const mongosDB = st.s0.getDB(testName);
+    assert.commandWorked(mongosDB.dropDatabase());
+
+    // Used by testPipeline to sort result documents. All _ids must be primitives.
+    function compareId(a, b) {
+        if (a._id < b._id) {
+            return -1;
+        }
+        if (a._id > b._id) {
+            return 1;
+        }
+        return 0;
+    }
+
+    // Helper for testing that pipeline returns correct set of results.
+    function testPipeline(pipeline, expectedResult, collection) {
+        assert.eq(collection.aggregate(pipeline).toArray().sort(compareId),
+                  expectedResult.sort(compareId));
+    }
+
+    // Shards and splits the collection 'coll' on _id.
+    function shardAndSplit(db, coll) {
+        // Shard the collection on _id.
+        assert.commandWorked(db.adminCommand({shardCollection: coll.getFullName(), key: {_id: 1}}));
+
+        // Split the collection into 2 chunks: [MinKey, 0), [0, MaxKey).
+        assert.commandWorked(db.adminCommand({split: coll.getFullName(), middle: {_id: 0}}));
+
+        // Move the [0, MaxKey) chunk to shard0001.
+        assert.commandWorked(db.adminCommand({
+            moveChunk: coll.getFullName(),
+            find: {_id: 1},
+            to: st.shard1.shardName,
+        }));
+    }
+
+    function runTest(coll, from, thirdColl, fourthColl) {
+        let db = null;  // Using the db variable is banned in this function.
+
+        coll.remove({});
+        from.remove({});
+        thirdColl.remove({});
+        fourthColl.remove({});
+
+        assert.writeOK(coll.insert({_id: 0, a: 1}));
+        assert.writeOK(coll.insert({_id: 1, a: null}));
+        assert.writeOK(coll.insert({_id: 2}));
+
+        assert.writeOK(from.insert({_id: 0, b: 1}));
+        assert.writeOK(from.insert({_id: 1, b: null}));
+        assert.writeOK(from.insert({_id: 2}));
+        //
+        // Basic functionality.
+        //
+
+        // "from" document added to "as" field if a == b, where nonexistent fields are treated as
+        // null.
+        let expectedResults = [
+            {_id: 0, a: 1, "same": [{_id: 0, b: 1}]},
+            {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+            {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+        ];
+        testPipeline([{$lookup: {localField: "a", foreignField: "b", from: "from", as: "same"}}],
+                     expectedResults,
+                     coll);
+
+        // If localField is nonexistent, it is treated as if it is null.
+        expectedResults = [
+            {_id: 0, a: 1, "same": [{_id: 1, b: null}, {_id: 2}]},
+            {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+            {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+        ];
+        testPipeline(
+            [{$lookup: {localField: "nonexistent", foreignField: "b", from: "from", as: "same"}}],
+            expectedResults,
+            coll);
+
+        // If foreignField is nonexistent, it is treated as if it is null.
+        expectedResults = [
+            {_id: 0, a: 1, "same": []},
+            {_id: 1, a: null, "same": [{_id: 0, b: 1}, {_id: 1, b: null}, {_id: 2}]},
+            {_id: 2, "same": [{_id: 0, b: 1}, {_id: 1, b: null}, {_id: 2}]}
+        ];
+        testPipeline(
+            [{$lookup: {localField: "a", foreignField: "nonexistent", from: "from", as: "same"}}],
+            expectedResults,
+            coll);
+
+        // If there are no matches or the from coll doesn't exist, the result is an empty array.
+        expectedResults =
+            [{_id: 0, a: 1, "same": []}, {_id: 1, a: null, "same": []}, {_id: 2, "same": []}];
+        testPipeline(
+            [{$lookup: {localField: "_id", foreignField: "nonexistent", from: "from", as: "same"}}],
+            expectedResults,
+            coll);
+        testPipeline(
+            [{$lookup: {localField: "a", foreignField: "b", from: "nonexistent", as: "same"}}],
+            expectedResults,
+            coll);
+
+        // If field name specified by "as" already exists, it is overwritten.
+        expectedResults = [
+            {_id: 0, "a": [{_id: 0, b: 1}]},
+            {_id: 1, "a": [{_id: 1, b: null}, {_id: 2}]},
+            {_id: 2, "a": [{_id: 1, b: null}, {_id: 2}]}
+        ];
+        testPipeline([{$lookup: {localField: "a", foreignField: "b", from: "from", as: "a"}}],
+                     expectedResults,
+                     coll);
+
+        // Running multiple $lookups in the same pipeline is allowed.
+        expectedResults = [
+            {_id: 0, a: 1, "c": [{_id: 0, b: 1}], "d": [{_id: 0, b: 1}]},
+            {
+              _id: 1,
+              a: null, "c": [{_id: 1, b: null}, {_id: 2}], "d": [{_id: 1, b: null}, {_id: 2}]
+            },
+            {_id: 2, "c": [{_id: 1, b: null}, {_id: 2}], "d": [{_id: 1, b: null}, {_id: 2}]}
+        ];
+        testPipeline(
+            [
+              {$lookup: {localField: "a", foreignField: "b", from: "from", as: "c"}},
+              {$project: {"a": 1, "c": 1}},
+              {$lookup: {localField: "a", foreignField: "b", from: "from", as: "d"}}
+            ],
+            expectedResults,
+            coll);
+
+        //
+        // Coalescing with $unwind.
+        //
+
+        // A normal $unwind with on the "as" field.
+        expectedResults = [
+            {_id: 0, a: 1, same: {_id: 0, b: 1}},
+            {_id: 1, a: null, same: {_id: 1, b: null}},
+            {_id: 1, a: null, same: {_id: 2}},
+            {_id: 2, same: {_id: 1, b: null}},
+            {_id: 2, same: {_id: 2}}
+        ];
+        testPipeline(
+            [
+              {$lookup: {localField: "a", foreignField: "b", from: "from", as: "same"}},
+              {$unwind: {path: "$same"}}
+            ],
+            expectedResults,
+            coll);
+
+        // An $unwind on the "as" field, with includeArrayIndex.
+        expectedResults = [
+            {_id: 0, a: 1, same: {_id: 0, b: 1}, index: NumberLong(0)},
+            {_id: 1, a: null, same: {_id: 1, b: null}, index: NumberLong(0)},
+            {_id: 1, a: null, same: {_id: 2}, index: NumberLong(1)},
+            {_id: 2, same: {_id: 1, b: null}, index: NumberLong(0)},
+            {_id: 2, same: {_id: 2}, index: NumberLong(1)},
+        ];
+        testPipeline(
+            [
+              {$lookup: {localField: "a", foreignField: "b", from: "from", as: "same"}},
+              {$unwind: {path: "$same", includeArrayIndex: "index"}}
+            ],
+            expectedResults,
+            coll);
+
+        // Normal $unwind with no matching documents.
+        expectedResults = [];
+        testPipeline(
+            [
+              {$lookup: {localField: "_id", foreignField: "nonexistent", from: "from", as: "same"}},
+              {$unwind: {path: "$same"}}
+            ],
+            expectedResults,
+            coll);
+
+        // $unwind with preserveNullAndEmptyArray with no matching documents.
+        expectedResults = [
+            {_id: 0, a: 1},
+            {_id: 1, a: null},
+            {_id: 2},
+        ];
+        testPipeline(
+            [
+              {$lookup: {localField: "_id", foreignField: "nonexistent", from: "from", as: "same"}},
+              {$unwind: {path: "$same", preserveNullAndEmptyArrays: true}}
+            ],
+            expectedResults,
+            coll);
+
+        // $unwind with preserveNullAndEmptyArray, some with matching documents, some without.
+        expectedResults = [
+            {_id: 0, a: 1},
+            {_id: 1, a: null, same: {_id: 0, b: 1}},
+            {_id: 2},
+        ];
+        testPipeline(
+            [
+              {$lookup: {localField: "_id", foreignField: "b", from: "from", as: "same"}},
+              {$unwind: {path: "$same", preserveNullAndEmptyArrays: true}}
+            ],
+            expectedResults,
+            coll);
+
+        // $unwind with preserveNullAndEmptyArray and includeArrayIndex, some with matching
+        // documents, some without.
+        expectedResults = [
+            {_id: 0, a: 1, index: null},
+            {_id: 1, a: null, same: {_id: 0, b: 1}, index: NumberLong(0)},
+            {_id: 2, index: null},
+        ];
+        testPipeline(
+            [
+              {$lookup: {localField: "_id", foreignField: "b", from: "from", as: "same"}},
+              {
+                $unwind:
+                    {path: "$same", preserveNullAndEmptyArrays: true, includeArrayIndex: "index"}
+              }
+            ],
+            expectedResults,
+            coll);
+
+        //
+        // Dependencies.
+        //
+
+        // If $lookup didn't add "localField" to its dependencies, this test would fail as the
+        // value of the "a" field would be lost and treated as null.
+        expectedResults = [
+            {_id: 0, "same": [{_id: 0, b: 1}]},
+            {_id: 1, "same": [{_id: 1, b: null}, {_id: 2}]},
+            {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+        ];
+        testPipeline(
+            [
+              {$lookup: {localField: "a", foreignField: "b", from: "from", as: "same"}},
+              {$project: {"same": 1}}
+            ],
+            expectedResults,
+            coll);
+
+        // If $lookup didn't add fields referenced by "let" variables to its dependencies, this test
+        // would fail as the value of the "a" field would be lost and treated as null.
+        expectedResults = [
+            {"_id": 0, "same": [{"_id": 0, "x": 1}, {"_id": 1, "x": 1}, {"_id": 2, "x": 1}]},
+            {
+              "_id": 1,
+              "same": [{"_id": 0, "x": null}, {"_id": 1, "x": null}, {"_id": 2, "x": null}]
+            },
+            {"_id": 2, "same": [{"_id": 0}, {"_id": 1}, {"_id": 2}]}
+        ];
+        testPipeline(
+            [
+              {
+                $lookup: {
+                    let : {var1: "$a"},
+                    pipeline: [{$project: {x: "$$var1"}}],
+                    from: "from",
+                    as: "same"
+                }
+              },
+              {$project: {"same": 1}}
+            ],
+            expectedResults,
+            coll);
+
+        //
+        // Dotted field paths.
+        //
+
+        coll.remove({});
+        assert.writeOK(coll.insert({_id: 0, a: 1}));
+        assert.writeOK(coll.insert({_id: 1, a: null}));
+        assert.writeOK(coll.insert({_id: 2}));
+        assert.writeOK(coll.insert({_id: 3, a: {c: 1}}));
+
+        from.remove({});
+        assert.writeOK(from.insert({_id: 0, b: 1}));
+        assert.writeOK(from.insert({_id: 1, b: null}));
+        assert.writeOK(from.insert({_id: 2}));
+        assert.writeOK(from.insert({_id: 3, b: {c: 1}}));
+        assert.writeOK(from.insert({_id: 4, b: {c: 2}}));
+
+        // Once without a dotted field.
+        let pipeline = [{$lookup: {localField: "a", foreignField: "b", from: "from", as: "same"}}];
+        expectedResults = [
+            {_id: 0, a: 1, "same": [{_id: 0, b: 1}]},
+            {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+            {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]},
+            {_id: 3, a: {c: 1}, "same": [{_id: 3, b: {c: 1}}]}
+        ];
+        testPipeline(pipeline, expectedResults, coll);
+
+        // Look up a dotted field.
+        pipeline = [{$lookup: {localField: "a.c", foreignField: "b.c", from: "from", as: "same"}}];
+        // All but the last document in 'coll' have a nullish value for 'a.c'.
+        expectedResults = [
+            {_id: 0, a: 1, same: [{_id: 0, b: 1}, {_id: 1, b: null}, {_id: 2}]},
+            {_id: 1, a: null, same: [{_id: 0, b: 1}, {_id: 1, b: null}, {_id: 2}]},
+            {_id: 2, same: [{_id: 0, b: 1}, {_id: 1, b: null}, {_id: 2}]},
+            {_id: 3, a: {c: 1}, same: [{_id: 3, b: {c: 1}}]}
+        ];
+        testPipeline(pipeline, expectedResults, coll);
+
+        // With an $unwind stage.
+        coll.remove({});
+        assert.writeOK(coll.insert({_id: 0, a: {b: 1}}));
+        assert.writeOK(coll.insert({_id: 1}));
+
+        from.remove({});
+        assert.writeOK(from.insert({_id: 0, target: 1}));
+
+        pipeline = [
+            {
+              $lookup: {
+                  localField: "a.b",
+                  foreignField: "target",
+                  from: "from",
+                  as: "same.documents",
+              }
+            },
+            {
+              // Expected input to $unwind:
+              // {_id: 0, a: {b: 1}, same: {documents: [{_id: 0, target: 1}]}}
+              // {_id: 1, same: {documents: []}}
+              $unwind: {
+                  path: "$same.documents",
+                  preserveNullAndEmptyArrays: true,
+                  includeArrayIndex: "c.d.e",
+              }
+            }
+        ];
+        expectedResults = [
+            {_id: 0, a: {b: 1}, same: {documents: {_id: 0, target: 1}}, c: {d: {e: NumberLong(0)}}},
+            {_id: 1, same: {}, c: {d: {e: null}}},
+        ];
+        testPipeline(pipeline, expectedResults, coll);
+
+        //
+        // Query-like local fields (SERVER-21287)
+        //
+
+        // This must only do an equality match rather than treating the value as a regex.
+        coll.remove({});
+        assert.writeOK(coll.insert({_id: 0, a: /a regex/}));
+
+        from.remove({});
+        assert.writeOK(from.insert({_id: 0, b: /a regex/}));
+        assert.writeOK(from.insert({_id: 1, b: "string that matches /a regex/"}));
+
+        pipeline = [
+            {
+              $lookup: {
+                  localField: "a",
+                  foreignField: "b",
+                  from: "from",
+                  as: "b",
+              }
+            },
+        ];
+        expectedResults = [{_id: 0, a: /a regex/, b: [{_id: 0, b: /a regex/}]}];
+        testPipeline(pipeline, expectedResults, coll);
+
+        //
+        // A local value of an array.
+        //
+
+        // Basic array corresponding to multiple documents.
+        coll.remove({});
+        assert.writeOK(coll.insert({_id: 0, a: [0, 1, 2]}));
+
+        from.remove({});
+        assert.writeOK(from.insert({_id: 0}));
+        assert.writeOK(from.insert({_id: 1}));
+
+        pipeline = [
+            {
+              $lookup: {
+                  localField: "a",
+                  foreignField: "_id",
+                  from: "from",
+                  as: "b",
+              }
+            },
+        ];
+        expectedResults = [{_id: 0, a: [0, 1, 2], b: [{_id: 0}, {_id: 1}]}];
+        testPipeline(pipeline, expectedResults, coll);
+
+        // Basic array corresponding to a single document.
+        coll.remove({});
+        assert.writeOK(coll.insert({_id: 0, a: [1]}));
+
+        from.remove({});
+        assert.writeOK(from.insert({_id: 0}));
+        assert.writeOK(from.insert({_id: 1}));
+
+        pipeline = [
+            {
+              $lookup: {
+                  localField: "a",
+                  foreignField: "_id",
+                  from: "from",
+                  as: "b",
+              }
+            },
+        ];
+        expectedResults = [{_id: 0, a: [1], b: [{_id: 1}]}];
+        testPipeline(pipeline, expectedResults, coll);
+
+        // Array containing regular expressions.
+        coll.remove({});
+        assert.writeOK(coll.insert({_id: 0, a: [/a regex/, /^x/]}));
+        assert.writeOK(coll.insert({_id: 1, a: [/^x/]}));
+
+        from.remove({});
+        assert.writeOK(from.insert({_id: 0, b: "should not match a regex"}));
+        assert.writeOK(from.insert({_id: 1, b: "xxxx"}));
+        assert.writeOK(from.insert({_id: 2, b: /a regex/}));
+        assert.writeOK(from.insert({_id: 3, b: /^x/}));
+
+        pipeline = [
+            {
+              $lookup: {
+                  localField: "a",
+                  foreignField: "b",
+                  from: "from",
+                  as: "b",
+              }
+            },
+        ];
+        expectedResults = [
+            {_id: 0, a: [/a regex/, /^x/], b: [{_id: 2, b: /a regex/}, {_id: 3, b: /^x/}]},
+            {_id: 1, a: [/^x/], b: [{_id: 3, b: /^x/}]}
+        ];
+        testPipeline(pipeline, expectedResults, coll);
+
+        // 'localField' references a field within an array of sub-objects.
+        coll.remove({});
+        assert.writeOK(coll.insert({_id: 0, a: [{b: 1}, {b: 2}]}));
+
+        from.remove({});
+        assert.writeOK(from.insert({_id: 0}));
+        assert.writeOK(from.insert({_id: 1}));
+        assert.writeOK(from.insert({_id: 2}));
+        assert.writeOK(from.insert({_id: 3}));
+
+        pipeline = [
+            {
+              $lookup: {
+                  localField: "a.b",
+                  foreignField: "_id",
+                  from: "from",
+                  as: "c",
+              }
+            },
+        ];
+
+        expectedResults = [{"_id": 0, "a": [{"b": 1}, {"b": 2}], "c": [{"_id": 1}, {"_id": 2}]}];
+        testPipeline(pipeline, expectedResults, coll);
+
+        //
+        // Test $lookup when the foreign collection is a view.
+        //
+        // TODO: Enable this test as part of SERVER-32548, fails whenever the foreign collection is
+        // sharded.
+        // coll.getDB().fromView.drop();
+        // assert.commandWorked(
+        //     coll.getDB().runCommand({create: "fromView", viewOn: "from", pipeline: []}));
+
+        // pipeline = [
+        //     {
+        //       $lookup: {
+        //           localField: "a.b",
+        //           foreignField: "_id",
+        //           from: "fromView",
+        //           as: "c",
+        //       }
+        //     },
+        // ];
+
+        // expectedResults = [{"_id": 0, "a": [{"b": 1}, {"b": 2}], "c": [{"_id": 1}, {"_id": 2}]}];
+        // testPipeline(pipeline, expectedResults, coll);
+
+        //
+        // Error cases.
+        //
+
+        // 'from', 'as', 'localField' and 'foreignField' must all be specified when run with
+        // localField/foreignField syntax.
+        assertErrorCode(coll,
+                        [{$lookup: {foreignField: "b", from: "from", as: "same"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {localField: "a", from: "from", as: "same"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {localField: "a", foreignField: "b", as: "same"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {localField: "a", foreignField: "b", from: "from"}}],
+                        ErrorCodes.FailedToParse);
+
+        // localField/foreignField and pipeline/let syntax must not be mixed.
+        assertErrorCode(coll,
+                        [{$lookup: {pipeline: [], foreignField: "b", from: "from", as: "as"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {pipeline: [], localField: "b", from: "from", as: "as"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(
+            coll,
+            [{$lookup: {pipeline: [], localField: "b", foreignField: "b", from: "from", as: "as"}}],
+            ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {let : {a: "$b"}, foreignField: "b", from: "from", as: "as"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {let : {a: "$b"}, localField: "b", from: "from", as: "as"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(
+            coll,
+            [{
+               $lookup:
+                   {let : {a: "$b"}, localField: "b", foreignField: "b", from: "from", as: "as"}
+            }],
+            ErrorCodes.FailedToParse);
+
+        // 'from', 'as', 'localField' and 'foreignField' must all be of type string.
+        assertErrorCode(coll,
+                        [{$lookup: {localField: 1, foreignField: "b", from: "from", as: "as"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {localField: "a", foreignField: 1, from: "from", as: "as"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {localField: "a", foreignField: "b", from: 1, as: "as"}}],
+                        ErrorCodes.FailedToParse);
+        assertErrorCode(coll,
+                        [{$lookup: {localField: "a", foreignField: "b", from: "from", as: 1}}],
+                        ErrorCodes.FailedToParse);
+
+        // The foreign collection must be a valid namespace.
+        assertErrorCode(coll,
+                        [{$lookup: {localField: "a", foreignField: "b", from: "", as: "as"}}],
+                        ErrorCodes.InvalidNamespace);
+        // $lookup's field must be an object.
+        assertErrorCode(coll, [{$lookup: "string"}], ErrorCodes.FailedToParse);
+    }
+
+    //
+    // Test unsharded local collection and unsharded foreign collection.
+    //
+    mongosDB.lookUp.drop();
+    mongosDB.from.drop();
+    mongosDB.thirdColl.drop();
+    mongosDB.fourthColl.drop();
+
+    runTest(mongosDB.lookUp, mongosDB.from, mongosDB.thirdColl, mongosDB.fourthColl);
+
+    // Verify that the command is sent only to the primary shard when both the local and foreign
+    // collections are unsharded.
+    assert(!assert
+                .commandWorked(mongosDB.lookup.explain().aggregate([{
+                    $lookup: {
+                        from: mongosDB.from.getName(),
+                        localField: "a",
+                        foreignField: "b",
+                        as: "results"
+                    }
+                }]))
+                .hasOwnProperty("shards"));
+    // Enable sharding on the test DB and ensure its primary is shard0000.
+    assert.commandWorked(mongosDB.adminCommand({enableSharding: mongosDB.getName()}));
+    st.ensurePrimaryShard(mongosDB.getName(), st.shard0.shardName);
+
+    //
+    // Test unsharded local collection and sharded foreign collection.
+    //
+
+    // Shard the foreign collection on _id.
+    shardAndSplit(mongosDB, mongosDB.from);
+    runTest(mongosDB.lookUp, mongosDB.from, mongosDB.thirdColl, mongosDB.fourthColl);
+
+    //
+    // Test sharded local collection and unsharded foreign collection.
+    //
+    mongosDB.from.drop();
+
+    // Shard the local collection on _id.
+    shardAndSplit(mongosDB, mongosDB.lookup);
+    runTest(mongosDB.lookUp, mongosDB.from, mongosDB.thirdColl, mongosDB.fourthColl);
+
+    //
+    // Test sharded local and foreign collections.
+    //
+
+    // Shard the foreign collection on _id.
+    shardAndSplit(mongosDB, mongosDB.from);
+    runTest(mongosDB.lookUp, mongosDB.from, mongosDB.thirdColl, mongosDB.fourthColl);
+
+    st.stop();
+}());
diff --git a/jstests/sharding/lookup_mongod_unaware.js b/jstests/sharding/lookup_mongod_unaware.js
new file mode 100644
index 00000000000..6333eec15de
--- /dev/null
+++ b/jstests/sharding/lookup_mongod_unaware.js
@@ -0,0 +1,168 @@
+// Tests the behavior of a $lookup when a shard contains incorrect routing information for the
+// local and/or foreign collections.  This includes when the shard thinks the collection is sharded
+// when it's not, and likewise when it thinks the collection is unsharded but is actually sharded.
+//
+// We restart a mongod to cause it to forget that a collection was sharded. When restarted, we
+// expect it to still have all the previous data.
+// @tags: [requires_persistence]
+(function() {
+    "use strict";
+
+    // Restarts the primary shard and ensures that it believes both collections are unsharded.
+    function restartPrimaryShard(rs, localColl, foreignColl) {
+        // Returns true if the shard is aware that the collection is sharded.
+        function hasRoutingInfoForNs(shardConn, coll) {
+            const res = shardConn.adminCommand({getShardVersion: coll, fullMetadata: true});
+            assert.commandWorked(res);
+            return res.metadata.collVersion != undefined;
+        }
+
+        rs.restart(0);
+        rs.awaitSecondaryNodes();
+        assert(!hasRoutingInfoForNs(rs.getPrimary(), localColl.getFullName()));
+        assert(!hasRoutingInfoForNs(rs.getPrimary(), foreignColl.getFullName()));
+    }
+
+    const testName = "lookup_stale_mongod";
+    const st = new ShardingTest({
+        shards: 2,
+        mongos: 2,
+        rs: {nodes: 1},
+    });
+
+    const mongos0DB = st.s0.getDB(testName);
+    const mongos0LocalColl = mongos0DB[testName + "_local"];
+    const mongos0ForeignColl = mongos0DB[testName + "_foreign"];
+
+    const mongos1DB = st.s1.getDB(testName);
+    const mongos1LocalColl = mongos1DB[testName + "_local"];
+    const mongos1ForeignColl = mongos1DB[testName + "_foreign"];
+
+    // Ensure that shard0 is the primary shard.
+    assert.commandWorked(mongos0DB.adminCommand({enableSharding: mongos0DB.getName()}));
+    st.ensurePrimaryShard(mongos0DB.getName(), st.shard0.shardName);
+
+    assert.writeOK(mongos0LocalColl.insert({_id: 0, a: 1}));
+    assert.writeOK(mongos0LocalColl.insert({_id: 1, a: null}));
+
+    assert.writeOK(mongos0ForeignColl.insert({_id: 0, b: 1}));
+    assert.writeOK(mongos0ForeignColl.insert({_id: 1, b: null}));
+
+    // Send writes through mongos1 such that it's aware of the collections and believes they are
+    // unsharded.
+    assert.writeOK(mongos1LocalColl.insert({_id: 2}));
+    assert.writeOK(mongos1ForeignColl.insert({_id: 2}));
+
+    const pipeline = [
+        {
+          $lookup:
+              {localField: "a", foreignField: "b", from: mongos0ForeignColl.getName(), as: "same"}
+        },
+        {$sort: {_id: 1}}
+    ];
+
+    // The results are expected to be correct if the $lookup stage is executed on the mongos which
+    // is aware that the collection is sharded.
+    const expectedResults = [
+        {_id: 0, a: 1, "same": [{_id: 0, b: 1}]},
+        {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+        {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+    ];
+
+    //
+    // Test unsharded local and sharded foreign collections, with the primary shard unaware that
+    // the foreign collection is sharded.
+    //
+
+    // Shard the foreign collection.
+    assert.commandWorked(
+        mongos0DB.adminCommand({shardCollection: mongos0ForeignColl.getFullName(), key: {_id: 1}}));
+
+    // Split the collection into 2 chunks: [MinKey, 1), [1, MaxKey).
+    assert.commandWorked(
+        mongos0DB.adminCommand({split: mongos0ForeignColl.getFullName(), middle: {_id: 1}}));
+
+    // Move the [minKey, 1) chunk to shard1.
+    assert.commandWorked(mongos0DB.adminCommand({
+        moveChunk: mongos0ForeignColl.getFullName(),
+        find: {_id: 0},
+        to: st.shard1.shardName,
+        _waitForDelete: true
+    }));
+
+    // Verify $lookup results through the fresh mongos.
+    restartPrimaryShard(st.rs0, mongos0LocalColl, mongos0ForeignColl);
+    assert.eq(mongos0LocalColl.aggregate(pipeline).toArray(), expectedResults);
+
+    // Verify $lookup results through mongos1, which is not aware that the local
+    // collection is sharded. The results are expected to be incorrect when both the mongos and
+    // primary shard incorrectly believe that a collection is unsharded.
+    // TODO: This should be fixed by SERVER-32629, likewise for the other aggregates in this file
+    // sent to the stale mongos.
+    restartPrimaryShard(st.rs0, mongos0LocalColl, mongos0ForeignColl);
+    assert.eq(mongos1LocalColl.aggregate(pipeline).toArray(), [
+        {_id: 0, a: 1, "same": []},
+        {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+        {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+    ]);
+
+    //
+    // Test sharded local and sharded foreign collections, with the primary shard unaware that
+    // either collection is sharded.
+    //
+
+    // Shard the local collection.
+    assert.commandWorked(
+        mongos0DB.adminCommand({shardCollection: mongos0LocalColl.getFullName(), key: {_id: 1}}));
+
+    // Split the collection into 2 chunks: [MinKey, 1), [1, MaxKey).
+    assert.commandWorked(
+        mongos0DB.adminCommand({split: mongos0LocalColl.getFullName(), middle: {_id: 1}}));
+
+    // Move the [minKey, 1) chunk to shard1.
+    assert.commandWorked(mongos0DB.adminCommand({
+        moveChunk: mongos0LocalColl.getFullName(),
+        find: {_id: 0},
+        to: st.shard1.shardName,
+        _waitForDelete: true
+    }));
+
+    // Verify $lookup results through the fresh mongos.
+    restartPrimaryShard(st.rs0, mongos0LocalColl, mongos0ForeignColl);
+    assert.eq(mongos0LocalColl.aggregate(pipeline).toArray(), expectedResults);
+
+    // Verify $lookup results through mongos1, which is not aware that the local
+    // collection is sharded. The results are expected to be incorrect when both the mongos and
+    // primary shard incorrectly believe that a collection is unsharded.
+    restartPrimaryShard(st.rs0, mongos0LocalColl, mongos0ForeignColl);
+    assert.eq(mongos1LocalColl.aggregate(pipeline).toArray(), [
+        {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+        {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+    ]);
+
+    //
+    // Test sharded local and unsharded foreign collections, with the primary shard unaware that
+    // the local collection is sharded.
+    //
+
+    // Recreate the foreign collection as unsharded.
+    mongos0ForeignColl.drop();
+    assert.writeOK(mongos0ForeignColl.insert({_id: 0, b: 1}));
+    assert.writeOK(mongos0ForeignColl.insert({_id: 1, b: null}));
+    assert.writeOK(mongos0ForeignColl.insert({_id: 2}));
+
+    // Verify $lookup results through the fresh mongos.
+    restartPrimaryShard(st.rs0, mongos0LocalColl, mongos0ForeignColl);
+    assert.eq(mongos0LocalColl.aggregate(pipeline).toArray(), expectedResults);
+
+    // Verify $lookup results through mongos1, which is not aware that the local
+    // collection is sharded. The results are expected to be incorrect when both the mongos and
+    // primary shard incorrectly believe that a collection is unsharded.
+    restartPrimaryShard(st.rs0, mongos0LocalColl, mongos0ForeignColl);
+    assert.eq(mongos1LocalColl.aggregate(pipeline).toArray(), [
+        {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+        {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+    ]);
+
+    st.stop();
+})();
diff --git a/jstests/sharding/lookup_stale_mongos.js b/jstests/sharding/lookup_stale_mongos.js
new file mode 100644
index 00000000000..3c713733b49
--- /dev/null
+++ b/jstests/sharding/lookup_stale_mongos.js
@@ -0,0 +1,130 @@
+// Tests the behavior of a $lookup when the mongos contains stale routing information for the
+// local and/or foreign collections.  This includes when mongos thinks the collection is sharded
+// when it's not, and likewise when mongos thinks the collection is unsharded but is actually
+// sharded.
+(function() {
+    "use strict";
+
+    const testName = "lookup_stale_mongos";
+    const st = new ShardingTest({
+        shards: 2,
+        mongos: 2,
+    });
+
+    const mongos0DB = st.s0.getDB(testName);
+    assert.commandWorked(mongos0DB.dropDatabase());
+    const mongos0LocalColl = mongos0DB[testName + "_local"];
+    const mongos0ForeignColl = mongos0DB[testName + "_foreign"];
+
+    const mongos1DB = st.s1.getDB(testName);
+    const mongos1LocalColl = mongos1DB[testName + "_local"];
+    const mongos1ForeignColl = mongos1DB[testName + "_foreign"];
+
+    const pipeline = [
+        {
+          $lookup:
+              {localField: "a", foreignField: "b", from: mongos1ForeignColl.getName(), as: "same"}
+        },
+        {$sort: {_id: 1}}
+    ];
+    const expectedResults = [
+        {_id: 0, a: 1, "same": [{_id: 0, b: 1}]},
+        {_id: 1, a: null, "same": [{_id: 1, b: null}, {_id: 2}]},
+        {_id: 2, "same": [{_id: 1, b: null}, {_id: 2}]}
+    ];
+
+    // Ensure that shard0 is the primary shard.
+    assert.commandWorked(mongos0DB.adminCommand({enableSharding: mongos0DB.getName()}));
+    st.ensurePrimaryShard(mongos0DB.getName(), st.shard0.shardName);
+
+    assert.writeOK(mongos0LocalColl.insert({_id: 0, a: 1}));
+    assert.writeOK(mongos0LocalColl.insert({_id: 1, a: null}));
+
+    assert.writeOK(mongos0ForeignColl.insert({_id: 0, b: 1}));
+    assert.writeOK(mongos0ForeignColl.insert({_id: 1, b: null}));
+
+    // Send writes through mongos1 such that it's aware of the collections and believes they are
+    // unsharded.
+    assert.writeOK(mongos1LocalColl.insert({_id: 2}));
+    assert.writeOK(mongos1ForeignColl.insert({_id: 2}));
+
+    //
+    // Test unsharded local and sharded foreign collections, with mongos unaware that the foreign
+    // collection is sharded.
+    //
+
+    // Shard the foreign collection through mongos0.
+    assert.commandWorked(
+        mongos0DB.adminCommand({shardCollection: mongos0ForeignColl.getFullName(), key: {_id: 1}}));
+
+    // Split the collection into 2 chunks: [MinKey, 1), [1, MaxKey).
+    assert.commandWorked(
+        mongos0DB.adminCommand({split: mongos0ForeignColl.getFullName(), middle: {_id: 1}}));
+
+    // Move the [minKey, 1) chunk to shard1.
+    assert.commandWorked(mongos0DB.adminCommand({
+        moveChunk: mongos0ForeignColl.getFullName(),
+        find: {_id: 0},
+        to: st.shard1.shardName,
+        _waitForDelete: true
+    }));
+
+    // Issue a $lookup through mongos1, which is unaware that the foreign collection is sharded.
+    assert.eq(mongos1LocalColl.aggregate(pipeline).toArray(), expectedResults);
+
+    //
+    // Test sharded local and sharded foreign collections, with mongos unaware that the local
+    // collection is sharded.
+    //
+
+    // Shard the local collection through mongos0.
+    assert.commandWorked(
+        mongos0DB.adminCommand({shardCollection: mongos0LocalColl.getFullName(), key: {_id: 1}}));
+
+    // Split the collection into 2 chunks: [MinKey, 1), [1, MaxKey).
+    assert.commandWorked(
+        mongos0DB.adminCommand({split: mongos0LocalColl.getFullName(), middle: {_id: 1}}));
+
+    // Move the [minKey, 1) chunk to shard1.
+    assert.commandWorked(mongos0DB.adminCommand({
+        moveChunk: mongos0LocalColl.getFullName(),
+        find: {_id: 0},
+        to: st.shard1.shardName,
+        _waitForDelete: true
+    }));
+
+    // Issue a $lookup through mongos1, which is unaware that the local collection is sharded.
+    assert.eq(mongos1LocalColl.aggregate(pipeline).toArray(), expectedResults);
+
+    //
+    // Test sharded local and unsharded foreign collections, with mongos unaware that the foreign
+    // collection is unsharded.
+    //
+
+    // Recreate the foreign collection as unsharded through mongos0.
+    mongos0ForeignColl.drop();
+    assert.writeOK(mongos0ForeignColl.insert({_id: 0, b: 1}));
+    assert.writeOK(mongos0ForeignColl.insert({_id: 1, b: null}));
+    assert.writeOK(mongos0ForeignColl.insert({_id: 2}));
+
+    // Issue a $lookup through mongos1, which is unaware that the foreign collection is now
+    // unsharded.
+    assert.eq(mongos1LocalColl.aggregate(pipeline).toArray(), expectedResults);
+
+    //
+    // Test unsharded local and foreign collections, with mongos unaware that the local
+    // collection is unsharded.
+    //
+
+    // Recreate the foreign collection as unsharded through mongos0.
+    mongos0LocalColl.drop();
+    assert.writeOK(mongos0LocalColl.insert({_id: 0, a: 1}));
+    assert.writeOK(mongos0LocalColl.insert({_id: 1, a: null}));
+    assert.writeOK(mongos0LocalColl.insert({_id: 2}));
+
+    // Issue a $lookup through mongos1, which is unaware that the local collection is now
+    // unsharded.
+    assert.eq(mongos1LocalColl.aggregate(pipeline).toArray(), expectedResults);
+
+    st.stop();
+})();