SERVER-58435 Implement the basics of translation of $lookup into SBE's nested loop join

4 files changed, 652 insertions, 18 deletions

diff --git a/src/mongo/db/query/SConscript b/src/mongo/db/query/SConscript
index 2332f06e8e3..1428041ae89 100644
--- a/src/mongo/db/query/SConscript
+++ b/src/mongo/db/query/SConscript
@@ -390,6 +390,7 @@ env.CppUnitTest(
         "sbe_and_hash_test.cpp",
         "sbe_and_sorted_test.cpp",
         "sbe_stage_builder_accumulator_test.cpp",
+        "sbe_stage_builder_lookup_test.cpp",
         "sbe_stage_builder_test_fixture.cpp",
         "sbe_stage_builder_test.cpp",
         "sbe_shard_filter_test.cpp",
diff --git a/src/mongo/db/query/sbe_stage_builder.cpp b/src/mongo/db/query/sbe_stage_builder.cpp
index afcc821784a..a834cf2c42c 100644
--- a/src/mongo/db/query/sbe_stage_builder.cpp
+++ b/src/mongo/db/query/sbe_stage_builder.cpp
@@ -2653,24 +2653,6 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     return {std::move(outStage), std::move(outputs)};
 }
 
-std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder::buildLookup(
-    const QuerySolutionNode* root, const PlanStageReqs& reqs) {
-    const auto lookupStage = static_cast<const EqLookupNode*>(root);
-    switch (lookupStage->lookupStrategy) {
-        case EqLookupNode::LookupStrategy::kHashJoin:
-            uasserted(5842602, "$lookup planning logic picked hash join");
-            break;
-        case EqLookupNode::LookupStrategy::kIndexedLoopJoin:
-            uasserted(5842603, "$lookup planning logic picked indexed loop join");
-            break;
-        case EqLookupNode::LookupStrategy::kNestedLoopJoin:
-            uasserted(5842604, "$lookup planning logic picked nested loop join");
-            break;
-        default:
-            MONGO_UNREACHABLE_TASSERT(5842605);
-    }
-    MONGO_UNREACHABLE_TASSERT(5842606);
-}
 std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots>
 SlotBasedStageBuilder::makeUnionForTailableCollScan(const QuerySolutionNode* root,
                                                     const PlanStageReqs& reqs) {
diff --git a/src/mongo/db/query/sbe_stage_builder_lookup.cpp b/src/mongo/db/query/sbe_stage_builder_lookup.cpp
new file mode 100644
index 00000000000..4dac1748299
--- /dev/null
+++ b/src/mongo/db/query/sbe_stage_builder_lookup.cpp
@@ -0,0 +1,270 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kQuery
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/query/sbe_stage_builder.h"
+
+#include <fmt/format.h>
+
+#include "mongo/db/exec/sbe/stages/loop_join.h"
+#include "mongo/db/query/sbe_stage_builder_coll_scan.h"
+#include "mongo/db/query/sbe_stage_builder_expression.h"
+#include "mongo/db/query/sbe_stage_builder_filter.h"
+#include "mongo/db/query/sbe_stage_builder_helpers.h"
+#include "mongo/db/query/sbe_stage_builder_index_scan.h"
+#include "mongo/db/query/sbe_stage_builder_projection.h"
+#include "mongo/db/query/util/make_data_structure.h"
+#include "mongo/logv2/log.h"
+
+namespace mongo::stage_builder {
+/**
+ * Helpers for building $lookup.
+ */
+namespace {
+using namespace sbe;
+using namespace sbe::value;
+
+std::unique_ptr<EExpression> replaceUndefinedWithNullOrPassthrough(SlotId slot) {
+    return makeE<EIf>(
+        makeE<ETypeMatch>(makeVariable(slot), getBSONTypeMask(TypeTags::bsonUndefined)),
+        makeConstant(TypeTags::Null, 0),
+        makeVariable(slot));
+}
+
+// Creates stages for building the key value. Missing keys are replaced with "null".
+// TODO SERVER-63690: implement handling of paths.
+std::pair<SlotId /*keySlot*/, std::unique_ptr<sbe::PlanStage>> buildLookupKey(
+    std::unique_ptr<sbe::PlanStage> inputStage,
+    SlotId recordSlot,
+    StringData fieldName,
+    const PlanNodeId nodeId,
+    SlotIdGenerator& slotIdGenerator) {
+    SlotId keySlot = slotIdGenerator.generate();
+    EvalStage innerBranch = makeProject(
+        EvalStage{std::move(inputStage), SlotVector{recordSlot}},
+        nodeId,
+        keySlot,
+        makeFunction("fillEmpty"_sd,
+                     makeFunction("getField"_sd, makeVariable(recordSlot), makeConstant(fieldName)),
+                     makeConstant(TypeTags::Null, 0)));
+
+    return {keySlot, std::move(innerBranch.stage)};
+}
+
+// Creates stages for the local side of $lookup.
+std::pair<SlotId /*localKey*/, std::unique_ptr<sbe::PlanStage>> buildLocalLookupBranch(
+    StageBuilderState& state,
+    std::unique_ptr<sbe::PlanStage> localInputStage,
+    SlotId localRecordSlot,
+    StringData localFieldName,
+    const PlanNodeId nodeId,
+    SlotIdGenerator& slotIdGenerator) {
+    auto [localKeySlot, localKeyStage] = buildLookupKey(
+        std::move(localInputStage), localRecordSlot, localFieldName, nodeId, slotIdGenerator);
+
+    // TODO SERVER-63691: repack local keys from an array into a set.
+    return std::make_pair(localKeySlot, std::move(localKeyStage));
+}
+
+// $lookup is an _outer_ left join that returns an empty array for "as" field rather than dropping
+// the unmatched local records. The branch that accumulates the matched records into an array
+// returns either 1 or 0 results, so to return an empty array for no-matches case we union this
+// branch with a const scan that produces an empty array but limit it to 1, so if the given branch
+// does produce a record, only that record is returned.
+std::pair<SlotId /*resultSlot*/, std::unique_ptr<sbe::PlanStage>> childResultOrEmptyArray(
+    std::unique_ptr<sbe::PlanStage> child,
+    SlotId childResultSlot,
+    const PlanNodeId nodeId,
+    SlotIdGenerator& slotIdGenerator) {
+    auto [emptyArrayTag, emptyArrayVal] = makeNewArray();
+    // Immediately take ownership of the new array (we could use a ValueGuard here but we'll
+    // need the constant below anyway).
+    std::unique_ptr<EExpression> emptyArrayConst = makeConstant(emptyArrayTag, emptyArrayVal);
+
+    SlotId emptyArraySlot = slotIdGenerator.generate();
+    std::unique_ptr<sbe::PlanStage> emptyArrayStage = makeProjectStage(
+        makeLimitCoScanTree(nodeId, 1), nodeId, emptyArraySlot, std::move(emptyArrayConst));
+
+    SlotId unionOutputSlot = slotIdGenerator.generate();
+    EvalStage unionStage =
+        makeUnion(makeVector(EvalStage{std::move(child), SlotVector{}},
+                             EvalStage{std::move(emptyArrayStage), SlotVector{}}),
+                  {makeSV(childResultSlot), makeSV(emptyArraySlot)} /*inputs*/,
+                  makeSV(unionOutputSlot),
+                  nodeId);
+
+    return std::make_pair(
+        unionOutputSlot,
+        std::move(makeLimitSkip(std::move(unionStage), nodeId, 1 /*limit*/).stage));
+}
+
+std::pair<SlotId /*matched docs*/, std::unique_ptr<sbe::PlanStage>> buildNljLookupStage(
+    StageBuilderState& state,
+    std::unique_ptr<sbe::PlanStage> localStage,
+    SlotId localRecordSlot,
+    StringData localFieldName,
+    std::unique_ptr<sbe::PlanStage> foreignStage,
+    SlotId foreignRecordSlot,
+    StringData foreignFieldName,
+    const PlanNodeId nodeId,
+    SlotIdGenerator& slotIdGenerator) {
+    // Build the outer branch that produces the correclated local key slot.
+    auto [localKeySlot, outerRootStage] = buildLocalLookupBranch(
+        state, std::move(localStage), localRecordSlot, localFieldName, nodeId, slotIdGenerator);
+
+    // Build the inner branch. It involves getting the key and then building the nested lookup join.
+    auto [foreignKeySlot, foreignKeyStage] = buildLookupKey(
+        std::move(foreignStage), foreignRecordSlot, foreignFieldName, nodeId, slotIdGenerator);
+    EvalStage innerBranch{std::move(foreignKeyStage), SlotVector{}};
+
+    // If the foreign key is an array, $lookup should match against any element of the array,
+    // so need to traverse into it, while applying the equality filter to each element. Contents
+    // of 'foreignKeySlot' will be overwritten with true/false result of the traversal - do we
+    // want to keep it?
+    {
+        SlotId traverseOutputSlot = slotIdGenerator.generate();
+        // "in" branch of the traverse applies the filter and populates the output slot with
+        // true/false. If the local key is an array check for membership rather than equality.
+        // Also, need to compare "undefined" in foreign as "null".
+        std::unique_ptr<EExpression> checkInTraverse = makeE<EIf>(
+            makeFunction("isArray"_sd, makeVariable(localKeySlot)),
+            makeFunction("isMember"_sd, makeVariable(foreignKeySlot), makeVariable(localKeySlot)),
+            makeBinaryOp(EPrimBinary::eq,
+                         makeVariable(localKeySlot),
+                         replaceUndefinedWithNullOrPassthrough(foreignKeySlot)));
+
+        SlotId innerTraverseOutputSlot = slotIdGenerator.generate();
+        EvalStage innerTraverseBranch = makeProject(makeLimitCoScanStage(nodeId),
+                                                    nodeId,
+                                                    innerTraverseOutputSlot,
+                                                    std::move(checkInTraverse));
+
+        innerBranch = makeTraverse(
+            std::move(innerBranch) /* "from" branch */,
+            std::move(innerTraverseBranch) /* "in" branch */,
+            foreignKeySlot /* inField */,
+            traverseOutputSlot /* outField */,
+            innerTraverseOutputSlot /* outFieldInner */,
+            makeVariable(innerTraverseOutputSlot) /* foldExpr */,
+            makeVariable(traverseOutputSlot) /* finalExpr: stop as soon as find a match */,
+            nodeId,
+            1 /*nestedArraysDepth*/);
+
+        // Add a filter that only lets through matched records.
+        std::unique_ptr<EExpression> predicate =
+            makeFunction("fillEmpty"_sd,
+                         makeVariable(traverseOutputSlot),
+                         makeConstant(TypeTags::Boolean, false));
+        innerBranch = makeFilter<false /*IsConst*/, false /*IsEof*/>(
+            std::move(innerBranch), std::move(predicate), nodeId);
+    }
+
+    // $lookup's aggregates the matching records into an array. We currently don't have a stage
+    // that could do this grouping _after_ NLJ, so we achieve it by having a hash_agg inside the
+    // inner branch that aggregates all matched records into a single accumulator. When there are
+    // no matches, return an empty array.
+    SlotId accumulatorSlot = slotIdGenerator.generate();
+    innerBranch = makeHashAgg(
+        std::move(innerBranch),
+        makeSV(), /*groupBy slots*/
+        makeEM(accumulatorSlot, makeFunction("addToArray"_sd, makeVariable(foreignRecordSlot))),
+        {} /*collatorSlot*/,
+        false /*allowDiskUse*/,
+        nodeId);
+    auto [innerResultSlot, innerRootStage] = childResultOrEmptyArray(
+        std::move(innerBranch.stage), accumulatorSlot, nodeId, slotIdGenerator);
+
+    // Connect the two branches with a nested loop join. For each outer record with a corresponding
+    // value in the 'localKeySlot', the inner branch will be executed and will place the result into
+    // 'innerResultSlot'.
+    std::unique_ptr<sbe::PlanStage> nlj =
+        makeS<LoopJoinStage>(std::move(outerRootStage),
+                             std::move(innerRootStage),
+                             makeSV(localRecordSlot) /*outerProjects*/,
+                             makeSV(localKeySlot) /*outerCorrelated*/,
+                             nullptr /*predicate*/,
+                             nodeId);
+
+    return {innerResultSlot, std::move(nlj)};
+}
+}  // namespace
+
+std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder::buildLookup(
+    const QuerySolutionNode* root, const PlanStageReqs& reqs) {
+    const auto eqLookupNode = static_cast<const EqLookupNode*>(root);
+
+    switch (eqLookupNode->lookupStrategy) {
+        case EqLookupNode::LookupStrategy::kHashJoin:
+            uasserted(5842602, "$lookup planning logic picked hash join");
+            break;
+        case EqLookupNode::LookupStrategy::kIndexedLoopJoin:
+            uasserted(5842603, "$lookup planning logic picked indexed loop join");
+            break;
+        case EqLookupNode::LookupStrategy::kNestedLoopJoin:
+            // TODO SERVER-63533: replace the check for number of children with proper access to the
+            // foreign collection. The check currently allows us to run unit tests.
+            if (eqLookupNode->children.size() == 2) {
+                const auto& localRoot = eqLookupNode->children[0];
+                auto [localStage, localOutputs] = build(localRoot, reqs);
+                sbe::value::SlotId localScanSlot = localOutputs.get(PlanStageSlots::kResult);
+
+                const auto& foreignRoot = eqLookupNode->children[1];
+                auto [foreignStage, foreignOutputs] = build(foreignRoot, reqs);
+                sbe::value::SlotId foreignScanSlot = foreignOutputs.get(PlanStageSlots::kResult);
+
+                auto [matchedSlot, nljStage] = buildNljLookupStage(_state,
+                                                                   std::move(localStage),
+                                                                   localScanSlot,
+                                                                   eqLookupNode->joinFieldLocal,
+                                                                   std::move(foreignStage),
+                                                                   foreignScanSlot,
+                                                                   eqLookupNode->joinFieldForeign,
+                                                                   eqLookupNode->nodeId(),
+                                                                   _slotIdGenerator);
+
+                PlanStageSlots outputs;
+                outputs.set(kResult, localScanSlot);  // TODO: create an object for $lookup result
+                outputs.set("local"_sd, localScanSlot);
+                outputs.set("matched"_sd, matchedSlot);
+                return {std::move(nljStage), std::move(outputs)};
+            } else {
+
+                uasserted(5842604, "$lookup planning logic picked nested loop join");
+                break;
+            }
+        default:
+            MONGO_UNREACHABLE_TASSERT(5842605);
+    }
+    MONGO_UNREACHABLE_TASSERT(5842606);
+}
+
+}  // namespace mongo::stage_builder
diff --git a/src/mongo/db/query/sbe_stage_builder_lookup_test.cpp b/src/mongo/db/query/sbe_stage_builder_lookup_test.cpp
new file mode 100644
index 00000000000..4817876257e
--- /dev/null
+++ b/src/mongo/db/query/sbe_stage_builder_lookup_test.cpp
@@ -0,0 +1,381 @@
+/**
+ *    Copyright (C) 2021-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+/**
+ * This file contains tests for building execution stages that implement $lookup operator.
+ */
+
+#include "mongo/db/exec/sbe/sbe_plan_stage_test.h"
+#include "mongo/db/exec/sbe/stages/loop_join.h"
+#include "mongo/db/pipeline/document_source_lookup.h"
+#include "mongo/db/query/query_solution.h"
+#include "mongo/db/query/sbe_stage_builder_test_fixture.h"
+#include "mongo/util/assert_util.h"
+
+#include "mongo/db/pipeline/expression_context.h"
+#include "mongo/db/pipeline/expression_context_for_test.h"
+#include "mongo/db/query/query_test_service_context.h"
+
+namespace mongo::sbe {
+using namespace value;
+
+struct LookupStageBuilderTest : public SbeStageBuilderTestFixture {
+    // VirtualScanNode wants the docs wrapped into BSONArray.
+    std::vector<BSONArray> prepInputForVirtualScanNode(const std::vector<BSONObj>& docs) {
+        std::vector<BSONArray> input;
+        input.reserve(docs.size());
+        for (const auto& doc : docs) {
+            input.emplace_back(BSON_ARRAY(doc));
+        }
+        return input;
+    }
+
+    // Constructs a QuerySolution consisting of a EqLookupNode on top of two VirtualScanNodes.
+    std::unique_ptr<QuerySolution> makeLookupSolution(const std::string& lookupSpec,
+                                                      const std::string& fromColl,
+                                                      const std::vector<BSONObj>& localDocs,
+                                                      const std::vector<BSONObj>& foreignDocs) {
+        auto expCtx = make_intrusive<ExpressionContextForTest>();
+        auto lookupNss = NamespaceString{"test", fromColl};
+        expCtx->setResolvedNamespace(lookupNss,
+                                     ExpressionContext::ResolvedNamespace{lookupNss, {}});
+
+        auto docSource =
+            DocumentSourceLookUp::createFromBson(fromjson(lookupSpec).firstElement(), expCtx);
+        auto docLookup = static_cast<DocumentSourceLookUp*>(docSource.get());
+
+        auto localVirtScanNode =
+            std::make_unique<VirtualScanNode>(prepInputForVirtualScanNode(localDocs),
+                                              VirtualScanNode::ScanType::kCollScan,
+                                              false /*hasRecordId*/);
+
+        auto lookupNode = std::make_unique<EqLookupNode>(std::move(localVirtScanNode),
+                                                         fromColl,
+                                                         docLookup->getLocalField()->fullPath(),
+                                                         docLookup->getForeignField()->fullPath(),
+                                                         docLookup->getAsField().fullPath());
+
+        auto foreignVirtScanNode =
+            std::make_unique<VirtualScanNode>(prepInputForVirtualScanNode(foreignDocs),
+                                              VirtualScanNode::ScanType::kCollScan,
+                                              false /*hasRecordId*/);
+        std::vector<std::unique_ptr<QuerySolutionNode>> additionalChildren;
+        additionalChildren.push_back(std::move(foreignVirtScanNode));
+        lookupNode->addChildren(std::move(additionalChildren));
+
+        return makeQuerySolution(std::move(lookupNode));
+    }
+
+    // Execute the stage tree and check the results.
+    void CheckNljResults(PlanStage* nljStage,
+                         SlotId localSlot,
+                         SlotId matchedSlot,
+                         const std::vector<std::pair<BSONObj, std::vector<BSONObj>>>& expected,
+                         bool debugPrint = false) {
+
+        auto ctx = makeCompileCtx();
+        prepareTree(ctx.get(), nljStage);
+        SlotAccessor* outer = nljStage->getAccessor(*ctx, localSlot);
+        SlotAccessor* inner = nljStage->getAccessor(*ctx, matchedSlot);
+
+        size_t i = 0;
+        for (auto st = nljStage->getNext(); st == PlanState::ADVANCED;
+             st = nljStage->getNext(), i++) {
+            auto [outerTag, outerVal] = outer->copyOrMoveValue();
+            ValueGuard outerGuard{outerTag, outerVal};
+            if (debugPrint) {
+                std::cout << i << " outer: " << std::make_pair(outerTag, outerVal) << std::endl;
+            }
+            if (i >= expected.size()) {
+                // We'll assert eventually that there were more actual results than expected.
+                continue;
+            }
+
+            auto [expectedOuterTag, expectedOuterVal] = copyValue(
+                TypeTags::bsonObject, bitcastFrom<const char*>(expected[i].first.objdata()));
+            ValueGuard expectedOuterGuard{expectedOuterTag, expectedOuterVal};
+
+            assertValuesEqual(outerTag, outerVal, expectedOuterTag, expectedOuterVal);
+
+            auto [innerTag, innerVal] = inner->copyOrMoveValue();
+            ValueGuard innerGuard{innerTag, innerVal};
+
+            ASSERT_EQ(innerTag, TypeTags::Array);
+            auto innerMatches = getArrayView(innerVal);
+
+            if (debugPrint) {
+                std::cout << "  inner:" << std::endl;
+                for (size_t m = 0; m < innerMatches->size(); m++) {
+                    auto [matchedTag, matchedVal] = innerMatches->getAt(m);
+                    std::cout << "  " << m << ": " << std::make_pair(matchedTag, matchedVal)
+                              << std::endl;
+                }
+            }
+
+            ASSERT_EQ(innerMatches->size(), expected[i].second.size());
+            for (size_t m = 0; m < innerMatches->size(); m++) {
+                auto [matchedTag, matchedVal] = innerMatches->getAt(m);
+                auto [expectedMatchTag, expectedMatchVal] =
+                    copyValue(TypeTags::bsonObject,
+                              bitcastFrom<const char*>(expected[i].second[m].objdata()));
+                ValueGuard expectedMatchGuard{expectedMatchTag, expectedMatchVal};
+                assertValuesEqual(matchedTag, matchedVal, expectedMatchTag, expectedMatchVal);
+            }
+        }
+        ASSERT_EQ(i, expected.size());
+        nljStage->close();
+    }
+
+    void runTest(const std::vector<BSONObj>& ldocs,
+                 const std::vector<BSONObj>& fdocs,
+                 const std::string& lkey,
+                 const std::string& fkey,
+                 const std::vector<std::pair<BSONObj, std::vector<BSONObj>>>& expected,
+                 bool debugPrint = false) {
+        const char* foreignCollName = "fromColl";
+        std::stringstream lookupSpec;
+        lookupSpec << "{$lookup: ";
+        lookupSpec << "  {";
+        lookupSpec << "    from: '" << foreignCollName << "', ";
+        lookupSpec << "    localField:   '" << lkey << "', ";
+        lookupSpec << "    foreignField: '" << fkey << "', ";
+        lookupSpec << "    as: 'matched'";
+        lookupSpec << "  }";
+        lookupSpec << "}";
+
+        auto solution = makeLookupSolution(lookupSpec.str(), foreignCollName, ldocs, fdocs);
+        auto [resultSlots, stage, data, _] = buildPlanStage(std::move(solution),
+                                                            false /*hasRecordId*/,
+                                                            nullptr /*shard filterer*/,
+                                                            nullptr /*collator*/);
+        if (debugPrint) {
+            debugPrintPlan(*stage);
+        }
+
+        CheckNljResults(stage.get(),
+                        data.outputs.get("local"),
+                        data.outputs.get("matched"),
+                        expected,
+                        debugPrint);
+    }
+
+    void debugPrintPlan(const PlanStage& stage, StringData header = "") {
+        std::cout << std::endl << "*** " << header << " ***" << std::endl;
+        std::cout << DebugPrinter{}.print(stage.debugPrint());
+        std::cout << std::endl;
+    }
+};
+
+TEST_F(LookupStageBuilderTest, NestedLoopJoin_Basic) {
+    const std::vector<BSONObj> ldocs = {
+        fromjson("{id:0, lkey:1}"),
+        fromjson("{id:1, lkey:12}"),
+        fromjson("{id:2, lkey:3}"),
+        fromjson("{id:3, lkey:[1,4]}"),
+    };
+
+    const std::vector<BSONObj> fdocs = {
+        fromjson("{id:0, fkey:1}"),
+        fromjson("{id:1, fkey:3}"),
+        fromjson("{id:2, fkey:[1,4,25]}"),
+        fromjson("{id:3, fkey:4}"),
+        fromjson("{id:4, fkey:[24,25,26]}"),
+        fromjson("{id:5, no_fkey:true}"),
+        fromjson("{id:6, fkey:null}"),
+        fromjson("{id:7, fkey:undefined}"),
+        fromjson("{id:8, fkey:[]}"),
+        fromjson("{id:9, fkey:[null]}"),
+    };
+
+    const std::vector<std::pair<BSONObj, std::vector<BSONObj>>> expected = {
+        {ldocs[0], {fdocs[0], fdocs[2]}},
+        {ldocs[1], {}},
+        {ldocs[2], {fdocs[1]}},
+        {ldocs[3], {fdocs[0], fdocs[2], fdocs[3]}},
+    };
+
+    runTest(ldocs, fdocs, "lkey", "fkey", expected);
+}
+
+TEST_F(LookupStageBuilderTest, NestedLoopJoin_LocalKey_Null) {
+    const std::vector<BSONObj> ldocs = {fromjson("{id:0, lkey:null}")};
+
+    const std::vector<BSONObj> fdocs = {fromjson("{id:0, fkey:1}"),
+                                        fromjson("{id:1, no_fkey:true}"),
+                                        fromjson("{id:2, fkey:null}"),
+                                        fromjson("{id:3, fkey:[null]}"),
+                                        fromjson("{id:4, fkey:undefined}"),
+                                        fromjson("{id:5, fkey:[undefined]}"),
+                                        fromjson("{id:6, fkey:[]}"),
+                                        fromjson("{id:7, fkey:[[]]}")};
+
+    std::vector<std::pair<BSONObj, std::vector<BSONObj>>> expected{
+        {ldocs[0], {fdocs[1], fdocs[2], fdocs[3], fdocs[4], fdocs[5]}},
+    };
+
+    runTest(ldocs, fdocs, "lkey", "fkey", expected);
+}
+
+TEST_F(LookupStageBuilderTest, NestedLoopJoin_LocalKey_Missing) {
+    const std::vector<BSONObj> ldocs = {fromjson("{id:0, no_lkey:true}")};
+
+    const std::vector<BSONObj> fdocs = {fromjson("{id:0, fkey:1}"),
+                                        fromjson("{id:1, no_fkey:true}"),
+                                        fromjson("{id:2, fkey:null}"),
+                                        fromjson("{id:3, fkey:[null]}"),
+                                        fromjson("{id:4, fkey:undefined}"),
+                                        fromjson("{id:5, fkey:[undefined]}"),
+                                        fromjson("{id:6, fkey:[]}"),
+                                        fromjson("{id:7, fkey:[[]]}")};
+
+    std::vector<std::pair<BSONObj, std::vector<BSONObj>>> expected = {
+        {ldocs[0], {fdocs[1], fdocs[2], fdocs[3], fdocs[4], fdocs[5]}},
+    };
+
+    runTest(ldocs, fdocs, "lkey", "fkey", expected);
+}
+
+TEST_F(LookupStageBuilderTest, NestedLoopJoin_EmptyArrays) {
+    const std::vector<BSONObj> ldocs = {
+        fromjson("{id:0, lkey:[]}"),
+        fromjson("{id:1, lkey:[[]]}"),
+    };
+    const std::vector<BSONObj> fdocs = {
+        fromjson("{id:0, fkey:1}"),
+        fromjson("{id:1, no_fkey:true}"),
+        fromjson("{id:2, fkey:null}"),
+        fromjson("{id:3, fkey:[null]}"),
+        fromjson("{id:4, fkey:undefined}"),
+        fromjson("{id:5, fkey:[undefined]}"),
+        fromjson("{id:6, fkey:[]}"),
+        fromjson("{id:7, fkey:[[]]}"),
+    };
+
+    std::vector<std::pair<BSONObj, std::vector<BSONObj>>> expected = {
+        {ldocs[0], {}},          // TODO SERVER-63368: fix this case if the ticket is declined
+        {ldocs[1], {fdocs[7]}},  // TODO SEVER-63700: it should be {fdocs[6], fdocs[7]}
+    };
+
+    runTest(ldocs, fdocs, "lkey", "fkey", expected);
+}
+
+TEST_F(LookupStageBuilderTest, NestedLoopJoin_LocalKey_SubFieldScalar) {
+    const std::vector<BSONObj> ldocs = {
+        fromjson("{id:0, nested:{lkey:1, other:3}}"),
+        fromjson("{id:1, nested:{no_lkey:true}}"),
+        fromjson("{id:2, nested:1}"),
+        fromjson("{id:3, lkey:1}"),
+        fromjson("{id:4, nested:{lkey:42}}"),
+    };
+    const std::vector<BSONObj> fdocs = {
+        fromjson("{id:0, fkey:1}"),
+        fromjson("{id:1, no_fkey:true}"),
+        fromjson("{id:2, fkey:3}"),
+        fromjson("{id:3, fkey:[1, 2]}"),
+    };
+
+    std::vector<std::pair<BSONObj, std::vector<BSONObj>>> expected = {
+        {ldocs[0], {fdocs[0], fdocs[3]}},
+        {ldocs[1], {fdocs[1]}},
+        {ldocs[2], {fdocs[1]}},
+        {ldocs[3], {fdocs[1]}},
+        {ldocs[4], {}},
+    };
+
+    // TODO SERVER-63690: enable this test.
+    // runTest(ldocs, fdocs, "nested.lkey", "fkey", expected);
+}
+
+TEST_F(LookupStageBuilderTest, NestedLoopJoin_LocalKey_SubFieldArray) {
+    const std::vector<BSONObj> ldocs = {
+        fromjson("{id:0, nested:[{lkey:1},{lkey:2}]}"),
+        fromjson("{id:1, nested:[{lkey:42}]}"),
+        fromjson("{id:2, nested:[{lkey:{other:1}}]}"),
+        fromjson("{id:3, nested:[{lkey:[]}]}"),
+        fromjson("{id:4, nested:[{other:3}]}"),
+        fromjson("{id:5, nested:[]}"),
+        fromjson("{id:6, nested:[[]]}"),
+        fromjson("{id:7, lkey:[1,2]}"),
+    };
+    const std::vector<BSONObj> fdocs = {
+        fromjson("{id:0, fkey:1}"),
+        fromjson("{id:1, fkey:2}"),
+        fromjson("{id:2, fkey:3}"),
+        fromjson("{id:3, fkey:[1, 4]}"),
+        fromjson("{id:4, no_fkey:true}"),
+        fromjson("{id:5, fkey:[]}"),
+        fromjson("{id:6, fkey:null}"),
+    };
+
+    //'expected' documents pre-SERVER-63368 behavior of the classic engine.
+    std::vector<std::pair<BSONObj, std::vector<BSONObj>>> expected = {
+        {ldocs[0], {fdocs[0], fdocs[1], fdocs[3]}},
+        {ldocs[1], {}},
+        {ldocs[2], {}},
+        {ldocs[3], {fdocs[4], fdocs[6]}},
+        {ldocs[4], {fdocs[4], fdocs[6]}},
+        {ldocs[5], {fdocs[4], fdocs[6]}},
+        {ldocs[6], {fdocs[4], fdocs[6]}},
+        {ldocs[7], {fdocs[4], fdocs[6]}},
+    };
+
+    // TODO SERVER-63690: enable this test.
+    // runTest(ldocs, fdocs, "nested.lkey", "fkey", expected, true);
+}
+
+TEST_F(LookupStageBuilderTest, NestedLoopJoin_LocalKey_PathWithNumber) {
+    const std::vector<BSONObj> ldocs = {
+        fromjson("{id:0, nested:[{lkey:1},{lkey:2}]}"),
+        fromjson("{id:1, nested:[{lkey:[2,3,1]}]}"),
+        fromjson("{id:2, nested:[{lkey:2},{lkey:1}]}"),
+        fromjson("{id:3, nested:[{lkey:[2,3]}]}"),
+        fromjson("{id:4, nested:{lkey:1}}"),
+        fromjson("{id:5, nested:{lkey:[1,2]}}"),
+        fromjson("{id:6, nested:[{other:1},{lkey:1}]}"),
+    };
+    const std::vector<BSONObj> fdocs = {
+        fromjson("{id:0, fkey:1}"),
+        fromjson("{id:1, fkey:3}"),
+        fromjson("{id:2, fkey:null}"),
+    };
+    //'expected' documents pre-SERVER-63368 behavior of the classic engine.
+    std::vector<std::pair<BSONObj, std::vector<BSONObj>>> expected = {
+        {ldocs[0], {fdocs[0]}},
+        {ldocs[1], {fdocs[0], fdocs[1]}},
+        {ldocs[2], {}},
+        {ldocs[3], {fdocs[1]}},
+        {ldocs[4], {fdocs[2]}},
+        {ldocs[5], {fdocs[2]}},
+        {ldocs[6], {fdocs[2]}},
+    };
+
+    // TODO SERVER-63690: either remove or enable this test.
+    // runTest(ldocs, fdocs, "nested.0.lkey", "fkey", expected, true);
+}
+}  // namespace mongo::sbe