SERVER-42436 Implement transformation from ProjectionAST to projection execution tree

16 files changed, 800 insertions, 276 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 5e35d6f415b..ace0a35332f 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -1017,6 +1017,7 @@ env.Library(
         'exec/plan_stage.cpp',
         'exec/projection.cpp',
         'exec/projection_exec.cpp',
+        'exec/projection_executor.cpp',
         'exec/queued_data_stage.cpp',
         'exec/record_store_fast_count.cpp',
         'exec/requires_all_indices_stage.cpp',
diff --git a/src/mongo/db/exec/SConscript b/src/mongo/db/exec/SConscript
index 8dd4ac5188a..cadce17d459 100644
--- a/src/mongo/db/exec/SConscript
+++ b/src/mongo/db/exec/SConscript
@@ -80,6 +80,7 @@ env.CppUnitTest(
         "find_projection_executor_test.cpp",
         "projection_exec_agg_test.cpp",
         "projection_exec_test.cpp",
+        "projection_executor_test.cpp",
         "queued_data_stage_test.cpp",
         "sort_test.cpp",
         "working_set_test.cpp",
diff --git a/src/mongo/db/exec/projection_executor.cpp b/src/mongo/db/exec/projection_executor.cpp
new file mode 100644
index 00000000000..91c56032cd6
--- /dev/null
+++ b/src/mongo/db/exec/projection_executor.cpp
@@ -0,0 +1,276 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/base/exact_cast.h"
+#include "mongo/db/exec/projection_executor.h"
+#include "mongo/db/pipeline/expression_find_internal.h"
+#include "mongo/db/pipeline/parsed_exclusion_projection.h"
+#include "mongo/db/pipeline/parsed_inclusion_projection.h"
+#include "mongo/db/query/projection_ast_path_tracking_visitor.h"
+#include "mongo/db/query/util/make_data_structure.h"
+
+namespace mongo::projection_executor {
+namespace {
+using ParsedAggregationProjection = parsed_aggregation_projection::ParsedAggregationProjection;
+using ParsedInclusionProjection = parsed_aggregation_projection::ParsedInclusionProjection;
+using ParsedExclusionProjection = parsed_aggregation_projection::ParsedExclusionProjection;
+
+constexpr auto kInclusion = projection_ast::ProjectType::kInclusion;
+constexpr auto kExclusion = projection_ast::ProjectType::kExclusion;
+constexpr auto kProjectionPostImageVarName =
+    parsed_aggregation_projection::ParsedAggregationProjection::kProjectionPostImageVarName;
+
+/**
+ * Holds data used to built a projection executor while walking an AST tree. This struct is attached
+ * to 'PathTrackingVisitorContext' and can be accessed by projection AST visitors to track the
+ * current context.
+ */
+template <typename Executor>
+struct ProjectionExecutorVisitorData {
+    // A projection executor returned upon completion of AST traversal.
+    std::unique_ptr<Executor> executor;
+    boost::intrusive_ptr<ExpressionContext> expCtx;
+    // A root replacement expression to be attached to the projection executor, if any. If there
+    // are multiple root replacement expressions in the AST, they will be chained together, so that
+    // one expression becomes an input to another.
+    boost::intrusive_ptr<Expression> rootReplacementExpression;
+
+    auto rootNode() const {
+        return executor->getRoot();
+    }
+
+    void setRootReplacementExpression(boost::intrusive_ptr<Expression> expr) {
+        rootReplacementExpression = expr;
+        executor->setRootReplacementExpression(rootReplacementExpression);
+    }
+};
+
+template <typename Executor>
+using ProjectionExecutorVisitorContext =
+    projection_ast::PathTrackingVisitorContext<ProjectionExecutorVisitorData<Executor>>;
+
+template <typename Executor>
+auto makeProjectionPreImageExpression(const ProjectionExecutorVisitorData<Executor>& data) {
+    return ExpressionFieldPath::parse(data.expCtx, "$$ROOT", data.expCtx->variablesParseState);
+}
+
+template <typename Executor>
+auto makeProjectionPostImageExpression(const ProjectionExecutorVisitorData<Executor>& data) {
+    return data.rootReplacementExpression
+        ? data.rootReplacementExpression
+        : ExpressionFieldPath::parse(
+              data.expCtx, "$$" + kProjectionPostImageVarName, data.expCtx->variablesParseState);
+}
+
+/**
+ * Creates a find()-style positional expression from the given AST 'node' to be applied to the
+ * 'path' on the input document. If the visitor 'data' already contains a root replacement
+ * expression, it will be used as an input operand to the new root replacement expression, otherwise
+ * a field path expressions will be created to access a projection post-image document.
+ */
+template <typename Executor>
+auto createFindPositionalExpression(projection_ast::ProjectionPositionalASTNode* node,
+                                    const ProjectionExecutorVisitorData<Executor>& data,
+                                    const FieldPath& path) {
+    invariant(node);
+
+    const auto& children = node->children();
+    invariant(children.size() == 1UL);
+
+    auto matchExprNode =
+        exact_pointer_cast<projection_ast::MatchExpressionASTNode*>(children[0].get());
+    invariant(matchExprNode);
+
+    return make_intrusive<ExpressionInternalFindPositional>(data.expCtx,
+                                                            makeProjectionPreImageExpression(data),
+                                                            makeProjectionPostImageExpression(data),
+                                                            path,
+                                                            matchExprNode->matchExpression());
+}
+
+/**
+ * Creates a find()-style $slice expression from the given AST 'node' to be applied to the
+ * 'path' on the input document. If the visitor 'data' already contains a root replacement
+ * expression, it will be used as an input operand to the new root replacement expression, otherwise
+ * a field path expressions will be created to access a projection post-image document.
+ */
+template <typename Executor>
+auto createFindSliceExpression(projection_ast::ProjectionSliceASTNode* node,
+                               const ProjectionExecutorVisitorData<Executor>& data,
+                               const FieldPath& path) {
+    invariant(node);
+
+    return make_intrusive<ExpressionInternalFindSlice>(
+        data.expCtx, makeProjectionPostImageExpression(data), path, node->skip(), node->limit());
+}
+
+/**
+ * Creates a find()-style $elemMatch expression from the given AST 'node' to be applied at the
+ * 'path' on the input document.
+ */
+template <typename Executor>
+auto createFindElemMatchExpression(projection_ast::ProjectionElemMatchASTNode* node,
+                                   const ProjectionExecutorVisitorData<Executor>& data,
+                                   const FieldPath& path) {
+    invariant(node);
+
+    const auto& children = node->children();
+    invariant(children.size() == 1UL);
+
+    auto matchExprNode =
+        exact_pointer_cast<projection_ast::MatchExpressionASTNode*>(children[0].get());
+    invariant(matchExprNode);
+
+    return make_intrusive<ExpressionInternalFindElemMatch>(data.expCtx,
+                                                           makeProjectionPreImageExpression(data),
+                                                           path,
+                                                           matchExprNode->matchExpression());
+}
+
+/**
+ * A projection AST visitor which creates inclusion or exclusion nodes in the projection execution
+ * tree by calling corresponding 'addProjectionForPath()' or 'addExpressionForPath()' on the root
+ * node of the tree while traversing the AST. If the AST contains root-replacement expressions, they
+ * will be attached to the projection executor.
+ *
+ * To track the current path in the projection, this visitor should be used with
+ * 'PathTrackingWalker' which will help to maintain the current path via
+ * 'PathTrackingVisitorContext'.
+ */
+template <typename Executor>
+class ProjectionExecutorVisitor final : public projection_ast::ProjectionASTVisitor {
+public:
+    ProjectionExecutorVisitor(ProjectionExecutorVisitorContext<Executor>* context)
+        : _context{context} {
+        invariant(_context);
+    }
+
+    void visit(projection_ast::ProjectionPositionalASTNode* node) final {
+        constexpr auto isInclusion = std::is_same_v<Executor, ParsedInclusionProjection>;
+        invariant(isInclusion);
+
+        const auto& path = _context->fullPath();
+        auto& userData = _context->data();
+
+        userData.rootNode()->addProjectionForPath(path.fullPath());
+        userData.setRootReplacementExpression(createFindPositionalExpression(node, userData, path));
+    }
+
+    void visit(projection_ast::ProjectionSliceASTNode* node) final {
+        const auto& path = _context->fullPath();
+        auto& userData = _context->data();
+
+        // A $slice expression can be applied to an exclusion projection. In this case we don't need
+        // to project out the path to which $slice is applied, since it will already be included
+        // into the output document.
+        if constexpr (std::is_same_v<Executor, ParsedInclusionProjection>) {
+            userData.rootNode()->addProjectionForPath(path.fullPath());
+        }
+
+        userData.setRootReplacementExpression(createFindSliceExpression(node, userData, path));
+    }
+
+    void visit(projection_ast::ProjectionElemMatchASTNode* node) final {
+        const auto& path = _context->fullPath();
+        const auto& userData = _context->data();
+
+        userData.rootNode()->addExpressionForPath(
+            path.fullPath(), createFindElemMatchExpression(node, userData, path));
+    }
+
+    void visit(projection_ast::ExpressionASTNode* node) final {
+        const auto& path = _context->fullPath();
+        const auto& userData = _context->data();
+
+        userData.rootNode()->addExpressionForPath(path.fullPath(), node->expression());
+    }
+
+    void visit(projection_ast::BooleanConstantASTNode* node) final {
+        const auto& path = _context->fullPath();
+        const auto& userData = _context->data();
+
+        // In an inclusion projection only the _id field can be excluded from the result document.
+        // If this is the case, then we don't need to include the field into the projection.
+        if constexpr (std::is_same_v<Executor, ParsedInclusionProjection>) {
+            const auto isIdField = path == "_id";
+            if (isIdField && !node->value()) {
+                return;
+            }
+            // In inclusion projection only _id field can be excluded, make sure this is the case.
+            invariant(!isIdField || node->value());
+        }
+
+        userData.rootNode()->addProjectionForPath(path.fullPath());
+    }
+
+    void visit(projection_ast::ProjectionPathASTNode* node) final {}
+    void visit(projection_ast::MatchExpressionASTNode* node) final {}
+
+private:
+    ProjectionExecutorVisitorContext<Executor>* _context;
+};
+
+/**
+ * A helper function which creates a 'ProjectionExecutorWalker' to walk a projection AST,
+ * starting at the node 'root', and build a projection executor of the specified type
+ * 'Executor'.
+ */
+template <typename Executor>
+auto buildProjectionExecutor(boost::intrusive_ptr<ExpressionContext> expCtx,
+                             projection_ast::ProjectionPathASTNode* root,
+                             const ProjectionPolicies policies) {
+    ProjectionExecutorVisitorContext<Executor> context{
+        {std::make_unique<Executor>(expCtx, policies), expCtx}};
+    ProjectionExecutorVisitor<Executor> executorVisitor{&context};
+    projection_ast::PathTrackingWalker walker{&context, {&executorVisitor}, {}};
+    projection_ast_walker::walk(&walker, root);
+    return std::move(context.data().executor);
+}
+}  // namespace
+
+std::unique_ptr<ParsedAggregationProjection> buildProjectionExecutor(
+    boost::intrusive_ptr<ExpressionContext> expCtx,
+    projection_ast::Projection* projection,
+    const ProjectionPolicies policies) {
+    invariant(projection);
+
+    switch (projection->type()) {
+        case kInclusion:
+            return buildProjectionExecutor<ParsedInclusionProjection>(
+                expCtx, projection->root(), policies);
+        case kExclusion:
+            return buildProjectionExecutor<ParsedExclusionProjection>(
+                expCtx, projection->root(), policies);
+        default:
+            MONGO_UNREACHABLE;
+    }
+}
+}  // namespace mongo::projection_executor
diff --git a/src/mongo/db/exec/projection_executor.h b/src/mongo/db/exec/projection_executor.h
new file mode 100644
index 00000000000..98e7fe89bfd
--- /dev/null
+++ b/src/mongo/db/exec/projection_executor.h
@@ -0,0 +1,45 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/pipeline/parsed_aggregation_projection.h"
+#include "mongo/db/query/projection_ast.h"
+#include "mongo/db/query/projection_ast_walker.h"
+
+namespace mongo::projection_executor {
+/**
+ * Builds a projection execution tree from the given 'projection' and using the given projection
+ * 'policies' by walking an AST tree starting at the root node stored within the 'projection'.
+ */
+std::unique_ptr<parsed_aggregation_projection::ParsedAggregationProjection> buildProjectionExecutor(
+    boost::intrusive_ptr<ExpressionContext> expCtx,
+    projection_ast::Projection* projection,
+    ProjectionPolicies policies);
+}  // namespace mongo::projection_executor
diff --git a/src/mongo/db/exec/projection_executor_test.cpp b/src/mongo/db/exec/projection_executor_test.cpp
new file mode 100644
index 00000000000..bea4dbbc710
--- /dev/null
+++ b/src/mongo/db/exec/projection_executor_test.cpp
@@ -0,0 +1,163 @@
+/**
+ *    Copyright (C) 2019-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/exec/document_value/document_value_test_util.h"
+#include "mongo/db/exec/projection_executor.h"
+#include "mongo/db/pipeline/aggregation_context_fixture.h"
+#include "mongo/db/query/projection_ast_util.h"
+#include "mongo/db/query/projection_parser.h"
+#include "mongo/unittest/unittest.h"
+
+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());
+        }
+
+        return projection_ast::parse(
+            getExpCtx(), proj, expr.getValue().get(), match.get_value_or({}), {});
+    }
+};
+
+TEST_F(ProjectionExecutorTest, CanProjectInclusionWithIdPath) {
+    auto projWithId = parseWithDefaultPolicies(fromjson("{a: 1, _id: 1}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &projWithId, {});
+    ASSERT_DOCUMENT_EQ(Document{fromjson("{_id: 123, a: 'abc'}")},
+                       executor->applyTransformation(
+                           Document{fromjson("{_id: 123, a: 'abc', b: 'def', c: 'ghi'}")}));
+
+    auto projWithoutId = parseWithDefaultPolicies(fromjson("{a: 1, _id: 0}"));
+    executor = buildProjectionExecutor(getExpCtx(), &projWithoutId, {});
+    ASSERT_DOCUMENT_EQ(Document{fromjson("{a: 'abc'}")},
+                       executor->applyTransformation(
+                           Document{fromjson("{_id: 123, a: 'abc', b: 'def', c: 'ghi'}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectInclusionUndottedPath) {
+    auto proj = parseWithDefaultPolicies(fromjson("{a: 1, b: 1}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(
+        Document{fromjson("{a: 'abc', b: 'def'}")},
+        executor->applyTransformation(Document{fromjson("{a: 'abc', b: 'def', c: 'ghi'}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectInclusionDottedPath) {
+    auto proj = parseWithDefaultPolicies(fromjson("{'a.b': 1, 'a.d': 1}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(
+        Document{fromjson("{a: {b: 'abc', d: 'ghi'}}")},
+        executor->applyTransformation(Document{fromjson("{a: {b: 'abc', c: 'def', d: 'ghi'}}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectExpression) {
+    auto proj = parseWithDefaultPolicies(fromjson("{c: {$add: ['$a', '$b']}}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(Document{fromjson("{c: 3}")},
+                       executor->applyTransformation(Document{fromjson("{a: 1, b: 2}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectExclusionWithIdPath) {
+    auto projWithoutId = parseWithDefaultPolicies(fromjson("{a: 0, _id: 0}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &projWithoutId, {});
+    ASSERT_DOCUMENT_EQ(Document{fromjson("{b: 'def', c: 'ghi'}")},
+                       executor->applyTransformation(
+                           Document{fromjson("{_id: 123, a: 'abc', b: 'def', c: 'ghi'}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectExclusionUndottedPath) {
+    auto proj = parseWithDefaultPolicies(fromjson("{a: 0, b: 0}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(
+        Document{fromjson("{c: 'ghi'}")},
+        executor->applyTransformation(Document{fromjson("{a: 'abc', b: 'def', c: 'ghi'}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectExclusionDottedPath) {
+    auto proj = parseWithDefaultPolicies(fromjson("{'a.b': 0, 'a.d': 0}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(
+        Document{fromjson("{a: {c: 'def'}}")},
+        executor->applyTransformation(Document{fromjson("{a: {b: 'abc', c: 'def', d: 'ghi'}}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectFindPositional) {
+    auto proj = parseWithDefaultPolicies(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 executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(
+        Document{fromjson("{a: [{b: 3}]}")},
+        executor->applyTransformation(Document{fromjson("{a: [{b: 1}, {b: 2}, {b: 3}]}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectFindElemMatchWithExclusion) {
+    auto proj = parseWithDefaultPolicies(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{
+                           fromjson("{a: [{b: 1}, {b: 2}, {b: 3}], c: 'abc', d: 'def'}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectFindSliceWithInclusion) {
+    auto proj = parseWithDefaultPolicies(fromjson("{'a.b': {$slice: [1,2]}, c: 1}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(
+        Document{fromjson("{a: {b: [2,3]}, c: 'abc'}")},
+        executor->applyTransformation(Document{fromjson("{a: {b: [1,2,3,4]}, c: 'abc'}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectFindSliceWithExclusion) {
+    auto proj = parseWithDefaultPolicies(fromjson("{'a.b': {$slice: [1,2]}, c: 0}"));
+    auto executor = buildProjectionExecutor(getExpCtx(), &proj, {});
+    ASSERT_DOCUMENT_EQ(
+        Document{fromjson("{a: {b: [2,3]}}")},
+        executor->applyTransformation(Document{fromjson("{a: {b: [1,2,3,4]}, c: 'abc'}")}));
+}
+
+TEST_F(ProjectionExecutorTest, CanProjectFindSliceAndPositional) {
+    auto proj = parseWithDefaultPolicies(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]}")},
+        executor->applyTransformation(Document{fromjson("{a: {b: [1,2,3,4]}, c: [5,6,7]}")}));
+}
+}  // namespace mongo::projection_executor
diff --git a/src/mongo/db/matcher/copyable_match_expression.h b/src/mongo/db/matcher/copyable_match_expression.h
index e9a9bb16346..3e7faacadda 100644
--- a/src/mongo/db/matcher/copyable_match_expression.h
+++ b/src/mongo/db/matcher/copyable_match_expression.h
@@ -56,12 +56,15 @@ public:
                             std::unique_ptr<const ExtensionsCallback> extensionsCallback =
                                 std::make_unique<ExtensionsCallbackNoop>(),
                             MatchExpressionParser::AllowedFeatureSet allowedFeatures =
-                                MatchExpressionParser::kDefaultSpecialFeatures)
+                                MatchExpressionParser::kDefaultSpecialFeatures,
+                            bool optimizeExpression = false)
         : _matchAST(matchAST), _extensionsCallback(std::move(extensionsCallback)) {
         StatusWithMatchExpression parseResult =
             MatchExpressionParser::parse(_matchAST, expCtx, *_extensionsCallback, allowedFeatures);
         uassertStatusOK(parseResult.getStatus());
-        _matchExpr = std::move(parseResult.getValue());
+        _matchExpr = optimizeExpression
+            ? MatchExpression::optimize(std::move(parseResult.getValue()))
+            : std::move(parseResult.getValue());
     }
 
     /**
@@ -75,7 +78,7 @@ public:
      * Overload * so that CopyableMatchExpression can be dereferenced as if it were a pointer to the
      * underlying MatchExpression.
      */
-    const MatchExpression& operator*() {
+    const MatchExpression& operator*() const {
         return *_matchExpr;
     }
 
@@ -83,7 +86,7 @@ public:
      * Overload -> so that CopyableMatchExpression can be dereferenced as if it were a pointer to
      * the underlying MatchExpression.
      */
-    const MatchExpression* operator->() {
+    const MatchExpression* operator->() const {
         return &(*_matchExpr);
     }
 
diff --git a/src/mongo/db/pipeline/expression_find_internal.h b/src/mongo/db/pipeline/expression_find_internal.h
index 61a696c5d9d..794d5ebc912 100644
--- a/src/mongo/db/pipeline/expression_find_internal.h
+++ b/src/mongo/db/pipeline/expression_find_internal.h
@@ -32,6 +32,7 @@
 #include <fmt/format.h>
 
 #include "mongo/db/exec/find_projection_executor.h"
+#include "mongo/db/matcher/copyable_match_expression.h"
 #include "mongo/db/pipeline/expression.h"
 
 namespace mongo {
@@ -47,10 +48,10 @@ public:
                                      boost::intrusive_ptr<Expression> preImageExpr,
                                      boost::intrusive_ptr<Expression> postImageExpr,
                                      FieldPath path,
-                                     const MatchExpression* matchExpr)
+                                     CopyableMatchExpression matchExpr)
         : Expression{expCtx, {preImageExpr, postImageExpr}},
           _path{std::move(path)},
-          _matchExpr{matchExpr} {}
+          _matchExpr{std::move(matchExpr)} {}
 
     Value evaluate(const Document& root, Variables* variables) const final {
         using namespace fmt::literals;
@@ -84,17 +85,33 @@ public:
         return {{_path.front().toString()}, {}};
     }
 
+    boost::intrusive_ptr<Expression> optimize() final {
+        for (const auto& child : _children) {
+            child->optimize();
+        }
+        // SERVER-43740: ideally we'd want to optimize '_matchExpr' here as well. However, given
+        // that the match expression is stored as a shared copyable expression in this class, and
+        // 'MatchExpression::optimize()' takes and returns a unique pointer on a match expression,
+        // there is no easy way to replace a copyable match expression with its optimized
+        // equivalent. So, for now we will assume that the copyable match expression is passed to
+        // this expression already optimized. Once we have MatchExpression and Expression combined,
+        // we should revisit this code and make sure that 'optimized()' method is called on
+        // _matchExpr.
+        return this;
+    }
+
 protected:
     void _doAddDependencies(DepsTracker* deps) const final {
-        _children[0]->addDependencies(deps);
-        _children[1]->addDependencies(deps);
-        deps->needWholeDocument = true;
+        for (const auto& child : _children) {
+            child->addDependencies(deps);
+        }
         _matchExpr->addDependencies(deps);
+        deps->needWholeDocument = true;
     }
 
 private:
     const FieldPath _path;
-    const MatchExpression* _matchExpr;
+    const CopyableMatchExpression _matchExpr;
 };
 
 /**
@@ -139,8 +156,17 @@ public:
         return {{_path.front().toString()}, {}};
     }
 
+    boost::intrusive_ptr<Expression> optimize() final {
+        invariant(_children.size() == 1ul);
+
+        _children[0]->optimize();
+        return this;
+    }
+
 protected:
     void _doAddDependencies(DepsTracker* deps) const final {
+        invariant(_children.size() == 1ul);
+
         _children[0]->addDependencies(deps);
         deps->needWholeDocument = true;
     }
@@ -160,7 +186,7 @@ public:
     ExpressionInternalFindElemMatch(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                                     boost::intrusive_ptr<Expression> child,
                                     FieldPath path,
-                                    std::unique_ptr<MatchExpression> matchExpr)
+                                    CopyableMatchExpression matchExpr)
         : Expression{expCtx, {child}}, _path{std::move(path)}, _matchExpr{std::move(matchExpr)} {}
 
     Value evaluate(const Document& root, Variables* variables) const final {
@@ -183,15 +209,32 @@ public:
         MONGO_UNREACHABLE;
     }
 
+    boost::intrusive_ptr<Expression> optimize() final {
+        invariant(_children.size() == 1ul);
+
+        _children[0]->optimize();
+        // SERVER-43740: ideally we'd want to optimize '_matchExpr' here as well. However, given
+        // that the match expression is stored as a shared copyable expression in this class, and
+        // 'MatchExpression::optimize()' takes and returns a unique pointer on a match expression,
+        // there is no easy way to replace a copyable match expression with its optimized
+        // equivalent. So, for now we will assume that the copyable match expression is passed to
+        // this expression already optimized. Once we have MatchExpression and Expression combined,
+        // we should revisit this code and make sure that 'optimized()' method is called on
+        // _matchExpr.
+        return this;
+    }
+
 protected:
     void _doAddDependencies(DepsTracker* deps) const final {
+        invariant(_children.size() == 1ul);
+
         _children[0]->addDependencies(deps);
-        deps->needWholeDocument = true;
         _matchExpr->addDependencies(deps);
+        deps->needWholeDocument = true;
     }
 
 private:
     const FieldPath _path;
-    const std::unique_ptr<MatchExpression> _matchExpr;
+    const CopyableMatchExpression _matchExpr;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/expression_find_internal_test.cpp b/src/mongo/db/pipeline/expression_find_internal_test.cpp
index 638bb76732e..0fe1e34e9c5 100644
--- a/src/mongo/db/pipeline/expression_find_internal_test.cpp
+++ b/src/mongo/db/pipeline/expression_find_internal_test.cpp
@@ -49,14 +49,18 @@ auto defineAndSetProjectionPostImageVariable(boost::intrusive_ptr<ExpressionCont
 
 class ExpressionInternalFindPositionalTest : public AggregationContextFixture {
 protected:
-    auto createExpression(const MatchExpression* matchExpr, const std::string& path) {
+    auto createExpression(BSONObj matchSpec, const std::string& path) {
+        auto matchExpr = CopyableMatchExpression{matchSpec,
+                                                 getExpCtx(),
+                                                 std::make_unique<ExtensionsCallbackNoop>(),
+                                                 MatchExpressionParser::kBanAllSpecialFeatures};
         auto expr = make_intrusive<ExpressionInternalFindPositional>(
             getExpCtx(),
             ExpressionFieldPath::parse(getExpCtx(), "$$ROOT", getExpCtx()->variablesParseState),
             ExpressionFieldPath::parse(
                 getExpCtx(), "$$" + kProjectionPostImageVarName, getExpCtx()->variablesParseState),
             path,
-            matchExpr);
+            std::move(matchExpr));
         return expr;
     }
 };
@@ -77,7 +81,12 @@ protected:
 
 class ExpressionInternalFindElemMatchTest : public AggregationContextFixture {
 protected:
-    auto createExpression(std::unique_ptr<MatchExpression> matchExpr, const std::string& path) {
+    auto createExpression(BSONObj matchSpec, const std::string& path) {
+        auto matchExpr = CopyableMatchExpression{matchSpec,
+                                                 getExpCtx(),
+                                                 std::make_unique<ExtensionsCallbackNoop>(),
+                                                 MatchExpressionParser::kBanAllSpecialFeatures};
+
         return make_intrusive<ExpressionInternalFindElemMatch>(
             getExpCtx(),
             ExpressionFieldPath::parse(getExpCtx(), "$$ROOT", getExpCtx()->variablesParseState),
@@ -90,9 +99,7 @@ TEST_F(ExpressionInternalFindPositionalTest, AppliesProjectionToPostImage) {
     defineAndSetProjectionPostImageVariable(getExpCtx(),
                                             Value{fromjson("{bar: 1, foo: [1,2,6,10]}")});
 
-    auto matchSpec = fromjson("{bar: 1, foo: {$gte: 5}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
-    auto expr = createExpression(matchExpr.get(), "foo");
+    auto expr = createExpression(fromjson("{bar: 1, foo: {$gte: 5}}"), "foo");
 
     ASSERT_DOCUMENT_EQ(
         Document{fromjson("{bar:1, foo: [6]}")},
@@ -103,9 +110,7 @@ TEST_F(ExpressionInternalFindPositionalTest, AppliesProjectionToPostImage) {
 TEST_F(ExpressionInternalFindPositionalTest, RecordsProjectionDependencies) {
     auto varId = defineAndSetProjectionPostImageVariable(
         getExpCtx(), Value{fromjson("{bar: 1, foo: [1,2,6,10]}")});
-    auto matchSpec = fromjson("{bar: 1, foo: {$gte: 5}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
-    auto expr = createExpression(matchExpr.get(), "foo");
+    auto expr = createExpression(fromjson("{bar: 1, foo: {$gte: 5}}"), "foo");
 
     DepsTracker deps;
     expr->addDependencies(&deps);
@@ -122,9 +127,7 @@ TEST_F(ExpressionInternalFindPositionalTest, AddsArrayUndottedPathToComputedPath
     defineAndSetProjectionPostImageVariable(getExpCtx(),
                                             Value{fromjson("{bar: 1, foo: [1,2,6,10]}")});
 
-    auto matchSpec = fromjson("{bar: 1, foo: {$gte: 5}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
-    auto expr = createExpression(matchExpr.get(), "foo");
+    auto expr = createExpression(fromjson("{bar: 1, foo: {$gte: 5}}"), "foo");
 
     DepsTracker deps;
     auto computedPaths = expr->getComputedPaths({});
@@ -139,9 +142,7 @@ TEST_F(ExpressionInternalFindPositionalTest,
     defineAndSetProjectionPostImageVariable(getExpCtx(),
                                             Value{fromjson("{bar: 1, foo: [1,2,6,10]}")});
 
-    auto matchSpec = fromjson("{bar: 1, 'foo.bar': {$gte: 5}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
-    auto expr = createExpression(matchExpr.get(), "foo.bar");
+    auto expr = createExpression(fromjson("{bar: 1, 'foo.bar': {$gte: 5}}"), "foo.bar");
 
     DepsTracker deps;
     auto computedPaths = expr->getComputedPaths({});
@@ -206,9 +207,7 @@ TEST_F(ExpressionInternalFindSliceTest, AddsTopLevelFieldOfArrayDottedPathToComp
 }
 
 TEST_F(ExpressionInternalFindElemMatchTest, AppliesProjectionToRootDocument) {
-    auto matchSpec = fromjson("{foo: {$elemMatch: {bar: {$gte: 5}}}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
-    auto expr = createExpression(std::move(matchExpr), "foo");
+    auto expr = createExpression(fromjson("{foo: {$elemMatch: {bar: {$gte: 5}}}}"), "foo");
 
     ASSERT_VALUE_EQ(Document{fromjson("{foo: [{bar: 6, z: 6}]}")}["foo"],
                     expr->evaluate(Document{fromjson("{foo: [{bar: 1, z: 1}, {bar: 2, z: 2}, "
@@ -217,9 +216,7 @@ TEST_F(ExpressionInternalFindElemMatchTest, AppliesProjectionToRootDocument) {
 }
 
 TEST_F(ExpressionInternalFindElemMatchTest, RecordsProjectionDependencies) {
-    auto matchSpec = fromjson("{foo: {$elemMatch: {bar: {$gte: 5}}}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
-    auto expr = createExpression(std::move(matchExpr), "foo");
+    auto expr = createExpression(fromjson("{foo: {$elemMatch: {bar: {$gte: 5}}}}"), "foo");
 
     DepsTracker deps;
     expr->addDependencies(&deps);
diff --git a/src/mongo/db/pipeline/parsed_exclusion_projection.h b/src/mongo/db/pipeline/parsed_exclusion_projection.h
index 93c754dc4d3..b25e97329d7 100644
--- a/src/mongo/db/pipeline/parsed_exclusion_projection.h
+++ b/src/mongo/db/pipeline/parsed_exclusion_projection.h
@@ -52,11 +52,7 @@ namespace parsed_aggregation_projection {
 class ExclusionNode final : public ProjectionNode {
 public:
     ExclusionNode(ProjectionPolicies policies, std::string pathToNode = "")
-        : ProjectionNode(policies, std::move(pathToNode)) {
-        // Computed fields are not supported by exclusion projections.
-        invariant(_policies.computedFieldsPolicy ==
-                  ProjectionPolicies::ComputedFieldsPolicy::kBanComputedFields);
-    }
+        : ProjectionNode(policies, std::move(pathToNode)) {}
 
     ExclusionNode* addOrGetChild(const std::string& field) {
         return static_cast<ExclusionNode*>(ProjectionNode::addOrGetChild(field));
@@ -93,19 +89,18 @@ class ParsedExclusionProjection : public ParsedAggregationProjection {
 public:
     ParsedExclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
                               ProjectionPolicies policies)
-        : ParsedAggregationProjection(
-              expCtx,
-              {policies.idPolicy,
-               policies.arrayRecursionPolicy,
-               ProjectionPolicies::ComputedFieldsPolicy::kBanComputedFields}),
-          _root(new ExclusionNode(_policies)) {}
+        : ParsedAggregationProjection(expCtx, policies), _root(new ExclusionNode(_policies)) {}
 
     TransformerType getType() const final {
         return TransformerType::kExclusionProjection;
     }
 
-    const ExclusionNode& getRoot() const {
-        return *_root;
+    const ExclusionNode* getRoot() const {
+        return _root.get();
+    }
+
+    ExclusionNode* getRoot() {
+        return _root.get();
     }
 
     Document serializeTransformation(
diff --git a/src/mongo/db/pipeline/parsed_find_projection_test.cpp b/src/mongo/db/pipeline/parsed_find_projection_test.cpp
index 971d57221db..66fd72ff24e 100644
--- a/src/mongo/db/pipeline/parsed_find_projection_test.cpp
+++ b/src/mongo/db/pipeline/parsed_find_projection_test.cpp
@@ -46,14 +46,17 @@ protected:
                          const std::string& path,
                          const Document& input) {
         auto proj = ParsedAggregationProjection::create(getExpCtx(), projSpec, {});
-        auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
+        auto matchExpr = CopyableMatchExpression{matchSpec,
+                                                 getExpCtx(),
+                                                 std::make_unique<ExtensionsCallbackNoop>(),
+                                                 MatchExpressionParser::kBanAllSpecialFeatures};
         auto expr = make_intrusive<ExpressionInternalFindPositional>(
             getExpCtx(),
             ExpressionFieldPath::parse(getExpCtx(), "$$ROOT", getExpCtx()->variablesParseState),
             ExpressionFieldPath::parse(
                 getExpCtx(), "$$" + kProjectionPostImageVarName, getExpCtx()->variablesParseState),
             path,
-            matchExpr.get());
+            std::move(matchExpr));
         proj->setRootReplacementExpression(expr);
         return proj->applyTransformation(input);
     }
@@ -103,15 +106,18 @@ TEST_F(PositionalProjectionExecutionTest, AppliesProjectionToPreImage) {
 
 TEST_F(PositionalProjectionExecutionTest, ShouldAddInclusionFieldsAndWholeDocumentToDependencies) {
     auto proj = ParsedAggregationProjection::create(getExpCtx(), fromjson("{bar: 1, _id: 0}"), {});
-    auto match = fromjson("{bar: 1, 'foo.bar': {$gte: 5}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(match, getExpCtx()));
+    auto matchSpec = fromjson("{bar: 1, 'foo.bar': {$gte: 5}}");
+    auto matchExpr = CopyableMatchExpression{matchSpec,
+                                             getExpCtx(),
+                                             std::make_unique<ExtensionsCallbackNoop>(),
+                                             MatchExpressionParser::kBanAllSpecialFeatures};
     auto expr = make_intrusive<ExpressionInternalFindPositional>(
         getExpCtx(),
         ExpressionFieldPath::parse(getExpCtx(), "$$ROOT", getExpCtx()->variablesParseState),
         ExpressionFieldPath::parse(
             getExpCtx(), "$$" + kProjectionPostImageVarName, getExpCtx()->variablesParseState),
         "foo.bar",
-        matchExpr.get());
+        std::move(matchExpr));
     proj->setRootReplacementExpression(expr);
 
     DepsTracker deps;
@@ -125,15 +131,18 @@ TEST_F(PositionalProjectionExecutionTest, ShouldAddInclusionFieldsAndWholeDocume
 
 TEST_F(PositionalProjectionExecutionTest, ShouldConsiderAllPathsAsModified) {
     auto proj = ParsedAggregationProjection::create(getExpCtx(), fromjson("{bar: 1, _id: 0}"), {});
-    auto match = fromjson("{bar: 1, 'foo.bar': {$gte: 5}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(match, getExpCtx()));
+    auto matchSpec = fromjson("{bar: 1, 'foo.bar': {$gte: 5}}");
+    auto matchExpr = CopyableMatchExpression{matchSpec,
+                                             getExpCtx(),
+                                             std::make_unique<ExtensionsCallbackNoop>(),
+                                             MatchExpressionParser::kBanAllSpecialFeatures};
     auto expr = make_intrusive<ExpressionInternalFindPositional>(
         getExpCtx(),
         ExpressionFieldPath::parse(getExpCtx(), "$$ROOT", getExpCtx()->variablesParseState),
         ExpressionFieldPath::parse(
             getExpCtx(), "$$" + kProjectionPostImageVarName, getExpCtx()->variablesParseState),
         "foo.bar",
-        matchExpr.get());
+        std::move(matchExpr));
     proj->setRootReplacementExpression(expr);
 
     auto modifiedPaths = proj->getModifiedPaths();
@@ -166,14 +175,17 @@ TEST_F(SliceProjectionExecutionTest, AppliesProjectionToPostImage) {
 TEST_F(SliceProjectionExecutionTest, CanApplySliceAndPositionalProjectionsTogether) {
     auto proj = ParsedAggregationProjection::create(getExpCtx(), fromjson("{foo: 1, bar: 1}"), {});
     auto matchSpec = fromjson("{foo: {$gte: 3}}");
-    auto matchExpr = uassertStatusOK(MatchExpressionParser::parse(matchSpec, getExpCtx()));
+    auto matchExpr = CopyableMatchExpression{matchSpec,
+                                             getExpCtx(),
+                                             std::make_unique<ExtensionsCallbackNoop>(),
+                                             MatchExpressionParser::kBanAllSpecialFeatures};
     auto positionalExpr = make_intrusive<ExpressionInternalFindPositional>(
         getExpCtx(),
         ExpressionFieldPath::parse(getExpCtx(), "$$ROOT", getExpCtx()->variablesParseState),
         ExpressionFieldPath::parse(
             getExpCtx(), "$$" + kProjectionPostImageVarName, getExpCtx()->variablesParseState),
         "foo",
-        matchExpr.get());
+        std::move(matchExpr));
     auto sliceExpr =
         make_intrusive<ExpressionInternalFindSlice>(getExpCtx(), positionalExpr, "bar", 1, 1);
     proj->setRootReplacementExpression(sliceExpr);
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection.h b/src/mongo/db/pipeline/parsed_inclusion_projection.h
index da316e8bbdd..a3f205b90b1 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection.h
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection.h
@@ -102,8 +102,12 @@ public:
         return TransformerType::kInclusionProjection;
     }
 
-    const InclusionNode& getRoot() const {
-        return *_root;
+    const InclusionNode* getRoot() const {
+        return _root.get();
+    }
+
+    InclusionNode* getRoot() {
+        return _root.get();
     }
 
     /**
diff --git a/src/mongo/db/query/projection.cpp b/src/mongo/db/query/projection.cpp
index d70c5b5894e..6ad92a8859f 100644
--- a/src/mongo/db/query/projection.cpp
+++ b/src/mongo/db/query/projection.cpp
@@ -32,10 +32,27 @@
 #include "mongo/base/exact_cast.h"
 #include "mongo/db/query/projection_ast_path_tracking_visitor.h"
 #include "mongo/db/query/projection_ast_walker.h"
+#include "mongo/db/query/util/make_data_structure.h"
 
 namespace mongo {
 namespace projection_ast {
 namespace {
+/**
+ * Holds data used for dependency analysis while walking an AST tree. This struct is attached to
+ * 'PathTrackingVisitorContext' and can be accessed by projection AST visitors to track the current
+ * context.
+ */
+struct DepsAnalysisData {
+    DepsTracker fieldDependencyTracker{DepsTracker::kAllMetadataAvailable};
+
+    void addRequiredField(const std::string& fieldName) {
+        fieldDependencyTracker.fields.insert(fieldName);
+    }
+
+    std::vector<std::string> requiredFields() const {
+        return {fieldDependencyTracker.fields.begin(), fieldDependencyTracker.fields.end()};
+    }
+};
 
 /**
  * Does "broad" analysis on the projection, about whether the entire document, or details from the
@@ -43,183 +60,140 @@ namespace {
  */
 class ProjectionAnalysisVisitor final : public ProjectionASTVisitor {
 public:
-    void visit(MatchExpressionASTNode* node) {}
-    void visit(ProjectionPathASTNode* node) {
+    ProjectionAnalysisVisitor(ProjectionDependencies* deps) : _deps(deps) {
+        invariant(_deps);
+    }
+
+    void visit(ProjectionPathASTNode* node) final {
         if (node->parent()) {
-            _deps.hasDottedPath = true;
+            _deps->hasDottedPath = true;
         }
     }
-    void visit(ProjectionPositionalASTNode* node) {
-        _deps.requiresMatchDetails = true;
-        _deps.requiresDocument = true;
+
+    void visit(ProjectionPositionalASTNode* node) final {
+        _deps->requiresMatchDetails = true;
+        _deps->requiresDocument = true;
     }
 
-    void visit(ProjectionSliceASTNode* node) {
-        _deps.requiresDocument = true;
+    void visit(ProjectionSliceASTNode* node) final {
+        _deps->requiresDocument = true;
     }
-    void visit(ProjectionElemMatchASTNode* node) {
-        _deps.requiresDocument = true;
+
+    void visit(ProjectionElemMatchASTNode* node) final {
+        _deps->requiresDocument = true;
     }
-    void visit(ExpressionASTNode* node) {
-        const Expression* expr = node->expression();
+
+    void visit(ExpressionASTNode* node) final {
+        const Expression* expr = node->expressionRaw();
         const ExpressionMeta* meta = dynamic_cast<const ExpressionMeta*>(expr);
 
         // Only {$meta: 'sortKey'} projections can be covered. Projections with any other expression
         // need the document.
         if (!(meta && meta->getMetaType() == DocumentMetadataFields::MetaType::kSortKey)) {
-            _deps.requiresDocument = true;
+            _deps->requiresDocument = true;
         }
     }
-    void visit(BooleanConstantASTNode* node) {}
 
-    ProjectionDependencies extractResult() {
-        return std::move(_deps);
-    }
+    void visit(BooleanConstantASTNode* node) final {}
+    void visit(MatchExpressionASTNode* node) final {}
 
 private:
-    ProjectionDependencies _deps;
+    ProjectionDependencies* _deps;
 };
 
 /**
  * Uses a DepsTracker to determine which fields are required from the projection.
+ *
+ * To track the current path in the projection, this visitor should be used with
+ * 'PathTrackingWalker' which will help to maintain the current path via
+ * 'PathTrackingVisitorContext'.
  */
-class DepsAnalysisPreVisitor final : public PathTrackingPreVisitor<DepsAnalysisPreVisitor> {
+class DepsAnalysisVisitor final : public ProjectionASTVisitor {
 public:
-    DepsAnalysisPreVisitor(PathTrackingVisitorContext<>* ctx)
-        : PathTrackingPreVisitor(ctx),
-          _fieldDependencyTracker(DepsTracker::kAllMetadataAvailable) {}
-
-    void doVisit(MatchExpressionASTNode* node) {
-        node->matchExpression()->addDependencies(&_fieldDependencyTracker);
+    DepsAnalysisVisitor(PathTrackingVisitorContext<DepsAnalysisData>* context) : _context{context} {
+        invariant(_context);
     }
 
-    void doVisit(ProjectionPathASTNode* node) {}
+    void visit(MatchExpressionASTNode* node) final {
+        node->matchExpression()->addDependencies(&_context->data().fieldDependencyTracker);
+    }
 
-    void doVisit(ProjectionPositionalASTNode* node) {
+    void visit(ProjectionPositionalASTNode* node) final {
         // Positional projection on a.b.c.$ may actually modify a, a.b, a.b.c, etc.
         // Treat the top-level field as a dependency.
-
         addTopLevelPathAsDependency();
     }
-    void doVisit(ProjectionSliceASTNode* node) {
+
+    void visit(ProjectionSliceASTNode* node) final {
         // find() $slice on a.b.c may modify a, a.b, and a.b.c if they're all arrays.
         // Treat the top-level field as a dependency.
         addTopLevelPathAsDependency();
     }
 
-    void doVisit(ProjectionElemMatchASTNode* node) {
-        const auto& fieldName = fullPath();
-        _fieldDependencyTracker.fields.insert(fieldName.fullPath());
+    void visit(ProjectionElemMatchASTNode* node) final {
+        addFullPathAsDependency();
     }
 
-    void doVisit(ExpressionASTNode* node) {
-        const auto fieldName = fullPath();
-
+    void visit(ExpressionASTNode* node) final {
         // The output of an expression on a dotted path depends on whether that field is an array.
         invariant(node->parent());
         if (!node->parent()->isRoot()) {
-            _fieldDependencyTracker.fields.insert(fieldName.fullPath());
+            addFullPathAsDependency();
         }
 
-        node->expression()->addDependencies(&_fieldDependencyTracker);
+        node->expression()->addDependencies(&_context->data().fieldDependencyTracker);
     }
 
-    void doVisit(BooleanConstantASTNode* node) {
+    void visit(BooleanConstantASTNode* node) final {
         // For inclusions, we depend on the field.
-        auto fieldName = fullPath();
         if (node->value()) {
-            _fieldDependencyTracker.fields.insert(fieldName.fullPath());
+            addFullPathAsDependency();
         }
     }
 
-    std::vector<std::string> requiredFields() {
-        return {_fieldDependencyTracker.fields.begin(), _fieldDependencyTracker.fields.end()};
-    }
-
-    DepsTracker* depsTracker() {
-        return &_fieldDependencyTracker;
-    }
+    void visit(ProjectionPathASTNode* node) final {}
 
 private:
     void addTopLevelPathAsDependency() {
-        FieldPath fp(fullPath());
-        _fieldDependencyTracker.fields.insert(fp.front().toString());
-    }
-
-    DepsTracker _fieldDependencyTracker;
-};
+        const auto& path = _context->fullPath();
 
-/**
- * Visitor which helps maintain the field path context for the deps analysis.
- */
-class DepsAnalysisPostVisitor final : public PathTrackingPostVisitor<DepsAnalysisPostVisitor> {
-public:
-    DepsAnalysisPostVisitor(PathTrackingVisitorContext<>* context)
-        : PathTrackingPostVisitor(context) {}
-
-    void doVisit(MatchExpressionASTNode* node) {}
-    void doVisit(ProjectionPathASTNode* node) {}
-    void doVisit(ProjectionPositionalASTNode* node) {}
-    void doVisit(ProjectionSliceASTNode* node) {}
-    void doVisit(ProjectionElemMatchASTNode* node) {}
-    void doVisit(ExpressionASTNode* node) {}
-    void doVisit(BooleanConstantASTNode* node) {}
-};
-
-/**
- * Walker for doing dependency analysis on the projection.
- */
-class DepsWalker final {
-public:
-    DepsWalker(ProjectType type)
-        : _depsPreVisitor(&_context), _depsPostVisitor(&_context), _projectionType(type) {}
-
-    void preVisit(ASTNode* node) {
-        node->acceptVisitor(&_generalAnalysisVisitor);
-        node->acceptVisitor(&_depsPreVisitor);
+        _context->data().addRequiredField(path.front().toString());
     }
 
-    void postVisit(ASTNode* node) {
-        node->acceptVisitor(&_depsPostVisitor);
+    void addFullPathAsDependency() {
+        const auto& path = _context->fullPath();
+
+        _context->data().addRequiredField(path.fullPath());
     }
 
-    void inVisit(long count, ASTNode* node) {}
+    PathTrackingVisitorContext<DepsAnalysisData>* _context;
+};
 
-    ProjectionDependencies done() {
-        ProjectionDependencies res = _generalAnalysisVisitor.extractResult();
+auto analyzeProjection(ProjectionPathASTNode* root, ProjectType type) {
+    ProjectionDependencies deps;
+    PathTrackingVisitorContext<DepsAnalysisData> context;
+    DepsAnalysisVisitor depsAnalysisVisitor{&context};
+    ProjectionAnalysisVisitor projectionAnalysisVisitor{&deps};
+    PathTrackingWalker walker{&context, {&depsAnalysisVisitor, &projectionAnalysisVisitor}, {}};
 
-        if (_projectionType == ProjectType::kInclusion) {
-            res.requiredFields = _depsPreVisitor.requiredFields();
-        } else {
-            invariant(_projectionType == ProjectType::kExclusion);
-            res.requiresDocument = true;
-        }
+    projection_ast_walker::walk(&walker, root);
 
-        auto* depsTracker = _depsPreVisitor.depsTracker();
-        res.needsTextScore = depsTracker->getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE);
-        res.needsGeoPoint =
-            depsTracker->getNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_POINT);
-        res.needsGeoDistance =
-            depsTracker->getNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_DISTANCE);
-        res.needsSortKey = depsTracker->getNeedsMetadata(DepsTracker::MetadataType::SORT_KEY);
+    const auto& userData = context.data();
+    const auto& tracker = userData.fieldDependencyTracker;
 
-        return res;
+    if (type == ProjectType::kInclusion) {
+        deps.requiredFields = userData.requiredFields();
+    } else {
+        invariant(type == ProjectType::kExclusion);
+        deps.requiresDocument = true;
     }
 
-private:
-    PathTrackingVisitorContext<> _context;
-    ProjectionAnalysisVisitor _generalAnalysisVisitor;
-
-    DepsAnalysisPreVisitor _depsPreVisitor;
-    DepsAnalysisPostVisitor _depsPostVisitor;
+    deps.needsTextScore = tracker.getNeedsMetadata(DepsTracker::MetadataType::TEXT_SCORE);
+    deps.needsGeoPoint = tracker.getNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_POINT);
+    deps.needsGeoDistance = tracker.getNeedsMetadata(DepsTracker::MetadataType::GEO_NEAR_DISTANCE);
+    deps.needsSortKey = tracker.getNeedsMetadata(DepsTracker::MetadataType::SORT_KEY);
 
-    ProjectType _projectionType;
-};
-
-ProjectionDependencies analyzeProjection(ProjectionPathASTNode* root, ProjectType type) {
-    DepsWalker walker(type);
-    projection_ast_walker::walk(&walker, root);
-    return walker.done();
+    return deps;
 }
 }  // namespace
 
diff --git a/src/mongo/db/query/projection_ast.h b/src/mongo/db/query/projection_ast.h
index a10a1100185..2c136ba0a00 100644
--- a/src/mongo/db/query/projection_ast.h
+++ b/src/mongo/db/query/projection_ast.h
@@ -30,18 +30,14 @@
 #pragma once
 
 #include "mongo/db/jsobj.h"
+#include "mongo/db/matcher/copyable_match_expression.h"
 #include "mongo/db/matcher/expression_parser.h"
 #include "mongo/db/pipeline/dependencies.h"
 #include "mongo/db/pipeline/expression.h"
 #include "mongo/db/query/projection_ast_visitor.h"
-#include "mongo/util/str.h"
 
 namespace mongo {
-
-class ProjectionASTVisitor;
-
 namespace projection_ast {
-
 /*
  * A tree representation of a projection. The main purpose of this class is to offer a typed,
  * walkable representation of a projection. It's mostly meant to be used while doing validation and
@@ -98,13 +94,10 @@ protected:
 
 class MatchExpressionASTNode final : public ASTNode {
 public:
-    MatchExpressionASTNode(BSONObj bson, std::unique_ptr<MatchExpression> me)
-        : _bson(bson), _matchExpr(std::move(me)) {}
+    MatchExpressionASTNode(CopyableMatchExpression matchExpr) : _matchExpr{matchExpr} {}
 
     MatchExpressionASTNode(const MatchExpressionASTNode& other)
-        // Performing a shallowClone() on the match expression and maintaining a pointer to the
-        // underlying BSON is equivalent to a deep clone.
-        : ASTNode(other), _bson(other._bson), _matchExpr(other._matchExpr->shallowClone()) {}
+        : ASTNode{other}, _matchExpr{other._matchExpr} {}
 
     std::unique_ptr<ASTNode> clone() const override final {
         return std::make_unique<MatchExpressionASTNode>(*this);
@@ -114,14 +107,12 @@ public:
         visitor->visit(this);
     }
 
-    const MatchExpression* matchExpression() const {
-        return _matchExpr.get();
+    CopyableMatchExpression matchExpression() const {
+        return _matchExpr;
     }
 
 private:
-    // Must carry the BSON around as well since _matchExpr maintains pointers into it.
-    BSONObj _bson;
-    std::unique_ptr<MatchExpression> _matchExpr;
+    CopyableMatchExpression _matchExpr;
 };
 
 class ProjectionPathASTNode final : public ASTNode {
@@ -245,10 +236,14 @@ public:
         return std::make_unique<ExpressionASTNode>(*this);
     }
 
-    const Expression* expression() const {
+    Expression* expressionRaw() const {
         return _expr.get();
     }
 
+    boost::intrusive_ptr<Expression> expression() const {
+        return _expr;
+    }
+
 private:
     boost::intrusive_ptr<Expression> _expr;
 };
@@ -272,7 +267,5 @@ public:
 private:
     bool _val;
 };
-
-
 }  // namespace projection_ast
 }  // namespace mongo
diff --git a/src/mongo/db/query/projection_ast_path_tracking_visitor.h b/src/mongo/db/query/projection_ast_path_tracking_visitor.h
index 2a7e4be1c1c..2108233ac8e 100644
--- a/src/mongo/db/query/projection_ast_path_tracking_visitor.h
+++ b/src/mongo/db/query/projection_ast_path_tracking_visitor.h
@@ -48,7 +48,10 @@ using PathTrackingDummyDefaultType = void*;
 template <class UserData = PathTrackingDummyDefaultType>
 class PathTrackingVisitorContext {
 public:
-    FieldPath fullPath() const {
+    PathTrackingVisitorContext() {}
+    PathTrackingVisitorContext(UserData data) : _data{std::move(data)} {}
+
+    auto fullPath() const {
         invariant(!_fieldNames.empty());
         invariant(!_fieldNames.top().empty());
 
@@ -59,7 +62,7 @@ public:
             FieldPath::getFullyQualifiedPath(_basePath->fullPath(), _fieldNames.top().front()));
     }
 
-    const boost::optional<FieldPath>& basePath() const {
+    const auto& basePath() const {
         return _basePath;
     }
 
@@ -80,7 +83,7 @@ public:
         _fieldNames.push(std::move(fields));
     }
 
-    UserData& data() {
+    auto& data() {
         return _data;
     }
 
@@ -96,88 +99,59 @@ private:
     boost::optional<FieldPath> _basePath;
 };
 
+namespace {
 /**
- * Base visitor used for maintaining field names while traversing the AST.
+ * A path tracking pre-visitor used for maintaining field names while traversing the AST.
  *
- * This is intended to be used with the projection AST walker (projection_ast::walk()). Users of
- * this class MUST use both the PathTrackingPreVisitor and PathTrackingPostVisitor in order to
- * correctly maintain the state about the current path being visited.
-
- * Derived classes can have custom behavior through doVisit() methods on each node type.
+ * This is intended to be used with the 'ProjectionPathTrackingWalker' only to correctly maintain
+ * the state about the current path being visited.
  */
-template <class Derived, class UserData = PathTrackingDummyDefaultType>
-class PathTrackingPreVisitor : public ProjectionASTVisitor {
+template <class UserData = PathTrackingDummyDefaultType>
+class PathTrackingPreVisitor final : public ProjectionASTVisitor {
 public:
-    PathTrackingPreVisitor(PathTrackingVisitorContext<UserData>* ctx) : _context(ctx) {}
-
-    void visit(MatchExpressionASTNode* node) override {
-        static_cast<Derived*>(this)->doVisit(node);
+    PathTrackingPreVisitor(PathTrackingVisitorContext<UserData>* context) : _context{context} {
+        invariant(_context);
     }
 
-    void visit(ProjectionPathASTNode* node) override {
+    void visit(ProjectionPathASTNode* node) final {
         if (node->parent()) {
             _context->setBasePath(_context->fullPath());
             _context->popFrontFieldName();
         }
 
         _context->pushFieldNames({node->fieldNames().begin(), node->fieldNames().end()});
-
-        static_cast<Derived*>(this)->doVisit(node);
     }
 
-    void visit(ProjectionPositionalASTNode* node) override {
-        static_cast<Derived*>(this)->doVisit(node);
-    }
-
-    void visit(ProjectionSliceASTNode* node) override {
-        static_cast<Derived*>(this)->doVisit(node);
-    }
-
-    void visit(ProjectionElemMatchASTNode* node) {
-        static_cast<Derived*>(this)->doVisit(node);
-    }
-
-    void visit(ExpressionASTNode* node) override {
-        static_cast<Derived*>(this)->doVisit(node);
-    }
-
-    void visit(BooleanConstantASTNode* node) override {
-        static_cast<Derived*>(this)->doVisit(node);
-    }
-
-protected:
-    const FieldPath fullPath() const {
-        return _context->fullPath();
-    }
-
-    UserData* data() const {
-        return _context->data;
-    }
+    void visit(MatchExpressionASTNode* node) final {}
+    void visit(ProjectionPositionalASTNode* node) final {}
+    void visit(ProjectionSliceASTNode* node) final {}
+    void visit(ProjectionElemMatchASTNode* node) final {}
+    void visit(ExpressionASTNode* node) final {}
+    void visit(BooleanConstantASTNode* node) final {}
 
 private:
     PathTrackingVisitorContext<UserData>* _context;
 };
 
 /**
- * Base post-visitor which helps maintain field names while traversing the AST.
+ * A path tracking post-visitor used for maintaining field names while traversing the AST.
+ *
+ * This is intended to be used with the 'PathTrackingWalker' only to correctly maintain the state
+ * about the current path being visited.
  */
-template <class Derived, class UserData = PathTrackingDummyDefaultType>
-class PathTrackingPostVisitor : public ProjectionASTVisitor {
+template <class UserData = PathTrackingDummyDefaultType>
+class PathTrackingPostVisitor final : public ProjectionASTVisitor {
 public:
-    PathTrackingPostVisitor(PathTrackingVisitorContext<UserData>* context) : _context(context) {}
-
-    void visit(MatchExpressionASTNode* node) override {
-        static_cast<Derived*>(this)->doVisit(node);
+    PathTrackingPostVisitor(PathTrackingVisitorContext<UserData>* context) : _context{context} {
+        invariant(_context);
     }
 
-    void visit(projection_ast::ProjectionPathASTNode* node) override {
-        static_cast<Derived*>(this)->doVisit(node);
-
+    void visit(projection_ast::ProjectionPathASTNode* node) final {
         _context->popFieldNames();
 
         if (_context->basePath()) {
             // Update the context variable tracking the current path being traversed.
-            const FieldPath& fp = *_context->basePath();
+            const auto& fp = *_context->basePath();
             if (fp.getPathLength() == 1) {
                 _context->setBasePath(boost::none);
             } else {
@@ -187,37 +161,75 @@ public:
         }
     }
 
-    void visit(ProjectionPositionalASTNode* node) override {
+    void visit(ProjectionPositionalASTNode* node) final {
         _context->popFrontFieldName();
-
-        static_cast<Derived*>(this)->doVisit(node);
     }
 
-    void visit(ProjectionSliceASTNode* node) override {
+    void visit(ProjectionSliceASTNode* node) final {
         _context->popFrontFieldName();
-
-        static_cast<Derived*>(this)->doVisit(node);
     }
 
-    void visit(ProjectionElemMatchASTNode* node) override {
+    void visit(ProjectionElemMatchASTNode* node) final {
         _context->popFrontFieldName();
-
-        static_cast<Derived*>(this)->doVisit(node);
     }
 
-    void visit(ExpressionASTNode* node) override {
+    void visit(ExpressionASTNode* node) final {
         _context->popFrontFieldName();
-        static_cast<Derived*>(this)->doVisit(node);
     }
 
-    void visit(BooleanConstantASTNode* node) override {
+    void visit(BooleanConstantASTNode* node) final {
         _context->popFrontFieldName();
-
-        static_cast<Derived*>(this)->doVisit(node);
     }
 
+    void visit(MatchExpressionASTNode* node) final {}
+
 private:
     PathTrackingVisitorContext<UserData>* _context;
 };
+}  // namespace
+
+/**
+ * A general path tracking walker to be used with projection AST visitors which need to track
+ * the current projection path. Visitors will be able to access the current path via
+ * 'PathTrackingVisitorContext', passed as 'context' parameter to the constructor of this class.
+ * The walker and the visitors must be initialized with the same 'context' pointer.
+ *
+ * The visitors specified in the 'preVisitors' and 'postVisitors' parameters will be visited in
+ * the same order as they were added to the vector.
+ */
+template <class UserData = PathTrackingDummyDefaultType>
+class PathTrackingWalker final {
+public:
+    PathTrackingWalker(PathTrackingVisitorContext<UserData>* context,
+                       std::vector<ProjectionASTVisitor*> preVisitors,
+                       std::vector<ProjectionASTVisitor*> postVisitors)
+        : _pathTrackingPreVisitor{context},
+          _pathTrackingPostVisitor{context},
+          _preVisitors{std::move(preVisitors)},
+          _postVisitors{std::move(postVisitors)} {
+        _preVisitors.insert(_preVisitors.begin(), &_pathTrackingPreVisitor);
+        _postVisitors.push_back(&_pathTrackingPostVisitor);
+    }
+
+    void preVisit(projection_ast::ASTNode* node) {
+        for (auto visitor : _preVisitors) {
+            node->acceptVisitor(visitor);
+        }
+    }
+
+    void postVisit(projection_ast::ASTNode* node) {
+        for (auto visitor : _postVisitors) {
+            node->acceptVisitor(visitor);
+        }
+    }
+
+    void inVisit(long count, projection_ast::ASTNode* node) {}
+
+private:
+    PathTrackingPreVisitor<UserData> _pathTrackingPreVisitor;
+    PathTrackingPostVisitor<UserData> _pathTrackingPostVisitor;
+    std::vector<ProjectionASTVisitor*> _preVisitors;
+    std::vector<ProjectionASTVisitor*> _postVisitors;
+};
 }  // namespace projection_ast
 }  // namespace mongo
diff --git a/src/mongo/db/query/projection_ast_util.cpp b/src/mongo/db/query/projection_ast_util.cpp
index 1eaf3bf3743..f2b4c2aa1ee 100644
--- a/src/mongo/db/query/projection_ast_util.cpp
+++ b/src/mongo/db/query/projection_ast_util.cpp
@@ -49,7 +49,8 @@ public:
     BSONPreVisitor(BSONVisitorContext* context) : _context(context) {}
 
     virtual void visit(MatchExpressionASTNode* node) {
-        node->matchExpression()->serialize(&_context->builder());
+        static_cast<const MatchExpressionASTNode*>(node)->matchExpression()->serialize(
+            &_context->builder());
         _context->fieldNames.top().pop_front();
     }
     virtual void visit(ProjectionPathASTNode* node) {
diff --git a/src/mongo/db/query/projection_parser.cpp b/src/mongo/db/query/projection_parser.cpp
index be6179d4efa..a64d2e7dbd8 100644
--- a/src/mongo/db/query/projection_parser.cpp
+++ b/src/mongo/db/query/projection_parser.cpp
@@ -222,15 +222,13 @@ Projection parse(boost::intrusive_ptr<ExpressionContext> expCtx,
                 BSONObj elemMatchObj = elem.wrap();
                 invariant(elemMatchObj.isOwned());
 
-                StatusWithMatchExpression statusWithMatcher =
-                    MatchExpressionParser::parse(elemMatchObj,
-                                                 expCtx,
-                                                 ExtensionsCallbackNoop(),
-                                                 MatchExpressionParser::kBanAllSpecialFeatures);
-                auto matcher = uassertStatusOK(std::move(statusWithMatcher));
-
-                auto matchNode =
-                    std::make_unique<MatchExpressionASTNode>(elemMatchObj, std::move(matcher));
+                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,
@@ -291,13 +289,19 @@ Projection parse(boost::intrusive_ptr<ExpressionContext> expCtx,
                 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>(queryObj, query->shallowClone())));
+                addNodeAtPath(&root,
+                              path,
+                              std::make_unique<ProjectionPositionalASTNode>(
+                                  std::make_unique<MatchExpressionASTNode>(matcher)));
 
                 hasPositional = true;
             }