SERVER-42988 allow agg-style object syntax in find() projection

18 files changed, 893 insertions, 321 deletions

diff --git a/jstests/aggregation/optimize_away_pipeline.js b/jstests/aggregation/optimize_away_pipeline.js
index d3634837ef0..f4c2c0ac483 100644
--- a/jstests/aggregation/optimize_away_pipeline.js
+++ b/jstests/aggregation/optimize_away_pipeline.js
@@ -478,12 +478,11 @@ if (!FixtureHelpers.isMongos(db) && isWiredTiger(db)) {
         expectedResult: [{_id: 1, x: 10}]
     });
     db.setProfilingLevel(0);
-    let profile = db.system.profile.find({}, {op: 1, ns: 1, comment: 'optimize_away_pipeline'})
-                      .sort({ts: 1})
-                      .toArray();
-    assert(arrayEq(
-        profile,
-        [{op: "command", ns: coll.getFullName()}, {op: "getmore", ns: coll.getFullName()}]));
+    let profile = db.system.profile.find({}, {op: 1, ns: 1}).sort({ts: 1}).toArray();
+    assert(
+        arrayEq(profile,
+                [{op: "command", ns: coll.getFullName()}, {op: "getmore", ns: coll.getFullName()}]),
+        profile);
     // Test getMore puts a correct namespace into profile data for a view with an optimized away
     // pipeline.
     if (!FixtureHelpers.isSharded(coll)) {
@@ -499,9 +498,7 @@ if (!FixtureHelpers.isMongos(db) && isWiredTiger(db)) {
             expectedResult: [{_id: 1, x: 10}]
         });
         db.setProfilingLevel(0);
-        profile = db.system.profile.find({}, {op: 1, ns: 1, comment: 'optimize_away_pipeline'})
-                      .sort({ts: 1})
-                      .toArray();
+        profile = db.system.profile.find({}, {op: 1, ns: 1}).sort({ts: 1}).toArray();
         assert(arrayEq(
             profile,
             [{op: "query", ns: view.getFullName()}, {op: "getmore", ns: view.getFullName()}]));
diff --git a/src/mongo/db/exec/projection_executor_test.cpp b/src/mongo/db/exec/projection_executor_test.cpp
index bea4dbbc710..7d3b7410205 100644
--- a/src/mongo/db/exec/projection_executor_test.cpp
+++ b/src/mongo/db/exec/projection_executor_test.cpp
@@ -40,15 +40,30 @@ namespace mongo::projection_executor {
 class ProjectionExecutorTest : public AggregationContextFixture {
 public:
     projection_ast::Projection parseWithDefaultPolicies(
-        const BSONObj& proj, boost::optional<BSONObj> match = boost::none) {
-        StatusWith<std::unique_ptr<MatchExpression>> expr{nullptr};
-        if (match) {
-            expr = MatchExpressionParser::parse(*match, getExpCtx());
-            uassertStatusOK(expr.getStatus());
+        const BSONObj& projectionBson, boost::optional<BSONObj> matchExprBson = boost::none) {
+        return parseWithPolicies(projectionBson, matchExprBson, ProjectionPolicies{});
+    }
+
+    projection_ast::Projection parseWithFindFeaturesEnabled(
+        const BSONObj& projectionBson, boost::optional<BSONObj> matchExprBson = boost::none) {
+        auto policy = ProjectionPolicies::findProjectionPolicies();
+        return parseWithPolicies(projectionBson, matchExprBson, policy);
+    }
+
+    projection_ast::Projection parseWithPolicies(const BSONObj& projectionBson,
+                                                 boost::optional<BSONObj> matchExprBson,
+                                                 ProjectionPolicies policies) {
+        StatusWith<std::unique_ptr<MatchExpression>> swMatchExpression(nullptr);
+        if (matchExprBson) {
+            swMatchExpression = MatchExpressionParser::parse(*matchExprBson, getExpCtx());
+            uassertStatusOK(swMatchExpression.getStatus());
         }
 
-        return projection_ast::parse(
-            getExpCtx(), proj, expr.getValue().get(), match.get_value_or({}), {});
+        return projection_ast::parse(getExpCtx(),
+                                     projectionBson,
+                                     swMatchExpression.getValue().get(),
+                                     matchExprBson.get_value_or(BSONObj()),
+                                     policies);
     }
 };
 
@@ -114,14 +129,15 @@ TEST_F(ProjectionExecutorTest, CanProjectExclusionDottedPath) {
 }
 
 TEST_F(ProjectionExecutorTest, CanProjectFindPositional) {
-    auto proj = parseWithDefaultPolicies(fromjson("{'a.b.$': 1}"), fromjson("{'a.b': {$gte: 3}}"));
+    auto proj =
+        parseWithFindFeaturesEnabled(fromjson("{'a.b.$': 1}"), fromjson("{'a.b': {$gte: 3}}"));
     auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
     ASSERT_DOCUMENT_EQ(Document{fromjson("{a: {b: [3]}}")},
                        executor->applyTransformation(Document{fromjson("{a: {b: [1,2,3,4]}}")}));
 }
 
 TEST_F(ProjectionExecutorTest, CanProjectFindElemMatchWithInclusion) {
-    auto proj = parseWithDefaultPolicies(fromjson("{a: {$elemMatch: {b: {$gte: 3}}}, c: 1}"));
+    auto proj = parseWithFindFeaturesEnabled(fromjson("{a: {$elemMatch: {b: {$gte: 3}}}, c: 1}"));
     auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
     ASSERT_DOCUMENT_EQ(
         Document{fromjson("{a: [{b: 3}]}")},
@@ -129,7 +145,7 @@ TEST_F(ProjectionExecutorTest, CanProjectFindElemMatchWithInclusion) {
 }
 
 TEST_F(ProjectionExecutorTest, CanProjectFindElemMatchWithExclusion) {
-    auto proj = parseWithDefaultPolicies(fromjson("{a: {$elemMatch: {b: {$gte: 3}}}, c: 0}"));
+    auto proj = parseWithFindFeaturesEnabled(fromjson("{a: {$elemMatch: {b: {$gte: 3}}}, c: 0}"));
     auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
     ASSERT_DOCUMENT_EQ(Document{fromjson("{a: [{b: 3}], d: 'def'}")},
                        executor->applyTransformation(Document{
@@ -137,7 +153,7 @@ TEST_F(ProjectionExecutorTest, CanProjectFindElemMatchWithExclusion) {
 }
 
 TEST_F(ProjectionExecutorTest, CanProjectFindSliceWithInclusion) {
-    auto proj = parseWithDefaultPolicies(fromjson("{'a.b': {$slice: [1,2]}, c: 1}"));
+    auto proj = parseWithFindFeaturesEnabled(fromjson("{'a.b': {$slice: [1,2]}, c: 1}"));
     auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
     ASSERT_DOCUMENT_EQ(
         Document{fromjson("{a: {b: [2,3]}, c: 'abc'}")},
@@ -145,7 +161,7 @@ TEST_F(ProjectionExecutorTest, CanProjectFindSliceWithInclusion) {
 }
 
 TEST_F(ProjectionExecutorTest, CanProjectFindSliceWithExclusion) {
-    auto proj = parseWithDefaultPolicies(fromjson("{'a.b': {$slice: [1,2]}, c: 0}"));
+    auto proj = parseWithFindFeaturesEnabled(fromjson("{'a.b': {$slice: [1,2]}, c: 0}"));
     auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
     ASSERT_DOCUMENT_EQ(
         Document{fromjson("{a: {b: [2,3]}}")},
@@ -153,8 +169,8 @@ TEST_F(ProjectionExecutorTest, CanProjectFindSliceWithExclusion) {
 }
 
 TEST_F(ProjectionExecutorTest, CanProjectFindSliceAndPositional) {
-    auto proj = parseWithDefaultPolicies(fromjson("{'a.b': {$slice: [1,2]}, 'c.$': 1}"),
-                                         fromjson("{c: {$gte: 6}}"));
+    auto proj = parseWithFindFeaturesEnabled(fromjson("{'a.b': {$slice: [1,2]}, 'c.$': 1}"),
+                                             fromjson("{c: {$gte: 6}}"));
     auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
     ASSERT_DOCUMENT_EQ(
         Document{fromjson("{a: {b: [2,3]}, c: [6]}")},
diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index 42afe075abb..916d4808e80 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -383,6 +383,7 @@ env.Library(
         'expression',
         'field_path',
         '$BUILD_DIR/mongo/db/matcher/expressions',
+        '$BUILD_DIR/mongo/db/query/projection_ast',
     ]
 )
 
diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index e69ff21aa5d..19658f2e005 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -191,6 +191,10 @@ intrusive_ptr<Expression> Expression::parseOperand(
     }
 }
 
+bool Expression::isExpressionName(StringData name) {
+    return parserMap.find(name) != parserMap.end();
+}
+
 namespace {
 /**
  * UTF-8 multi-byte code points consist of one leading byte of the form 11xxxxxx, and potentially
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index 7b8953c8bb1..21fff1d3ce9 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -223,6 +223,11 @@ public:
         BSONElement exprElement,
         const VariablesParseState& vps);
 
+    /**
+     * Return whether 'name' refers to an expression in the language.
+     */
+    static bool isExpressionName(StringData name);
+
     /*
       Produce a field path std::string with the field prefix removed.
 
diff --git a/src/mongo/db/pipeline/field_path.cpp b/src/mongo/db/pipeline/field_path.cpp
index 6eb9fe46d0b..52652246b39 100644
--- a/src/mongo/db/pipeline/field_path.cpp
+++ b/src/mongo/db/pipeline/field_path.cpp
@@ -104,4 +104,38 @@ void FieldPath::uassertValidFieldName(StringData fieldName) {
     uassert(
         16412, "FieldPath field names may not contain '.'.", fieldName.find('.') == string::npos);
 }
+
+FieldPath FieldPath::concat(const FieldPath& tail) const {
+    const FieldPath& head = *this;
+
+    std::string concat;
+    const auto expectedStringSize = _fieldPath.size() + 1 + tail._fieldPath.size();
+    concat.reserve(expectedStringSize);
+    concat.insert(concat.begin(), head._fieldPath.begin(), head._fieldPath.end());
+    concat.push_back('.');
+    concat.insert(concat.end(), tail._fieldPath.begin(), tail._fieldPath.end());
+    invariant(concat.size() == expectedStringSize);
+
+    std::vector<size_t> newDots;
+    // Subtract 2 since both contain std::string::npos at the beginning and the entire size at
+    // the end. Add one because we inserted a dot in order to concatenate the two paths.
+    const auto expectedDotSize =
+        head._fieldPathDotPosition.size() + tail._fieldPathDotPosition.size() - 2 + 1;
+    newDots.reserve(expectedDotSize);
+
+    // The first one in head._fieldPathDotPosition is npos. The last one, is, conveniently, the
+    // size of head fieldPath, which also happens to be the index at which we added a new dot.
+    newDots.insert(
+        newDots.begin(), head._fieldPathDotPosition.begin(), head._fieldPathDotPosition.end());
+
+    invariant(tail._fieldPathDotPosition.size() >= 2);
+    for (size_t i = 1; i < tail._fieldPathDotPosition.size(); ++i) {
+        // Move each index back by size of the first field path, plus one, for the newly added dot.
+        newDots.push_back(tail._fieldPathDotPosition[i] + head._fieldPath.size() + 1);
+    }
+    invariant(newDots.back() == concat.size());
+    invariant(newDots.size() == expectedDotSize);
+
+    return FieldPath(std::move(concat), std::move(newDots));
+}
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/field_path.h b/src/mongo/db/pipeline/field_path.h
index f502b2f49c4..afb2a17244c 100644
--- a/src/mongo/db/pipeline/field_path.h
+++ b/src/mongo/db/pipeline/field_path.h
@@ -138,7 +138,12 @@ public:
         return FieldPath(getSubpath(getPathLength() - 2));
     }
 
+    FieldPath concat(const FieldPath& tail) const;
+
 private:
+    FieldPath(std::string string, std::vector<size_t> dots)
+        : _fieldPath(std::move(string)), _fieldPathDotPosition(std::move(dots)) {}
+
     static const char prefix = '$';
 
     // Contains the full field path, with each field delimited by a '.' character.
diff --git a/src/mongo/db/pipeline/field_path_test.cpp b/src/mongo/db/pipeline/field_path_test.cpp
index b869bc38d27..72d58e93be3 100644
--- a/src/mongo/db/pipeline/field_path_test.cpp
+++ b/src/mongo/db/pipeline/field_path_test.cpp
@@ -197,5 +197,34 @@ TEST(FieldPathTest, GetSubpath) {
     ASSERT_EQUALS("foo.bar.baz", path.getSubpath(2));
 }
 
+void checkConcatWorks(const FieldPath& head, const FieldPath& tail) {
+    FieldPath concat = head.concat(tail);
+    ASSERT(concat == FieldPath::getFullyQualifiedPath(head.fullPath(), tail.fullPath()));
+    ASSERT_EQ(concat.getPathLength(), head.getPathLength() + tail.getPathLength());
+
+    const auto expectedTail = head.getPathLength() == 1
+        ? tail
+        : FieldPath::getFullyQualifiedPath(head.tail().fullPath(), tail.fullPath());
+    ASSERT(FieldPath(concat.tail()) == expectedTail);
+
+    ASSERT_EQ(concat.front(), head.front());
+    ASSERT_EQ(concat.back(), tail.back());
+    for (size_t i = 0; i < concat.getPathLength(); i++) {
+        const auto expected = (i < head.getPathLength())
+            ? head.getFieldName(i)
+            : tail.getFieldName(i - head.getPathLength());
+        ASSERT_EQ(concat.getFieldName(i), expected);
+    }
+}
+
+TEST(FieldPathTest, Concat) {
+    checkConcatWorks("abc", "cde");
+    checkConcatWorks("abc.ef", "cde.ab");
+    checkConcatWorks("abc.$id", "cde");
+    checkConcatWorks("abc", "$id.x");
+    checkConcatWorks("some.long.path.with.many.parts", "another.long.ish.path");
+    checkConcatWorks("$db", "$id");
+    checkConcatWorks("$db.$id", "$id.$db");
+}
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/parsed_aggregation_projection.cpp b/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
index 51ecff7684d..561af013548 100644
--- a/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
+++ b/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
@@ -27,6 +27,8 @@
  *    it in the license file.
  */
 
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kQuery
+
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/pipeline/parsed_aggregation_projection.h"
@@ -40,8 +42,10 @@
 #include "mongo/db/pipeline/field_path.h"
 #include "mongo/db/pipeline/parsed_exclusion_projection.h"
 #include "mongo/db/pipeline/parsed_inclusion_projection.h"
+#include "mongo/db/query/projection_parser.h"
 #include "mongo/stdx/unordered_set.h"
 #include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
 #include "mongo/util/str.h"
 
 namespace mongo {
@@ -321,6 +325,19 @@ std::unique_ptr<ParsedAggregationProjection> ParsedAggregationProjection::create
 
     // Actually parse the specification.
     parsedProject->parse(spec);
+
+    // If we got here, it means that parsing using ParsedAggProjection::parse() succeeded.
+    // Since the query system is currently between two implementations of projection, we want to
+    // make sure that the old version and new version agree on which projections are valid. Check
+    // that the new parser also finds this projection valid.
+    try {
+        projection_ast::parse(expCtx, spec, policies);
+    } catch (const DBException& e) {
+        error() << "old parser found projection " << spec << " valid but new parser returned "
+                << e.toStatus();
+        MONGO_UNREACHABLE;
+    }
+
     return parsedProject;
 }
 }  // namespace parsed_aggregation_projection
diff --git a/src/mongo/db/query/SConscript b/src/mongo/db/query/SConscript
index 2ab42d4d3d1..c687aea2b82 100644
--- a/src/mongo/db/query/SConscript
+++ b/src/mongo/db/query/SConscript
@@ -28,9 +28,6 @@ env.Library(
         "planner_wildcard_helpers.cpp",
         "planner_analysis.cpp",
         "planner_ixselect.cpp",
-        "projection.cpp",
-        "projection_ast_util.cpp",
-        "projection_parser.cpp",
         "query_planner.cpp",
         "expression_index.cpp",
         "index_bounds.cpp",
@@ -54,6 +51,7 @@ env.Library(
         "collation/collator_factory_interface",
         "collation/collator_interface",
         "command_request_response",
+        "projection_ast",
         "query_knobs",
     ],
     LIBDEPS_PRIVATE=[
@@ -62,6 +60,19 @@ env.Library(
 )
 
 env.Library(
+    target='projection_ast',
+    source=[
+        "projection.cpp",
+        "projection_ast_util.cpp",
+        "projection_parser.cpp",
+    ],
+    LIBDEPS=[
+        "$BUILD_DIR/mongo/base",
+        "$BUILD_DIR/mongo/db/matcher/expressions",
+    ],
+)
+
+env.Library(
     target='distinct_command_idl',
     source=[
         env.Idlc('distinct_command.idl')[0],
diff --git a/src/mongo/db/query/canonical_query.cpp b/src/mongo/db/query/canonical_query.cpp
index fdf06b7c849..5722bc90f6b 100644
--- a/src/mongo/db/query/canonical_query.cpp
+++ b/src/mongo/db/query/canonical_query.cpp
@@ -242,8 +242,11 @@ Status CanonicalQuery::init(OperationContext* opCtx,
     if (!_qr->getProj().isEmpty()) {
         Status newParserStatus = Status::OK();
         try {
-            _proj.emplace(projection_ast::parse(
-                expCtx, _qr->getProj(), _root.get(), _qr->getFilter(), ProjectionPolicies{}));
+            _proj.emplace(projection_ast::parse(expCtx,
+                                                _qr->getProj(),
+                                                _root.get(),
+                                                _qr->getFilter(),
+                                                ProjectionPolicies::findProjectionPolicies()));
         } catch (const DBException& e) {
             newParserStatus = e.toStatus();
         }
diff --git a/src/mongo/db/query/canonical_query_encoder_test.cpp b/src/mongo/db/query/canonical_query_encoder_test.cpp
index 1d647910ce8..d95f6626613 100644
--- a/src/mongo/db/query/canonical_query_encoder_test.cpp
+++ b/src/mongo/db/query/canonical_query_encoder_test.cpp
@@ -134,7 +134,10 @@ TEST(CanonicalQueryEncoderTest, ComputeKey) {
     testComputeKey("{}", "{}", "{a: 0}", "an");
     testComputeKey("{}", "{}", "{a: false}", "an");
     testComputeKey("{}", "{}", "{a: 99}", "an|_id-a");
-    testComputeKey("{}", "{}", "{a: 'foo'}", "an|_id-a");
+    // TODO SERVER-44118: Since "foo" is a computed field it's treated as an expression and means
+    // the entire document is required. This should not be the case.
+    testComputeKey("{}", "{}", "{a: 'foo'}", "an");
+
     // $slice defaults to exclusion.
     testComputeKey("{}", "{}", "{a: {$slice: [3, 5]}}", "an");
     testComputeKey("{}", "{}", "{a: {$slice: [3, 5]}, b: 0}", "an");
@@ -148,7 +151,9 @@ TEST(CanonicalQueryEncoderTest, ComputeKey) {
 
     testComputeKey("{}", "{}", "{a: {$slice: [3, 5]}, b: {$elemMatch: {x: 2}}}", "an");
 
-    testComputeKey("{}", "{}", "{a: ObjectId('507f191e810c19729de860ea')}", "an|_id-a");
+    // Since ObjectId is a literal value and is treated as an Expression, the entire document is
+    // required. TODO SERVER-44118: The entire document shouldn't be necessary.
+    testComputeKey("{}", "{}", "{a: ObjectId('507f191e810c19729de860ea')}", "an");
     testComputeKey("{a: 1}", "{}", "{'a.$': 1}", "eqa");
     testComputeKey("{a: 1}", "{}", "{a: 1}", "eqa|_id-a");
 
@@ -183,8 +188,9 @@ TEST(CanonicalQueryEncoderTest, ComputeKeyEscaped) {
     // Field name in projection.
     testComputeKey("{}", "{}", "{'a,[]~|-<>': 1}", "an|_id-a\\,\\[\\]\\~\\|\\-<>");
 
-    // Value in projection.
-    testComputeKey("{}", "{}", "{a: 'foo,[]~|-<>'}", "an|_id-a");
+    // String literal provided as value. TODO SERVER-44118: Since this is treated as an Expression,
+    // the entire document is required.
+    testComputeKey("{}", "{}", "{a: 'foo,[]~|-<>'}", "an");
 }
 
 // Cache keys for $geoWithin queries with legacy and GeoJSON coordinates should
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index ced708e90dc..9be58ae42b2 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -652,8 +652,12 @@ StatusWith<unique_ptr<PlanStage>> applyProjection(OperationContext* opCtx,
                                                   unique_ptr<PlanStage> root) {
     invariant(!projObj.isEmpty());
 
-    projection_ast::Projection proj = projection_ast::parse(
-        cq->getExpCtx(), projObj.getOwned(), cq->root(), cq->getQueryObj(), ProjectionPolicies{});
+    projection_ast::Projection proj =
+        projection_ast::parse(cq->getExpCtx(),
+                              projObj.getOwned(),
+                              cq->root(),
+                              cq->getQueryObj(),
+                              ProjectionPolicies::findProjectionPolicies());
 
     {
         // The query system is in the process of migrating from one projection
diff --git a/src/mongo/db/query/projection_ast.h b/src/mongo/db/query/projection_ast.h
index 8a279fbbf8c..3f85c85c419 100644
--- a/src/mongo/db/query/projection_ast.h
+++ b/src/mongo/db/query/projection_ast.h
@@ -74,11 +74,16 @@ public:
         return _children;
     }
 
+    ASTNode* child(size_t index) const {
+        invariant(index < _children.size());
+        return _children[index].get();
+    }
+
     const ASTNode* parent() const {
         return _parent;
     }
 
-    const bool isRoot() const {
+    bool isRoot() const {
         return !_parent;
     }
 
@@ -156,6 +161,21 @@ public:
         _fieldNames.push_back(fieldName.toString());
     }
 
+    /**
+     * Remove a node which is a direct child of this tree. Returns true if anything was removed,
+     * false otherwise.
+     */
+    bool removeChild(StringData fieldName) {
+        if (auto it = std::find(_fieldNames.begin(), _fieldNames.end(), fieldName);
+            it != _fieldNames.end()) {
+            _children.erase(_children.begin() + std::distance(_fieldNames.begin(), it));
+            _fieldNames.erase(it);
+            return true;
+        }
+
+        return false;
+    }
+
     const std::vector<std::string>& fieldNames() const {
         return _fieldNames;
     }
diff --git a/src/mongo/db/query/projection_ast_test.cpp b/src/mongo/db/query/projection_ast_test.cpp
index 3b0733bf497..ac3592a8d3b 100644
--- a/src/mongo/db/query/projection_ast_test.cpp
+++ b/src/mongo/db/query/projection_ast_test.cpp
@@ -52,6 +52,18 @@ class ProjectionASTTest : public AggregationContextFixture {
 public:
     Projection parseWithDefaultPolicies(const BSONObj& projectionBson,
                                         boost::optional<BSONObj> matchExprBson = boost::none) {
+        return parseWithPolicies(projectionBson, matchExprBson, ProjectionPolicies{});
+    }
+
+    Projection parseWithFindFeaturesEnabled(const BSONObj& projectionBson,
+                                            boost::optional<BSONObj> matchExprBson = boost::none) {
+        auto policy = ProjectionPolicies::findProjectionPolicies();
+        return parseWithPolicies(projectionBson, matchExprBson, policy);
+    }
+
+    Projection parseWithPolicies(const BSONObj& projectionBson,
+                                 boost::optional<BSONObj> matchExprBson,
+                                 ProjectionPolicies policies) {
         StatusWith<std::unique_ptr<MatchExpression>> swMatchExpression(nullptr);
         if (matchExprBson) {
             swMatchExpression = MatchExpressionParser::parse(*matchExprBson, getExpCtx());
@@ -62,7 +74,7 @@ public:
                                      projectionBson,
                                      swMatchExpression.getValue().get(),
                                      matchExprBson.get_value_or(BSONObj()),
-                                     ProjectionPolicies{});
+                                     policies);
     }
 };
 
@@ -91,37 +103,69 @@ TEST_F(ProjectionASTTest, TestParsingTypeInclusion) {
     ASSERT(projWithoutId.type() == ProjectType::kInclusion);
 }
 
+TEST_F(ProjectionASTTest, TestParsingTypeInclusionIdOnly) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{_id: 1}"));
+    ASSERT(proj.type() == ProjectType::kInclusion);
+}
+
 TEST_F(ProjectionASTTest, TestParsingTypeInclusionWithNesting) {
     Projection proj = parseWithDefaultPolicies(fromjson("{'a.b': 1, 'a.c': 1}"));
     ASSERT(proj.type() == ProjectType::kInclusion);
 }
 
+TEST_F(ProjectionASTTest, TestParsingTypeInclusionWithNestingObjectSyntax) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: {b: 1, c: 1}}"));
+    ASSERT(proj.type() == ProjectType::kInclusion);
+}
+
 TEST_F(ProjectionASTTest, TestParsingTypeExclusion) {
     Projection p = parseWithDefaultPolicies(fromjson("{a: 0, _id: 0}"));
     ASSERT(p.type() == ProjectType::kExclusion);
 }
 
+TEST_F(ProjectionASTTest, TestParsingTypeExclusionWithNesting) {
+    Projection p = parseWithDefaultPolicies(fromjson("{'a.b': 0, _id: 0}"));
+    ASSERT(p.type() == ProjectType::kExclusion);
+}
+
+TEST_F(ProjectionASTTest, TestParsingTypeExclusionWithIdIncluded) {
+    {
+        Projection p = parseWithDefaultPolicies(fromjson("{'a.b': 0, _id: 1}"));
+        ASSERT(p.type() == ProjectType::kExclusion);
+    }
+
+    {
+        Projection p = parseWithDefaultPolicies(fromjson("{_id: 1, 'a.b': 0}"));
+        ASSERT(p.type() == ProjectType::kExclusion);
+    }
+}
+
+TEST_F(ProjectionASTTest, TestParsingTypeExclusionWithNestingObjectSyntax) {
+    Projection p = parseWithDefaultPolicies(fromjson("{a: {b: 0}, _id: 0}"));
+    ASSERT(p.type() == ProjectType::kExclusion);
+}
+
 TEST_F(ProjectionASTTest, TestParsingTypeOnlyElemMatch) {
-    Projection p = parseWithDefaultPolicies(fromjson("{a: {$elemMatch: {foo: 3}}}"));
+    Projection p = parseWithFindFeaturesEnabled(fromjson("{a: {$elemMatch: {foo: 3}}}"));
     ASSERT(p.type() == ProjectType::kInclusion);
 }
 
 TEST_F(ProjectionASTTest, TestParsingTypeOnlySlice) {
-    Projection p = parseWithDefaultPolicies(fromjson("{a: {$slice: 1}}"));
+    Projection p = parseWithFindFeaturesEnabled(fromjson("{a: {$slice: 1}}"));
     ASSERT(p.type() == ProjectType::kExclusion);
 }
 
 TEST_F(ProjectionASTTest, TestParsingTypeOnlyElemMatchAndSlice) {
     {
         Projection p =
-            parseWithDefaultPolicies(fromjson("{a: {$elemMatch: {foo: 3}}, b: {$slice: 1}}"));
+            parseWithFindFeaturesEnabled(fromjson("{a: {$elemMatch: {foo: 3}}, b: {$slice: 1}}"));
         ASSERT(p.type() == ProjectType::kExclusion);
     }
 
     // The order shouldn't matter.
     {
         Projection p =
-            parseWithDefaultPolicies(fromjson("{a: {$slice: 1}, b: {$elemMatch: {foo: 3}}}"));
+            parseWithFindFeaturesEnabled(fromjson("{a: {$slice: 1}, b: {$elemMatch: {foo: 3}}}"));
         ASSERT(p.type() == ProjectType::kExclusion);
     }
 }
@@ -153,35 +197,43 @@ TEST_F(ProjectionASTTest, TestParsingInclusionWithMeta) {
 TEST_F(ProjectionASTTest, TestInvalidProjectionWithInclusionsAndExclusions) {
     ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{a: 1, b: 0}")), DBException, 31254);
     ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{a: 0, b: 1}")), DBException, 31253);
+}
+
+TEST_F(ProjectionASTTest, TestInvalidProjectionWithExpressionInExclusion) {
     ASSERT_THROWS_CODE(
         parseWithDefaultPolicies(fromjson("{a: 0, b: {$add: [1, 2]}}")), DBException, 31252);
 }
 
+TEST_F(ProjectionASTTest, TestInvalidProjectionWithLiteralInExclusion) {
+    ASSERT_THROWS_CODE(
+        parseWithDefaultPolicies(fromjson("{a: 0, b: 'hallo world'}")), DBException, 31310);
+}
+
 TEST_F(ProjectionASTTest, TestInvalidProjectionWithPositionalAndElemMatch) {
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a.$': 1, b: {$elemMatch: {foo: 'bar'}}}"),
-                                 fromjson("{a: 1}")),
+        parseWithFindFeaturesEnabled(fromjson("{'a.$': 1, b: {$elemMatch: {foo: 'bar'}}}"),
+                                     fromjson("{a: 1}")),
         DBException,
         31255);
 
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{b: {$elemMatch: {foo: 'bar'}}, 'a.$': 1}"),
-                                 fromjson("{a: 1}")),
+        parseWithFindFeaturesEnabled(fromjson("{b: {$elemMatch: {foo: 'bar'}}, 'a.$': 1}"),
+                                     fromjson("{a: 1}")),
         DBException,
         31256);
 }
 
 TEST_F(ProjectionASTTest, TestCloningWithPositionalAndSlice) {
     Projection proj =
-        parseWithDefaultPolicies(fromjson("{'a.b': 1, b: 1, 'c.d.$': 1, f: {$slice: [1, 2]}}"),
-                                 fromjson("{'c.d': {$gt: 1}}"));
+        parseWithFindFeaturesEnabled(fromjson("{'a.b': 1, b: 1, 'c.d.$': 1, f: {$slice: [1, 2]}}"),
+                                     fromjson("{'c.d': {$gt: 1}}"));
     ASSERT(proj.type() == ProjectType::kInclusion);
     assertCanClone(std::move(proj));
 }
 
 TEST_F(ProjectionASTTest, TestCloningWithElemMatch) {
     Projection proj =
-        parseWithDefaultPolicies(fromjson("{'a.b': 1, b: 1, f: {$elemMatch: {foo: 'bar'}}}"));
+        parseWithFindFeaturesEnabled(fromjson("{'a.b': 1, b: 1, f: {$elemMatch: {foo: 'bar'}}}"));
     ASSERT(proj.type() == ProjectType::kInclusion);
     assertCanClone(std::move(proj));
 }
@@ -194,7 +246,7 @@ TEST_F(ProjectionASTTest, TestCloningWithExpression) {
 }
 
 TEST_F(ProjectionASTTest, TestDebugBSONWithPositionalAndSliceSkipLimit) {
-    Projection proj = parseWithDefaultPolicies(
+    Projection proj = parseWithFindFeaturesEnabled(
         fromjson("{'a.b': 1, b: 1, 'c.d.$': 1, f: {$slice: [1, 2]}}"), fromjson("{'c.d': 1}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
@@ -203,10 +255,27 @@ TEST_F(ProjectionASTTest, TestDebugBSONWithPositionalAndSliceSkipLimit) {
     ASSERT_BSONOBJ_EQ(output, expected);
 }
 
+TEST_F(ProjectionASTTest, TestDebugBSONWithNestedDollarPrefixedFields) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{c: {$id: 1}}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{c: {$id: true}, _id: true}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
+TEST_F(ProjectionASTTest, TestDebugBSONWithNestedPositional) {
+    Projection proj =
+        parseWithFindFeaturesEnabled(fromjson("{c: {'d.$': 1}}"), fromjson("{'c.d': 1}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{c: {'d.$': {'c.d': {$eq: 1}}}, _id: true}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
 TEST_F(ProjectionASTTest, TestDebugBSONWithPositionalInTheMiddleOfFieldPaths) {
     // Should treat "c.d.$.e" the same as "c.d.$".
-    Projection proj = parseWithDefaultPolicies(fromjson("{'a.b': 1, b: 1, 'c.d.$.e': 1}"),
-                                               fromjson("{'c.d': 1}"));
+    Projection proj = parseWithFindFeaturesEnabled(fromjson("{'a.b': 1, b: 1, 'c.d.$.e': 1}"),
+                                                   fromjson("{'c.d': 1}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
     BSONObj expected =
@@ -216,8 +285,8 @@ TEST_F(ProjectionASTTest, TestDebugBSONWithPositionalInTheMiddleOfFieldPaths) {
 
 TEST_F(ProjectionASTTest, TestDebugBSONWithPositionalInTheMiddleOfFieldPathsWithDollarPrefixField) {
     // Should treat "c.$id.$.e" the same as "c.$id.$".
-    Projection proj = parseWithDefaultPolicies(fromjson("{'a.b': 1, b: 1, 'c.$id.$.e': 1}"),
-                                               fromjson("{'c.$id': 1}"));
+    Projection proj = parseWithFindFeaturesEnabled(fromjson("{'a.b': 1, b: 1, 'c.$id.$.e': 1}"),
+                                                   fromjson("{'c.$id': 1}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
     BSONObj expected =
@@ -226,7 +295,7 @@ TEST_F(ProjectionASTTest, TestDebugBSONWithPositionalInTheMiddleOfFieldPathsWith
 }
 
 TEST_F(ProjectionASTTest, TestDebugBSONWithSliceLimit) {
-    Projection proj = parseWithDefaultPolicies(fromjson("{'a.b': 1, b: 1, f: {$slice: 2}}"));
+    Projection proj = parseWithFindFeaturesEnabled(fromjson("{'a.b': 1, b: 1, f: {$slice: 2}}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
     BSONObj expected = fromjson("{a: {b: true}, b: true, f: {$slice: 2}, _id: true}");
@@ -235,7 +304,7 @@ TEST_F(ProjectionASTTest, TestDebugBSONWithSliceLimit) {
 
 TEST_F(ProjectionASTTest, TestDebugBSONWithElemMatch) {
     Projection proj =
-        parseWithDefaultPolicies(fromjson("{'a.b': 1, b: 1, f: {$elemMatch: {foo: 'bar'}}}"));
+        parseWithFindFeaturesEnabled(fromjson("{'a.b': 1, b: 1, f: {$elemMatch: {foo: 'bar'}}}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
     BSONObj expected =
@@ -253,7 +322,15 @@ TEST_F(ProjectionASTTest, TestDebugBSONWithExpression) {
     ASSERT_BSONOBJ_EQ(output, expected);
 }
 
-TEST_F(ProjectionASTTest, TestDebugBSONWithExclusion) {
+TEST_F(ProjectionASTTest, TestDebugBSONWithSimpleInclusion) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: 1, b: 1}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{a: true, b: true, _id: true}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
+TEST_F(ProjectionASTTest, TestDebugBSONWithSimpleExclusion) {
     Projection proj = parseWithDefaultPolicies(fromjson("{a: 0, b: 0}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
@@ -261,8 +338,40 @@ TEST_F(ProjectionASTTest, TestDebugBSONWithExclusion) {
     ASSERT_BSONOBJ_EQ(output, expected);
 }
 
+TEST_F(ProjectionASTTest, TestDebugBSONWithObjectSyntaxInclusion) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: {b: {d: 1}, c: 1}, f: 1}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{a: {b: {d: true}, c: true}, f: true, _id : true}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
+TEST_F(ProjectionASTTest, TestDebugBSONWithObjectSyntaxExclusion) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: {b: {d: 0}, c: 0}, f: 0}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{a: {b: {d: false}, c: false}, f: false}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
+TEST_F(ProjectionASTTest, TestDebugBSONWithMixedSyntaxInclusion) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: {'b.d': 1, c: 1}, f: 1}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{a: {b: {d: true}, c: true}, f: true, _id: true}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
+TEST_F(ProjectionASTTest, TestDebugBSONWithMixedSyntaxExclusion) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: {'b.d': 0, c: 0}, f: 0}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{a: {b: {d: false}, c: false}, f: false}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
 TEST_F(ProjectionASTTest, TestDebugBSONWithOnlyElemMatch) {
-    Projection proj = parseWithDefaultPolicies(fromjson("{a: {$elemMatch: {foo: 3}}}"));
+    Projection proj = parseWithFindFeaturesEnabled(fromjson("{a: {$elemMatch: {foo: 3}}}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
     BSONObj expected = fromjson("{a: {$elemMatch: {foo: {$eq: 3}}}, _id: true}");
@@ -270,13 +379,29 @@ TEST_F(ProjectionASTTest, TestDebugBSONWithOnlyElemMatch) {
 }
 
 TEST_F(ProjectionASTTest, TestDebugBSONWithOnlySlice) {
-    Projection proj = parseWithDefaultPolicies(fromjson("{a: {$slice: 1}}"));
+    Projection proj = parseWithFindFeaturesEnabled(fromjson("{a: {$slice: 1}}"));
 
     BSONObj output = projection_ast::astToDebugBSON(proj.root());
     BSONObj expected = fromjson("{a: {$slice: 1}}");
     ASSERT_BSONOBJ_EQ(output, expected);
 }
 
+TEST_F(ProjectionASTTest, TestDebugBSONWithLiteralValue) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: 'abc'}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{a: {$const: 'abc'}, _id: true}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
+TEST_F(ProjectionASTTest, TestDebugBSONWithNestedLiteralValue) {
+    Projection proj = parseWithDefaultPolicies(fromjson("{a: {b: 'abc'}}"));
+
+    BSONObj output = projection_ast::astToDebugBSON(proj.root());
+    BSONObj expected = fromjson("{a: {b: {$const: 'abc'}}, _id: true}");
+    ASSERT_BSONOBJ_EQ(output, expected);
+}
+
 TEST_F(ProjectionASTTest, ParserErrorsOnCollisionNestedFieldFirst) {
     ASSERT_THROWS_CODE(
         parseWithDefaultPolicies(fromjson("{'a.b': 1, d: 1, a: 1}")), DBException, 31250);
@@ -287,73 +412,69 @@ TEST_F(ProjectionASTTest, ParserErrorsOnCollisionNestedFieldLast) {
         parseWithDefaultPolicies(fromjson("{a: 1, d: 1, 'a.b': 1}")), DBException, 31249);
 }
 
-TEST_F(ProjectionASTTest, ParserErrorsOnInvalidSliceArguments) {
-    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{a: {$slice: ['not a number', 123]}}")),
-                       DBException,
-                       31257);
-
-    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{a: {$slice: [123, 'not a number']}}")),
-                       DBException,
-                       31258);
-
-    ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{a: {$slice: [123, -5]}}")), DBException, 31259);
-}
-
 TEST_F(ProjectionASTTest, ParserErrorsOnSliceWithWrongNumberOfArguments) {
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a': {$slice: []}}")), DBException, 31272);
-    ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a': {$slice: [1, 2, 3]}}")), DBException, 31272);
+        parseWithFindFeaturesEnabled(fromjson("{'a': {$slice: []}}")), DBException, 28667);
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a': {$slice: [1, 2, 3, 4]}}")),
+                       DBException,
+                       28667);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnSliceWithWrongArgumentType) {
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a': {$slice: 'hello world'}}")),
+                       DBException,
+                       28667);
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a': {$slice: 'hello world'}}")), DBException, 31273);
-    ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a': {$slice: {foo: 1}}}")), DBException, 31273);
+        parseWithFindFeaturesEnabled(fromjson("{'a': {$slice: {foo: 1}}}")), DBException, 28667);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnInvalidElemMatchArgument) {
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{a: {$elemMatch: []}}")), DBException, 31274);
+        parseWithFindFeaturesEnabled(fromjson("{a: {$elemMatch: []}}")), DBException, 31274);
 
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{a: {$elemMatch: 'string'}}")), DBException, 31274);
+        parseWithFindFeaturesEnabled(fromjson("{a: {$elemMatch: 'string'}}")), DBException, 31274);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnElemMatchOnDottedField) {
-    ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a.b': {$elemMatch: {b: 1}}}")), DBException, 31275);
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a.b': {$elemMatch: {b: 1}}}")),
+                       DBException,
+                       31275);
+
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{a: {x: {$elemMatch: {b: 1}}}}")),
+                       DBException,
+                       31275);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnMultiplePositionalInOnePath) {
-    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{'a.$.b.$': 1}"), fromjson("{a: 1}")),
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a.$.b.$': 1}"), fromjson("{a: 1}")),
                        DBException,
                        31287);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnMultiplePositionalInProjection) {
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a.$': 1, 'b.$': 1}"), fromjson("{a: 1, b: 1}")),
+        parseWithFindFeaturesEnabled(fromjson("{'a.$': 1, 'b.$': 1}"), fromjson("{a: 1, b: 1}")),
         DBException,
         31276);
 
-    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{'a.b.$.': 1}"), fromjson("{a: 1}")),
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a.b.$.': 1}"), fromjson("{a: 1}")),
                        DBException,
                        31270);
 
-    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{'a.$.b.$': 1}"), fromjson("{a: 1}")),
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a.$.b.$': 1}"), fromjson("{a: 1}")),
                        DBException,
                        31287);
 
-    ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a.$.$': 1}"), fromjson("{a: 1}")), DBException, 31287);
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a.$.$': 1}"), fromjson("{a: 1}")),
+                       DBException,
+                       31287);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnPositionalProjectionNotMatchingQuery) {
-    ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a.$': 1}"), fromjson("{b: 1}")), DBException, 31277);
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a.$': 1}"), fromjson("{b: 1}")),
+                       DBException,
+                       31277);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnSubfieldPrefixedByDbRefField) {
@@ -363,39 +484,91 @@ TEST_F(ProjectionASTTest, ParserErrorsOnSubfieldPrefixedByDbRefField) {
 
 TEST_F(ProjectionASTTest, ParserErrorsOnJustPositionalProjection) {
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'$': 1}"), fromjson("{a: 1}")), DBException, 31277);
+        parseWithFindFeaturesEnabled(fromjson("{'$': 1}"), fromjson("{a: 1}")), DBException, 31277);
 
     // {$: 1} is an invalid match expression.
-    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{$: 1}"), fromjson("{$: 1}")),
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{$: 1}"), fromjson("{$: 1}")),
                        DBException,
                        ErrorCodes::BadValue);
 
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'$': 1}"), fromjson("{}")), DBException, 31277);
+        parseWithFindFeaturesEnabled(fromjson("{'$': 1}"), fromjson("{}")), DBException, 31277);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnPositionalAndSlice) {
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a.$': {$slice: 1}}")), DBException, 31271);
+        parseWithFindFeaturesEnabled(fromjson("{'a.$': {$slice: 1}}")), DBException, 31271);
 }
 
 TEST_F(ProjectionASTTest, ParserErrorsOnPositionalOnElemMatch) {
+    ASSERT_THROWS_CODE(parseWithFindFeaturesEnabled(fromjson("{'a.$': {$elemMatch: {b: 1}}}")),
+                       DBException,
+                       31271);
+}
+
+TEST_F(ProjectionASTTest, ParserErrorsOnPositionalOnLiteral) {
     ASSERT_THROWS_CODE(
-        parseWithDefaultPolicies(fromjson("{'a.$': {$elemMatch: {b: 1}}}")), DBException, 31271);
+        parseWithFindFeaturesEnabled(fromjson("{'a.$': 'literal string'}")), DBException, 31308);
+}
+
+TEST_F(ProjectionASTTest, ParserErrorsOnPositionalAndSubObj) {
+    ASSERT_THROWS_CODE(
+        parseWithFindFeaturesEnabled(fromjson("{'a': {'b.$': {c: 1}}}")), DBException, 31271);
 }
 
 TEST_F(ProjectionASTTest, ParserDoesNotErrorOnPositionalOfDbRefField) {
-    Projection idProj =
-        parseWithDefaultPolicies(fromjson("{'a.$id.b.$': 1, x: 1}"), fromjson("{'a.$id.b': 1}"));
+    Projection idProj = parseWithFindFeaturesEnabled(fromjson("{'a.$id.b.$': 1, x: 1}"),
+                                                     fromjson("{'a.$id.b': 1}"));
     ASSERT(idProj.type() == ProjectType::kInclusion);
 
-    Projection dbProj =
-        parseWithDefaultPolicies(fromjson("{'a.$db.b.$': 1, x: 1}"), fromjson("{'a.$db.b': 1}"));
+    Projection dbProj = parseWithFindFeaturesEnabled(fromjson("{'a.$db.b.$': 1, x: 1}"),
+                                                     fromjson("{'a.$db.b': 1}"));
     ASSERT(dbProj.type() == ProjectType::kInclusion);
 
-    Projection refProj =
-        parseWithDefaultPolicies(fromjson("{'a.$ref.b.$': 1, x: 1}"), fromjson("{'a.$ref.b': 1}"));
+    Projection refProj = parseWithFindFeaturesEnabled(fromjson("{'a.$ref.b.$': 1, x: 1}"),
+                                                      fromjson("{'a.$ref.b': 1}"));
     ASSERT(refProj.type() == ProjectType::kInclusion);
 }
 
+TEST_F(ProjectionASTTest, ShouldThrowWhenParsingInvalidExpression) {
+    ASSERT_THROWS(parseWithDefaultPolicies(BSON("a" << BSON("$gt" << BSON("bad"
+                                                                          << "arguments")))),
+                  AssertionException);
+}
+
+TEST_F(ProjectionASTTest, ShouldThrowWhenParsingUnknownExpression) {
+    ASSERT_THROWS_CODE(
+        parseWithDefaultPolicies(BSON("a" << BSON("$fakeExpression" << BSON("bad"
+                                                                            << "arguments")))),
+        AssertionException,
+        31325);
+}
+
+TEST_F(ProjectionASTTest, ShouldThrowWhenParsingSliceInvalidWithFindFeaturesOff) {
+    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{a: {$slice: {wrongType: 1}}}")),
+                       AssertionException,
+                       28667);
+}
+
+TEST_F(ProjectionASTTest, ShouldSucceedWhenParsingAggSliceWithFindFeaturesOff) {
+    auto proj = parseWithDefaultPolicies(fromjson("{a: {$slice: ['$a', 3]}}"));
+    ASSERT(proj.type() == ProjectType::kInclusion);
+}
+
+TEST_F(ProjectionASTTest, ShouldThrowWhenParsingElemMatchWithFindFeaturesOff) {
+    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{a: {$elemMatch: {foo: 3}}}")),
+                       AssertionException,
+                       31325);
+}
+
+TEST_F(ProjectionASTTest, ShouldThrowWhenParsingPositionalWithFindFeaturesOff) {
+    ASSERT_THROWS_CODE(parseWithDefaultPolicies(fromjson("{'a.$': 1}"), fromjson("{a: {$gt: 1}}")),
+                       AssertionException,
+                       31324);
+    ASSERT_THROWS_CODE(
+        parseWithDefaultPolicies(fromjson("{a: {'b.$': 1}}"), fromjson("{'a.b': {$gt: 1}}")),
+        AssertionException,
+        31324);
+}
+
 }  // namespace
diff --git a/src/mongo/db/query/projection_parser.cpp b/src/mongo/db/query/projection_parser.cpp
index 9642b2456d2..727ec0df089 100644
--- a/src/mongo/db/query/projection_parser.cpp
+++ b/src/mongo/db/query/projection_parser.cpp
@@ -35,6 +35,23 @@
 namespace mongo {
 namespace projection_ast {
 namespace {
+
+/**
+ * Returns whether an element's type implies that the element is an inclusion or exclusion.
+ */
+bool isInclusionOrExclusionType(BSONType type) {
+    switch (type) {
+        case BSONType::Bool:
+        case BSONType::NumberInt:
+        case BSONType::NumberLong:
+        case BSONType::NumberDouble:
+        case BSONType::NumberDecimal:
+            return true;
+        default:
+            return false;
+    }
+}
+
 void addNodeAtPathHelper(ProjectionPathASTNode* root,
                          const FieldPath& path,
                          size_t componentIndex,
@@ -80,7 +97,8 @@ void addNodeAtPath(ProjectionPathASTNode* root,
  * Return a pair {begin, end} indicating where the first positional operator in the string is.
  * If there is no positional operator, returns boost::none.
  */
-boost::optional<std::pair<size_t, size_t>> findFirstPositionalOperator(StringData fullPath) {
+using PositionalProjectionLocation = boost::optional<std::pair<size_t, size_t>>;
+PositionalProjectionLocation findFirstPositionalOperator(StringData fullPath) {
     size_t first = fullPath.find(".$.");
     if (first != std::string::npos) {
         return {{first, first + 3}};
@@ -129,225 +147,424 @@ bool isPrefixOf(StringData first, StringData second) {
 
     return second.startsWith(first) && second[first.size()] == '.';
 }
-}  // namespace
 
-Projection parse(boost::intrusive_ptr<ExpressionContext> expCtx,
-                 const BSONObj& obj,
-                 const MatchExpression* const query,
-                 const BSONObj& queryObj,
-                 ProjectionPolicies policies) {
-    // TODO: SERVER-42988 Support agg syntax with nesting.
+struct ParseContext {
+    const boost::intrusive_ptr<ExpressionContext> expCtx;
 
-    ProjectionPathASTNode root;
+    // Properties of the projection/query.
+    const MatchExpression* const query = nullptr;
+    const BSONObj& queryObj;
 
-    bool idSpecified = false;
+    const BSONObj& spec;
+    const ProjectionPolicies policies;
 
+    // Properties of the projection that need to be stored for later checks.
+    bool idSpecified = false;
     bool hasPositional = false;
     bool hasElemMatch = false;
     bool hasFindSlice = false;
     boost::optional<ProjectType> type;
 
-    for (auto&& elem : obj) {
-        idSpecified |=
-            elem.fieldNameStringData() == "_id" || elem.fieldNameStringData().startsWith("_id.");
+    // Whether there's an {_id: 1} projection.
+    bool idIncludedEntirely = false;
+};
+
+void attemptToParseFindSlice(ParseContext* parseCtx,
+                             const FieldPath& path,
+                             const BSONObj& subObj,
+                             ProjectionPathASTNode* parent) {
+    if (subObj.firstElement().isNumber()) {
+        addNodeAtPath(parent,
+                      path,
+                      std::make_unique<ProjectionSliceASTNode>(boost::none,
+                                                               subObj.firstElement().numberInt()));
+    } else if (subObj.firstElementType() == BSONType::Array) {
+        BSONObj arr = subObj.firstElement().embeddedObject();
+        uassert(31272, "$slice array argument should be of form [skip, limit]", arr.nFields() == 2);
+
+        BSONObjIterator it(arr);
+        BSONElement skipElt = it.next();
+        BSONElement limitElt = it.next();
+        uassert(31257,
+                str::stream() << "$slice expects the skip argument to be a number, got "
+                              << skipElt.type(),
+                skipElt.isNumber());
+        uassert(31258,
+                str::stream() << "$slice expects the limit argument to be a number, got "
+                              << limitElt.type(),
+                limitElt.isNumber());
+
+        uassert(31259,
+                str::stream() << "$slice limit must be positive, got " << limitElt.numberInt(),
+                limitElt.numberInt() > 0);
+        addNodeAtPath(
+            parent,
+            path,
+            std::make_unique<ProjectionSliceASTNode>(skipElt.numberInt(), limitElt.numberInt()));
+    } else {
+        uasserted(31273, "$slice only supports numbers and [skip, limit] arrays");
+    }
+
+    parseCtx->hasFindSlice = true;
+}
 
-        const auto firstPositionalProjection =
-            findFirstPositionalOperator(elem.fieldNameStringData());
-
-        if (elem.type() == BSONType::Object) {
-            BSONObj subObj = elem.embeddedObject();
-
-            // Make sure this isn't a positional operator. It's illegal to combine positional with
-            // any expression.
-            uassert(31271,
-                    "positional projection cannot be used with an expression",
-                    !firstPositionalProjection);
-
-            FieldPath path(elem.fieldNameStringData());
-
-            if (subObj.firstElementFieldNameStringData() == "$slice") {
-                if (subObj.firstElement().isNumber()) {
-                    addNodeAtPath(&root,
-                                  path,
-                                  std::make_unique<ProjectionSliceASTNode>(
-                                      boost::none, subObj.firstElement().numberInt()));
-                } else if (subObj.firstElementType() == BSONType::Array) {
-                    BSONObj arr = subObj.firstElement().embeddedObject();
-                    uassert(31272,
-                            "$slice array argument should be of form [skip, limit]",
-                            arr.nFields() == 2);
-
-                    BSONObjIterator it(arr);
-                    BSONElement skipElt = it.next();
-                    BSONElement limitElt = it.next();
-                    uassert(31257,
-                            str::stream() << "$slice expects the skip argument to be a number, got "
-                                          << skipElt.type(),
-                            skipElt.isNumber());
-                    uassert(31258,
-                            str::stream()
-                                << "$slice expects the limit argument to be a number, got "
-                                << limitElt.type(),
-                            limitElt.isNumber());
-
-
-                    uassert(31259,
-                            str::stream()
-                                << "$slice limit must be positive, got " << limitElt.numberInt(),
-                            limitElt.numberInt() > 0);
-                    addNodeAtPath(&root,
-                                  path,
-                                  std::make_unique<ProjectionSliceASTNode>(skipElt.numberInt(),
-                                                                           limitElt.numberInt()));
-                } else {
-                    uasserted(31273, "$slice only supports numbers and [skip, limit] arrays");
-                }
-
-                hasFindSlice = true;
-            } else if (subObj.firstElementFieldNameStringData() == "$elemMatch") {
-                // Validate $elemMatch arguments and dependencies.
-                uassert(31274,
-                        str::stream() << "elemMatch: Invalid argument, object required, but got "
-                                      << subObj.firstElementType(),
-                        subObj.firstElementType() == BSONType::Object);
-
-                uassert(
-                    31255, "Cannot specify positional operator and $elemMatch.", !hasPositional);
-
-                uassert(31275,
-                        "Cannot use $elemMatch projection on a nested field.",
-                        !str::contains(elem.fieldName(), '.'));
-
-                // Create a MatchExpression for the elemMatch.
-                BSONObj elemMatchObj = elem.wrap();
-                invariant(elemMatchObj.isOwned());
-
-                auto matcher =
-                    CopyableMatchExpression{elemMatchObj,
-                                            expCtx,
-                                            std::make_unique<ExtensionsCallbackNoop>(),
-                                            MatchExpressionParser::kBanAllSpecialFeatures,
-                                            true /* optimize expression */};
-                auto matchNode = std::make_unique<MatchExpressionASTNode>(std::move(matcher));
-
-                addNodeAtPath(&root,
-                              path,
-                              std::make_unique<ProjectionElemMatchASTNode>(std::move(matchNode)));
-                hasElemMatch = true;
-            } else {
-                const bool isMeta = subObj.firstElementFieldNameStringData() == "$meta";
-
-                uassert(31252,
-                        "Cannot use expression other than $meta in exclusion projection",
-                        !type || *type == ProjectType::kInclusion || isMeta);
-
-                if (!isMeta) {
-                    type = ProjectType::kInclusion;
-                }
-
-                auto expr =
-                    Expression::parseExpression(expCtx, subObj, expCtx->variablesParseState);
-                addNodeAtPath(&root, path, std::make_unique<ExpressionASTNode>(expr));
+/**
+ * If the object matches the form of an expression ({<valid expression name>: <arguments>}) then
+ * attempts to parse it as an expression and add it to the tree.
+
+ * Returns whether the object matched the form an actual expression.
+ */
+bool attemptToParseGenericExpression(ParseContext* parseCtx,
+                                     const FieldPath& path,
+                                     const BSONObj& subObj,
+                                     ProjectionPathASTNode* parent) {
+    if (!Expression::isExpressionName(subObj.firstElementFieldNameStringData())) {
+        return false;
+    }
+
+    // It must be an expression.
+    const bool isMeta = subObj.firstElementFieldNameStringData() == "$meta";
+    uassert(31252,
+            "Cannot use expression other than $meta in exclusion projection",
+            !parseCtx->type || *parseCtx->type == ProjectType::kInclusion || isMeta);
+
+    if (!isMeta) {
+        parseCtx->type = ProjectType::kInclusion;
+    }
+
+    auto expr = Expression::parseExpression(
+        parseCtx->expCtx, subObj, parseCtx->expCtx->variablesParseState);
+    addNodeAtPath(parent, path, std::make_unique<ExpressionASTNode>(expr));
+    return true;
+}
+
+
+/**
+ * Treats the given object as either a find()-only expression ($slice, $elemMatch) if allowed,
+ * or as a generic agg expression, and adds it to the the tree.
+ *
+ * Returns whether or not the object matched the form of an expression. If not,
+ * the sub object must represent a sub-projection (e.g. {a: {b: 1}}).
+ */
+bool parseSubObjectAsExpression(ParseContext* parseCtx,
+                                const FieldPath& path,
+                                const BSONObj& subObj,
+                                ProjectionPathASTNode* parent) {
+
+    if (parseCtx->policies.findOnlyFeaturesAllowed()) {
+        if (subObj.firstElementFieldNameStringData() == "$slice") {
+            Status findSliceStatus = Status::OK();
+            try {
+                attemptToParseFindSlice(parseCtx, path, subObj, parent);
+                return true;
+            } catch (const DBException& exn) {
+                findSliceStatus = exn.toStatus();
             }
 
-        } else if (elem.trueValue()) {
-            if (!firstPositionalProjection) {
-                FieldPath path(elem.fieldNameStringData());
-                addNodeAtPath(&root, path, std::make_unique<BooleanConstantASTNode>(true));
-            } else {
-                uassert(31276,
-                        "Cannot specify more than one positional projection per query.",
-                        !hasPositional);
-
-                uassert(31256, "Cannot specify positional operator and $elemMatch.", !hasElemMatch);
-
-                StringData matchField = str::before(elem.fieldNameStringData(), '.');
-                uassert(31277,
-                        str::stream()
-                            << "Positional projection '" << elem.fieldName() << "' does not "
-                            << "match the query document.",
-                        query && hasPositionalOperatorMatch(query, matchField));
-
-                // Check that the path does not end with ".$." which can be interpreted as the
-                // positional projection.
-                uassert(31270,
-                        str::stream() << "Path cannot end with '.$.'",
-                        !elem.fieldNameStringData().endsWith(".$."));
-
-                const auto [firstPositionalBegin, firstPositionalEnd] = *firstPositionalProjection;
-
-                // See if there's another positional operator after the first one. If there is,
-                // it's invalid.
-                StringData remainingPathAfterPositional =
-                    elem.fieldNameStringData().substr(firstPositionalEnd);
-                uassert(31287,
-                        str::stream() << "Cannot use positional operator twice: "
-                                      << elem.fieldNameStringData(),
-                        findFirstPositionalOperator(remainingPathAfterPositional) == boost::none);
-
-                // Get everything up to the first positional operator.
-                StringData pathWithoutPositionalOperator =
-                    elem.fieldNameStringData().substr(0, firstPositionalBegin);
-
-                auto matcher =
-                    CopyableMatchExpression{queryObj,
-                                            expCtx,
-                                            std::make_unique<ExtensionsCallbackNoop>(),
-                                            MatchExpressionParser::kBanAllSpecialFeatures,
-                                            true /* optimize expression */};
-
-                FieldPath path(pathWithoutPositionalOperator);
-                invariant(query);
-                addNodeAtPath(&root,
-                              path,
-                              std::make_unique<ProjectionPositionalASTNode>(
-                                  std::make_unique<MatchExpressionASTNode>(matcher)));
-
-                hasPositional = true;
+            try {
+                attemptToParseGenericExpression(parseCtx, path, subObj, parent);
+            } catch (DBException& exn) {
+                exn.addContext(str::stream()
+                               << "Invalid $slice syntax. The given syntax " << subObj
+                               << " did not match the find() syntax because :: " << findSliceStatus
+                               << " :: "
+                               << "The given syntax did not match the expression $slice syntax.");
+                throw;
             }
 
-            uassert(31253,
-                    str::stream() << "Cannot do inclusion on field " << elem.fieldNameStringData()
-                                  << " in exclusion projection",
-                    !type || *type == ProjectType::kInclusion);
-            type = ProjectType::kInclusion;
+            return true;
+        } else if (subObj.firstElementFieldNameStringData() == "$elemMatch") {
+            // Validate $elemMatch arguments and dependencies.
+            uassert(31274,
+                    str::stream() << "elemMatch: Invalid argument, object required, but got "
+                                  << subObj.firstElementType(),
+                    subObj.firstElementType() == BSONType::Object);
+
+            uassert(31255,
+                    "Cannot specify positional operator and $elemMatch.",
+                    !parseCtx->hasPositional);
+
+            uassert(31275,
+                    "Cannot use $elemMatch projection on a nested field.",
+                    path.getPathLength() == 1 && parent->isRoot());
+
+            // Create a MatchExpression for the elemMatch.
+            BSONObj elemMatchObj = BSON(path.fullPath() << subObj);
+            invariant(elemMatchObj.isOwned());
+
+            auto matcher = CopyableMatchExpression{elemMatchObj,
+                                                   parseCtx->expCtx,
+                                                   std::make_unique<ExtensionsCallbackNoop>(),
+                                                   MatchExpressionParser::kBanAllSpecialFeatures,
+                                                   true /* optimize expression */};
+            auto matchNode = std::make_unique<MatchExpressionASTNode>(std::move(matcher));
+
+            addNodeAtPath(
+                parent, path, std::make_unique<ProjectionElemMatchASTNode>(std::move(matchNode)));
+            parseCtx->hasElemMatch = true;
+            return true;
+        }
+    }
+
+    return attemptToParseGenericExpression(parseCtx, path, subObj, parent);
+}
+
+/**
+ * Treats the given element as an inclusion projection, and update the tree as necessary.
+ */
+void parseInclusion(ParseContext* ctx,
+                    BSONElement elem,
+                    ProjectionPathASTNode* parent,
+                    boost::optional<FieldPath> fullPathToParent,
+                    PositionalProjectionLocation firstPositionalProjection) {
+    // There are special rules about _id being included. _id may be included in both inclusion and
+    // exclusion projections.
+    const bool isTopLevelIdProjection = elem.fieldNameStringData() == "_id" && parent->isRoot();
+
+    if (!firstPositionalProjection) {
+        FieldPath path(elem.fieldNameStringData());
+        addNodeAtPath(parent, path, std::make_unique<BooleanConstantASTNode>(true));
+
+        if (isTopLevelIdProjection) {
+            ctx->idIncludedEntirely = true;
+        }
+    } else {
+        uassert(31276,
+                "Cannot specify more than one positional projection per query.",
+                !ctx->hasPositional);
+
+        uassert(31256, "Cannot specify positional operator and $elemMatch.", !ctx->hasElemMatch);
+
+        StringData matchField = fullPathToParent ? fullPathToParent->front()
+                                                 : str::before(elem.fieldNameStringData(), '.');
+        uassert(31277,
+                str::stream() << "Positional projection '" << elem.fieldName() << "' does not "
+                              << "match the query document.",
+                ctx->query && hasPositionalOperatorMatch(ctx->query, matchField));
+
+        // Check that the path does not end with ".$." which can be interpreted as the
+        // positional projection.
+        uassert(31270,
+                str::stream() << "Path cannot end with '.$.'",
+                !elem.fieldNameStringData().endsWith(".$."));
+
+        const auto [firstPositionalBegin, firstPositionalEnd] = *firstPositionalProjection;
+
+        // See if there's another positional operator after the first one. If there is,
+        // it's invalid.
+        StringData remainingPathAfterPositional =
+            elem.fieldNameStringData().substr(firstPositionalEnd);
+        uassert(31287,
+                str::stream() << "Cannot use positional operator twice: "
+                              << elem.fieldNameStringData(),
+                findFirstPositionalOperator(remainingPathAfterPositional) == boost::none &&
+                    remainingPathAfterPositional != "$");
+
+        // Get everything up to the first positional operator.
+        StringData pathWithoutPositionalOperator =
+            elem.fieldNameStringData().substr(0, firstPositionalBegin);
+
+        FieldPath path(pathWithoutPositionalOperator);
+
+        auto matcher = CopyableMatchExpression{ctx->queryObj,
+                                               ctx->expCtx,
+                                               std::make_unique<ExtensionsCallbackNoop>(),
+                                               MatchExpressionParser::kBanAllSpecialFeatures,
+                                               true /* optimize expression */};
+
+        invariant(ctx->query);
+        addNodeAtPath(parent,
+                      path,
+                      std::make_unique<ProjectionPositionalASTNode>(
+                          std::make_unique<MatchExpressionASTNode>(matcher)));
+
+        ctx->hasPositional = true;
+    }
+
+    if (!isTopLevelIdProjection) {
+        uassert(31253,
+                str::stream() << "Cannot do inclusion on field " << elem.fieldNameStringData()
+                              << " in exclusion projection",
+                !ctx->type || *ctx->type == ProjectType::kInclusion);
+        ctx->type = ProjectType::kInclusion;
+    }
+}
+
+/**
+ * Treates the given element as an exclusion projection and updates the tree as necessary.
+ */
+void parseExclusion(ParseContext* ctx, BSONElement elem, ProjectionPathASTNode* parent) {
+    invariant(!elem.trueValue());
+    FieldPath path(elem.fieldNameStringData());
+    addNodeAtPath(parent, path, std::make_unique<BooleanConstantASTNode>(false));
+
+    const bool isTopLevelIdProjection = elem.fieldNameStringData() == "_id" && parent->isRoot();
+    if (!isTopLevelIdProjection) {
+        uassert(31254,
+                str::stream() << "Cannot do exclusion on field " << elem.fieldNameStringData()
+                              << " in inclusion projection",
+                !ctx->type || *ctx->type == ProjectType::kExclusion);
+        ctx->type = ProjectType::kExclusion;
+    }
+}
+
+/**
+ * Treats the given element as a literal value (e.g. {a: "foo"}) and updates the tree as necessary.
+ */
+void parseLiteral(ParseContext* ctx, BSONElement elem, ProjectionPathASTNode* parent) {
+    auto expr = Expression::parseOperand(ctx->expCtx, elem, ctx->expCtx->variablesParseState);
+
+    FieldPath pathFromParent(elem.fieldNameStringData());
+    addNodeAtPath(parent, pathFromParent, std::make_unique<ExpressionASTNode>(expr));
+
+    uassert(31310,
+            str::stream() << "Cannot use an expression " << elem << " in an exclusion projection",
+            !ctx->type || *ctx->type == ProjectType::kInclusion);
+    ctx->type = ProjectType::kInclusion;
+}
+
+// Mutually recursive with parseSubObject().
+void parseElement(ParseContext* ctx,
+                  BSONElement elem,
+                  boost::optional<FieldPath> fullPathToParent,
+                  ProjectionPathASTNode* parent);
+
+/**
+ * Parses the given object and updates the tree as necessary.
+ * This function will do the work of determining whether the sub object should be treated as an
+ * expression or subprojection.
+ */
+void parseSubObject(ParseContext* ctx,
+                    StringData objFieldName,
+                    boost::optional<FieldPath> fullPathToParent,
+                    const BSONObj& obj,
+                    ProjectionPathASTNode* parent) {
+    FieldPath path(objFieldName);
+
+    if (obj.nFields() == 1 && obj.firstElementFieldNameStringData().startsWith("$")) {
+        // Maybe it's an expression.
+        const bool isExpression = parseSubObjectAsExpression(ctx, path, obj, parent);
+
+        if (isExpression) {
+            return;
+        }
+
+        // It was likely intended to be an expression. Check if it's a valid field path or not to
+        // confirm.
+        try {
+            FieldPath fp(obj.firstElementFieldNameStringData());
+        } catch (const DBException&) {
+            uasserted(31325,
+                      str::stream()
+                          << "Unknown expression " << obj.firstElementFieldNameStringData());
+        }
+    }
+
+    // It's not an expression. Create a node to represent the new layer in the tree.
+    ProjectionPathASTNode* newParent = nullptr;
+    {
+        auto ownedChild = std::make_unique<ProjectionPathASTNode>();
+        newParent = ownedChild.get();
+        parent->addChild(objFieldName, std::move(ownedChild));
+    }
+
+    const FieldPath fullPathToNewParent = fullPathToParent ? fullPathToParent->concat(path) : path;
+    for (auto&& elem : obj) {
+        parseElement(ctx, elem, fullPathToNewParent, newParent);
+    }
+}
+
+/**
+ * Determine what "type" this element of the projection is (inclusion, exclusion, sub object,
+ * literal) and update the tree accordingly.
+ */
+void parseElement(ParseContext* ctx,
+                  BSONElement elem,
+                  boost::optional<FieldPath> fullPathToParent,
+                  ProjectionPathASTNode* parent) {
+    const auto firstPositionalProjection = findFirstPositionalOperator(elem.fieldNameStringData());
+
+    // If there is a positional projection, find only features must be enabled.
+    uassert(31324,
+            "Cannot use positional projection in aggregation projection",
+            (!firstPositionalProjection || ctx->policies.findOnlyFeaturesAllowed()));
+
+    if (elem.type() == BSONType::Object) {
+        BSONObj subObj = elem.embeddedObject();
+
+        // Make sure this isn't a positional operator. It's illegal to combine positional with
+        // any expression.
+        uassert(31271,
+                "positional projection cannot be used with an expression or sub object",
+                static_cast<bool>(!firstPositionalProjection));
+
+        parseSubObject(ctx, elem.fieldNameStringData(), fullPathToParent, subObj, parent);
+    } else if (isInclusionOrExclusionType(elem.type())) {
+        if (elem.trueValue()) {
+            parseInclusion(ctx, elem, parent, fullPathToParent, firstPositionalProjection);
         } else {
-            invariant(!elem.trueValue());
-            FieldPath path(elem.fieldNameStringData());
-            addNodeAtPath(&root, path, std::make_unique<BooleanConstantASTNode>(false));
-
-            if (elem.fieldNameStringData() != "_id") {
-                uassert(31254,
-                        str::stream() << "Cannot do exclusion on field "
-                                      << elem.fieldNameStringData() << "in inclusion projection",
-                        !type || *type == ProjectType::kExclusion);
-                type = ProjectType::kExclusion;
-            }
+            parseExclusion(ctx, elem, parent);
         }
+    } else {
+        uassert(31308,
+                "positional projection cannot be used with a literal",
+                static_cast<bool>(!firstPositionalProjection));
+
+        parseLiteral(ctx, elem, parent);
+    }
+}
+}  // namespace
+
+Projection parse(boost::intrusive_ptr<ExpressionContext> expCtx,
+                 const BSONObj& obj,
+                 const MatchExpression* const query,
+                 const BSONObj& queryObj,
+                 ProjectionPolicies policies) {
+    ProjectionPathASTNode root;
+
+    ParseContext ctx{expCtx, query, queryObj, obj, policies};
+
+    for (auto&& elem : obj) {
+        ctx.idSpecified |=
+            elem.fieldNameStringData() == "_id" || elem.fieldNameStringData().startsWith("_id.");
+        parseElement(&ctx, elem, boost::none, &root);
     }
 
     // find() defaults about inclusion/exclusion. These rules are preserved for compatibility
     // reasons. If there are no explicit inclusion/exclusion fields, the type depends on which
-    // expressions (if any) are used.
-    if (!type) {
-        if (hasFindSlice) {
-            type = ProjectType::kExclusion;
-        } else if (hasElemMatch) {
-            type = ProjectType::kInclusion;
+    // find() expressions (if any) are used in the following order: $slice, $elemMatch, $meta.
+    if (!ctx.type) {
+        if (ctx.idIncludedEntirely) {
+            // The projection {_id: 1} is considered an inclusion. The ParseContext's type field was
+            // not marked as such, because a projection {_id: 1, a: 0} is also valid, but considered
+            // exclusion.
+            ctx.type = ProjectType::kInclusion;
+        } else if (ctx.hasFindSlice) {
+            // If the projection has only find() expressions, then $slice has highest precedence.
+            ctx.type = ProjectType::kExclusion;
+        } else if (ctx.hasElemMatch) {
+            // $elemMatch has next-highest precedent.
+            ctx.type = ProjectType::kInclusion;
         } else {
             // This happens only when the projection is entirely $meta expressions.
-            type = ProjectType::kExclusion;
+            ctx.type = ProjectType::kExclusion;
         }
     }
-    invariant(type);
+    invariant(ctx.type);
 
-    if (!idSpecified && policies.idPolicy == ProjectionPolicies::DefaultIdPolicy::kIncludeId &&
-        *type == ProjectType::kInclusion) {
+    if (!ctx.idSpecified && policies.idPolicy == ProjectionPolicies::DefaultIdPolicy::kIncludeId &&
+        *ctx.type == ProjectType::kInclusion) {
         // Add a node to the root indicating that _id is included.
         addNodeAtPath(&root, "_id", std::make_unique<BooleanConstantASTNode>(true));
     }
 
-    return Projection{std::move(root), *type};
+    if (*ctx.type == ProjectType::kExclusion && ctx.idSpecified && ctx.idIncludedEntirely) {
+        // The user explicitly included _id in an exclusion projection. This is legal syntax, but
+        // the node indicating that _id is included doesn't need to be in the tree.
+        invariant(root.removeChild("_id"));
+    }
+
+    return Projection{std::move(root), *ctx.type};
 }
 
 Projection parse(boost::intrusive_ptr<ExpressionContext> expCtx,
diff --git a/src/mongo/db/query/projection_policies.h b/src/mongo/db/query/projection_policies.h
index c72790037ca..3326e2e3f92 100644
--- a/src/mongo/db/query/projection_policies.h
+++ b/src/mongo/db/query/projection_policies.h
@@ -52,22 +52,43 @@ struct ProjectionPolicies {
     // projections. Computed fields are implicitly prohibited by exclusion projections.
     enum class ComputedFieldsPolicy { kBanComputedFields, kAllowComputedFields };
 
+    // Whether $elemMatch, find() $slice and positional projection are allowed.
+    enum class FindOnlyFeaturesPolicy { kBanFindOnlyFeatures, kAllowFindOnlyFeatures };
+
     static const DefaultIdPolicy kDefaultIdPolicyDefault = DefaultIdPolicy::kIncludeId;
     static const ArrayRecursionPolicy kArrayRecursionPolicyDefault =
         ArrayRecursionPolicy::kRecurseNestedArrays;
     static const ComputedFieldsPolicy kComputedFieldsPolicyDefault =
         ComputedFieldsPolicy::kAllowComputedFields;
+    static const FindOnlyFeaturesPolicy kFindOnlyFeaturesPolicyDefault =
+        FindOnlyFeaturesPolicy::kBanFindOnlyFeatures;
+
+    static ProjectionPolicies findProjectionPolicies() {
+        return ProjectionPolicies{
+            ProjectionPolicies::kDefaultIdPolicyDefault,
+            ProjectionPolicies::kArrayRecursionPolicyDefault,
+            ProjectionPolicies::kComputedFieldsPolicyDefault,
+            ProjectionPolicies::FindOnlyFeaturesPolicy::kAllowFindOnlyFeatures};
+    }
 
-    ProjectionPolicies(DefaultIdPolicy idPolicy = kDefaultIdPolicyDefault,
-                       ArrayRecursionPolicy arrayRecursionPolicy = kArrayRecursionPolicyDefault,
-                       ComputedFieldsPolicy computedFieldsPolicy = kComputedFieldsPolicyDefault)
+    ProjectionPolicies(
+        DefaultIdPolicy idPolicy = kDefaultIdPolicyDefault,
+        ArrayRecursionPolicy arrayRecursionPolicy = kArrayRecursionPolicyDefault,
+        ComputedFieldsPolicy computedFieldsPolicy = kComputedFieldsPolicyDefault,
+        FindOnlyFeaturesPolicy findOnlyFeaturesPolicy = kFindOnlyFeaturesPolicyDefault)
         : idPolicy(idPolicy),
           arrayRecursionPolicy(arrayRecursionPolicy),
-          computedFieldsPolicy(computedFieldsPolicy) {}
+          computedFieldsPolicy(computedFieldsPolicy),
+          findOnlyFeaturesPolicy(findOnlyFeaturesPolicy) {}
 
     const DefaultIdPolicy idPolicy;
     const ArrayRecursionPolicy arrayRecursionPolicy;
     const ComputedFieldsPolicy computedFieldsPolicy;
+    const FindOnlyFeaturesPolicy findOnlyFeaturesPolicy;
+
+    bool findOnlyFeaturesAllowed() const {
+        return findOnlyFeaturesPolicy == FindOnlyFeaturesPolicy::kAllowFindOnlyFeatures;
+    }
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/query/projection_test.cpp b/src/mongo/db/query/projection_test.cpp
index c4180584f2f..12e33792114 100644
--- a/src/mongo/db/query/projection_test.cpp
+++ b/src/mongo/db/query/projection_test.cpp
@@ -50,7 +50,9 @@ using projection_ast::Projection;
 /**
  * Helper for creating projections.
  */
-projection_ast::Projection createProjection(const BSONObj& query, const BSONObj& projObj) {
+projection_ast::Projection createProjection(const BSONObj& query,
+                                            const BSONObj& projObj,
+                                            ProjectionPolicies policies = {}) {
     QueryTestServiceContext serviceCtx;
     auto opCtx = serviceCtx.makeOperationContext();
     const CollatorInterface* collator = nullptr;
@@ -61,7 +63,7 @@ projection_ast::Projection createProjection(const BSONObj& query, const BSONObj&
     ASSERT_OK(statusWithMatcher.getStatus());
     std::unique_ptr<MatchExpression> queryMatchExpr = std::move(statusWithMatcher.getValue());
     projection_ast::Projection res =
-        projection_ast::parse(expCtx, projObj, queryMatchExpr.get(), query, ProjectionPolicies{});
+        projection_ast::parse(expCtx, projObj, queryMatchExpr.get(), query, policies);
 
     return res;
 }
@@ -72,6 +74,12 @@ projection_ast::Projection createProjection(const char* queryStr, const char* pr
     return createProjection(query, projObj);
 }
 
+projection_ast::Projection createFindProjection(const char* queryStr, const char* projStr) {
+    BSONObj query = fromjson(queryStr);
+    BSONObj projObj = fromjson(projStr);
+    return createProjection(query, projObj, ProjectionPolicies::findProjectionPolicies());
+}
+
 void assertInvalidProjection(const char* queryStr, const char* projStr) {
     BSONObj query = fromjson(queryStr);
     BSONObj projObj = fromjson(projStr);
@@ -174,18 +182,18 @@ TEST(QueryProjectionTest, InvalidElemMatchExprProjection) {
 }
 
 TEST(QueryProjectionTest, ValidPositionalOperatorProjections) {
-    createProjection("{a: 1}", "{'a.$': 1}");
-    createProjection("{a: 1}", "{'a.foo.bar.$': 1}");
-    createProjection("{'a.b.c': 1}", "{'a.b.c.$': 1}");
-    createProjection("{'a.b.c': 1}", "{'a.e.f.$': 1}");
-    createProjection("{a: {b: 1}}", "{'a.$': 1}");
-    createProjection("{a: 1, b: 1}}", "{'a.$': 1}");
-    createProjection("{a: 1, b: 1}}", "{'b.$': 1}");
-    createProjection("{$and: [{a: 1}, {b: 1}]}", "{'a.$': 1}");
-    createProjection("{$and: [{a: 1}, {b: 1}]}", "{'b.$': 1}");
-    createProjection("{$or: [{a: 1}, {b: 1}]}", "{'a.$': 1}");
-    createProjection("{$or: [{a: 1}, {b: 1}]}", "{'b.$': 1}");
-    createProjection("{$and: [{$or: [{a: 1}, {$and: [{b: 1}, {c: 1}]}]}]}", "{'c.d.f.$': 1}");
+    createFindProjection("{a: 1}", "{'a.$': 1}");
+    createFindProjection("{a: 1}", "{'a.foo.bar.$': 1}");
+    createFindProjection("{'a.b.c': 1}", "{'a.b.c.$': 1}");
+    createFindProjection("{'a.b.c': 1}", "{'a.e.f.$': 1}");
+    createFindProjection("{a: {b: 1}}", "{'a.$': 1}");
+    createFindProjection("{a: 1, b: 1}}", "{'a.$': 1}");
+    createFindProjection("{a: 1, b: 1}}", "{'b.$': 1}");
+    createFindProjection("{$and: [{a: 1}, {b: 1}]}", "{'a.$': 1}");
+    createFindProjection("{$and: [{a: 1}, {b: 1}]}", "{'b.$': 1}");
+    createFindProjection("{$or: [{a: 1}, {b: 1}]}", "{'a.$': 1}");
+    createFindProjection("{$or: [{a: 1}, {b: 1}]}", "{'b.$': 1}");
+    createFindProjection("{$and: [{$or: [{a: 1}, {$and: [{b: 1}, {c: 1}]}]}]}", "{'c.d.f.$': 1}");
 }
 
 // Some match expressions (eg. $where) do not override MatchExpression::path()
@@ -266,7 +274,8 @@ TEST(QueryProjectionTest, SortKeyMetaProjectionDoesNotRequireDocument) {
 }
 
 TEST(QueryProjectionTest, SortKeyMetaAndSlice) {
-    auto proj = createProjection("{}", "{a: 1, foo: {$meta: 'sortKey'}, _id: 0, b: {$slice: 1}}");
+    auto proj =
+        createFindProjection("{}", "{a: 1, foo: {$meta: 'sortKey'}, _id: 0, b: {$slice: 1}}");
 
     ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
     ASSERT_TRUE(proj.requiresDocument());
@@ -276,8 +285,8 @@ TEST(QueryProjectionTest, SortKeyMetaAndSlice) {
 }
 
 TEST(QueryProjectionTest, SortKeyMetaAndElemMatch) {
-    auto proj =
-        createProjection("{}", "{a: 1, foo: {$meta: 'sortKey'}, _id: 0, b: {$elemMatch: {a: 1}}}");
+    auto proj = createFindProjection(
+        "{}", "{a: 1, foo: {$meta: 'sortKey'}, _id: 0, b: {$elemMatch: {a: 1}}}");
 
     ASSERT_TRUE(proj.metadataDeps()[DocumentMetadataFields::kSortKey]);
     ASSERT_TRUE(proj.requiresDocument());
@@ -331,17 +340,17 @@ TEST(QueryProjectionTest, ExclusionProjectionPreservesNonExcludedFields) {
 }
 
 TEST(QueryProjectionTest, PositionalProjectionDoesNotPreserveField) {
-    auto proj = createProjection("{a: {$elemMatch: {$eq: 0}}}", "{'a.$': 1}");
+    auto proj = createFindProjection("{a: {$elemMatch: {$eq: 0}}}", "{'a.$': 1}");
     ASSERT_FALSE(proj.isFieldRetainedExactly("a"));
 }
 
 TEST(QueryProjectionTest, PositionalProjectionDoesNotPreserveChild) {
-    auto proj = createProjection("{a: {$elemMatch: {$eq: 0}}}", "{'a.$': 1}");
+    auto proj = createFindProjection("{a: {$elemMatch: {$eq: 0}}}", "{'a.$': 1}");
     ASSERT_FALSE(proj.isFieldRetainedExactly("a.b"));
 }
 
 TEST(QueryProjectionTest, PositionalProjectionDoesNotPreserveParent) {
-    auto proj = createProjection("{'a.b': {$elemMatch: {$eq: 0}}}", "{'a.b.$': 1}");
+    auto proj = createFindProjection("{'a.b': {$elemMatch: {$eq: 0}}}", "{'a.b.$': 1}");
     ASSERT_FALSE(proj.isFieldRetainedExactly("a"));
 }
 
@@ -402,7 +411,7 @@ TEST(QueryProjectionTest, IdExclusionWithInclusionProjectionDoesNotPreserveId) {
 }
 
 TEST(QueryProjectionTest, PositionalProjectionDoesNotPreserveFields) {
-    auto proj = createProjection("{a: 1}", "{'a.$': 1}");
+    auto proj = createFindProjection("{a: 1}", "{'a.$': 1}");
     ASSERT_FALSE(proj.isFieldRetainedExactly("b"));
     ASSERT_FALSE(proj.isFieldRetainedExactly("a"));
     ASSERT_FALSE(proj.isFieldRetainedExactly("a.b"));
@@ -410,7 +419,7 @@ TEST(QueryProjectionTest, PositionalProjectionDoesNotPreserveFields) {
 }
 
 TEST(QueryProjectionTest, PositionalProjectionWithIdExclusionDoesNotPreserveFields) {
-    auto proj = createProjection("{a: 1}", "{_id: 0, 'a.$': 1}");
+    auto proj = createFindProjection("{a: 1}", "{_id: 0, 'a.$': 1}");
     ASSERT_FALSE(proj.isFieldRetainedExactly("b"));
     ASSERT_FALSE(proj.isFieldRetainedExactly("a"));
     ASSERT_FALSE(proj.isFieldRetainedExactly("a.b"));
@@ -418,7 +427,7 @@ TEST(QueryProjectionTest, PositionalProjectionWithIdExclusionDoesNotPreserveFiel
 }
 
 TEST(QueryProjectionTest, PositionalProjectionWithIdInclusionPreservesId) {
-    auto proj = createProjection("{a: 1}", "{_id: 1, 'a.$': 1}");
+    auto proj = createFindProjection("{a: 1}", "{_id: 1, 'a.$': 1}");
     ASSERT_FALSE(proj.isFieldRetainedExactly("b"));
     ASSERT_FALSE(proj.isFieldRetainedExactly("a"));
     ASSERT_FALSE(proj.isFieldRetainedExactly("a.b"));
@@ -441,14 +450,14 @@ TEST(QueryProjectionTest, DBRefProjections) {
     createProjection(BSONObj(), BSON("$id" << 1));
     createProjection(BSONObj(), BSON("$ref" << 1));
     // dotted before
-    createProjection("{}", "{'a.$ref': 1}");
-    createProjection("{}", "{'a.$id': 1}");
-    createProjection("{}", "{'a.$db': 1}");
+    createFindProjection("{}", "{'a.$ref': 1}");
+    createFindProjection("{}", "{'a.$id': 1}");
+    createFindProjection("{}", "{'a.$db': 1}");
     // dotted after
     createProjection("{}", "{'$id.a': 1}");
     // position operator on $id
     // $ref and $db hold the collection and database names respectively,
     // so these fields cannot be arrays.
-    createProjection("{'a.$id': {$elemMatch: {x: 1}}}", "{'a.$id.$': 1}");
+    createFindProjection("{'a.$id': {$elemMatch: {x: 1}}}", "{'a.$id.$': 1}");
 }
 }  // namespace