SERVER-68970 Update sbe_stage_builder_projection.cpp to use traverseP

14 files changed, 645 insertions, 250 deletions

diff --git a/src/mongo/db/exec/sbe/expressions/expression.cpp b/src/mongo/db/exec/sbe/expressions/expression.cpp
index a787dea1284..d337cd8fa14 100644
--- a/src/mongo/db/exec/sbe/expressions/expression.cpp
+++ b/src/mongo/db/exec/sbe/expressions/expression.cpp
@@ -398,6 +398,7 @@ static stdx::unordered_map<std::string, BuiltinFn> kBuiltinFunctions = {
     {"newArrayFromRange",
      BuiltinFn{[](size_t n) { return n == 3; }, vm::Builtin::newArrayFromRange, false}},
     {"newObj", BuiltinFn{[](size_t n) { return n % 2 == 0; }, vm::Builtin::newObj, false}},
+    {"makeBsonObj", BuiltinFn{[](size_t n) { return n >= 2; }, vm::Builtin::makeBsonObj, false}},
     {"ksToString", BuiltinFn{[](size_t n) { return n == 1; }, vm::Builtin::ksToString, false}},
     {"ks",
      BuiltinFn{[](size_t n) { return n >= 3 && n < Ordering::kMaxCompoundIndexKeys + 3; },
diff --git a/src/mongo/db/exec/sbe/makeobj_enums.h b/src/mongo/db/exec/sbe/makeobj_enums.h
new file mode 100644
index 00000000000..966b2dc48b0
--- /dev/null
+++ b/src/mongo/db/exec/sbe/makeobj_enums.h
@@ -0,0 +1,38 @@
+/**
+ *    Copyright (C) 2022-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.
+ */
+
+#pragma once
+
+namespace mongo::sbe {
+
+enum class MakeObjFieldBehavior { drop, keep };
+
+enum class MakeObjOutputType { object, bsonObject };
+
+}  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/stages/makeobj.cpp b/src/mongo/db/exec/sbe/stages/makeobj.cpp
index a0254e813f1..b1d55813b84 100644
--- a/src/mongo/db/exec/sbe/stages/makeobj.cpp
+++ b/src/mongo/db/exec/sbe/stages/makeobj.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/makeobj_spec.h"
 #include "mongo/util/str.h"
 
 namespace mongo::sbe {
@@ -109,18 +110,12 @@ void MakeObjStageBase<O>::prepare(CompileCtx& ctx) {
     if (_rootSlot) {
         _root = _children[0]->getAccessor(ctx, *_rootSlot);
     }
-    for (auto& p : _fields) {
-        // Mark the values from _fields with 'std::numeric_limits<size_t>::max()'.
-        auto [it, inserted] = _allFieldsMap.emplace(p, std::numeric_limits<size_t>::max());
-        uassert(4822818, str::stream() << "duplicate field: " << p, inserted);
-    }
+
+    _allFieldsMap = value::MakeObjSpec::buildAllFieldsMap(_fields, _projectFields);
 
     for (size_t idx = 0; idx < _projectFields.size(); ++idx) {
-        auto& p = _projectFields[idx];
-        // Mark the values from _projectFields with their corresponding index.
-        auto [it, inserted] = _allFieldsMap.emplace(p, idx);
-        uassert(4822819, str::stream() << "duplicate field: " << p, inserted);
-        _projects.emplace_back(p, _children[0]->getAccessor(ctx, _projectVars[idx]));
+        _projects.emplace_back(_projectFields[idx],
+                               _children[0]->getAccessor(ctx, _projectVars[idx]));
     }
 
     _compiled = true;
@@ -273,46 +268,9 @@ void MakeObjStageBase<MakeObjOutputType::bsonObject>::produceObject() {
         auto [tag, val] = _root->getViewOfValue();
 
         size_t nFieldsNeededIfInclusion = _fields.size();
-        if (tag == value::TypeTags::bsonObject) {
-            if (!(nFieldsNeededIfInclusion == 0 && _fieldBehavior == FieldBehavior::keep)) {
-                auto be = value::bitcastTo<const char*>(val);
-                // Skip document length.
-                be += 4;
-                while (*be != 0) {
-                    auto sv = bson::fieldNameView(be);
-                    auto key = StringMapHasher{}.hashed_key(StringData(sv));
-
-                    auto nextBe = bson::advance(be, sv.size());
-
-                    if (!isFieldProjectedOrRestricted(key)) {
-                        bob.append(BSONElement(be, sv.size() + 1, nextBe - be));
-                        --nFieldsNeededIfInclusion;
-                    }
-
-                    if (nFieldsNeededIfInclusion == 0 && _fieldBehavior == FieldBehavior::keep) {
-                        break;
-                    }
-
-                    be = nextBe;
-                }
-            }
-        } else if (tag == value::TypeTags::Object) {
-            if (!(nFieldsNeededIfInclusion == 0 && _fieldBehavior == FieldBehavior::keep)) {
-                auto objRoot = value::getObjectView(val);
-                for (size_t idx = 0; idx < objRoot->size(); ++idx) {
-                    auto key = StringMapHasher{}.hashed_key(StringData(objRoot->field(idx)));
-
-                    if (!isFieldProjectedOrRestricted(key)) {
-                        auto [tag, val] = objRoot->getAt(idx);
-                        bson::appendValueToBsonObj(bob, objRoot->field(idx), tag, val);
-                        --nFieldsNeededIfInclusion;
-                    }
-
-                    if (nFieldsNeededIfInclusion == 0 && _fieldBehavior == FieldBehavior::keep) {
-                        break;
-                    }
-                }
-            }
+        if (value::isObject(tag)) {
+            value::MakeObjSpec::keepOrDropFields(
+                tag, val, _fieldBehavior, nFieldsNeededIfInclusion, _allFieldsMap, bob);
         } else {
             for (size_t idx = 0; idx < _projects.size(); ++idx) {
                 projectField(&bob, idx);
diff --git a/src/mongo/db/exec/sbe/stages/makeobj.h b/src/mongo/db/exec/sbe/stages/makeobj.h
index 0094f41af5a..16575b5a456 100644
--- a/src/mongo/db/exec/sbe/stages/makeobj.h
+++ b/src/mongo/db/exec/sbe/stages/makeobj.h
@@ -29,15 +29,11 @@
 
 #pragma once
 
+#include "mongo/db/exec/sbe/makeobj_enums.h"
 #include "mongo/db/exec/sbe/stages/stages.h"
 #include "mongo/db/exec/sbe/values/value.h"
 
 namespace mongo::sbe {
-
-enum class MakeObjFieldBehavior { drop, keep };
-
-enum class MakeObjOutputType { object, bsonObject };
-
 /**
  * Base stage for creating a bsonObject or object.
  *
diff --git a/src/mongo/db/exec/sbe/values/makeobj_spec.h b/src/mongo/db/exec/sbe/values/makeobj_spec.h
new file mode 100644
index 00000000000..9e6ee5a9119
--- /dev/null
+++ b/src/mongo/db/exec/sbe/values/makeobj_spec.h
@@ -0,0 +1,181 @@
+/**
+ *    Copyright (C) 2022-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.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/sbe/makeobj_enums.h"
+#include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/value.h"
+
+namespace mongo::sbe::value {
+/**
+ * MakeObjSpec is a wrapper around a FieldBehavior value, a list of fields, a list of projected
+ * fields, and a StringMap object.
+ */
+class MakeObjSpec {
+public:
+    using FieldBehavior = MakeObjFieldBehavior;
+
+    // GetArgFn is a function pointer type used by produceBsonObject().
+    using GetArgFn = std::function<std::pair<value::TypeTags, value::Value>(size_t idx)>;
+
+    MakeObjSpec(boost::optional<FieldBehavior> fieldBehavior,
+                std::vector<std::string> fields,
+                std::vector<std::string> projectFields)
+        : _fieldBehavior(fieldBehavior),
+          _fields(fields),
+          _projectFields(projectFields),
+          _allFieldsMap(buildAllFieldsMap()) {}
+
+    size_t getApproximateSize() const;
+
+private:
+    StringMap<size_t> buildAllFieldsMap() const;
+
+    void keepOrDropFields(value::TypeTags rootTag,
+                          value::Value rootVal,
+                          UniqueBSONObjBuilder& bob) const;
+
+public:
+    static StringMap<size_t> buildAllFieldsMap(const std::vector<std::string>& fields,
+                                               const std::vector<std::string>& projectFields) {
+        StringMap<size_t> m;
+
+        for (auto& p : fields) {
+            // Mark the values from fields with 'std::numeric_limits<size_t>::max()'.
+            auto [it, inserted] = m.emplace(p, std::numeric_limits<size_t>::max());
+            uassert(6897000, str::stream() << "duplicate field: " << p, inserted);
+        }
+
+        for (size_t idx = 0; idx < projectFields.size(); ++idx) {
+            auto& p = projectFields[idx];
+            // Mark the values from projectFields with their corresponding index.
+            auto [it, inserted] = m.emplace(p, idx);
+            uassert(6897001, str::stream() << "duplicate field: " << p, inserted);
+        }
+
+        return m;
+    }
+
+    static void keepOrDropFields(value::TypeTags rootTag,
+                                 value::Value rootVal,
+                                 const boost::optional<FieldBehavior>& fieldBehavior,
+                                 size_t nFieldsNeededIfInclusion,
+                                 const StringMap<size_t>& allFieldsMap,
+                                 UniqueBSONObjBuilder& bob) {
+        auto isFieldProjectedOrRestricted = [&](const StringMapHashedKey& key) -> bool {
+            bool foundKey = false;
+            bool projected = false;
+            bool restricted = false;
+
+            if (!allFieldsMap.empty()) {
+                if (auto it = allFieldsMap.find(key); it != allFieldsMap.end()) {
+                    foundKey = true;
+                    projected = it->second != std::numeric_limits<size_t>::max();
+                    restricted = *fieldBehavior != FieldBehavior::keep;
+                }
+            }
+            if (!foundKey) {
+                restricted = *fieldBehavior == FieldBehavior::keep;
+            }
+
+            return projected || restricted;
+        };
+
+        if (rootTag == value::TypeTags::bsonObject) {
+            if (!(nFieldsNeededIfInclusion == 0 && fieldBehavior == FieldBehavior::keep)) {
+                auto be = value::bitcastTo<const char*>(rootVal);
+                // Skip document length.
+                be += 4;
+                while (*be != 0) {
+                    auto sv = bson::fieldNameView(be);
+                    auto key = StringMapHasher{}.hashed_key(StringData(sv));
+                    auto nextBe = bson::advance(be, sv.size());
+
+                    if (!isFieldProjectedOrRestricted(key)) {
+                        bob.append(BSONElement(be, sv.size() + 1, nextBe - be));
+                        --nFieldsNeededIfInclusion;
+                    }
+
+                    if (nFieldsNeededIfInclusion == 0 && fieldBehavior == FieldBehavior::keep) {
+                        break;
+                    }
+
+                    be = nextBe;
+                }
+            }
+        } else if (rootTag == value::TypeTags::Object) {
+            if (!(nFieldsNeededIfInclusion == 0 && fieldBehavior == FieldBehavior::keep)) {
+                auto objRoot = value::getObjectView(rootVal);
+                for (size_t idx = 0; idx < objRoot->size(); ++idx) {
+                    auto key = StringMapHasher{}.hashed_key(StringData(objRoot->field(idx)));
+
+                    if (!isFieldProjectedOrRestricted(key)) {
+                        auto [tag, val] = objRoot->getAt(idx);
+                        bson::appendValueToBsonObj(bob, objRoot->field(idx), tag, val);
+                        --nFieldsNeededIfInclusion;
+                    }
+
+                    if (nFieldsNeededIfInclusion == 0 && fieldBehavior == FieldBehavior::keep) {
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    // This method will invoke the 'getArg' function pointer to retrieve the value for each
+    // projected field (if any).
+    std::pair<value::TypeTags, value::Value> produceBsonObject(value::TypeTags rootTag,
+                                                               value::Value rootVal,
+                                                               const GetArgFn& getArg) const {
+        UniqueBSONObjBuilder bob;
+        if (value::isObject(rootTag)) {
+            keepOrDropFields(rootTag, rootVal, bob);
+        }
+
+        for (size_t idx = 0; idx < _projectFields.size(); ++idx) {
+            auto [fieldTag, fieldVal] = getArg(idx);
+            bson::appendValueToBsonObj(bob, _projectFields[idx], fieldTag, fieldVal);
+        }
+
+        bob.doneFast();
+        char* data = bob.bb().release().release();
+        return {value::TypeTags::bsonObject, value::bitcastFrom<char*>(data)};
+    }
+
+    std::string toString() const;
+
+private:
+    const boost::optional<FieldBehavior> _fieldBehavior;
+    const std::vector<std::string> _fields;
+    const std::vector<std::string> _projectFields;
+    const StringMap<size_t> _allFieldsMap;
+};
+}  // namespace mongo::sbe::value
diff --git a/src/mongo/db/exec/sbe/values/slot.cpp b/src/mongo/db/exec/sbe/values/slot.cpp
index 0274df1b6e7..e9f6e08f313 100644
--- a/src/mongo/db/exec/sbe/values/slot.cpp
+++ b/src/mongo/db/exec/sbe/values/slot.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/exec/js_function.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/makeobj_spec.h"
 #include "mongo/db/exec/sbe/values/sort_spec.h"
 #include "mongo/db/exec/sbe/values/value_builder.h"
 #include "mongo/db/storage/key_string.h"
@@ -705,6 +706,9 @@ int getApproximateSize(TypeTags tag, Value val) {
         case TypeTags::sortSpec:
             result += getSortSpecView(val)->getApproximateSize();
             break;
+        case TypeTags::makeObjSpec:
+            result += getMakeObjSpecView(val)->getApproximateSize();
+            break;
         case TypeTags::indexBounds:
             result += size_estimator::estimate(*getIndexBoundsView(val));
             break;
diff --git a/src/mongo/db/exec/sbe/values/value.cpp b/src/mongo/db/exec/sbe/values/value.cpp
index 23f5f46e1e6..aa24586f58b 100644
--- a/src/mongo/db/exec/sbe/values/value.cpp
+++ b/src/mongo/db/exec/sbe/values/value.cpp
@@ -33,7 +33,9 @@
 
 #include "mongo/base/compare_numbers.h"
 #include "mongo/db/exec/js_function.h"
+#include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/makeobj_spec.h"
 #include "mongo/db/exec/sbe/values/sort_spec.h"
 #include "mongo/db/exec/sbe/values/value_builder.h"
 #include "mongo/db/exec/sbe/values/value_printer.h"
@@ -193,6 +195,47 @@ size_t SortSpec::getApproximateSize() const {
     return size;
 }
 
+size_t MakeObjSpec::getApproximateSize() const {
+    auto size = sizeof(MakeObjSpec);
+    size += size_estimator::estimate(_fields);
+    size += size_estimator::estimate(_projectFields);
+    size += size_estimator::estimate(_allFieldsMap);
+    return size;
+}
+
+StringMap<size_t> MakeObjSpec::buildAllFieldsMap() const {
+    return buildAllFieldsMap(_fields, _projectFields);
+}
+
+void MakeObjSpec::keepOrDropFields(value::TypeTags rootTag,
+                                   value::Value rootVal,
+                                   UniqueBSONObjBuilder& bob) const {
+    keepOrDropFields(rootTag, rootVal, _fieldBehavior, _fields.size(), _allFieldsMap, bob);
+}
+
+std::string MakeObjSpec::toString() const {
+    StringBuilder builder;
+    builder << (_fieldBehavior == MakeObjSpec::FieldBehavior::keep ? "keep" : "drop") << ", [";
+
+    for (size_t i = 0; i < _fields.size(); ++i) {
+        if (i != 0) {
+            builder << ", ";
+        }
+        builder << '"' << _fields[i] << '"';
+    }
+    builder << "], [";
+
+    for (size_t i = 0; i < _projectFields.size(); ++i) {
+        if (i != 0) {
+            builder << ", ";
+        }
+        builder << '"' << _projectFields[i] << '"';
+    }
+    builder << "]";
+
+    return builder.str();
+}
+
 std::pair<TypeTags, Value> makeCopyJsFunction(const JsFunction& jsFunction) {
     auto ownedJsFunction = bitcastFrom<JsFunction*>(new JsFunction(jsFunction));
     return {TypeTags::jsFunction, ownedJsFunction};
@@ -213,6 +256,11 @@ std::pair<TypeTags, Value> makeCopySortSpec(const SortSpec& ss) {
     return {TypeTags::sortSpec, ssCopy};
 }
 
+std::pair<TypeTags, Value> makeCopyMakeObjSpec(const MakeObjSpec& mos) {
+    auto mosCopy = bitcastFrom<MakeObjSpec*>(new MakeObjSpec(mos));
+    return {TypeTags::makeObjSpec, mosCopy};
+}
+
 std::pair<TypeTags, Value> makeCopyCollator(const CollatorInterface& collator) {
     auto collatorCopy = bitcastFrom<CollatorInterface*>(collator.clone().release());
     return {TypeTags::collator, collatorCopy};
@@ -291,6 +339,9 @@ void releaseValueDeep(TypeTags tag, Value val) noexcept {
         case TypeTags::sortSpec:
             delete getSortSpecView(val);
             break;
+        case TypeTags::makeObjSpec:
+            delete getMakeObjSpecView(val);
+            break;
         case TypeTags::collator:
             delete getCollatorView(val);
             break;
diff --git a/src/mongo/db/exec/sbe/values/value.h b/src/mongo/db/exec/sbe/values/value.h
index feae93a69ee..082ee4d3523 100644
--- a/src/mongo/db/exec/sbe/values/value.h
+++ b/src/mongo/db/exec/sbe/values/value.h
@@ -76,6 +76,7 @@ using IndexKeysInclusionSet = std::bitset<Ordering::kMaxCompoundIndexKeys>;
 
 namespace value {
 class SortSpec;
+class MakeObjSpec;
 
 static constexpr size_t kNewUUIDLength = 16;
 
@@ -157,6 +158,9 @@ enum class TypeTags : uint8_t {
     // Pointer to a SortSpec object.
     sortSpec,
 
+    // Pointer to a MakeObjSpec object.
+    makeObjSpec,
+
     // Pointer to a IndexBounds object.
     indexBounds,
 
@@ -1210,6 +1214,10 @@ inline SortSpec* getSortSpecView(Value val) noexcept {
     return reinterpret_cast<SortSpec*>(val);
 }
 
+inline MakeObjSpec* getMakeObjSpecView(Value val) noexcept {
+    return reinterpret_cast<MakeObjSpec*>(val);
+}
+
 inline IndexBounds* getIndexBoundsView(Value val) noexcept {
     return reinterpret_cast<IndexBounds*>(val);
 }
@@ -1336,6 +1344,8 @@ std::pair<TypeTags, Value> makeCopyFtsMatcher(const fts::FTSMatcher&);
 
 std::pair<TypeTags, Value> makeCopySortSpec(const SortSpec&);
 
+std::pair<TypeTags, Value> makeCopyMakeObjSpec(const MakeObjSpec&);
+
 std::pair<TypeTags, Value> makeCopyCollator(const CollatorInterface& collator);
 
 std::pair<TypeTags, Value> makeCopyIndexBounds(const IndexBounds& collator);
@@ -1406,6 +1416,8 @@ inline std::pair<TypeTags, Value> copyValue(TypeTags tag, Value val) {
             return makeCopyFtsMatcher(*getFtsMatcherView(val));
         case TypeTags::sortSpec:
             return makeCopySortSpec(*getSortSpecView(val));
+        case TypeTags::makeObjSpec:
+            return makeCopyMakeObjSpec(*getMakeObjSpecView(val));
         case TypeTags::collator:
             return makeCopyCollator(*getCollatorView(val));
         case TypeTags::indexBounds:
diff --git a/src/mongo/db/exec/sbe/values/value_printer.cpp b/src/mongo/db/exec/sbe/values/value_printer.cpp
index a04d1407930..090735671e5 100644
--- a/src/mongo/db/exec/sbe/values/value_printer.cpp
+++ b/src/mongo/db/exec/sbe/values/value_printer.cpp
@@ -27,6 +27,7 @@
  *    it in the license file.
  */
 #include "mongo/db/exec/sbe/values/value_printer.h"
+#include "mongo/db/exec/sbe/values/makeobj_spec.h"
 #include "mongo/db/exec/sbe/values/sort_spec.h"
 #include "mongo/db/exec/sbe/values/value.h"
 #include "mongo/platform/basic.h"
@@ -154,6 +155,9 @@ void ValuePrinter<T>::writeTagToStream(TypeTags tag) {
         case TypeTags::sortSpec:
             stream << "sortSpec";
             break;
+        case TypeTags::makeObjSpec:
+            stream << "makeObjSpec";
+            break;
         case TypeTags::indexBounds:
             stream << "indexBounds";
             break;
@@ -454,7 +458,7 @@ void ValuePrinter<T>::writeValueToStream(TypeTags tag, Value val, size_t depth)
             stream << ')';
             break;
         }
-        case value::TypeTags::ftsMatcher: {
+        case TypeTags::ftsMatcher: {
             auto ftsMatcher = getFtsMatcherView(val);
             stream << "FtsMatcher(";
             writeObjectToStream(ftsMatcher->query().toBSON());
@@ -466,6 +470,9 @@ void ValuePrinter<T>::writeValueToStream(TypeTags tag, Value val, size_t depth)
             writeObjectToStream(getSortSpecView(val)->getPattern());
             stream << ')';
             break;
+        case TypeTags::makeObjSpec:
+            stream << "MakeObjSpec(" << getMakeObjSpecView(val)->toString() << ")";
+            break;
         case TypeTags::indexBounds:
             // When calling toString() we don't know if the index has a non-simple collation or
             // not. Passing false could produce invalid UTF-8, which is not acceptable when we are
diff --git a/src/mongo/db/exec/sbe/vm/vm.cpp b/src/mongo/db/exec/sbe/vm/vm.cpp
index bb4125471ca..76d3bf4af80 100644
--- a/src/mongo/db/exec/sbe/vm/vm.cpp
+++ b/src/mongo/db/exec/sbe/vm/vm.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/exec/sbe/accumulator_sum_value_enum.h"
 #include "mongo/db/exec/sbe/values/arith_common.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/makeobj_spec.h"
 #include "mongo/db/exec/sbe/values/sbe_pattern_value_cmp.h"
 #include "mongo/db/exec/sbe/values/sort_spec.h"
 #include "mongo/db/exec/sbe/values/value.h"
@@ -4167,6 +4168,32 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinGenerateSortKey
                 new KeyString::Value(ss->generateSortKey(obj, collator)))};
 }
 
+std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinMakeBsonObj(ArityType arity) {
+    invariant(arity >= 2);
+    tassert(6897002,
+            str::stream() << "Unsupported number of arguments passed to makeBsonObj(): " << arity,
+            arity >= 2);
+
+    auto [mosOwned, mosTag, mosVal] = getFromStack(0);
+    auto [objOwned, objTag, objVal] = getFromStack(1);
+
+    if (mosTag != value::TypeTags::makeObjSpec) {
+        return {false, value::TypeTags::Nothing, 0};
+    }
+
+    auto mos = value::getMakeObjSpecView(mosVal);
+
+    // For produceBsonObject()'s 'getArg' parameter, we pass in a lambda that will retrieve the
+    // value of each projected field from the VM stack.
+    auto [tag, val] = mos->produceBsonObject(
+        objTag, objVal, [&](size_t idx) -> std::pair<value::TypeTags, value::Value> {
+            auto [_, tag, val] = getFromStack(2 + idx);
+            return {tag, val};
+        });
+
+    return {true, tag, val};
+}
+
 std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinReverseArray(ArityType arity) {
     invariant(arity == 1);
     auto [inputOwned, inputType, inputVal] = getFromStack(0);
@@ -4637,6 +4664,8 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::dispatchBuiltin(Builti
             return builtinFtsMatch(arity);
         case Builtin::generateSortKey:
             return builtinGenerateSortKey(arity);
+        case Builtin::makeBsonObj:
+            return builtinMakeBsonObj(arity);
         case Builtin::tsSecond:
             return builtinTsSecond(arity);
         case Builtin::tsIncrement:
diff --git a/src/mongo/db/exec/sbe/vm/vm.h b/src/mongo/db/exec/sbe/vm/vm.h
index 46793fb4f0d..02f7146c42f 100644
--- a/src/mongo/db/exec/sbe/vm/vm.h
+++ b/src/mongo/db/exec/sbe/vm/vm.h
@@ -611,6 +611,7 @@ enum class Builtin : uint8_t {
     hash,
     ftsMatch,
     generateSortKey,
+    makeBsonObj,
     tsSecond,
     tsIncrement,
     typeMatch,
@@ -1208,6 +1209,7 @@ private:
     std::tuple<bool, value::TypeTags, value::Value> builtinHash(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinFtsMatch(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinGenerateSortKey(ArityType arity);
+    std::tuple<bool, value::TypeTags, value::Value> builtinMakeBsonObj(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinTsSecond(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinTsIncrement(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinTypeMatch(ArityType arity);
diff --git a/src/mongo/db/query/sbe_stage_builder_eval_frame.h b/src/mongo/db/query/sbe_stage_builder_eval_frame.h
index 35ffa9e316d..c66f3dba17a 100644
--- a/src/mongo/db/query/sbe_stage_builder_eval_frame.h
+++ b/src/mongo/db/query/sbe_stage_builder_eval_frame.h
@@ -32,6 +32,7 @@
 #include <stack>
 
 #include "mongo/db/exec/sbe/expressions/expression.h"
+#include "mongo/stdx/variant.h"
 
 namespace mongo::stage_builder {
 
@@ -44,57 +45,75 @@ class EvalExpr {
 public:
     EvalExpr() = default;
 
-    EvalExpr(EvalExpr&& e) : _expr(std::move(e._expr)), _slot(e._slot) {
-        e._slot = boost::none;
+    EvalExpr(EvalExpr&& e) : _exprOrSlot(std::move(e._exprOrSlot)) {
+        e.reset();
     }
 
-    EvalExpr(std::unique_ptr<sbe::EExpression>&& e) : _expr(std::move(e)) {}
+    EvalExpr(std::unique_ptr<sbe::EExpression>&& e) : _exprOrSlot(std::move(e)) {}
 
-    EvalExpr(sbe::value::SlotId s) : _expr(sbe::makeE<sbe::EVariable>(s)), _slot(s) {}
+    EvalExpr(sbe::value::SlotId s) : _exprOrSlot(s) {}
 
     EvalExpr& operator=(EvalExpr&& e) {
         if (this == &e) {
             return *this;
         }
 
-        _expr = std::move(e._expr);
-        _slot = e._slot;
-        e._slot = boost::none;
+        _exprOrSlot = std::move(e._exprOrSlot);
+        e.reset();
         return *this;
     }
 
     EvalExpr& operator=(std::unique_ptr<sbe::EExpression>&& e) {
-        _expr = std::move(e);
-        _slot = boost::none;
+        _exprOrSlot = std::move(e);
+        e.reset();
         return *this;
     }
 
     EvalExpr& operator=(sbe::value::SlotId s) {
-        _expr = sbe::makeE<sbe::EVariable>(s);
-        _slot = s;
+        _exprOrSlot = s;
         return *this;
     }
 
+    boost::optional<sbe::value::SlotId> getSlot() const {
+        return hasSlot() ? boost::make_optional(stdx::get<sbe::value::SlotId>(_exprOrSlot))
+                         : boost::none;
+    }
+
+    bool hasSlot() const {
+        return stdx::holds_alternative<sbe::value::SlotId>(_exprOrSlot);
+    }
+
+    EvalExpr clone() const {
+        if (hasSlot()) {
+            return stdx::get<sbe::value::SlotId>(_exprOrSlot);
+        }
+
+        const auto& expr = stdx::get<std::unique_ptr<sbe::EExpression>>(_exprOrSlot);
+
+        tassert(
+            6897007, "Unexpected: clone() method invoked on null EvalExpr", expr.get() != nullptr);
+
+        return expr->clone();
+    }
+
     explicit operator bool() const {
-        return static_cast<bool>(_expr);
+        return hasSlot() || stdx::get<std::unique_ptr<sbe::EExpression>>(_exprOrSlot) != nullptr;
     }
 
     void reset() {
-        _expr.reset();
-        _slot = boost::none;
+        _exprOrSlot = std::unique_ptr<sbe::EExpression>();
     }
 
     std::unique_ptr<sbe::EExpression> extractExpr() {
-        return std::move(_expr);
-    }
+        if (hasSlot()) {
+            return sbe::makeE<sbe::EVariable>(stdx::get<sbe::value::SlotId>(_exprOrSlot));
+        }
 
-    boost::optional<sbe::value::SlotId> getSlot() const {
-        return _slot;
+        return std::move(stdx::get<std::unique_ptr<sbe::EExpression>>(_exprOrSlot));
     }
 
 private:
-    std::unique_ptr<sbe::EExpression> _expr;
-    boost::optional<sbe::value::SlotId> _slot;
+    stdx::variant<std::unique_ptr<sbe::EExpression>, sbe::value::SlotId> _exprOrSlot;
 };
 
 /**
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.cpp b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
index cb61aa76a3d..5a27986d4e5 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
@@ -325,8 +325,8 @@ std::pair<sbe::value::SlotId, EvalStage> projectEvalExpr(
     PlanNodeId planNodeId,
     sbe::value::SlotIdGenerator* slotIdGenerator) {
     // If expr's value is already in a slot, return the slot.
-    if (expr.getSlot()) {
-        return {*expr.getSlot(), stageOrLimitCoScan(std::move(stage), planNodeId)};
+    if (expr.hasSlot()) {
+        return {*expr.getSlot(), std::move(stage)};
     }
 
     // If expr's value is an expression, create a ProjectStage to evaluate the expression
diff --git a/src/mongo/db/query/sbe_stage_builder_projection.cpp b/src/mongo/db/query/sbe_stage_builder_projection.cpp
index e45fa11ddd3..6e3fcff50c0 100644
--- a/src/mongo/db/query/sbe_stage_builder_projection.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_projection.cpp
@@ -42,6 +42,7 @@
 #include "mongo/db/exec/sbe/stages/traverse.h"
 #include "mongo/db/exec/sbe/stages/union.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/makeobj_spec.h"
 #include "mongo/db/matcher/expression_array.h"
 #include "mongo/db/query/sbe_stage_builder_expression.h"
 #include "mongo/db/query/sbe_stage_builder_filter.h"
@@ -74,30 +75,32 @@ enum class EvalMode {
 // be nullptr, in this case 'slot' is assigned in 'evalStage' of the current nested level.
 class ProjectEval {
 public:
-    ProjectEval(EvalMode mode) : _slot{sbe::value::SlotId(0)}, _expr{nullptr}, _mode(mode) {}
+    ProjectEval(EvalMode mode) : _mode(mode) {}
 
-    ProjectEval(sbe::value::SlotId slot, ExpressionType expr)
-        : _slot{slot}, _expr{std::move(expr)}, _mode{EvalMode::EvaluateField} {}
+    ProjectEval(EvalExpr expr) : _expr{std::move(expr)}, _mode{EvalMode::EvaluateField} {}
 
-    sbe::value::SlotId slot() const {
-        return _slot;
+    EvalMode mode() const {
+        return _mode;
     }
 
-    const ExpressionType& expr() const {
-        return _expr;
+    const bool hasSlot() const {
+        return _expr.getSlot().has_value();
     }
 
-    EvalMode mode() const {
-        return _mode;
+    sbe::value::SlotId slot() const {
+        return *_expr.getSlot();
     }
 
     ExpressionType extractExpr() {
+        return _expr.extractExpr();
+    }
+
+    EvalExpr extractEvalExpr() {
         return std::move(_expr);
     }
 
 private:
-    sbe::value::SlotId _slot;
-    ExpressionType _expr;
+    EvalExpr _expr;
     EvalMode _mode;
 };
 
@@ -112,25 +115,39 @@ struct PositionalProjectionData {
 struct ProjectionTraversalVisitorContext {
     // Represents current projection level. Created each time visitor encounters path projection.
     struct NestedLevel {
-        NestedLevel(sbe::value::SlotId inputSlot,
+        NestedLevel(EvalExpr inputExpr,
                     std::list<std::string> fields,
-                    PlanNodeId planNodeId)
-            : inputSlot(inputSlot),
+                    boost::optional<sbe::FrameId> lambdaFrame)
+            : inputExpr(std::move(inputExpr)),
               fields(std::move(fields)),
-              evalStage(makeLimitCoScanStage(planNodeId)) {}
+              lambdaFrame(std::move(lambdaFrame)) {}
+
+        sbe::value::SlotId getInputSlot() const {
+            tassert(6897003, "Expected input EvalExpr to be a slot", inputExpr.hasSlot());
+            return *inputExpr.getSlot();
+        }
+
+        EvalExpr getInputEvalExpr() const {
+            return inputExpr.clone();
+        }
+
+        ExpressionType getInputExpr() const {
+            return getInputEvalExpr().extractExpr();
+        }
 
-        // The input slot for the current level. This is the parent sub-document for each of the
-        // projected fields at the current level.
-        sbe::value::SlotId inputSlot;
+        // The input expression for the current level. This is the parent sub-document for each of
+        // the projected fields at the current level. 'inputExpr' can be a slot or a local variable.
+        EvalExpr inputExpr;
         // The fields names at the current projection level.
         std::list<std::string> fields;
+        // The lambda frame associated with the current level.
+        boost::optional<sbe::FrameId> lambdaFrame;
         // A traversal sub-tree which combines traversals for each of the fields at the current
         // level.
         PlanStageType evalStage;
         // Vector containing expressions for each of the projections at the current level. There is
         // an eval for each of the fields in the current nested level.
         std::vector<ProjectEval> evals;
-
         // Whether or not any subtree of this level has a computed field.
         bool subtreeContainsComputedField = false;
     };
@@ -147,8 +164,12 @@ struct ProjectionTraversalVisitorContext {
         levels.top().fields.pop_front();
     }
 
-    bool isLastLevel() {
-        return levels.size() == 1;
+    size_t numLevels() const {
+        return levels.size();
+    }
+
+    bool isLastLevel() const {
+        return numLevels() == 1;
     }
 
     auto& topLevel() {
@@ -166,9 +187,10 @@ struct ProjectionTraversalVisitorContext {
         levels.pop();
     }
 
-    void pushLevel(std::list<std::string> fields) {
-        levels.push(
-            {levels.size() <= 1 ? inputSlot : state.slotId(), std::move(fields), planNodeId});
+    void pushLevel(std::list<std::string> fields,
+                   EvalExpr expr,
+                   boost::optional<sbe::FrameId> lambdaFrame = boost::none) {
+        levels.push({std::move(expr), std::move(fields), lambdaFrame});
     }
 
     std::pair<sbe::value::SlotId, PlanStageType> done() {
@@ -176,8 +198,13 @@ struct ProjectionTraversalVisitorContext {
         auto& evals = topLevelEvals();
         invariant(evals.size() == 1);
         auto& eval = evals[0];
-        invariant(eval.mode() == EvalMode::EvaluateField && !eval.expr());
-        return {eval.slot(), std::move(topLevel().evalStage)};
+
+        invariant(eval.mode() == EvalMode::EvaluateField);
+
+        return projectEvalExpr(eval.extractEvalExpr(),
+                               std::move(topLevel().evalStage),
+                               planNodeId,
+                               state.slotIdGenerator);
     }
 
     ProjectionTraversalVisitorContext(StageBuilderState& state,
@@ -185,13 +212,15 @@ struct ProjectionTraversalVisitorContext {
                                       projection_ast::ProjectType projectType,
                                       PlanStageType inputStage,
                                       sbe::value::SlotId inputSlot,
-                                      sbe::value::SlotId preImageSlot)
+                                      sbe::value::SlotId preImageSlot,
+                                      bool isBasicProjection)
         : state(state),
           planNodeId(planNodeId),
           projectType(projectType),
           inputSlot(inputSlot),
-          preImageSlot(preImageSlot) {
-        pushLevel({});
+          preImageSlot(preImageSlot),
+          isBasicProjection(isBasicProjection) {
+        levels.push({inputSlot, {}, boost::none});
         topLevel().evalStage = std::move(inputStage);
     }
 
@@ -208,6 +237,8 @@ struct ProjectionTraversalVisitorContext {
 
     std::stack<NestedLevel> levels;
 
+    bool isBasicProjection = false;
+
     // Flag indicating if $slice operator is used in the projection.
     bool hasSliceProjection = false;
 
@@ -228,7 +259,17 @@ public:
     }
 
     void visit(const projection_ast::ProjectionPathASTNode* node) final {
-        _context->pushLevel({node->fieldNames().begin(), node->fieldNames().end()});
+        auto isBasicProj = _context->isBasicProjection;
+        auto lambdaFrame =
+            isBasicProj ? boost::make_optional(_context->state.frameId()) : boost::none;
+
+        auto expr = isBasicProj
+            ? EvalExpr{makeVariable(*lambdaFrame, 0)}
+            : EvalExpr{_context->numLevels() <= 1 ? _context->inputSlot : _context->state.slotId()};
+
+        _context->pushLevel(
+            {node->fieldNames().begin(), node->fieldNames().end()}, std::move(expr), lambdaFrame);
+
         _context->currentFieldPath.push_back(_context->topFrontField());
     }
 
@@ -285,67 +326,47 @@ private:
 namespace {
 using FieldVector = std::vector<std::string>;
 
-std::tuple<sbe::value::SlotVector, FieldVector, FieldVector, FieldVector, PlanStageType>
-prepareFieldEvals(ProjectionTraversalVisitorContext* context,
-                  const projection_ast::ProjectionPathASTNode* node) {
+std::tuple<FieldVector, FieldVector, FieldVector, std::vector<EvalExpr>> prepareFieldEvals(
+    const FieldVector& fieldNames, std::vector<ProjectEval>& evals) {
     // Ensure that there is eval for each of the field names.
-    auto& evals = context->topLevelEvals();
-    const auto& fieldNames = node->fieldNames();
     invariant(evals.size() == fieldNames.size());
 
+    FieldVector keepFields;
+    FieldVector restrictFields;
+    FieldVector projectFields;
+    std::vector<EvalExpr> projectExprs;
+
     // Walk through all the fields at the current nested level and,
-    //    * For exclusion projections populate the 'restrictFields' array to be passed to the
+    //    * For exclusion projections, populate the 'restrictFields' array to be passed to the
     //      mkobj stage, which constructs an output document for the current nested level.
-    //    * For inclusion projections,
-    //         - Populates 'projectFields' and 'projectSlots' vectors holding field names to
-    //           project, and slots to access evaluated projection values.
-    //         - Populates 'projects' map to actually project out the values.
-    sbe::value::SlotMap<std::unique_ptr<sbe::EExpression>> projects;
-    sbe::value::SlotVector projectSlots;
-    std::vector<std::string> projectFields;
-    std::vector<std::string> restrictFields;
-    std::vector<std::string> keepFields;
+    //    * For inclusion projections, populate the 'keepFields' array to be passed to the
+    //      mkobj stage, and also populate the 'projectFields' and 'projectExprs' vectors with
+    //      the field names and the projection values (represented as EvalExprs).
     for (size_t i = 0; i < fieldNames.size(); i++) {
         auto& fieldName = fieldNames[i];
         auto& eval = evals[i];
 
         switch (eval.mode()) {
-            case EvalMode::IgnoreField:
-                // Nothing to do with this field.
-                break;
             case EvalMode::RestrictField:
-                // This is an exclusion projection and we need put the field name to the vector of
-                // restricted fields.
                 restrictFields.push_back(fieldName);
                 break;
             case EvalMode::KeepField:
                 keepFields.push_back(fieldName);
                 break;
+            case EvalMode::IgnoreField:
+                break;
             case EvalMode::EvaluateField: {
-                // We need to evaluate value and add a field with it in the resulting object.
-                projectSlots.push_back(eval.slot());
                 projectFields.push_back(fieldName);
-
-                if (eval.expr()) {
-                    projects.emplace(eval.slot(), eval.extractExpr());
-                }
+                projectExprs.emplace_back(eval.extractEvalExpr());
                 break;
             }
         }
     }
 
-    auto evalStage{std::move(context->topLevel().evalStage)};
-
-    // If we have something to actually project, then inject a projection stage.
-    if (!projects.empty()) {
-        evalStage = makeProject(std::move(evalStage), std::move(projects), context->planNodeId);
-    }
-
-    return {std::move(projectSlots),
-            std::move(projectFields),
+    return {std::move(keepFields),
             std::move(restrictFields),
-            std::move(keepFields),
-            std::move(evalStage)};
+            std::move(projectFields),
+            std::move(projectExprs)};
 }
 
 }  // namespace
@@ -380,15 +401,7 @@ public:
                                                 _context->inputSlot,
                                                 _context->planNodeId);
 
-        if (auto slot = expr.getSlot(); slot) {
-            // If the expression is already bound to a slot, just use that slot
-            _context->topLevelEvals().emplace_back(*slot, nullptr);
-        } else {
-            // If the expression is not bound to a slot yet, allocate a new slot and push the
-            // slot/expr pair into topLevelEvals().
-            _context->topLevelEvals().emplace_back(_context->state.slotId(), expr.extractExpr());
-        }
-
+        _context->topLevelEvals().emplace_back(std::move(expr));
         _context->topLevel().evalStage = std::move(stage);
     }
 
@@ -400,93 +413,134 @@ public:
         _context->currentFieldPath.pop_back();
         invariant(_context->topLevel().fields.empty());
 
-        auto [projectSlots, projectFields, restrictFields, keepFields, childLevelStage] =
-            prepareFieldEvals(_context, node);
+        auto [keepFields, dropFields, projectFields, projectExprs] =
+            prepareFieldEvals(node->fieldNames(), _context->topLevelEvals());
+
+        // Generate a document for the current nested level.
+        const bool isInclusion = _context->projectType == projection_ast::ProjectType::kInclusion;
+
+        auto [fieldBehavior, fieldVector] = isInclusion
+            ? std::make_pair(sbe::value::MakeObjSpec::FieldBehavior::keep, std::move(keepFields))
+            : std::make_pair(sbe::value::MakeObjSpec::FieldBehavior::drop, std::move(dropFields));
+
+        auto childLevelStage = std::move(_context->topLevel().evalStage);
+        tassert(6897004,
+                "Expected inner input stage to be null",
+                !_context->isBasicProjection || childLevelStage.stage == nullptr);
+
+        auto lambdaFrame = _context->topLevel().lambdaFrame;
+        tassert(6897005,
+                "Expected lambda frame to be set",
+                !_context->isBasicProjection || lambdaFrame);
+
+        auto [childInputSlot, childInputExpr] = [&] {
+            auto evalExpr = _context->topLevel().getInputEvalExpr();
+            auto slot = evalExpr.getSlot();
+            tassert(6897006,
+                    "Expected input EvalExpr to be a slot",
+                    _context->isBasicProjection || slot);
+            return std::make_pair(slot, evalExpr.extractExpr());
+        }();
+
+        const bool containsComputedField = _context->topLevel().subtreeContainsComputedField;
 
-        // Finally, inject an mkobj stage to generate a document for the current nested level. For
-        // inclusion projection also add a filter stage on top to filter out input values for
-        // nested traversal if they don't result in documents.
-        auto childLevelInputSlot = _context->topLevel().inputSlot;
-        auto childLevelResultSlot = _context->state.slotId();
-        if (_context->projectType == projection_ast::ProjectType::kInclusion) {
-            auto mkBsonStage = makeMkBsonObj(std::move(childLevelStage),
-                                             childLevelResultSlot,
-                                             childLevelInputSlot,
-                                             sbe::MakeBsonObjStage::FieldBehavior::keep,
-                                             std::move(keepFields),
-                                             std::move(projectFields),
-                                             std::move(projectSlots),
-                                             true,
-                                             false,
-                                             _context->planNodeId);
-
-            if (_context->topLevel().subtreeContainsComputedField) {
-                // Projections of computed fields should always be applied to elements of an array,
-                // even if the elements aren't objects. For example:
-                // projection: {a: {b: "x"}}
-                // document: {a: [1,2,3]}
-                // result: {a: [{b: "x"}, {b: "x"}, {b: "x"}, {b: "x"}]}
-
-                childLevelStage = std::move(mkBsonStage);
-            } else {
-                // There are no computed fields, only inclusions. So anything that's not a document
-                // will get projected out. Example:
-                // projection: {a: {b: 1}}
-                // document: {a: [1, {b: 2}, 3]}
-                // result: {a: [{b: 2}]}
-
-                childLevelStage =
-                    makeFilter<true>(std::move(mkBsonStage),
-                                     makeFunction("isObject"_sd, makeVariable(childLevelInputSlot)),
-                                     _context->planNodeId);
-            }
+        // We've finished extracting what we need from the child level, so pop if off the stack.
+        _context->popLevel();
 
-        } else {
-            childLevelStage = makeMkBsonObj(std::move(childLevelStage),
-                                            childLevelResultSlot,
-                                            childLevelInputSlot,
-                                            sbe::MakeBsonObjStage::FieldBehavior::drop,
-                                            std::move(restrictFields),
-                                            std::move(projectFields),
-                                            std::move(projectSlots),
-                                            false,
-                                            true,
-                                            _context->planNodeId);
+        // If the child's 'subtreeContainsComputedField' flag was set, then propagate it to the
+        // parent level.
+        _context->topLevel().subtreeContainsComputedField =
+            _context->topLevel().subtreeContainsComputedField || containsComputedField;
+
+        // Create a makeBsonObj() expression to generate the document for the current nested level.
+        auto makeObjSpecExpr = makeConstant(
+            sbe::value::TypeTags::makeObjSpec,
+            sbe::value::bitcastFrom<sbe::value::MakeObjSpec*>(new sbe::value::MakeObjSpec(
+                fieldBehavior, std::move(fieldVector), std::move(projectFields))));
+
+        auto args = sbe::makeEs(std::move(makeObjSpecExpr), childInputExpr->clone());
+        for (auto& expr : projectExprs) {
+            args.push_back(expr.extractExpr());
         }
 
-        // We are done with the child level. Now we need to extract corresponding field from parent
-        // level, traverse it and assign value to 'childLevelInputSlot'.
-        {
-            const bool containsComputedField = _context->topLevel().subtreeContainsComputedField;
-            _context->popLevel();
-            _context->topLevel().subtreeContainsComputedField =
-                _context->topLevel().subtreeContainsComputedField || containsComputedField;
+        auto innerExpr = sbe::makeE<sbe::EFunction>("makeBsonObj", std::move(args));
+
+        if (!isInclusion || !containsComputedField) {
+            // If this is an inclusion projection and with no computed fields, then anything that's
+            // not an object should get filtered out. Example:
+            // projection: {a: {b: 1}}
+            // document: {a: [1, {b: 2}, 3]}
+            // result: {a: [{b: 2}]}
+            //
+            // If this is an inclusion projection with 1 or more computed fields, then projections
+            // of computed fields should always be applied even if the values aren't objects.
+            // Example:
+            // projection: {a: {b: "x"}}
+            // document: {a: [1,2,3]}
+            // result: {a: [{b: "x"}, {b: "x"}, {b: "x"}, {b: "x"}]}
+            //
+            // If this is an exclusion projection, then anything that is not an object should be
+            // preserved as-is.
+            innerExpr = sbe::makeE<sbe::EIf>(makeFunction("isObject", childInputExpr->clone()),
+                                             std::move(innerExpr),
+                                             isInclusion && !containsComputedField
+                                                 ? makeConstant(sbe::value::TypeTags::Nothing, 0)
+                                                 : childInputExpr->clone());
         }
 
-        auto parentLevelInputSlot = _context->topLevel().inputSlot;
-        auto parentLevelStage{std::move(_context->topLevel().evalStage)};
-        if (!_context->isLastLevel()) {
-            parentLevelStage = makeProject(std::move(parentLevelStage),
-                                           _context->planNodeId,
-                                           childLevelInputSlot,
-                                           makeFunction("getField"_sd,
-                                                        makeVariable(parentLevelInputSlot),
-                                                        makeConstant(_context->topFrontField())));
+        auto parentInputExpr = _context->topLevel().getInputExpr();
+        auto fromExpr = _context->isLastLevel()
+            ? std::move(parentInputExpr)
+            : makeFunction("getField"_sd,
+                           std::move(parentInputExpr),
+                           makeConstant(_context->topFrontField()));
+
+        if (_context->isBasicProjection) {
+            // If this is a basic projection, we can make use of traverseP().
+            auto traversePExpr =
+                makeFunction("traverseP",
+                             std::move(fromExpr),
+                             sbe::makeE<sbe::ELocalLambda>(*lambdaFrame, std::move(innerExpr)),
+                             makeConstant(sbe::value::TypeTags::Nothing, 0));
+
+            _context->topLevelEvals().emplace_back(std::move(traversePExpr));
+
+            return;
         }
 
-        auto parentLevelResultSlot = _context->state.slotId();
-        parentLevelStage = makeTraverse(std::move(parentLevelStage),
-                                        std::move(childLevelStage),
-                                        childLevelInputSlot,
-                                        parentLevelResultSlot,
-                                        childLevelResultSlot,
-                                        nullptr,
-                                        nullptr,
-                                        _context->planNodeId,
-                                        boost::none);
+        // For non-basic projections, childLevelStage may be non-null, so we have to use
+        // TraverseStage.
+        auto outputSlot = _context->state.slotId();
+        auto innerResultSlot = _context->state.slotId();
 
-        _context->topLevel().evalStage = std::move(parentLevelStage);
-        _context->topLevelEvals().emplace_back(parentLevelResultSlot, nullptr);
+        auto parentLevelStage = std::move(_context->topLevel().evalStage);
+
+        // Assign 'fromExpr' to childInputSlot. We will use childInputSlot as the input
+        // to TraverseStage, and the TraverseStage will feed childInputSlot (or the array
+        // elements from childInputSlot if it's an array) as input to innerBranch which
+        // evaluates the call to mkBson().
+        auto fromBranch = makeProject(std::move(parentLevelStage),
+                                      _context->planNodeId,
+                                      *childInputSlot,
+                                      std::move(fromExpr));
+
+        auto innerBranch = makeProject(std::move(childLevelStage),
+                                       _context->planNodeId,
+                                       innerResultSlot,
+                                       std::move(innerExpr));
+
+        auto traverseStage = makeTraverse(std::move(fromBranch),
+                                          std::move(innerBranch),
+                                          *childInputSlot,
+                                          outputSlot,
+                                          innerResultSlot,
+                                          nullptr,
+                                          nullptr,
+                                          _context->planNodeId,
+                                          boost::none);
+
+        _context->topLevelEvals().emplace_back(outputSlot);
+        _context->topLevel().evalStage = std::move(traverseStage);
     }
 
     void visit(const projection_ast::ProjectionPositionalASTNode* node) final {
@@ -633,13 +687,11 @@ public:
         inBranch = makeFilter<true>(
             std::move(inBranch), makeVariable(traversingAnArrayFlagSlot), _context->planNodeId);
 
-        auto inputDocumentSlot = _context->topLevel().inputSlot;
-        sbe::EVariable inputDocumentVariable{inputDocumentSlot};
         auto fromBranch = makeProject(std::move(_context->topLevel().evalStage),
                                       _context->planNodeId,
                                       inputArraySlot,
                                       makeFunction("getField"_sd,
-                                                   inputDocumentVariable.clone(),
+                                                   _context->topLevel().getInputExpr(),
                                                    makeConstant(_context->topFrontField())));
 
         fromBranch = makeProject(std::move(fromBranch),
@@ -670,8 +722,8 @@ public:
                                  makeConstant(sbe::value::TypeTags::Nothing, 0),
                                  makeVariable(filteredArraySlot)));
 
+        _context->topLevelEvals().emplace_back(resultSlot);
         _context->topLevel().evalStage = std::move(resultStage);
-        _context->topLevelEvals().emplace_back(resultSlot, nullptr);
     }
 
     void visit(const projection_ast::MatchExpressionASTNode* node) final {}
@@ -699,11 +751,35 @@ public:
 
         // All field paths without $slice operator are marked using 'EvalMode::IgnoreField' (see
         // other methods of this visitor). This makes 'prepareFieldEvals' function to populate
-        // 'projectSlots' and 'projectFields' only with evals for $slice operators if there are
-        // any.  We do not remove any fields in the plan generated by this visitor, so the
+        // 'projectFields' and 'projectExprs' only with evals for $slice operators if there are
+        // any. We do not remove any fields in the plan generated by this visitor, so the
         // 'restrictFields' and 'keepFields' return values are not used.
-        auto [projectSlots, projectFields, restrictFields, keepFields, childLevelStage] =
-            prepareFieldEvals(_context, node);
+        auto [keepFields, restrictFields, projectFields, projectExprs] =
+            prepareFieldEvals(node->fieldNames(), _context->topLevelEvals());
+
+        auto childLevelStage{std::move(_context->topLevel().evalStage)};
+
+        sbe::value::SlotMap<ExpressionType> projectsMap;
+        sbe::value::SlotVector projectSlots;
+
+        for (auto&& expr : projectExprs) {
+            sbe::value::SlotId slot;
+            if (!expr.hasSlot()) {
+                slot = _context->state.slotId();
+                projectsMap.emplace(slot, expr.extractExpr());
+            } else {
+                slot = *expr.getSlot();
+            }
+
+            projectSlots.push_back(slot);
+        }
+
+        // If we have something to actually project, then inject a projection stage.
+        if (!projectsMap.empty()) {
+            childLevelStage = makeProject(
+                std::move(childLevelStage), std::move(projectsMap), _context->planNodeId);
+        }
+
         invariant(restrictFields.empty());
         invariant(keepFields.empty());
 
@@ -747,7 +823,7 @@ public:
         //
         // Construct mkobj stage which adds fields evaluating $slice operator ('projectFields' and
         // 'projectSlots') to the already constructed object from all previous operators.
-        auto childLevelInputSlot = _context->topLevel().inputSlot;
+        auto childLevelInputSlot = _context->topLevel().getInputSlot();
         auto childLevelObjSlot = _context->state.slotId();
         childLevelStage = makeMkBsonObj(std::move(childLevelStage),
                                         childLevelObjSlot,
@@ -775,7 +851,7 @@ public:
         // level, traverse it and assign value to 'childLevelInputSlot'.
         _context->popLevel();
 
-        auto parentLevelInputSlot = _context->topLevel().inputSlot;
+        auto parentLevelInputSlot = _context->topLevel().getInputSlot();
         auto parentLevelStage{std::move(_context->topLevel().evalStage)};
         if (!_context->isLastLevel()) {
             // Extract value of the current field from the object in 'parentLevelInputSlot'.
@@ -804,8 +880,8 @@ public:
                                         _context->planNodeId,
                                         1 /* nestedArraysDepth */);
 
+        _context->topLevelEvals().emplace_back(parentLevelResultSlot);
         _context->topLevel().evalStage = std::move(parentLevelStage);
-        _context->topLevelEvals().emplace_back(parentLevelResultSlot, nullptr);
     }
 
     void visit(const projection_ast::ProjectionPositionalASTNode* node) final {
@@ -818,10 +894,9 @@ public:
     void visit(const projection_ast::ProjectionSliceASTNode* node) final {
         using namespace std::literals;
 
-        auto arrayFromField =
-            makeFunction("getField"_sd,
-                         sbe::makeE<sbe::EVariable>(_context->topLevel().inputSlot),
-                         makeConstant(_context->topFrontField()));
+        auto arrayFromField = makeFunction("getField"_sd,
+                                           _context->topLevel().getInputExpr(),
+                                           makeConstant(_context->topFrontField()));
         auto binds = sbe::makeEs(std::move(arrayFromField));
         auto frameId = _context->state.frameId();
         sbe::EVariable arrayVariable{frameId, 0};
@@ -841,7 +916,7 @@ public:
         auto sliceExpr =
             sbe::makeE<sbe::ELocalBind>(frameId, std::move(binds), std::move(extractSubArrayExpr));
 
-        _context->topLevelEvals().emplace_back(_context->state.slotId(), std::move(sliceExpr));
+        _context->topLevelEvals().emplace_back(std::move(sliceExpr));
     }
 
     void visit(const projection_ast::ProjectionElemMatchASTNode* node) final {
@@ -1138,6 +1213,25 @@ std::pair<sbe::value::SlotId, PlanStageType> generatePositionalProjection(
 
     return {nextFieldResultSlot, std::move(resultStage)};
 }
+
+bool projectionIsBasic(const projection_ast::ASTNode* node) {
+    if (!node)
+        return true;
+
+    auto path = dynamic_cast<const projection_ast::ProjectionPathASTNode*>(node);
+    auto boolean = dynamic_cast<const projection_ast::BooleanConstantASTNode*>(node);
+
+    if (!path && !boolean) {
+        return false;
+    }
+
+    for (size_t i = 0; i < node->children().size(); ++i) {
+        if (!projectionIsBasic(node->child(i))) {
+            return false;
+        }
+    }
+    return true;
+}
 }  // namespace
 
 std::pair<sbe::value::SlotId, EvalStage> generateProjection(
@@ -1146,8 +1240,11 @@ std::pair<sbe::value::SlotId, EvalStage> generateProjection(
     EvalStage stage,
     sbe::value::SlotId inputVar,
     PlanNodeId planNodeId) {
+    auto projType = projection->type();
+    bool isBasicProj = projectionIsBasic(projection->root());
+
     ProjectionTraversalVisitorContext context{
-        state, planNodeId, projection->type(), std::move(stage), inputVar, inputVar};
+        state, planNodeId, projType, std::move(stage), inputVar, inputVar, isBasicProj};
     ProjectionTraversalPreVisitor preVisitor{&context};
     ProjectionTraversalInVisitor inVisitor{&context};
     ProjectionTraversalPostVisitor postVisitor{&context};
@@ -1162,7 +1259,7 @@ std::pair<sbe::value::SlotId, EvalStage> generateProjection(
         // of it for $slice operator. This second tree modifies resulting objects from from other
         // operators to include fields with $slice operator.
         ProjectionTraversalVisitorContext sliceContext{
-            state, planNodeId, projection->type(), std::move(resultStage), resultSlot, inputVar};
+            state, planNodeId, projType, std::move(resultStage), resultSlot, inputVar, isBasicProj};
         ProjectionTraversalPreVisitor slicePreVisitor{&sliceContext};
         ProjectionTraversalInVisitor sliceInVisitor{&sliceContext};
         SliceProjectionTraversalPostVisitor slicePostVisitor{&sliceContext};