SERVER-35325 Implement key generation for "allPaths" indexes

22 files changed, 2307 insertions, 512 deletions

diff --git a/src/mongo/db/exec/projection_exec_agg.cpp b/src/mongo/db/exec/projection_exec_agg.cpp
index 30aa62969b4..54b5132f118 100644
--- a/src/mongo/db/exec/projection_exec_agg.cpp
+++ b/src/mongo/db/exec/projection_exec_agg.cpp
@@ -42,13 +42,35 @@ public:
     using ProjectionParseMode = ParsedAggregationProjection::ProjectionParseMode;
     using TransformerType = TransformerInterface::TransformerType;
 
-    ProjectionExecutor(BSONObj projSpec) {
+    ProjectionExecutor(BSONObj projSpec,
+                       DefaultIdPolicy defaultIdPolicy,
+                       ArrayRecursionPolicy arrayRecursionPolicy) {
         // Construct a dummy ExpressionContext for ParsedAggregationProjection. It's OK to set the
         // ExpressionContext's OperationContext and CollatorInterface to 'nullptr' here; since we
         // ban computed fields from the projection, the ExpressionContext will never be used.
         boost::intrusive_ptr<ExpressionContext> expCtx(new ExpressionContext(nullptr, nullptr));
+
+        // Default projection behaviour is to include _id if the projection spec omits it. If the
+        // caller has specified that we should *exclude* _id by default, do so here. We translate
+        // DefaultIdPolicy to ParsedAggregationProjection::ProjectionDefaultIdPolicy in order to
+        // avoid exposing internal aggregation types to the query system.
+        ParsedAggregationProjection::ProjectionDefaultIdPolicy idPolicy =
+            (defaultIdPolicy == ProjectionExecAgg::DefaultIdPolicy::kIncludeId
+                 ? ParsedAggregationProjection::ProjectionDefaultIdPolicy::kIncludeId
+                 : ParsedAggregationProjection::ProjectionDefaultIdPolicy::kExcludeId);
+
+        // By default, $project will recurse through nested arrays. If the caller has specified that
+        // it should not, we inhibit it from doing so here. We separate this class' internal enum
+        // ArrayRecursionPolicy from ParsedAggregationProjection::ProjectionArrayRecursionPolicy
+        // in order to avoid exposing aggregation types to the query system.
+        ParsedAggregationProjection::ProjectionArrayRecursionPolicy recursionPolicy =
+            (arrayRecursionPolicy == ArrayRecursionPolicy::kRecurseNestedArrays
+                 ? ParsedAggregationProjection::ProjectionArrayRecursionPolicy::kRecurseNestedArrays
+                 : ParsedAggregationProjection::ProjectionArrayRecursionPolicy::
+                       kDoNotRecurseNestedArrays);
+
         _projection = ParsedAggregationProjection::create(
-            expCtx, projSpec, ProjectionParseMode::kBanComputedFields);
+            expCtx, projSpec, idPolicy, recursionPolicy, ProjectionParseMode::kBanComputedFields);
     }
 
     ProjectionType getType() const {
@@ -73,9 +95,12 @@ ProjectionExecAgg::ProjectionExecAgg(BSONObj projSpec, std::unique_ptr<Projectio
 
 ProjectionExecAgg::~ProjectionExecAgg() = default;
 
-std::unique_ptr<ProjectionExecAgg> ProjectionExecAgg::create(BSONObj projSpec) {
-    return std::unique_ptr<ProjectionExecAgg>(
-        new ProjectionExecAgg(projSpec, std::make_unique<ProjectionExecutor>(projSpec)));
+std::unique_ptr<ProjectionExecAgg> ProjectionExecAgg::create(BSONObj projSpec,
+                                                             DefaultIdPolicy defaultIdPolicy,
+                                                             ArrayRecursionPolicy recursionPolicy) {
+    return std::unique_ptr<ProjectionExecAgg>(new ProjectionExecAgg(
+        projSpec,
+        std::make_unique<ProjectionExecutor>(projSpec, defaultIdPolicy, recursionPolicy)));
 }
 
 ProjectionExecAgg::ProjectionType ProjectionExecAgg::getType() const {
diff --git a/src/mongo/db/exec/projection_exec_agg.h b/src/mongo/db/exec/projection_exec_agg.h
index a71f611beda..b885ca6808c 100644
--- a/src/mongo/db/exec/projection_exec_agg.h
+++ b/src/mongo/db/exec/projection_exec_agg.h
@@ -40,9 +40,25 @@ namespace mongo {
  */
 class ProjectionExecAgg {
 public:
+    // Allows the caller to specify how the projection should handle nested arrays; that is, an
+    // array whose immediate parent is itself an array. For example, in the case of sample document
+    // {a: [1, 2, [3, 4], {b: [5, 6]}]} the array [3, 4] is a nested array. The array [5, 6] is not,
+    // because there is an intervening object between it and its closest array ancestor.
+    enum class ArrayRecursionPolicy { kRecurseNestedArrays, kDoNotRecurseNestedArrays };
+
+    // Allows the caller to indicate whether the projection should default to including or excluding
+    // the _id field in the event that the projection spec does not specify the desired behavior.
+    // For instance, given a projection {a: 1}, specifying 'kExcludeId' is equivalent to projecting
+    // {a: 1, _id: 0} while 'kIncludeId' is equivalent to the projection {a: 1, _id: 1}. If the user
+    // explicitly specifies a projection on _id, then this will override the default policy; for
+    // instance, {a: 1, _id: 0} will exclude _id for both 'kExcludeId' and 'kIncludeId'.
+    enum class DefaultIdPolicy { kIncludeId, kExcludeId };
+
     enum class ProjectionType { kInclusionProjection, kExclusionProjection };
 
-    static std::unique_ptr<ProjectionExecAgg> create(BSONObj projSpec);
+    static std::unique_ptr<ProjectionExecAgg> create(BSONObj projSpec,
+                                                     DefaultIdPolicy defaultIdPolicy,
+                                                     ArrayRecursionPolicy recursionPolicy);
 
     ~ProjectionExecAgg();
 
@@ -52,6 +68,10 @@ public:
         return _projSpec;
     }
 
+    const BSONObj& getSpec() const {
+        return _projSpec;
+    }
+
     BSONObj applyProjection(BSONObj inputDoc) const;
 
 private:
diff --git a/src/mongo/db/exec/projection_exec_agg_test.cpp b/src/mongo/db/exec/projection_exec_agg_test.cpp
index a8dcde138d3..8503e7a50f8 100644
--- a/src/mongo/db/exec/projection_exec_agg_test.cpp
+++ b/src/mongo/db/exec/projection_exec_agg_test.cpp
@@ -38,69 +38,88 @@
 namespace mongo {
 namespace {
 
+using ArrayRecursionPolicy = ProjectionExecAgg::ArrayRecursionPolicy;
+using DefaultIdPolicy = ProjectionExecAgg::DefaultIdPolicy;
+
 template <typename T>
 BSONObj wrapInLiteral(const T& arg) {
     return BSON("$literal" << arg);
 }
 
+// Helper to simplify the creation of a ProjectionExecAgg which includes _id and recurses nested
+// arrays by default.
+std::unique_ptr<ProjectionExecAgg> makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+    BSONObj projSpec) {
+    return ProjectionExecAgg::create(
+        projSpec, DefaultIdPolicy::kIncludeId, ArrayRecursionPolicy::kRecurseNestedArrays);
+}
+
 //
 // Error cases.
 //
 
 TEST(ProjectionExecAggErrors, ShouldRejectMixOfInclusionAndComputedFields) {
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a" << true << "b" << wrapInLiteral(1))),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << true << "b" << wrapInLiteral(1))),
                   AssertionException);
 
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a" << wrapInLiteral(1) << "b" << true)),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << wrapInLiteral(1) << "b" << true)),
                   AssertionException);
 
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a.b" << true << "a.c" << wrapInLiteral(1))),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a.b" << true << "a.c" << wrapInLiteral(1))),
                   AssertionException);
 
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a.b" << wrapInLiteral(1) << "a.c" << true)),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a.b" << wrapInLiteral(1) << "a.c" << true)),
                   AssertionException);
 
-    ASSERT_THROWS(
-        ProjectionExecAgg::create(BSON("a" << BSON("b" << true << "c" << wrapInLiteral(1)))),
-        AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << BSON("b" << true << "c" << wrapInLiteral(1)))),
+                  AssertionException);
 
-    ASSERT_THROWS(
-        ProjectionExecAgg::create(BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << true))),
-        AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << true))),
+                  AssertionException);
 }
 
 TEST(ProjectionExecAggErrors, ShouldRejectMixOfExclusionAndComputedFields) {
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a" << false << "b" << wrapInLiteral(1))),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << false << "b" << wrapInLiteral(1))),
                   AssertionException);
 
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a" << wrapInLiteral(1) << "b" << false)),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << wrapInLiteral(1) << "b" << false)),
                   AssertionException);
 
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a.b" << false << "a.c" << wrapInLiteral(1))),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a.b" << false << "a.c" << wrapInLiteral(1))),
                   AssertionException);
 
-    ASSERT_THROWS(ProjectionExecAgg::create(BSON("a.b" << wrapInLiteral(1) << "a.c" << false)),
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a.b" << wrapInLiteral(1) << "a.c" << false)),
                   AssertionException);
 
-    ASSERT_THROWS(
-        ProjectionExecAgg::create(BSON("a" << BSON("b" << false << "c" << wrapInLiteral(1)))),
-        AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << BSON("b" << false << "c" << wrapInLiteral(1)))),
+                  AssertionException);
 
-    ASSERT_THROWS(
-        ProjectionExecAgg::create(BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << false))),
-        AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << false))),
+                  AssertionException);
 }
 
 TEST(ProjectionExecAggErrors, ShouldRejectOnlyComputedFields) {
-    ASSERT_THROWS(
-        ProjectionExecAgg::create(BSON("a" << wrapInLiteral(1) << "b" << wrapInLiteral(1))),
-        AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a" << wrapInLiteral(1) << "b" << wrapInLiteral(1))),
+                  AssertionException);
 
-    ASSERT_THROWS(
-        ProjectionExecAgg::create(BSON("a.b" << wrapInLiteral(1) << "a.c" << wrapInLiteral(1))),
-        AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+                      BSON("a.b" << wrapInLiteral(1) << "a.c" << wrapInLiteral(1))),
+                  AssertionException);
 
-    ASSERT_THROWS(ProjectionExecAgg::create(
+    ASSERT_THROWS(makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
                       BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << wrapInLiteral(1)))),
                   AssertionException);
 }
@@ -108,39 +127,50 @@ TEST(ProjectionExecAggErrors, ShouldRejectOnlyComputedFields) {
 // Valid projections.
 
 TEST(ProjectionExecAggType, ShouldAcceptInclusionProjection) {
-    auto parsedProject = ProjectionExecAgg::create(BSON("a" << true));
+    auto parsedProject =
+        makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(BSON("a" << true));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kInclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("_id" << false << "a" << true));
+    parsedProject = makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+        BSON("_id" << false << "a" << true));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kInclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("_id" << false << "a.b.c" << true));
+    parsedProject = makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+        BSON("_id" << false << "a.b.c" << true));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kInclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("_id.x" << true));
+    parsedProject =
+        makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(BSON("_id.x" << true));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kInclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("_id" << BSON("x" << true)));
+    parsedProject = makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+        BSON("_id" << BSON("x" << true)));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kInclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("x" << BSON("_id" << true)));
+    parsedProject = makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+        BSON("x" << BSON("_id" << true)));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kInclusionProjection);
 }
 
 TEST(ProjectionExecAggType, ShouldAcceptExclusionProjection) {
-    auto parsedProject = ProjectionExecAgg::create(BSON("a" << false));
+    auto parsedProject =
+        makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(BSON("a" << false));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kExclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("_id.x" << false));
+    parsedProject =
+        makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(BSON("_id.x" << false));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kExclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("_id" << BSON("x" << false)));
+    parsedProject = makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+        BSON("_id" << BSON("x" << false)));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kExclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("x" << BSON("_id" << false)));
+    parsedProject = makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(
+        BSON("x" << BSON("_id" << false)));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kExclusionProjection);
 
-    parsedProject = ProjectionExecAgg::create(BSON("_id" << false));
+    parsedProject =
+        makeProjectionWithDefaultIdInclusionAndNestedArrayRecursion(BSON("_id" << false));
     ASSERT(parsedProject->getType() == ProjectionExecAgg::ProjectionType::kExclusionProjection);
 }
 
diff --git a/src/mongo/db/index/SConscript b/src/mongo/db/index/SConscript
index 3ec4e5a7d90..5fbf8868460 100644
--- a/src/mongo/db/index/SConscript
+++ b/src/mongo/db/index/SConscript
@@ -19,6 +19,7 @@ env.Library(
 env.Library(
         target='key_generator',
         source=[
+            'all_paths_key_generator.cpp',
             'btree_key_generator.cpp',
             'expression_keys_private.cpp',
             'sort_key_generator.cpp',
@@ -30,6 +31,7 @@ env.Library(
             '$BUILD_DIR/mongo/db/geo/geoparser',
             '$BUILD_DIR/mongo/db/index_names',
             '$BUILD_DIR/mongo/db/mongohasher',
+            '$BUILD_DIR/mongo/db/projection_exec_agg',
             '$BUILD_DIR/mongo/db/query/collation/collator_interface',
             '$BUILD_DIR/third_party/s2/s2',
             'expression_params',
@@ -57,6 +59,7 @@ env.Library(
 env.CppUnitTest(
         target='key_generator_test',
         source=[
+            'all_paths_key_generator_test.cpp',
             '2d_key_generator_test.cpp',
             'btree_key_generator_test.cpp',
             'hash_key_generator_test.cpp',
diff --git a/src/mongo/db/index/all_paths_access_method.cpp b/src/mongo/db/index/all_paths_access_method.cpp
index d2d143ebfff..46a3d432e4e 100644
--- a/src/mongo/db/index/all_paths_access_method.cpp
+++ b/src/mongo/db/index/all_paths_access_method.cpp
@@ -26,24 +26,26 @@
  *    it in the license file.
  */
 
+#include "mongo/platform/basic.h"
+
 #include "mongo/db/index/all_paths_access_method.h"
 
 #include "mongo/db/catalog/index_catalog_entry.h"
-#include "mongo/db/jsobj.h"
 
 namespace mongo {
 
-// Standard AllPaths implementation below.
 AllPathsAccessMethod::AllPathsAccessMethod(IndexCatalogEntry* allPathsState,
                                            SortedDataInterface* btree)
-    : IndexAccessMethod(allPathsState, btree) {
-    // TODO: SERVER-35325: Implement AllPathsAcessMethod.
-}
+    : IndexAccessMethod(allPathsState, btree),
+      _keyGen(
+          _descriptor->keyPattern(), _descriptor->pathProjection(), _btreeState->getCollator()) {}
 
 void AllPathsAccessMethod::doGetKeys(const BSONObj& obj,
                                      BSONObjSet* keys,
                                      MultikeyPaths* multikeyPaths) const {
-    // TODO: SERVER-35325: Implement AllPathsAcessMethod.
+    // TODO SERVER-35748: Until MultikeyPaths has been updated to facilitate 'allPaths' indexes, we
+    // use AllPathsKeyGenerator::MultikeyPathsMock to separate multikey paths from RecordId keys.
+    auto multikeyPathsMock = SimpleBSONObjComparator::kInstance.makeBSONObjSet();
+    _keyGen.generateKeys(obj, keys, &multikeyPathsMock);
 }
-
 }  // namespace mongo
diff --git a/src/mongo/db/index/all_paths_access_method.h b/src/mongo/db/index/all_paths_access_method.h
index f51c173ffac..926d8ed1166 100644
--- a/src/mongo/db/index/all_paths_access_method.h
+++ b/src/mongo/db/index/all_paths_access_method.h
@@ -28,14 +28,15 @@
 
 #pragma once
 
+#include "mongo/db/index/all_paths_key_generator.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/jsobj.h"
 
 namespace mongo {
 
 /**
- * The IndexAccessMethod for an AllPaths index.
- * Any index created with {"field.$**": 1} or {"$**": 1} uses this.
+ * Class which is responsible for generating and providing access to AllPaths index keys. Any index
+ * created with { "$**": ±1 } or { "path.$**": ±1 } uses this class.
  */
 class AllPathsAccessMethod : public IndexAccessMethod {
 public:
@@ -43,6 +44,7 @@ public:
 
 private:
     void doGetKeys(const BSONObj& obj, BSONObjSet* keys, MultikeyPaths* multikeyPaths) const final;
-};
 
+    const AllPathsKeyGenerator _keyGen;
+};
 }  // namespace mongo
diff --git a/src/mongo/db/index/all_paths_key_generator.cpp b/src/mongo/db/index/all_paths_key_generator.cpp
new file mode 100644
index 00000000000..c894188436c
--- /dev/null
+++ b/src/mongo/db/index/all_paths_key_generator.cpp
@@ -0,0 +1,167 @@
+/**
+ * Copyright (C) 2018 MongoDB Inc.
+ *
+ * This program is free software: you can redistribute it and/or  modify
+ * it under the terms of the GNU Affero General Public License, version 3,
+ * as published by the Free Software Foundation.
+ *
+ * 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
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * 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 GNU Affero General 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/index/all_paths_key_generator.h"
+
+#include "mongo/db/jsobj.h"
+#include "mongo/db/query/collation/collation_index_key.h"
+
+namespace mongo {
+namespace {
+
+// If the user does not specify any projection, then we default to a projection of {_id: 0} in order
+// to prevent the _id field from being indexed, since it already has its own dedicated index.
+static const BSONObj kDefaultProjection = BSON("_id"_sd << 0);
+
+// If the enclosing object is an array, then the current element's fieldname is the array index, so
+// we omit this when computing the full path. Otherwise, the full path is the pathPrefix plus the
+// element's fieldname.
+void pushPathComponent(BSONElement elem, bool enclosingObjIsArray, FieldRef* pathPrefix) {
+    if (!enclosingObjIsArray) {
+        pathPrefix->appendPart(elem.fieldNameStringData());
+    }
+}
+
+// If the enclosing object is not an array, then the final path component should be its field name.
+// Verify that this is the case and then pop it off the FieldRef.
+void popPathComponent(BSONElement elem, bool enclosingObjIsArray, FieldRef* pathToElem) {
+    if (!enclosingObjIsArray) {
+        invariant(pathToElem->getPart(pathToElem->numParts() - 1) == elem.fieldNameStringData());
+        pathToElem->removeLastPart();
+    }
+}
+}  // namespace
+
+constexpr StringData AllPathsKeyGenerator::kSubtreeSuffix;
+
+AllPathsKeyGenerator::AllPathsKeyGenerator(BSONObj keyPattern,
+                                           BSONObj pathProjection,
+                                           const CollatorInterface* collator)
+    : _collator(collator), _keyPattern(keyPattern) {
+    // We should never have a key pattern that contains more than a single element.
+    invariant(_keyPattern.nFields() == 1);
+
+    // The _keyPattern is either { "$**": ±1 } for all paths or { "path.$**": ±1 } for a single
+    // subtree. If we are indexing a single subtree, then we will project just that path.
+    auto indexRoot = _keyPattern.firstElement().fieldNameStringData();
+    auto suffixPos = indexRoot.find(kSubtreeSuffix);
+
+    // If we're indexing a single subtree, we can't also specify a path projection.
+    invariant(suffixPos == std::string::npos || pathProjection.isEmpty());
+
+    // If this is a subtree projection, the projection spec is { "path.to.subtree": 1 }. Otherwise,
+    // we use the path projection from the original command object. If the path projection is empty
+    // we default to {_id: 0}, since empty projections are illegal and will be rejected when parsed.
+    auto projSpec = (suffixPos != std::string::npos
+                         ? BSON(indexRoot.substr(0, suffixPos) << 1)
+                         : pathProjection.isEmpty() ? kDefaultProjection : pathProjection);
+
+    // If the projection spec does not explicitly specify _id, we exclude it by default. We also
+    // prevent the projection from recursing through nested arrays, in order to ensure that the
+    // output document aligns with the match system's expectations.
+    _projExec = ProjectionExecAgg::create(
+        projSpec,
+        ProjectionExecAgg::DefaultIdPolicy::kExcludeId,
+        ProjectionExecAgg::ArrayRecursionPolicy::kDoNotRecurseNestedArrays);
+}
+
+void AllPathsKeyGenerator::generateKeys(BSONObj inputDoc,
+                                        BSONObjSet* keys,
+                                        MultikeyPathsMock* multikeyPaths) const {
+    FieldRef workingPath;
+    _traverseAllPaths(
+        _projExec->applyProjection(inputDoc), false, &workingPath, keys, multikeyPaths);
+}
+
+void AllPathsKeyGenerator::_traverseAllPaths(BSONObj obj,
+                                             bool objIsArray,
+                                             FieldRef* path,
+                                             BSONObjSet* keys,
+                                             MultikeyPathsMock* multikeyPaths) const {
+    for (const auto elem : obj) {
+        // If this element is an empty object, fast-path skip it.
+        if (elem.type() == BSONType::Object && elem.Obj().isEmpty())
+            continue;
+
+        // Append the element's fieldname to the path, if the enclosing object is not an array.
+        pushPathComponent(elem, objIsArray, path);
+
+        switch (elem.type()) {
+            case BSONType::Array:
+                // If this is a nested array, we don't descend it but instead index it as a value.
+                if (_addKeyForNestedArray(elem, *path, objIsArray, keys))
+                    break;
+
+                // Add an entry for the multi-key path, and then fall through to BSONType::Object.
+                _addMultiKey(*path, multikeyPaths);
+
+            case BSONType::Object:
+                _traverseAllPaths(
+                    elem.Obj(), elem.type() == BSONType::Array, path, keys, multikeyPaths);
+                break;
+
+            default:
+                _addKey(elem, *path, keys);
+        }
+
+        // Remove the element's fieldname from the path, if it was pushed onto it earlier.
+        popPathComponent(elem, objIsArray, path);
+    }
+}
+
+bool AllPathsKeyGenerator::_addKeyForNestedArray(BSONElement elem,
+                                                 const FieldRef& fullPath,
+                                                 bool enclosingObjIsArray,
+                                                 BSONObjSet* keys) const {
+    // If this element is an array whose parent is also an array, index it as a value.
+    if (enclosingObjIsArray && elem.type() == BSONType::Array) {
+        _addKey(elem, fullPath, keys);
+        return true;
+    }
+    return false;
+}
+
+void AllPathsKeyGenerator::_addKey(BSONElement elem,
+                                   const FieldRef& fullPath,
+                                   BSONObjSet* keys) const {
+    // AllPaths keys are of the form { "": "path.to.field", "": <collation-aware value> }
+    BSONObjBuilder bob;
+    bob.append("", fullPath.dottedField());
+    CollationIndexKey::collationAwareIndexKeyAppend(elem, _collator, &bob);
+    keys->insert(bob.obj());
+}
+
+void AllPathsKeyGenerator::_addMultiKey(const FieldRef& fullPath,
+                                        MultikeyPathsMock* multikeyPaths) const {
+    // Multikey paths are denoted by an entry of the form { "": 1, "": "path.to.array" }.
+    multikeyPaths->insert(BSON("" << 1 << "" << fullPath.dottedField()));
+}
+
+}  // namespace mongo
diff --git a/src/mongo/db/index/all_paths_key_generator.h b/src/mongo/db/index/all_paths_key_generator.h
new file mode 100644
index 00000000000..278a2e81d1a
--- /dev/null
+++ b/src/mongo/db/index/all_paths_key_generator.h
@@ -0,0 +1,91 @@
+/**
+ * Copyright (C) 2018 MongoDB Inc.
+ *
+ * This program is free software: you can redistribute it and/or  modify
+ * it under the terms of the GNU Affero General Public License, version 3,
+ * as published by the Free Software Foundation.
+ *
+ * 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
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * 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 GNU Affero General Public License in all respects
+ * for all of the code used other than as permitted herein. If you modify
+ * file(s) with this exception, you may extend this exception to your
+ * version of the file(s), but you are not obligated to do so. If you do not
+ * wish to do so, delete this exception statement from your version. If you
+ * delete this exception statement from all source files in the program,
+ * then also delete it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/projection_exec_agg.h"
+#include "mongo/db/field_ref.h"
+#include "mongo/db/query/collation/collator_interface.h"
+
+namespace mongo {
+
+/**
+ * This class is responsible for generating an aggregation projection based on the keyPattern and
+ * pathProjection specs, and for subsequently extracting the set of all path-value pairs for each
+ * document.
+ */
+class AllPathsKeyGenerator {
+public:
+    static constexpr StringData kSubtreeSuffix = ".$**"_sd;
+
+    /**
+     * TODO SERVER-35748: Currently, the MultikeyPaths structure used by IndexAccessMethod is not
+     * suitable for tracking multikey paths in AllPaths indexes. In order to keep multikey paths
+     * separate from RecordId keys, and to ensure that both this key generator and the
+     * AllPathsIndexAccessMethod can be trivially switched over to using the new MultikeyPaths
+     * tracker once it is implemented, we use a mock MultikeyPaths here.
+     */
+    using MultikeyPathsMock = BSONObjSet;
+
+    AllPathsKeyGenerator(BSONObj keyPattern,
+                         BSONObj pathProjection,
+                         const CollatorInterface* collator);
+
+    /**
+     * Applies the appropriate AllPaths projection to the input doc, and then adds one key-value
+     * pair to the BSONObjSet 'keys' for each leaf node in the post-projection document:
+     *      { '': 'path.to.field', '': <collation-aware-field-value> }
+     * Also adds one entry to 'multikeyPaths' for each array encountered in the post-projection
+     * document, in the following format:
+     *      { '': 1, '': 'path.to.array' }
+     */
+    void generateKeys(BSONObj inputDoc, BSONObjSet* keys, MultikeyPathsMock* multikeyPaths) const;
+
+private:
+    // Traverses every path of the post-projection document, adding keys to the set as it goes.
+    void _traverseAllPaths(BSONObj obj,
+                           bool objIsArray,
+                           FieldRef* path,
+                           BSONObjSet* keys,
+                           MultikeyPathsMock* multikeyPaths) const;
+
+    // Helper functions to format the entry appropriately before adding it to the key/path tracker.
+    void _addMultiKey(const FieldRef& fullPath, MultikeyPathsMock* multikeyPaths) const;
+    void _addKey(BSONElement elem, const FieldRef& fullPath, BSONObjSet* keys) const;
+
+    // Helper to check whether the element is a nested array, and conditionally add it to 'keys'.
+    bool _addKeyForNestedArray(BSONElement elem,
+                               const FieldRef& fullPath,
+                               bool enclosingObjIsArray,
+                               BSONObjSet* keys) const;
+
+    std::unique_ptr<ProjectionExecAgg> _projExec;
+    const CollatorInterface* _collator;
+    const BSONObj _keyPattern;
+};
+}  // namespace mongo
diff --git a/src/mongo/db/index/all_paths_key_generator_test.cpp b/src/mongo/db/index/all_paths_key_generator_test.cpp
new file mode 100644
index 00000000000..59b0d32cd9c
--- /dev/null
+++ b/src/mongo/db/index/all_paths_key_generator_test.cpp
@@ -0,0 +1,906 @@
+/**
+ * Copyright (C) 2018 MongoDB Inc.
+ *
+ * This program is free software: you can redistribute it and/or  modify
+ * it under the terms of the GNU Affero General Public License, version 3,
+ * as published by the Free Software Foundation.
+ *
+ * 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
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * 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 GNU Affero General Public License in all respects
+ * for all of the code used other than as permitted herein. If you modify
+ * file(s) with this exception, you may extend this exception to your
+ * version of the file(s), but you are not obligated to do so. If you do not
+ * wish to do so, delete this exception statement from your version. If you
+ * delete this exception statement from all source files in the program,
+ * then also delete it in the license file.
+ */
+
+#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kIndex
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/bson/json.h"
+#include "mongo/db/index/all_paths_key_generator.h"
+#include "mongo/db/query/collation/collator_interface_mock.h"
+#include "mongo/unittest/unittest.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+namespace {
+
+BSONObjSet makeKeySet(std::initializer_list<BSONObj> init = {}) {
+    return SimpleBSONObjComparator::kInstance.makeBSONObjSet(std::move(init));
+}
+
+std::string dumpKeyset(const BSONObjSet& objs) {
+    std::stringstream ss;
+    ss << "[ ";
+    for (BSONObjSet::iterator i = objs.begin(); i != objs.end(); ++i) {
+        ss << i->toString() << " ";
+    }
+    ss << "]";
+
+    return ss.str();
+}
+
+bool assertKeysetsEqual(const BSONObjSet& expectedKeys, const BSONObjSet& actualKeys) {
+    if (expectedKeys.size() != actualKeys.size()) {
+        log() << "Expected: " << dumpKeyset(expectedKeys) << ", "
+              << "Actual: " << dumpKeyset(actualKeys);
+        return false;
+    }
+
+    if (!std::equal(expectedKeys.begin(),
+                    expectedKeys.end(),
+                    actualKeys.begin(),
+                    SimpleBSONObjComparator::kInstance.makeEqualTo())) {
+        log() << "Expected: " << dumpKeyset(expectedKeys) << ", "
+              << "Actual: " << dumpKeyset(actualKeys);
+        return false;
+    }
+
+    return true;
+}
+
+// Full-document tests with no projection.
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, ExtractTopLevelKey) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+    auto inputDoc = fromjson("{a: 1}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}")});
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, ExtractKeysFromNestedObject) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+    auto inputDoc = fromjson("{a: {b: 'one', c: 2}}");
+
+    auto expectedKeys =
+        makeKeySet({fromjson("{'': 'a.b', '': 'one'}"), fromjson("{'': 'a.c', '': 2}")});
+
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, DoNotExtractKeyForEmptyObject) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+    auto inputDoc = fromjson("{a: 1, b: {}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}")});
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, ExtractMultikeyPath) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+    auto inputDoc = fromjson("{a: [1, 2, {b: 'one', c: 2}, {d: 3}]}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.d', '': 3}")});
+
+    auto expectedMultikeyPaths = makeKeySet({fromjson("{'': 1, '': 'a'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, ExtractMultikeyPathAndDedupKeys) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+    auto inputDoc = fromjson("{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {d: 3}]}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.d', '': 3}")});
+
+    auto expectedMultikeyPaths = makeKeySet({fromjson("{'': 1, '': 'a'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, ExtractZeroElementMultikeyPath) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+    auto inputDoc = fromjson("{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {d: 3}], e: []}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.d', '': 3}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, ExtractNestedMultikeyPaths) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+
+    // Note: the 'e' array is nested within a subdocument in the enclosing 'a' array; it will
+    // generate a separate multikey entry 'a.e' and index keys for each of its elements. The raw
+    // array nested directly within the 'a' array will not, because the indexing system does not
+    // descend nested arrays without an intervening path component.
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]]}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'two'}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'a.e', '': 5}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorFullDocumentTest, ExtractMixedPathTypesAndAllSubpaths) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+
+    // Tests a mix of multikey paths, various duplicate-key scenarios, and deeply-nested paths.
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'two'}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'a.e', '': 5}"),
+                                    fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}"),
+                                    fromjson("{'': 'l', '': 'string'}")});
+
+    auto expectedMultikeyPaths = makeKeySet({fromjson("{'': 1, '': 'a'}"),
+                                             fromjson("{'': 1, '': 'a.e'}"),
+                                             fromjson("{'': 1, '': 'g.h.j'}"),
+                                             fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+// Single-subtree implicit projection.
+
+TEST(AllPathsKeyGeneratorSingleSubtreeTest, ExtractSubtreeWithSinglePathComponent) {
+    AllPathsKeyGenerator keyGen{fromjson("{'g.$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'g.h.j'}"), fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorSingleSubtreeTest, ExtractSubtreeWithMultiplePathComponents) {
+    AllPathsKeyGenerator keyGen{fromjson("{'g.h.$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'g.h.j'}"), fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorSingleSubtreeTest, ExtractMultikeySubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'g.h.j.$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'g.h.j'}"), fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorSingleSubtreeTest, ExtractNestedMultikeySubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'a.e.$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    // We project through the 'a' array to the nested 'e' array. Both 'a' and 'a.e' are added as
+    // multikey paths.
+    auto expectedKeys =
+        makeKeySet({fromjson("{'': 'a.e', '': 4}"), fromjson("{'': 'a.e', '': 5}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+// Explicit inclusion tests.
+
+TEST(AllPathsKeyGeneratorInclusionTest, InclusionProjectionSingleSubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{g: 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'g.h.j'}"), fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorInclusionTest, InclusionProjectionNestedSubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'g.h': 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'g.h.j'}"), fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorInclusionTest, InclusionProjectionMultikeySubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'g.h.j': 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'g.h.j'}"), fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorInclusionTest, InclusionProjectionNestedMultikeySubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'a.e': 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys =
+        makeKeySet({fromjson("{'': 'a.e', '': 4}"), fromjson("{'': 'a.e', '': 5}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorInclusionTest, InclusionProjectionMultipleSubtrees) {
+    AllPathsKeyGenerator keyGen{
+        fromjson("{'$**': 1}"), fromjson("{'a.b': 1, 'a.c': 1, 'a.e': 1, 'g.h.i': 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'two'}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'a.e', '': 5}"),
+                                    fromjson("{'': 'g.h.i', '': 9}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+// Explicit exclusion tests.
+
+TEST(AllPathsKeyGeneratorExclusionTest, ExclusionProjectionSingleSubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{g: 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'two'}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'a.e', '': 5}"),
+                                    fromjson("{'': 'l', '': 'string'}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorExclusionTest, ExclusionProjectionNestedSubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'g.h': 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'two'}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'a.e', '': 5}"),
+                                    fromjson("{'': 'l', '': 'string'}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorExclusionTest, ExclusionProjectionMultikeySubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'g.h.j': 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'two'}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'a.e', '': 5}"),
+                                    fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}"),
+                                    fromjson("{'': 'l', '': 'string'}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorExclusionTest, ExclusionProjectionNestedMultikeySubtree) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'a.e': 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 'one'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'two'}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}"),
+                                    fromjson("{'': 'g.h.k', '': 12}"),
+                                    fromjson("{'': 'l', '': 'string'}")});
+
+    auto expectedMultikeyPaths = makeKeySet({fromjson("{'': 1, '': 'a'}"),
+                                             fromjson("{'': 1, '': 'g.h.j'}"),
+                                             fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorExclusionTest, ExclusionProjectionMultipleSubtrees) {
+    AllPathsKeyGenerator keyGen{
+        fromjson("{'$**': 1}"), fromjson("{'a.b': 0, 'a.c': 0, 'a.e': 0, 'g.h.i': 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{a: [1, 2, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: 3, e: [4, 5]}, [6, 7, {f: 8}]], "
+        "g: {h: {i: 9, j: [10, {k: 11}, {k: [11.5]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': 2}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11.5}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}"),
+                                    fromjson("{'': 'l', '': 'string'}")});
+
+    auto expectedMultikeyPaths = makeKeySet({fromjson("{'': 1, '': 'a'}"),
+                                             fromjson("{'': 1, '': 'g.h.j'}"),
+                                             fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+// Test _id inclusion and exclusion behaviour.
+
+TEST(AllPathsKeyGeneratorIdTest, ExcludeIdFieldIfProjectionIsEmpty) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 1}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, ExcludeIdFieldForSingleSubtreeKeyPattern) {
+    AllPathsKeyGenerator keyGen{fromjson("{'a.$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 1}"),
+                                    fromjson("{'': 'a.e', '': 4}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, PermitIdFieldAsSingleSubtreeKeyPattern) {
+    AllPathsKeyGenerator keyGen{fromjson("{'_id.$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys =
+        makeKeySet({fromjson("{'': '_id.id1', '': 1}"), fromjson("{'': '_id.id2', '': 2}")});
+
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, PermitIdSubfieldAsSingleSubtreeKeyPattern) {
+    AllPathsKeyGenerator keyGen{fromjson("{'_id.id1.$**': 1}"), {}, nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': '_id.id1', '': 1}")});
+
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, ExcludeIdFieldByDefaultForInclusionProjection) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{a: 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 1}"),
+                                    fromjson("{'': 'a.e', '': 4}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, PermitIdSubfieldInclusionInExplicitProjection) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'_id.id1': 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': '_id.id1', '': 1}")});
+
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, ExcludeIdFieldByDefaultForExclusionProjection) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{a: 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys =
+        makeKeySet({fromjson("{'': 'g.h.i', '': 9}"), fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, PermitIdSubfieldExclusionInExplicitProjection) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{'_id.id1': 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': '_id.id2', '': 2}"),
+                                    fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 1}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, IncludeIdFieldIfExplicitlySpecifiedInProjection) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{_id: 1, a: 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': '_id.id1', '': 1}"),
+                                    fromjson("{'': '_id.id2', '': 2}"),
+                                    fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 1}"),
+                                    fromjson("{'': 'a.e', '': 4}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, ExcludeIdFieldIfExplicitlySpecifiedInProjection) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{_id: 0, a: 1}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 1}"),
+                                    fromjson("{'': 'a.e', '': 4}")});
+
+    auto expectedMultikeyPaths =
+        makeKeySet({fromjson("{'': 1, '': 'a'}"), fromjson("{'': 1, '': 'a.e'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+TEST(AllPathsKeyGeneratorIdTest, IncludeIdFieldIfExplicitlySpecifiedInExclusionProjection) {
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), fromjson("{_id: 1, a: 0}"), nullptr};
+
+    auto inputDoc = fromjson(
+        "{_id: {id1: 1, id2: 2}, a: [1, {b: 1, e: [4]}, [6, 7, {f: 8}]], g: {h: {i: 9, k: 12.0}}}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': '_id.id1', '': 1}"),
+                                    fromjson("{'': '_id.id2', '': 2}"),
+                                    fromjson("{'': 'g.h.i', '': 9}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}")});
+
+    auto expectedMultikeyPaths = makeKeySet();
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+// Collation tests.
+
+TEST(AllPathsKeyGeneratorCollationTest, CollationMixedPathAndKeyTypes) {
+    CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kReverseString);
+    AllPathsKeyGenerator keyGen{fromjson("{'$**': 1}"), {}, &collator};
+
+    // Verify that the collation is only applied to String values, but all types are indexed.
+    auto dateVal = "{'$date': 1529453450288}"_sd;
+    auto oidVal = "{'$oid': '520e6431b7fa4ea22d6b1872'}"_sd;
+    auto tsVal = "{'$timestamp': {'t': 1, 'i': 100}}"_sd;
+    auto undefVal = "{'$undefined': true}"_sd;
+
+    auto inputDoc =
+        fromjson("{a: [1, null, {b: 'one', c: 2}, {c: 2, d: 3}, {c: 'two', d: " + dateVal +
+                 ", e: [4, " + oidVal + "]}, [6, 7, {f: 8}]], g: {h: {i: " + tsVal +
+                 ", j: [10, {k: 11}, {k: [" + undefVal + "]}], k: 12.0}}, l: 'string'}");
+
+    auto expectedKeys = makeKeySet({fromjson("{'': 'a', '': 1}"),
+                                    fromjson("{'': 'a', '': null}"),
+                                    fromjson("{'': 'a', '': [6, 7, {f: 8}]}"),
+                                    fromjson("{'': 'a.b', '': 'eno'}"),
+                                    fromjson("{'': 'a.c', '': 2}"),
+                                    fromjson("{'': 'a.c', '': 'owt'}"),
+                                    fromjson("{'': 'a.d', '': 3}"),
+                                    fromjson("{'': 'a.d', '': " + dateVal + "}"),
+                                    fromjson("{'': 'a.e', '': 4}"),
+                                    fromjson("{'': 'a.e', '': " + oidVal + "}"),
+                                    fromjson("{'': 'g.h.i', '': " + tsVal + "}"),
+                                    fromjson("{'': 'g.h.j', '': 10}"),
+                                    fromjson("{'': 'g.h.j.k', '': 11}"),
+                                    fromjson("{'': 'g.h.j.k', '': " + undefVal + "}"),
+                                    fromjson("{'': 'g.h.k', '': 12.0}"),
+                                    fromjson("{'': 'l', '': 'gnirts'}")});
+
+    auto expectedMultikeyPaths = makeKeySet({fromjson("{'': 1, '': 'a'}"),
+                                             fromjson("{'': 1, '': 'a.e'}"),
+                                             fromjson("{'': 1, '': 'g.h.j'}"),
+                                             fromjson("{'': 1, '': 'g.h.j.k'}")});
+
+    auto outputKeys = makeKeySet();
+    auto multikeyPathsMock = makeKeySet();
+    keyGen.generateKeys(inputDoc, &outputKeys, &multikeyPathsMock);
+
+    ASSERT(assertKeysetsEqual(expectedKeys, outputKeys));
+    ASSERT(assertKeysetsEqual(expectedMultikeyPaths, multikeyPathsMock));
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/index/index_descriptor.cpp b/src/mongo/db/index/index_descriptor.cpp
index 76cccb142ae..1514acd6f81 100644
--- a/src/mongo/db/index/index_descriptor.cpp
+++ b/src/mongo/db/index/index_descriptor.cpp
@@ -90,6 +90,7 @@ constexpr StringData IndexDescriptor::kKeyPatternFieldName;
 constexpr StringData IndexDescriptor::kLanguageOverrideFieldName;
 constexpr StringData IndexDescriptor::kNamespaceFieldName;
 constexpr StringData IndexDescriptor::kPartialFilterExprFieldName;
+constexpr StringData IndexDescriptor::kPathProjectionFieldName;
 constexpr StringData IndexDescriptor::kSparseFieldName;
 constexpr StringData IndexDescriptor::kStorageEngineFieldName;
 constexpr StringData IndexDescriptor::kTextVersionFieldName;
diff --git a/src/mongo/db/index/index_descriptor.h b/src/mongo/db/index/index_descriptor.h
index b62cbeb9e38..d6945fe7301 100644
--- a/src/mongo/db/index/index_descriptor.h
+++ b/src/mongo/db/index/index_descriptor.h
@@ -1,32 +1,32 @@
 // index_descriptor.cpp
 
 /**
-*    Copyright (C) 2013 10gen Inc.
-*
-*    This program is free software: you can redistribute it and/or  modify
-*    it under the terms of the GNU Affero General Public License, version 3,
-*    as published by the Free Software Foundation.
-*
-*    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
-*    GNU Affero General Public License for more details.
-*
-*    You should have received a copy of the GNU Affero General Public License
-*    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*
-*    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 GNU Affero General 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.
-*/
+ *    Copyright (C) 2013 10gen Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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
 
@@ -77,6 +77,7 @@ public:
     static constexpr StringData kLanguageOverrideFieldName = "language_override"_sd;
     static constexpr StringData kNamespaceFieldName = "ns"_sd;
     static constexpr StringData kPartialFilterExprFieldName = "partialFilterExpression"_sd;
+    static constexpr StringData kPathProjectionFieldName = "starPathsTempName"_sd;
     static constexpr StringData kSparseFieldName = "sparse"_sd;
     static constexpr StringData kStorageEngineFieldName = "storageEngine"_sd;
     static constexpr StringData kTextVersionFieldName = "textIndexVersion"_sd;
@@ -93,6 +94,7 @@ public:
           _infoObj(infoObj.getOwned()),
           _numFields(infoObj.getObjectField(IndexDescriptor::kKeyPatternFieldName).nFields()),
           _keyPattern(infoObj.getObjectField(IndexDescriptor::kKeyPatternFieldName).getOwned()),
+          _projection(infoObj.getObjectField(IndexDescriptor::kPathProjectionFieldName).getOwned()),
           _indexName(infoObj.getStringField(IndexDescriptor::kIndexNameFieldName)),
           _parentNS(infoObj.getStringField(IndexDescriptor::kNamespaceFieldName)),
           _isIdIndex(isIdIndexPattern(_keyPattern)),
@@ -148,6 +150,13 @@ public:
     }
 
     /**
+     * Return the path projection spec, if one exists. This is only applicable for '$**' indexes.
+     */
+    const BSONObj& pathProjection() const {
+        return _projection;
+    }
+
+    /**
      * Test only command for testing behavior resulting from an incorrect key
      * pattern.
      */
@@ -278,6 +287,7 @@ private:
 
     int64_t _numFields;  // How many fields are indexed?
     BSONObj _keyPattern;
+    BSONObj _projection;
     std::string _indexName;
     std::string _parentNS;
     std::string _indexNamespace;
diff --git a/src/mongo/db/pipeline/document_source_project.cpp b/src/mongo/db/pipeline/document_source_project.cpp
index f5d560c4c9d..61bdfb41fe2 100644
--- a/src/mongo/db/pipeline/document_source_project.cpp
+++ b/src/mongo/db/pipeline/document_source_project.cpp
@@ -41,6 +41,9 @@ namespace mongo {
 using boost::intrusive_ptr;
 using parsed_aggregation_projection::ParsedAggregationProjection;
 
+using ProjectionArrayRecursionPolicy = ParsedAggregationProjection::ProjectionArrayRecursionPolicy;
+using ProjectionDefaultIdPolicy = ParsedAggregationProjection::ProjectionDefaultIdPolicy;
+
 REGISTER_DOCUMENT_SOURCE(project,
                          LiteParsedDocumentSourceDefault::parse,
                          DocumentSourceProject::createFromBson);
@@ -50,7 +53,10 @@ intrusive_ptr<DocumentSource> DocumentSourceProject::create(
     const bool isIndependentOfAnyCollection = false;
     intrusive_ptr<DocumentSource> project(new DocumentSourceSingleDocumentTransformation(
         expCtx,
-        ParsedAggregationProjection::create(expCtx, projectSpec),
+        ParsedAggregationProjection::create(expCtx,
+                                            projectSpec,
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays),
         "$project",
         isIndependentOfAnyCollection));
     return project;
diff --git a/src/mongo/db/pipeline/parsed_add_fields.h b/src/mongo/db/pipeline/parsed_add_fields.h
index a7fa9c3e0eb..18e8cfbe514 100644
--- a/src/mongo/db/pipeline/parsed_add_fields.h
+++ b/src/mongo/db/pipeline/parsed_add_fields.h
@@ -51,8 +51,15 @@ namespace parsed_aggregation_projection {
  */
 class ParsedAddFields : public ParsedAggregationProjection {
 public:
+    /**
+     * TODO SERVER-25510: The ParsedAggregationProjection _id and array-recursion policies are not
+     * applicable to the $addFields "projection" stage. We make them non-configurable here.
+     */
     ParsedAddFields(const boost::intrusive_ptr<ExpressionContext>& expCtx)
-        : ParsedAggregationProjection(expCtx), _root(new InclusionNode()) {}
+        : ParsedAggregationProjection(expCtx,
+                                      ProjectionDefaultIdPolicy::kIncludeId,
+                                      ProjectionArrayRecursionPolicy::kRecurseNestedArrays),
+          _root(new InclusionNode(_arrayRecursionPolicy)) {}
 
     /**
      * Creates the data needed to perform an AddFields.
diff --git a/src/mongo/db/pipeline/parsed_aggregation_projection.cpp b/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
index 5e106efe1e4..f600da57322 100644
--- a/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
+++ b/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
@@ -236,19 +236,30 @@ private:
                 elem.isBoolean() || elem.isNumber() ||
                     _parseMode != ProjectionParseMode::kBanComputedFields);
 
-        if ((elem.isBoolean() || elem.isNumber()) && !elem.trueValue()) {
-            // A top-level exclusion of "_id" is allowed in either an inclusion projection or an
-            // exclusion projection, so doesn't affect '_parsedType'.
-            if (pathToElem.fullPath() != "_id") {
-                uassert(40178,
-                        str::stream() << "Bad projection specification, cannot exclude fields "
-                                         "other than '_id' in an inclusion projection: "
+        if (pathToElem.fullPath() == "_id") {
+            // If the _id field is a computed value, then this must be an inclusion projection. If
+            // it is numeric or boolean, then this does not determine the projection type, due to
+            // the fact that inclusions may explicitly exclude _id and exclusions may include _id.
+            if (!elem.isBoolean() && !elem.isNumber()) {
+                uassert(ErrorCodes::FailedToParse,
+                        str::stream() << "Bad projection specification, '_id' may not be a "
+                                         "computed field in an exclusion projection: "
                                       << _rawObj.toString(),
                         !_parsedType ||
-                            (*_parsedType ==
-                             TransformerInterface::TransformerType::kExclusionProjection));
-                _parsedType = TransformerInterface::TransformerType::kExclusionProjection;
+                            _parsedType ==
+                                TransformerInterface::TransformerType::kInclusionProjection);
+                _parsedType = TransformerInterface::TransformerType::kInclusionProjection;
             }
+        } else if ((elem.isBoolean() || elem.isNumber()) && !elem.trueValue()) {
+            // If this is an excluded field other than '_id', ensure that the projection type has
+            // not already been set to kInclusionProjection.
+            uassert(40178,
+                    str::stream() << "Bad projection specification, cannot exclude fields "
+                                     "other than '_id' in an inclusion projection: "
+                                  << _rawObj.toString(),
+                    !_parsedType || (*_parsedType ==
+                                     TransformerInterface::TransformerType::kExclusionProjection));
+            _parsedType = TransformerInterface::TransformerType::kExclusionProjection;
         } else {
             // A boolean true, a truthy numeric value, or any expression can only be used with an
             // inclusion projection. Note that literal values like "string" or null are also treated
@@ -310,6 +321,8 @@ private:
 std::unique_ptr<ParsedAggregationProjection> ParsedAggregationProjection::create(
     const boost::intrusive_ptr<ExpressionContext>& expCtx,
     const BSONObj& spec,
+    ProjectionDefaultIdPolicy defaultIdPolicy,
+    ProjectionArrayRecursionPolicy arrayRecursionPolicy,
     ProjectionParseMode parseMode) {
     // Check that the specification was valid. Status returned is unspecific because validate()
     // is used by the $addFields stage as well as $project.
@@ -325,8 +338,10 @@ std::unique_ptr<ParsedAggregationProjection> ParsedAggregationProjection::create
     // We can't use make_unique() here, since the branches have different types.
     std::unique_ptr<ParsedAggregationProjection> parsedProject(
         projectionType == TransformerType::kInclusionProjection
-            ? static_cast<ParsedAggregationProjection*>(new ParsedInclusionProjection(expCtx))
-            : static_cast<ParsedAggregationProjection*>(new ParsedExclusionProjection(expCtx)));
+            ? static_cast<ParsedAggregationProjection*>(
+                  new ParsedInclusionProjection(expCtx, defaultIdPolicy, arrayRecursionPolicy))
+            : static_cast<ParsedAggregationProjection*>(
+                  new ParsedExclusionProjection(expCtx, defaultIdPolicy, arrayRecursionPolicy)));
 
     // Actually parse the specification.
     parsedProject->parse(spec);
diff --git a/src/mongo/db/pipeline/parsed_aggregation_projection.h b/src/mongo/db/pipeline/parsed_aggregation_projection.h
index f3ec7f23c85..6c0db1070d7 100644
--- a/src/mongo/db/pipeline/parsed_aggregation_projection.h
+++ b/src/mongo/db/pipeline/parsed_aggregation_projection.h
@@ -140,6 +140,20 @@ private:
  */
 class ParsedAggregationProjection : public TransformerInterface {
 public:
+    // Allows the caller to indicate whether the projection should default to including or excluding
+    // the _id field in the event that the projection spec does not specify the desired behavior.
+    // For instance, given a projection {a: 1}, specifying 'kExcludeId' is equivalent to projecting
+    // {a: 1, _id: 0} while 'kIncludeId' is equivalent to the projection {a: 1, _id: 1}. If the user
+    // explicitly specifies a projection on _id, then this will override the default policy; for
+    // instance, {a: 1, _id: 0} will exclude _id for both 'kExcludeId' and 'kIncludeId'.
+    enum class ProjectionDefaultIdPolicy { kIncludeId, kExcludeId };
+
+    // Allows the caller to specify how the projection should handle nested arrays; that is, an
+    // array whose immediate parent is itself an array. For example, in the case of sample document
+    // {a: [1, 2, [3, 4], {b: [5, 6]}]} the array [3, 4] is a nested array. The array [5, 6] is not,
+    // because there is an intervening object between it and its closest array ancestor.
+    enum class ProjectionArrayRecursionPolicy { kRecurseNestedArrays, kDoNotRecurseNestedArrays };
+
     // Allows the caller to specify whether computed fields should be allowed within inclusion
     // projections; they are implicitly prohibited within exclusion projections.
     enum class ProjectionParseMode {
@@ -155,6 +169,8 @@ public:
     static std::unique_ptr<ParsedAggregationProjection> create(
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
         const BSONObj& spec,
+        ProjectionDefaultIdPolicy defaultIdPolicy,
+        ProjectionArrayRecursionPolicy arrayRecursionPolicy,
         ProjectionParseMode parseRules = ProjectionParseMode::kAllowComputedFields);
 
     virtual ~ParsedAggregationProjection() = default;
@@ -187,8 +203,12 @@ public:
     }
 
 protected:
-    ParsedAggregationProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx)
-        : _expCtx(expCtx){};
+    ParsedAggregationProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                ProjectionDefaultIdPolicy defaultIdPolicy,
+                                ProjectionArrayRecursionPolicy arrayRecursionPolicy)
+        : _expCtx(expCtx),
+          _arrayRecursionPolicy(arrayRecursionPolicy),
+          _defaultIdPolicy(defaultIdPolicy){};
 
     /**
      * Apply the projection to 'input'.
@@ -196,6 +216,9 @@ protected:
     virtual Document applyProjection(const Document& input) const = 0;
 
     boost::intrusive_ptr<ExpressionContext> _expCtx;
+
+    ProjectionArrayRecursionPolicy _arrayRecursionPolicy;
+    ProjectionDefaultIdPolicy _defaultIdPolicy;
 };
 }  // namespace parsed_aggregation_projection
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp b/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp
index d7d50ea7316..3054838f6a9 100644
--- a/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp
+++ b/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp
@@ -44,6 +44,8 @@ namespace mongo {
 namespace parsed_aggregation_projection {
 namespace {
 
+using ProjectionArrayRecursionPolicy = ParsedAggregationProjection::ProjectionArrayRecursionPolicy;
+using ProjectionDefaultIdPolicy = ParsedAggregationProjection::ProjectionDefaultIdPolicy;
 using ProjectionParseMode = ParsedAggregationProjection::ProjectionParseMode;
 
 template <typename T>
@@ -51,6 +53,17 @@ BSONObj wrapInLiteral(const T& arg) {
     return BSON("$literal" << arg);
 }
 
+// Helper to simplify the creation of a ParsedAggregationProjection which includes _id and recurses
+// nested arrays by default.
+std::unique_ptr<ParsedAggregationProjection> makeProjectionWithDefaultPolicies(BSONObj projSpec) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    return ParsedAggregationProjection::create(
+        expCtx,
+        projSpec,
+        ProjectionDefaultIdPolicy::kIncludeId,
+        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+}
+
 //
 // Error cases.
 //
@@ -58,73 +71,69 @@ BSONObj wrapInLiteral(const T& arg) {
 TEST(ParsedAggregationProjectionErrors, ShouldRejectDuplicateFieldNames) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     // Include/exclude the same field twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << true << "a" << true)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << true << "a" << true)),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << false << "a" << false)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << BSON("b" << false << "b" << false))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << false << "a" << false)),
                   AssertionException);
-
-    // Mix of include/exclude and adding a field.
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << wrapInLiteral(1) << "a" << true)),
-        AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << false << "a" << wrapInLiteral(0))),
+        makeProjectionWithDefaultPolicies(BSON("a" << BSON("b" << false << "b" << false))),
         AssertionException);
 
-    // Adding the same field twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << wrapInLiteral(1) << "a" << wrapInLiteral(0))),
+    // Mix of include/exclude and adding a field.
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << wrapInLiteral(1) << "a" << true)),
+                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << false << "a" << wrapInLiteral(0))),
                   AssertionException);
+
+    // Adding the same field twice.
+    ASSERT_THROWS(
+        makeProjectionWithDefaultPolicies(BSON("a" << wrapInLiteral(1) << "a" << wrapInLiteral(0))),
+        AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectDuplicateIds) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     // Include/exclude _id twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("_id" << true << "_id" << true)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("_id" << true << "_id" << true)),
+                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("_id" << false << "_id" << false)),
                   AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("_id" << false << "_id" << false)),
-        AssertionException);
 
     // Mix of including/excluding and adding _id.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("_id" << wrapInLiteral(1) << "_id" << true)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("_id" << false << "_id" << wrapInLiteral(0))),
-                  AssertionException);
+    ASSERT_THROWS(
+        makeProjectionWithDefaultPolicies(BSON("_id" << wrapInLiteral(1) << "_id" << true)),
+        AssertionException);
+    ASSERT_THROWS(
+        makeProjectionWithDefaultPolicies(BSON("_id" << false << "_id" << wrapInLiteral(0))),
+        AssertionException);
 
     // Adding _id twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("_id" << wrapInLiteral(1) << "_id" << wrapInLiteral(0))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("_id" << wrapInLiteral(1) << "_id" << wrapInLiteral(0))),
                   AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectFieldsWithSharedPrefix) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     // Include/exclude Fields with a shared prefix.
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << true << "a.b" << true)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << true << "a.b" << true)),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a.b" << false << "a" << false)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a.b" << false << "a" << false)),
                   AssertionException);
 
     // Mix of include/exclude and adding a shared prefix.
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << wrapInLiteral(1) << "a.b" << true)),
+                  AssertionException);
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << wrapInLiteral(1) << "a.b" << true)),
+        makeProjectionWithDefaultPolicies(BSON("a.b" << false << "a" << wrapInLiteral(0))),
         AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a.b" << false << "a" << wrapInLiteral(0))),
-                  AssertionException);
 
     // Adding a shared prefix twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << wrapInLiteral(1) << "a.b" << wrapInLiteral(0))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << wrapInLiteral(1) << "a.b" << wrapInLiteral(0))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a.b.c.d" << wrapInLiteral(1) << "a.b.c" << wrapInLiteral(0))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a.b.c.d" << wrapInLiteral(1) << "a.b.c" << wrapInLiteral(0))),
                   AssertionException);
 }
 
@@ -132,44 +141,39 @@ TEST(ParsedAggregationProjectionErrors, ShouldRejectPathConflictsWithNonAlphaNum
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     // Include/exclude non-alphanumeric fields with a shared prefix. First assert that the non-
     // alphanumeric fields are accepted when no prefixes are present.
-    ASSERT(ParsedAggregationProjection::create(
-        expCtx, BSON("a.b-c" << true << "a.b" << true << "a.b?c" << true << "a.b c" << true)));
-    ASSERT(ParsedAggregationProjection::create(
-        expCtx, BSON("a.b c" << false << "a.b?c" << false << "a.b" << false << "a.b-c" << false)));
+    ASSERT(makeProjectionWithDefaultPolicies(
+        BSON("a.b-c" << true << "a.b" << true << "a.b?c" << true << "a.b c" << true)));
+    ASSERT(makeProjectionWithDefaultPolicies(
+        BSON("a.b c" << false << "a.b?c" << false << "a.b" << false << "a.b-c" << false)));
 
     // Then assert that we throw when we introduce a prefixed field.
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(
-            expCtx,
+        makeProjectionWithDefaultPolicies(
             BSON("a.b-c" << true << "a.b" << true << "a.b?c" << true << "a.b c" << true << "a.b.d"
                          << true)),
         AssertionException);
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(
-            expCtx,
-            BSON("a.b.d" << false << "a.b c" << false << "a.b?c" << false << "a.b" << false
-                         << "a.b-c"
-                         << false)),
+        makeProjectionWithDefaultPolicies(BSON(
+            "a.b.d" << false << "a.b c" << false << "a.b?c" << false << "a.b" << false << "a.b-c"
+                    << false)),
         AssertionException);
 
     // Adding the same field twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a.b?c" << wrapInLiteral(1) << "a.b?c" << wrapInLiteral(0))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a.b?c" << wrapInLiteral(1) << "a.b?c" << wrapInLiteral(0))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a.b c" << wrapInLiteral(0) << "a.b c" << wrapInLiteral(1))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a.b c" << wrapInLiteral(0) << "a.b c" << wrapInLiteral(1))),
                   AssertionException);
 
     // Mix of include/exclude and adding a shared prefix.
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(
-            expCtx,
+        makeProjectionWithDefaultPolicies(
             BSON("a.b-c" << true << "a.b" << wrapInLiteral(1) << "a.b?c" << true << "a.b c" << true
                          << "a.b.d"
                          << true)),
         AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx,
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
                       BSON("a.b.d" << false << "a.b c" << false << "a.b?c" << false << "a.b"
                                    << wrapInLiteral(0)
                                    << "a.b-c"
@@ -177,8 +181,7 @@ TEST(ParsedAggregationProjectionErrors, ShouldRejectPathConflictsWithNonAlphaNum
                   AssertionException);
 
     // Adding a shared prefix twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx,
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
                       BSON("a.b-c" << wrapInLiteral(1) << "a.b" << wrapInLiteral(1) << "a.b?c"
                                    << wrapInLiteral(1)
                                    << "a.b c"
@@ -186,8 +189,7 @@ TEST(ParsedAggregationProjectionErrors, ShouldRejectPathConflictsWithNonAlphaNum
                                    << "a.b.d"
                                    << wrapInLiteral(0))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx,
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
                       BSON("a.b.d" << wrapInLiteral(1) << "a.b c" << wrapInLiteral(1) << "a.b?c"
                                    << wrapInLiteral(1)
                                    << "a.b"
@@ -200,232 +202,206 @@ TEST(ParsedAggregationProjectionErrors, ShouldRejectPathConflictsWithNonAlphaNum
 TEST(ParsedAggregationProjectionErrors, ShouldRejectMixOfIdAndSubFieldsOfId) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     // Include/exclude _id twice.
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("_id" << true << "_id.x" << true)),
+                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("_id.x" << false << "_id" << false)),
+                  AssertionException);
+
+    // Mix of including/excluding and adding _id.
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("_id" << true << "_id.x" << true)),
+        makeProjectionWithDefaultPolicies(BSON("_id" << wrapInLiteral(1) << "_id.x" << true)),
         AssertionException);
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("_id.x" << false << "_id" << false)),
+        makeProjectionWithDefaultPolicies(BSON("_id.x" << false << "_id" << wrapInLiteral(0))),
         AssertionException);
 
-    // Mix of including/excluding and adding _id.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("_id" << wrapInLiteral(1) << "_id.x" << true)),
+    // Adding _id twice.
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("_id" << wrapInLiteral(1) << "_id.x" << wrapInLiteral(0))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("_id.x" << false << "_id" << wrapInLiteral(0))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("_id.b.c.d" << wrapInLiteral(1) << "_id.b.c" << wrapInLiteral(0))),
                   AssertionException);
+}
 
-    // Adding _id twice.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("_id" << wrapInLiteral(1) << "_id.x" << wrapInLiteral(0))),
-                  AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(
-            expCtx, BSON("_id.b.c.d" << wrapInLiteral(1) << "_id.b.c" << wrapInLiteral(0))),
-        AssertionException);
+TEST(ParsedAggregationProjectionErrors, ShouldAllowMixOfIdInclusionAndExclusion) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+
+    // Mixing "_id" inclusion with exclusion.
+    auto parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << true << "a" << false));
+    ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
+
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("a" << false << "_id" << true));
+    ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
+
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << true << "a.b.c" << false));
+    ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectMixOfInclusionAndExclusion) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     // Simple mix.
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << true << "b" << false)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << true << "b" << false)),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << false << "b" << true)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << false << "b" << true)),
                   AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << BSON("b" << false << "c" << true))),
-        AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("_id" << BSON("b" << false << "c" << true))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << BSON("b" << false << "c" << true))),
                   AssertionException);
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("_id.b" << false << "a.c" << true)),
+        makeProjectionWithDefaultPolicies(BSON("_id" << BSON("b" << false << "c" << true))),
         AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("_id.b" << false << "a.c" << true)),
+                  AssertionException);
 
     // Mix while also adding a field.
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << true << "b" << wrapInLiteral(1) << "c" << false)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << true << "b" << wrapInLiteral(1) << "c" << false)),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << false << "b" << wrapInLiteral(1) << "c" << true)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << false << "b" << wrapInLiteral(1) << "c" << true)),
                   AssertionException);
 
-    // Mixing "_id" inclusion with exclusion.
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("_id" << true << "a" << false)),
+    // Mix of "_id" subfield inclusion and exclusion.
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("_id.x" << true << "a.b.c" << false)),
                   AssertionException);
+}
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << false << "_id" << true)),
+TEST(ParsedAggregationProjectionErrors, ShouldRejectMixOfExclusionAndComputedFields) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << false << "b" << wrapInLiteral(1))),
                   AssertionException);
 
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("_id" << true << "a.b.c" << false)),
-        AssertionException);
-
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("_id.x" << true << "a.b.c" << false)),
-        AssertionException);
-}
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << wrapInLiteral(1) << "b" << false)),
+                  AssertionException);
 
-TEST(ParsedAggregationProjectionType, ShouldRejectMixOfExclusionAndComputedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << false << "b" << wrapInLiteral(1))),
+        makeProjectionWithDefaultPolicies(BSON("a.b" << false << "a.c" << wrapInLiteral(1))),
         AssertionException);
 
     ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << wrapInLiteral(1) << "b" << false)),
+        makeProjectionWithDefaultPolicies(BSON("a.b" << wrapInLiteral(1) << "a.c" << false)),
         AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a.b" << false << "a.c" << wrapInLiteral(1))),
-                  AssertionException);
-
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a.b" << wrapInLiteral(1) << "a.c" << false)),
-                  AssertionException);
-
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << BSON("b" << false << "c" << wrapInLiteral(1)))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << BSON("b" << false << "c" << wrapInLiteral(1)))),
                   AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << false))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << false))),
                   AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectDottedFieldInSubDocument) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << BSON("b.c" << true))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << BSON("b.c" << true))),
+                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << BSON("b.c" << wrapInLiteral(1)))),
                   AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << BSON("b.c" << wrapInLiteral(1)))),
-        AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectFieldNamesStartingWithADollar) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("$dollar" << 0)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("$dollar" << 1)),
-                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("$dollar" << 0)), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("$dollar" << 1)), AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("b.$dollar" << 0)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("b.$dollar" << 1)),
-                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("b.$dollar" << 0)), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("b.$dollar" << 1)), AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("b" << BSON("$dollar" << 0))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("b" << BSON("$dollar" << 0))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("b" << BSON("$dollar" << 1))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("b" << BSON("$dollar" << 1))),
                   AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("$add" << 0)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("$add" << 1)),
-                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("$add" << 0)), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("$add" << 1)), AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectTopLevelExpressions) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("$add" << BSON_ARRAY(4 << 2))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("$add" << BSON_ARRAY(4 << 2))),
                   AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectExpressionWithMultipleFieldNames) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << BSON("$add" << BSON_ARRAY(4 << 2) << "b" << 1))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << BSON("$add" << BSON_ARRAY(4 << 2) << "b" << 1))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << BSON("b" << 1 << "$add" << BSON_ARRAY(4 << 2)))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << BSON("b" << 1 << "$add" << BSON_ARRAY(4 << 2)))),
+                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << BSON("b" << BSON("c" << 1 << "$add" << BSON_ARRAY(4 << 2))))),
+                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << BSON("b" << BSON("$add" << BSON_ARRAY(4 << 2) << "c" << 1)))),
                   AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(
-            expCtx, BSON("a" << BSON("b" << BSON("c" << 1 << "$add" << BSON_ARRAY(4 << 2))))),
-        AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(
-            expCtx, BSON("a" << BSON("b" << BSON("$add" << BSON_ARRAY(4 << 2) << "c" << 1)))),
-        AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectEmptyProjection) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSONObj()), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSONObj()), AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldRejectEmptyNestedObject) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << BSONObj())),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << BSONObj())), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << false << "b" << BSONObj())),
                   AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << false << "b" << BSONObj())),
-        AssertionException);
-    ASSERT_THROWS(
-        ParsedAggregationProjection::create(expCtx, BSON("a" << true << "b" << BSONObj())),
-        AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a.b" << BSONObj())),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << true << "b" << BSONObj())),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << BSON("b" << BSONObj()))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a.b" << BSONObj())), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << BSON("b" << BSONObj()))),
                   AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldErrorOnInvalidExpression) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << false << "b" << BSON("$unknown" << BSON_ARRAY(4 << 2)))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << false << "b" << BSON("$unknown" << BSON_ARRAY(4 << 2)))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(
-                      expCtx, BSON("a" << true << "b" << BSON("$unknown" << BSON_ARRAY(4 << 2)))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(
+                      BSON("a" << true << "b" << BSON("$unknown" << BSON_ARRAY(4 << 2)))),
                   AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldErrorOnInvalidFieldPath) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     // Empty field names.
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("" << wrapInLiteral(2))),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("" << true)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("" << false)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("" << wrapInLiteral(2))),
                   AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("" << true)), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("" << false)), AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << BSON("" << true))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << BSON("" << true))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a" << BSON("" << false))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a" << BSON("" << false))),
                   AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("" << BSON("a" << true))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("" << BSON("a" << true))),
                   AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("" << BSON("a" << false))),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("" << BSON("a" << false))),
                   AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a." << true)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("a." << false)),
-                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a." << true)), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("a." << false)), AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON(".a" << true)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON(".a" << false)),
-                  AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON(".a" << true)), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON(".a" << false)), AssertionException);
 
     // Not testing field names with null bytes, since that is invalid BSON, and won't make it to the
     // $project stage without a previous error.
 
     // Field names starting with '$'.
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("$x" << wrapInLiteral(2))),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("c.$d" << true)),
-                  AssertionException);
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx, BSON("c.$d" << false)),
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("$x" << wrapInLiteral(2))),
                   AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("c.$d" << true)), AssertionException);
+    ASSERT_THROWS(makeProjectionWithDefaultPolicies(BSON("c.$d" << false)), AssertionException);
 }
 
 TEST(ParsedAggregationProjectionErrors, ShouldNotErrorOnTwoNestedFields) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedAggregationProjection::create(expCtx, BSON("a.b" << true << "a.c" << true));
-    ParsedAggregationProjection::create(expCtx, BSON("a.b" << true << "a" << BSON("c" << true)));
+    makeProjectionWithDefaultPolicies(BSON("a.b" << true << "a.c" << true));
+    makeProjectionWithDefaultPolicies(BSON("a.b" << true << "a" << BSON("c" << true)));
 }
 
 //
@@ -434,202 +410,259 @@ TEST(ParsedAggregationProjectionErrors, ShouldNotErrorOnTwoNestedFields) {
 
 TEST(ParsedAggregationProjectionType, ShouldDefaultToInclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << true));
+    auto parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << true));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << wrapInLiteral(1)));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << wrapInLiteral(1)));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("a" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldDetectExclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << false));
+    auto parsedProject = makeProjectionWithDefaultPolicies(BSON("a" << false));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id.x" << false));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id.x" << false));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << BSON("x" << false)));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << BSON("x" << false)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("x" << BSON("_id" << false)));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("x" << BSON("_id" << false)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << false));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << false));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldDetectInclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << true));
+    auto parsedProject = makeProjectionWithDefaultPolicies(BSON("a" << true));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject =
-        ParsedAggregationProjection::create(expCtx, BSON("_id" << false << "a" << true));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << false << "a" << true));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject =
-        ParsedAggregationProjection::create(expCtx, BSON("_id" << false << "a.b.c" << true));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << false << "a.b.c" << true));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id.x" << true));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id.x" << true));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << BSON("x" << true)));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << BSON("x" << true)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("x" << BSON("_id" << true)));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("x" << BSON("_id" << true)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldTreatOnlyComputedFieldsAsAnInclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << wrapInLiteral(1)));
+    auto parsedProject = makeProjectionWithDefaultPolicies(BSON("a" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id" << false << "a" << wrapInLiteral(1)));
+    parsedProject =
+        makeProjectionWithDefaultPolicies(BSON("_id" << false << "a" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id" << false << "a.b.c" << wrapInLiteral(1)));
+    parsedProject =
+        makeProjectionWithDefaultPolicies(BSON("_id" << false << "a.b.c" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id.x" << wrapInLiteral(1)));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id.x" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject =
-        ParsedAggregationProjection::create(expCtx, BSON("_id" << BSON("x" << wrapInLiteral(1))));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("_id" << BSON("x" << wrapInLiteral(1))));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject =
-        ParsedAggregationProjection::create(expCtx, BSON("x" << BSON("_id" << wrapInLiteral(1))));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("x" << BSON("_id" << wrapInLiteral(1))));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldAllowMixOfInclusionAndComputedFields) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     auto parsedProject =
-        ParsedAggregationProjection::create(expCtx, BSON("a" << true << "b" << wrapInLiteral(1)));
+        makeProjectionWithDefaultPolicies(BSON("a" << true << "b" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("a.b" << true << "a.c" << wrapInLiteral(1)));
+    parsedProject =
+        makeProjectionWithDefaultPolicies(BSON("a.b" << true << "a.c" << wrapInLiteral(1)));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("a" << BSON("b" << true << "c" << wrapInLiteral(1))));
+    parsedProject = makeProjectionWithDefaultPolicies(
+        BSON("a" << BSON("b" << true << "c" << wrapInLiteral(1))));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(expCtx,
-                                                        BSON("a" << BSON("b" << true << "c"
-                                                                             << "stringLiteral")));
+    parsedProject = makeProjectionWithDefaultPolicies(BSON("a" << BSON("b" << true << "c"
+                                                                           << "stringLiteral")));
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 }
 
-TEST(ParsedAggregationProjectionType, ShouldRejectMixOfInclusionAndComputedFieldsInStrictMode) {
+TEST(ParsedAggregationProjectionType, ShouldRejectMixOfInclusionAndBannedComputedFields) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx,
-                                                      BSON("a" << true << "b" << wrapInLiteral(1)),
-                                                      ProjectionParseMode::kBanComputedFields),
-                  AssertionException);
+    ASSERT_THROWS(
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("a" << true << "b" << wrapInLiteral(1)),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields),
+        AssertionException);
 
     ASSERT_THROWS(
         ParsedAggregationProjection::create(expCtx,
                                             BSON("a.b" << true << "a.c" << wrapInLiteral(1)),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
                                             ProjectionParseMode::kBanComputedFields),
         AssertionException);
 
     ASSERT_THROWS(ParsedAggregationProjection::create(
                       expCtx,
                       BSON("a" << BSON("b" << true << "c" << wrapInLiteral(1))),
+                      ProjectionDefaultIdPolicy::kIncludeId,
+                      ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
                       ProjectionParseMode::kBanComputedFields),
                   AssertionException);
 
-    ASSERT_THROWS(ParsedAggregationProjection::create(expCtx,
-                                                      BSON("a" << BSON("b" << true << "c"
-                                                                           << "stringLiteral")),
-                                                      ProjectionParseMode::kBanComputedFields),
-                  AssertionException);
+    ASSERT_THROWS(
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("a" << BSON("b" << true << "c"
+                                                                 << "stringLiteral")),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields),
+        AssertionException);
 }
 
-TEST(ParsedAggregationProjectionType, ShouldRejectOnlyComputedFieldsInStrictMode) {
+TEST(ParsedAggregationProjectionType, ShouldRejectOnlyComputedFieldsWhenComputedFieldsAreBanned) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     ASSERT_THROWS(ParsedAggregationProjection::create(
                       expCtx,
                       BSON("a" << wrapInLiteral(1) << "b" << wrapInLiteral(2)),
+                      ProjectionDefaultIdPolicy::kIncludeId,
+                      ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
                       ProjectionParseMode::kBanComputedFields),
                   AssertionException);
 
     ASSERT_THROWS(ParsedAggregationProjection::create(
                       expCtx,
                       BSON("a.b" << wrapInLiteral(1) << "a.c" << wrapInLiteral(2)),
+                      ProjectionDefaultIdPolicy::kIncludeId,
+                      ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
                       ProjectionParseMode::kBanComputedFields),
                   AssertionException);
 
     ASSERT_THROWS(ParsedAggregationProjection::create(
                       expCtx,
                       BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << wrapInLiteral(2))),
+                      ProjectionDefaultIdPolicy::kIncludeId,
+                      ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
                       ProjectionParseMode::kBanComputedFields),
                   AssertionException);
 
     ASSERT_THROWS(ParsedAggregationProjection::create(
                       expCtx,
                       BSON("a" << BSON("b" << wrapInLiteral(1) << "c" << wrapInLiteral(2))),
+                      ProjectionDefaultIdPolicy::kIncludeId,
+                      ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
                       ProjectionParseMode::kBanComputedFields),
                   AssertionException);
 }
 
-TEST(ParsedAggregationProjectionType, ShouldAcceptInclusionProjectionInStrictMode) {
+TEST(ParsedAggregationProjectionType, ShouldAcceptInclusionProjectionWhenComputedFieldsAreBanned) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("a" << true), ProjectionParseMode::kBanComputedFields);
+    auto parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("a" << true),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id" << false << "a" << true), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("_id" << false << "a" << true),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id" << false << "a.b.c" << true), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("_id" << false << "a.b.c" << true),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id.x" << true), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("_id.x" << true),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id" << BSON("x" << true)), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("_id" << BSON("x" << true)),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("x" << BSON("_id" << true)), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("x" << BSON("_id" << true)),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kInclusionProjection);
 }
 
-TEST(ParsedAggregationProjectionType, ShouldAcceptExclusionProjectionInStrictMode) {
+TEST(ParsedAggregationProjectionType, ShouldAcceptExclusionProjectionWhenComputedFieldsAreBanned) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    auto parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("a" << false), ProjectionParseMode::kBanComputedFields);
+    auto parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("a" << false),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id.x" << false), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("_id.x" << false),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id" << BSON("x" << false)), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("_id" << BSON("x" << false)),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("x" << BSON("_id" << false)), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("x" << BSON("_id" << false)),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 
-    parsedProject = ParsedAggregationProjection::create(
-        expCtx, BSON("_id" << false), ProjectionParseMode::kBanComputedFields);
+    parsedProject =
+        ParsedAggregationProjection::create(expCtx,
+                                            BSON("_id" << false),
+                                            ProjectionDefaultIdPolicy::kIncludeId,
+                                            ProjectionArrayRecursionPolicy::kRecurseNestedArrays,
+                                            ProjectionParseMode::kBanComputedFields);
     ASSERT(parsedProject->getType() == TransformerInterface::TransformerType::kExclusionProjection);
 }
 
@@ -637,8 +670,7 @@ TEST(ParsedAggregationProjectionType, ShouldCoerceNumericsToBools) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     std::vector<Value> zeros = {Value(0), Value(0LL), Value(0.0), Value(Decimal128(0))};
     for (auto&& zero : zeros) {
-        auto parsedProject =
-            ParsedAggregationProjection::create(expCtx, Document{{"a", zero}}.toBson());
+        auto parsedProject = makeProjectionWithDefaultPolicies(Document{{"a", zero}}.toBson());
         ASSERT(parsedProject->getType() ==
                TransformerInterface::TransformerType::kExclusionProjection);
     }
@@ -646,8 +678,7 @@ TEST(ParsedAggregationProjectionType, ShouldCoerceNumericsToBools) {
     std::vector<Value> nonZeroes = {
         Value(1), Value(-1), Value(3), Value(1LL), Value(1.0), Value(Decimal128(1))};
     for (auto&& nonZero : nonZeroes) {
-        auto parsedProject =
-            ParsedAggregationProjection::create(expCtx, Document{{"a", nonZero}}.toBson());
+        auto parsedProject = makeProjectionWithDefaultPolicies(Document{{"a", nonZero}}.toBson());
         ASSERT(parsedProject->getType() ==
                TransformerInterface::TransformerType::kInclusionProjection);
     }
diff --git a/src/mongo/db/pipeline/parsed_exclusion_projection.cpp b/src/mongo/db/pipeline/parsed_exclusion_projection.cpp
index 4dffd87f64f..40b561f24c7 100644
--- a/src/mongo/db/pipeline/parsed_exclusion_projection.cpp
+++ b/src/mongo/db/pipeline/parsed_exclusion_projection.cpp
@@ -43,7 +43,8 @@ namespace parsed_aggregation_projection {
 // ExclusionNode.
 //
 
-ExclusionNode::ExclusionNode(std::string pathToNode) : _pathToNode(std::move(pathToNode)) {}
+ExclusionNode::ExclusionNode(ProjectionArrayRecursionPolicy recursionPolicy, std::string pathToNode)
+    : _arrayRecursionPolicy(recursionPolicy), _pathToNode(std::move(pathToNode)) {}
 
 Document ExclusionNode::serialize() const {
     MutableDocument output;
@@ -90,8 +91,8 @@ ExclusionNode* ExclusionNode::getChild(std::string field) const {
 ExclusionNode* ExclusionNode::addChild(std::string field) {
     auto pathToChild = _pathToNode.empty() ? field : _pathToNode + "." + field;
 
-    auto emplacedPair = _children.emplace(
-        std::make_pair(std::move(field), stdx::make_unique<ExclusionNode>(pathToChild)));
+    auto emplacedPair = _children.emplace(std::make_pair(
+        std::move(field), stdx::make_unique<ExclusionNode>(_arrayRecursionPolicy, pathToChild)));
 
     // emplacedPair is a pair<iterator position, bool inserted>.
     invariant(emplacedPair.second);
@@ -112,7 +113,12 @@ Value ExclusionNode::applyProjectionToValue(Value val) const {
             // instead will result in {a: [{b: 0}, {b: 1}]}.
             std::vector<Value> values = val.getArray();
             for (auto it = values.begin(); it != values.end(); it++) {
-                *it = applyProjectionToValue(*it);
+                // If this is a nested array and our policy is to not recurse, leave the array
+                // as-is. Otherwise, descend into the array and project each element individually.
+                const bool shouldSkip = it->isArray() &&
+                    _arrayRecursionPolicy ==
+                        ProjectionArrayRecursionPolicy::kDoNotRecurseNestedArrays;
+                *it = (shouldSkip ? *it : applyProjectionToValue(*it));
             }
             return Value(std::move(values));
         }
@@ -145,23 +151,29 @@ Document ParsedExclusionProjection::applyProjection(const Document& inputDoc) co
 }
 
 void ParsedExclusionProjection::parse(const BSONObj& spec, ExclusionNode* node, size_t depth) {
+    bool idSpecified = false;
+
     for (auto elem : spec) {
-        const auto fieldName = elem.fieldNameStringData().toString();
+        const auto fieldName = elem.fieldNameStringData();
 
-        // A $ should have been detected in ParsedAggregationProjection's parsing before we get
-        // here.
+        // A $ should have been detected by ParsedAggregationProjection before we get here.
         invariant(fieldName[0] != '$');
 
+        // Track whether the projection spec specifies a desired behavior for the _id field.
+        idSpecified = idSpecified || fieldName == "_id"_sd || fieldName.startsWith("_id."_sd);
+
         switch (elem.type()) {
             case BSONType::Bool:
             case BSONType::NumberInt:
             case BSONType::NumberLong:
             case BSONType::NumberDouble:
             case BSONType::NumberDecimal: {
-                // We have already verified this is an exclusion projection.
-                invariant(!elem.trueValue());
-
-                node->excludePath(FieldPath(fieldName));
+                // We have already verified this is an exclusion projection. _id is the only field
+                // which is permitted to be explicitly included here.
+                invariant(!elem.trueValue() || elem.fieldNameStringData() == "_id"_sd);
+                if (!elem.trueValue()) {
+                    node->excludePath(FieldPath(fieldName));
+                }
                 break;
             }
             case BSONType::Object: {
@@ -195,6 +207,12 @@ void ParsedExclusionProjection::parse(const BSONObj& spec, ExclusionNode* node,
             default: { MONGO_UNREACHABLE; }
         }
     }
+
+    // If _id was not specified, then doing nothing will cause it to be included. If the default _id
+    // policy is kExcludeId, we add a new entry for _id to the ExclusionNode tree here.
+    if (!idSpecified && _defaultIdPolicy == ProjectionDefaultIdPolicy::kExcludeId) {
+        _root->excludePath({FieldPath("_id")});
+    }
 }
 
 }  // namespace parsed_aggregation_projection
diff --git a/src/mongo/db/pipeline/parsed_exclusion_projection.h b/src/mongo/db/pipeline/parsed_exclusion_projection.h
index 381143f2390..8f4799d69c5 100644
--- a/src/mongo/db/pipeline/parsed_exclusion_projection.h
+++ b/src/mongo/db/pipeline/parsed_exclusion_projection.h
@@ -49,7 +49,10 @@ namespace parsed_aggregation_projection {
  */
 class ExclusionNode {
 public:
-    ExclusionNode(std::string pathToNode = "");
+    using ProjectionArrayRecursionPolicy =
+        ParsedAggregationProjection::ProjectionArrayRecursionPolicy;
+
+    ExclusionNode(ProjectionArrayRecursionPolicy recursionPolicy, std::string pathToNode = "");
 
     /**
      * Serialize this exclusion.
@@ -84,6 +87,8 @@ private:
     // Fields excluded at this level.
     stdx::unordered_set<std::string> _excludedFields;
 
+    ProjectionArrayRecursionPolicy _arrayRecursionPolicy;
+
     std::string _pathToNode;
     stdx::unordered_map<std::string, std::unique_ptr<ExclusionNode>> _children;
 };
@@ -97,8 +102,11 @@ private:
  */
 class ParsedExclusionProjection : public ParsedAggregationProjection {
 public:
-    ParsedExclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx)
-        : ParsedAggregationProjection(expCtx), _root(new ExclusionNode()) {}
+    ParsedExclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                              ProjectionDefaultIdPolicy defaultIdPolicy,
+                              ProjectionArrayRecursionPolicy arrayRecursionPolicy)
+        : ParsedAggregationProjection(expCtx, defaultIdPolicy, arrayRecursionPolicy),
+          _root(new ExclusionNode(_arrayRecursionPolicy)) {}
 
     TransformerType getType() const final {
         return TransformerType::kExclusionProjection;
diff --git a/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp b/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp
index c5f31b1ad67..6287f64d374 100644
--- a/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp
+++ b/src/mongo/db/pipeline/parsed_exclusion_projection_test.cpp
@@ -48,8 +48,21 @@
 namespace mongo {
 namespace parsed_aggregation_projection {
 namespace {
+
+using ProjectionArrayRecursionPolicy = ParsedAggregationProjection::ProjectionArrayRecursionPolicy;
+using ProjectionDefaultIdPolicy = ParsedAggregationProjection::ProjectionDefaultIdPolicy;
+
 using std::vector;
 
+// Helper to simplify the creation of a ParsedExclusionProjection which includes _id and recurses
+// nested arrays by default.
+ParsedExclusionProjection makeExclusionProjectionWithDefaultPolicies() {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    return ParsedExclusionProjection(expCtx,
+                                     ProjectionDefaultIdPolicy::kIncludeId,
+                                     ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+}
+
 //
 // Errors.
 //
@@ -57,31 +70,32 @@ using std::vector;
 DEATH_TEST(ExclusionProjection,
            ShouldRejectComputedField,
            "Invariant failure fieldName[0] != '$'") {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
     // Top-level expression.
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << false << "b" << BSON("$literal" << 1)));
 }
 
 DEATH_TEST(ExclusionProjection,
-           ShouldFailWhenGivenIncludedField,
-           "Invariant failure !elem.trueValue()") {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+           ShouldFailWhenGivenIncludedNonIdField,
+           "Invariant failure !elem.trueValue() || elem.fieldNameStringData() == \"_id\"_sd") {
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << true));
 }
 
-DEATH_TEST(ExclusionProjection,
-           ShouldFailWhenGivenIncludedId,
-           "Invariant failure !elem.trueValue()") {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+DEATH_TEST(ExclusionProjectionExecutionTest,
+           ShouldFailWhenGivenIncludedIdSubfield,
+           "Invariant failure !elem.trueValue() || elem.fieldNameStringData() == \"_id\"_sd") {
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
+    exclusion.parse(BSON("_id.id1" << true));
+}
+
+TEST(ExclusionProjection, ShouldAllowExplicitIdInclusionInExclusionSpec) {
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("_id" << true << "a" << false));
 }
 
 TEST(ExclusionProjection, ShouldSerializeToEquivalentProjection) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(
         fromjson("{a: 0, b: {c: NumberLong(0), d: 0.0}, 'x.y': false, _id: NumberInt(0)}"));
 
@@ -111,8 +125,7 @@ TEST(ExclusionProjection, ShouldNotAddAnyDependencies) {
     // need to include the "a" in the dependencies of this projection, since it will just be ignored
     // later. If there are no later stages, then we will finish the dependency computation
     // cycle without full knowledge of which fields are needed, and thus include all the fields.
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("_id" << false << "a" << false << "b.c" << false << "x.y.z" << false));
 
     DepsTracker deps;
@@ -124,8 +137,7 @@ TEST(ExclusionProjection, ShouldNotAddAnyDependencies) {
 }
 
 TEST(ExclusionProjection, ShouldReportExcludedFieldsAsModified) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("_id" << false << "a" << false << "b.c" << false));
 
     auto modifiedPaths = exclusion.getModifiedPaths();
@@ -137,8 +149,7 @@ TEST(ExclusionProjection, ShouldReportExcludedFieldsAsModified) {
 }
 
 TEST(ExclusionProjection, ShouldReportExcludedFieldsAsModifiedWhenSpecifiedAsNestedObj) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << BSON("b" << false << "c" << BSON("d" << false))));
 
     auto modifiedPaths = exclusion.getModifiedPaths();
@@ -153,8 +164,7 @@ TEST(ExclusionProjection, ShouldReportExcludedFieldsAsModifiedWhenSpecifiedAsNes
 //
 
 TEST(ExclusionProjectionExecutionTest, ShouldExcludeTopLevelField) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << false));
 
     // More than one field in document.
@@ -179,8 +189,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldExcludeTopLevelField) {
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldCoerceNumericsToBools) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << Value(0) << "b" << Value(0LL) << "c" << Value(0.0) << "d"
                              << Value(Decimal128(0))));
 
@@ -191,8 +200,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldCoerceNumericsToBools) {
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldPreserveOrderOfExistingFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("second" << false));
     auto result = exclusion.applyProjection(Document{{"first", 0}, {"second", 1}, {"third", 2}});
     auto expectedResult = Document{{"first", 0}, {"third", 2}};
@@ -200,8 +208,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldPreserveOrderOfExistingFields) {
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldImplicitlyIncludeId) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << false));
     auto result = exclusion.applyProjection(Document{{"a", 1}, {"b", 2}, {"_id", "ID"_sd}});
     auto expectedResult = Document{{"b", 2}, {"_id", "ID"_sd}};
@@ -209,8 +216,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldImplicitlyIncludeId) {
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldExcludeIdIfExplicitlyExcluded) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << false << "_id" << false));
     auto result = exclusion.applyProjection(Document{{"a", 1}, {"b", 2}, {"_id", "ID"_sd}});
     auto expectedResult = Document{{"b", 2}};
@@ -218,8 +224,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldExcludeIdIfExplicitlyExcluded) {
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldExcludeIdAndKeepAllOtherFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("_id" << false));
     auto result = exclusion.applyProjection(Document{{"a", 1}, {"b", 2}, {"_id", "ID"_sd}});
     auto expectedResult = Document{{"a", 1}, {"b", 2}};
@@ -231,8 +236,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldExcludeIdAndKeepAllOtherFields) {
 //
 
 TEST(ExclusionProjectionExecutionTest, ShouldExcludeSubFieldsOfId) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("_id.x" << false << "_id" << BSON("y" << false)));
     auto result = exclusion.applyProjection(
         Document{{"_id", Document{{"x", 1}, {"y", 2}, {"z", 3}}}, {"a", 1}});
@@ -241,8 +245,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldExcludeSubFieldsOfId) {
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldExcludeSimpleDottedFieldFromSubDoc) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a.b" << false));
 
     // More than one field in sub document.
@@ -267,8 +270,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldExcludeSimpleDottedFieldFromSubDoc)
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldNotCreateSubDocIfDottedExcludedFieldDoesNotExist) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("sub.target" << false));
 
     // Should not add the path if it doesn't exist.
@@ -283,8 +285,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldNotCreateSubDocIfDottedExcludedFiel
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldApplyDottedExclusionToEachElementInArray) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a.b" << false));
 
     std::vector<Value> nestedValues = {
@@ -307,8 +308,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldApplyDottedExclusionToEachElementIn
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldAllowMixedNestedAndDottedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     // Exclude all of "a.b", "a.c", "a.d", and "a.e".
     exclusion.parse(
         BSON("a.b" << false << "a.c" << false << "a" << BSON("d" << false << "e" << false)));
@@ -319,8 +319,7 @@ TEST(ExclusionProjectionExecutionTest, ShouldAllowMixedNestedAndDottedFields) {
 }
 
 TEST(ExclusionProjectionExecutionTest, ShouldAlwaysKeepMetadataFromOriginalDoc) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedExclusionProjection exclusion(expCtx);
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
     exclusion.parse(BSON("a" << false));
 
     MutableDocument inputDocBuilder(Document{{"_id", "ID"_sd}, {"a", 1}});
@@ -335,6 +334,195 @@ TEST(ExclusionProjectionExecutionTest, ShouldAlwaysKeepMetadataFromOriginalDoc)
     ASSERT_DOCUMENT_EQ(result, expectedDoc.freeze());
 }
 
+//
+// _id exclusion policy.
+//
+
+TEST(ExclusionProjectionExecutionTest, ShouldIncludeIdByDefault) {
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
+    exclusion.parse(BSON("a" << false));
+
+    auto result = exclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{{"_id", 2}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldIncludeIdWithExplicitPolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    exclusion.parse(BSON("a" << false));
+
+    auto result = exclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{{"_id", 2}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldExcludeIdWithExplicitPolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kExcludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    exclusion.parse(BSON("a" << false));
+
+    auto result = exclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldOverrideIncludePolicyWithExplicitExcludeIdSpec) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    exclusion.parse(BSON("_id" << false << "a" << false));
+
+    auto result = exclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldOverrideExcludePolicyWithExplicitIncludeIdSpec) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kExcludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    exclusion.parse(BSON("_id" << true << "a" << false));
+
+    auto result = exclusion.applyProjection(Document{{"_id", 2}, {"a", 3}, {"b", 4}});
+    auto expectedResult = Document{{"_id", 2}, {"b", 4}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldAllowExclusionOfIdSubfieldWithDefaultIncludePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    exclusion.parse(BSON("_id.id1" << false << "a" << false));
+
+    auto result = exclusion.applyProjection(
+        Document{{"_id", Document{{"id1", 1}, {"id2", 2}}}, {"a", 3}, {"b", 4}});
+    auto expectedResult = Document{{"_id", Document{{"id2", 2}}}, {"b", 4}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldAllowExclusionOfIdSubfieldWithDefaultExcludePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kExcludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    exclusion.parse(BSON("_id.id1" << false << "a" << false));
+
+    auto result = exclusion.applyProjection(
+        Document{{"_id", Document{{"id1", 1}, {"id2", 2}}}, {"a", 3}, {"b", 4}});
+    auto expectedResult = Document{{"_id", Document{{"id2", 2}}}, {"b", 4}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+//
+// Nested array recursion.
+//
+
+TEST(ExclusionProjectionExecutionTest, ShouldRecurseNestedArraysByDefault) {
+    auto exclusion = makeExclusionProjectionWithDefaultPolicies();
+    exclusion.parse(BSON("a.b" << false));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => {a: [1, {c: 3}, [{c: 5}], {d: 6}]}
+    auto result = exclusion.applyProjection(
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}});
+
+    auto expectedResult = Document{{"a",
+                                    vector<Value>{Value(1),
+                                                  Value(Document{{"c", 3}}),
+                                                  Value(vector<Value>{Value(Document{{"c", 5}})}),
+                                                  Value(Document{{"d", 6}})}}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldRecurseNestedArraysForExplicitProRecursePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    exclusion.parse(BSON("a.b" << false));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => {a: [1, {c: 3}, [{c: 5}], {d: 6}]}
+    auto result = exclusion.applyProjection(
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}});
+
+    auto expectedResult = Document{{"a",
+                                    vector<Value>{Value(1),
+                                                  Value(Document{{"c", 3}}),
+                                                  Value(vector<Value>{Value(Document{{"c", 5}})}),
+                                                  Value(Document{{"d", 6}})}}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldNotRecurseNestedArraysForNoRecursePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kDoNotRecurseNestedArrays);
+    exclusion.parse(BSON("a.b" << false));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => {a: [1, {c: 3}, [{b: 4, c: 5}], {d: 6}]}
+    auto result = exclusion.applyProjection(
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}});
+
+    auto expectedResult =
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(ExclusionProjectionExecutionTest, ShouldNotRetainNestedArraysIfNoRecursionNeeded) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedExclusionProjection exclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kDoNotRecurseNestedArrays);
+    exclusion.parse(BSON("a" << false));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => {}
+    const auto inputDoc =
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}};
+
+    auto result = exclusion.applyProjection(inputDoc);
+    const auto expectedResult = Document{};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
 }  // namespace
 }  // namespace parsed_aggregation_projection
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection.cpp b/src/mongo/db/pipeline/parsed_inclusion_projection.cpp
index 45a9998e3ff..7e0c6bb05f7 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection.cpp
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection.cpp
@@ -43,7 +43,8 @@ using std::unique_ptr;
 // InclusionNode
 //
 
-InclusionNode::InclusionNode(std::string pathToNode) : _pathToNode(std::move(pathToNode)) {}
+InclusionNode::InclusionNode(ProjectionArrayRecursionPolicy recursionPolicy, std::string pathToNode)
+    : _arrayRecursionPolicy(recursionPolicy), _pathToNode(std::move(pathToNode)) {}
 
 void InclusionNode::optimize() {
     for (auto&& expressionIt : _expressions) {
@@ -129,7 +130,11 @@ Value InclusionNode::applyInclusionsToValue(Value inputValue) const {
     } else if (inputValue.getType() == BSONType::Array) {
         std::vector<Value> values = inputValue.getArray();
         for (auto it = values.begin(); it != values.end(); ++it) {
-            *it = applyInclusionsToValue(*it);
+            // If this is a nested array and our policy is to not recurse, remove the array.
+            // Otherwise, descend into the array and project each element individually.
+            const bool shouldSkip = it->isArray() &&
+                _arrayRecursionPolicy == ProjectionArrayRecursionPolicy::kDoNotRecurseNestedArrays;
+            *it = (shouldSkip ? Value() : applyInclusionsToValue(*it));
         }
         return Value(std::move(values));
     } else {
@@ -222,8 +227,9 @@ InclusionNode* InclusionNode::addChild(string field) {
     invariant(!str::contains(field, "."));
     _orderToProcessAdditionsAndChildren.push_back(field);
     auto childPath = FieldPath::getFullyQualifiedPath(_pathToNode, field);
-    auto insertedPair = _children.emplace(
-        std::make_pair(std::move(field), stdx::make_unique<InclusionNode>(std::move(childPath))));
+    auto insertedPair = _children.emplace(std::make_pair(
+        std::move(field),
+        stdx::make_unique<InclusionNode>(_arrayRecursionPolicy, std::move(childPath))));
     return insertedPair.first->second.get();
 }
 
@@ -262,15 +268,15 @@ void InclusionNode::addComputedPaths(std::set<std::string>* computedPaths,
 //
 
 void ParsedInclusionProjection::parse(const BSONObj& spec) {
-    // It is illegal to specify a projection with no output fields.
+    // It is illegal to specify an inclusion with no output fields.
     bool atLeastOneFieldInOutput = false;
 
-    // Tracks whether or not we should implicitly include "_id".
+    // Tracks whether or not we should apply the default _id projection policy.
     bool idSpecified = false;
 
     for (auto elem : spec) {
         auto fieldName = elem.fieldNameStringData();
-        idSpecified = idSpecified || fieldName == "_id" || fieldName.startsWith("_id.");
+        idSpecified = idSpecified || fieldName == "_id"_sd || fieldName.startsWith("_id."_sd);
         if (fieldName == "_id") {
             const bool idIsExcluded = (!elem.trueValue() && (elem.isNumber() || elem.isBoolean()));
             if (idIsExcluded) {
@@ -327,9 +333,13 @@ void ParsedInclusionProjection::parse(const BSONObj& spec) {
     }
 
     if (!idSpecified) {
-        // "_id" wasn't specified, so include it by default.
-        atLeastOneFieldInOutput = true;
-        _root->addIncludedField(FieldPath("_id"));
+        // _id wasn't specified, so apply the default _id projection policy here.
+        if (_defaultIdPolicy == ProjectionDefaultIdPolicy::kExcludeId) {
+            _idExcluded = true;
+        } else {
+            atLeastOneFieldInOutput = true;
+            _root->addIncludedField(FieldPath("_id"));
+        }
     }
 
     uassert(16403,
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection.h b/src/mongo/db/pipeline/parsed_inclusion_projection.h
index 81fdd0d1db2..644a44409fd 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection.h
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection.h
@@ -52,7 +52,10 @@ namespace parsed_aggregation_projection {
  */
 class InclusionNode {
 public:
-    InclusionNode(std::string pathToNode = "");
+    using ProjectionArrayRecursionPolicy =
+        ParsedAggregationProjection::ProjectionArrayRecursionPolicy;
+
+    InclusionNode(ProjectionArrayRecursionPolicy recursionPolicy, std::string pathToNode = "");
 
     /**
      * Optimize any computed expressions.
@@ -156,6 +159,8 @@ private:
      */
     bool subtreeContainsComputedFields() const;
 
+    ProjectionArrayRecursionPolicy _arrayRecursionPolicy;
+
     std::string _pathToNode;
 
     // Our projection semantics are such that all field additions need to be processed in the order
@@ -184,8 +189,11 @@ private:
  */
 class ParsedInclusionProjection : public ParsedAggregationProjection {
 public:
-    ParsedInclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx)
-        : ParsedAggregationProjection(expCtx), _root(new InclusionNode()) {}
+    ParsedInclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                              ProjectionDefaultIdPolicy defaultIdPolicy,
+                              ProjectionArrayRecursionPolicy arrayRecursionPolicy)
+        : ParsedAggregationProjection(expCtx, defaultIdPolicy, arrayRecursionPolicy),
+          _root(new InclusionNode(_arrayRecursionPolicy)) {}
 
     TransformerType getType() const final {
         return TransformerType::kInclusionProjection;
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp b/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp
index 46d28083680..3996db1036a 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp
@@ -40,11 +40,16 @@
 #include "mongo/db/pipeline/document_value_test_util.h"
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/db/pipeline/value.h"
+#include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 
 namespace mongo {
 namespace parsed_aggregation_projection {
 namespace {
+
+using ProjectionArrayRecursionPolicy = ParsedAggregationProjection::ProjectionArrayRecursionPolicy;
+using ProjectionDefaultIdPolicy = ParsedAggregationProjection::ProjectionDefaultIdPolicy;
+
 using std::vector;
 
 template <typename T>
@@ -52,23 +57,43 @@ BSONObj wrapInLiteral(const T& arg) {
     return BSON("$literal" << arg);
 }
 
-TEST(InclusionProjection, ShouldThrowWhenParsingInvalidExpression) {
+// Helper to simplify the creation of a ParsedInclusionProjection which includes _id and recurses
+// nested arrays by default.
+ParsedInclusionProjection makeInclusionProjectionWithDefaultPolicies() {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    return ParsedInclusionProjection(expCtx,
+                                     ProjectionDefaultIdPolicy::kIncludeId,
+                                     ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+}
+
+DEATH_TEST(InclusionProjection,
+           ShouldFailWhenGivenExcludedNonIdField,
+           "Invariant failure elem.trueValue()") {
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
+    inclusion.parse(BSON("a" << false));
+}
+
+DEATH_TEST(InclusionProjection,
+           ShouldFailWhenGivenIncludedIdSubfield,
+           "Invariant failure elem.trueValue()") {
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
+    inclusion.parse(BSON("_id.id1" << false));
+}
+
+TEST(InclusionProjection, ShouldThrowWhenParsingInvalidExpression) {
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     ASSERT_THROWS(inclusion.parse(BSON("a" << BSON("$gt" << BSON("bad"
                                                                  << "arguments")))),
                   AssertionException);
 }
 
 TEST(InclusionProjection, ShouldRejectProjectionWithNoOutputFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     ASSERT_THROWS(inclusion.parse(BSON("_id" << false)), AssertionException);
 }
 
 TEST(InclusionProjection, ShouldAddIncludedFieldsToDependencies) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("_id" << false << "a" << true << "x.y" << true));
 
     DepsTracker deps;
@@ -81,8 +106,7 @@ TEST(InclusionProjection, ShouldAddIncludedFieldsToDependencies) {
 }
 
 TEST(InclusionProjection, ShouldAddIdToDependenciesIfNotSpecified) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true));
 
     DepsTracker deps;
@@ -94,8 +118,7 @@ TEST(InclusionProjection, ShouldAddIdToDependenciesIfNotSpecified) {
 }
 
 TEST(InclusionProjection, ShouldAddDependenciesOfComputedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a"
                          << "$a"
                          << "x"
@@ -111,8 +134,7 @@ TEST(InclusionProjection, ShouldAddDependenciesOfComputedFields) {
 }
 
 TEST(InclusionProjection, ShouldAddPathToDependenciesForNestedComputedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("x.y"
                          << "$z"));
 
@@ -129,8 +151,7 @@ TEST(InclusionProjection, ShouldAddPathToDependenciesForNestedComputedFields) {
 }
 
 TEST(InclusionProjection, ShouldSerializeToEquivalentProjection) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(fromjson("{a: {$add: ['$a', 2]}, b: {d: 3}, 'x.y': {$literal: 4}}"));
 
     // Adds implicit "_id" inclusion, converts numbers to bools, serializes expressions.
@@ -148,8 +169,7 @@ TEST(InclusionProjection, ShouldSerializeToEquivalentProjection) {
 }
 
 TEST(InclusionProjection, ShouldSerializeExplicitExclusionOfId) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("_id" << false << "a" << true));
 
     // Adds implicit "_id" inclusion, converts numbers to bools, serializes expressions.
@@ -167,8 +187,7 @@ TEST(InclusionProjection, ShouldSerializeExplicitExclusionOfId) {
 
 
 TEST(InclusionProjection, ShouldOptimizeTopLevelExpressions) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << BSON("$add" << BSON_ARRAY(1 << 2))));
 
     inclusion.optimize();
@@ -186,8 +205,7 @@ TEST(InclusionProjection, ShouldOptimizeTopLevelExpressions) {
 }
 
 TEST(InclusionProjection, ShouldOptimizeNestedExpressions) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.b" << BSON("$add" << BSON_ARRAY(1 << 2))));
 
     inclusion.optimize();
@@ -206,8 +224,7 @@ TEST(InclusionProjection, ShouldOptimizeNestedExpressions) {
 }
 
 TEST(InclusionProjection, ShouldReportThatAllExceptIncludedFieldsAreModified) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON(
         "a" << wrapInLiteral("computedVal") << "b.c" << wrapInLiteral("computedVal") << "d" << true
             << "e.f"
@@ -226,8 +243,7 @@ TEST(InclusionProjection, ShouldReportThatAllExceptIncludedFieldsAreModified) {
 }
 
 TEST(InclusionProjection, ShouldReportThatAllExceptIncludedFieldsAreModifiedWithIdExclusion) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("_id" << false << "a" << wrapInLiteral("computedVal") << "b.c"
                                << wrapInLiteral("computedVal")
                                << "d"
@@ -254,8 +270,7 @@ TEST(InclusionProjection, ShouldReportThatAllExceptIncludedFieldsAreModifiedWith
 //
 
 TEST(InclusionProjectionExecutionTest, ShouldIncludeTopLevelField) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true));
 
     // More than one field in document.
@@ -280,8 +295,7 @@ TEST(InclusionProjectionExecutionTest, ShouldIncludeTopLevelField) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldAddComputedTopLevelField) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("newField" << wrapInLiteral("computedVal")));
     auto result = inclusion.applyProjection(Document{});
     auto expectedResult = Document{{"newField", "computedVal"_sd}};
@@ -294,8 +308,7 @@ TEST(InclusionProjectionExecutionTest, ShouldAddComputedTopLevelField) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldApplyBothInclusionsAndComputedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "newField" << wrapInLiteral("computedVal")));
     auto result = inclusion.applyProjection(Document{{"a", 1}});
     auto expectedResult = Document{{"a", 1}, {"newField", "computedVal"_sd}};
@@ -303,8 +316,7 @@ TEST(InclusionProjectionExecutionTest, ShouldApplyBothInclusionsAndComputedField
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldIncludeFieldsInOrderOfInputDoc) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("first" << true << "second" << true << "third" << true));
     auto inputDoc = Document{{"second", 1}, {"first", 0}, {"third", 2}};
     auto result = inclusion.applyProjection(inputDoc);
@@ -312,8 +324,7 @@ TEST(InclusionProjectionExecutionTest, ShouldIncludeFieldsInOrderOfInputDoc) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldApplyComputedFieldsInOrderSpecified) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("firstComputed" << wrapInLiteral("FIRST") << "secondComputed"
                                          << wrapInLiteral("SECOND")));
     auto result = inclusion.applyProjection(Document{{"first", 0}, {"second", 1}, {"third", 2}});
@@ -322,8 +333,7 @@ TEST(InclusionProjectionExecutionTest, ShouldApplyComputedFieldsInOrderSpecified
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldImplicitlyIncludeId) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true));
     auto result = inclusion.applyProjection(Document{{"_id", "ID"_sd}, {"a", 1}, {"b", 2}});
     auto expectedResult = Document{{"_id", "ID"_sd}, {"a", 1}};
@@ -336,8 +346,7 @@ TEST(InclusionProjectionExecutionTest, ShouldImplicitlyIncludeId) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldImplicitlyIncludeIdWithComputedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("newField" << wrapInLiteral("computedVal")));
     auto result = inclusion.applyProjection(Document{{"_id", "ID"_sd}, {"a", 1}});
     auto expectedResult = Document{{"_id", "ID"_sd}, {"newField", "computedVal"_sd}};
@@ -345,8 +354,7 @@ TEST(InclusionProjectionExecutionTest, ShouldImplicitlyIncludeIdWithComputedFiel
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldIncludeIdIfExplicitlyIncluded) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "_id" << true << "b" << true));
     auto result =
         inclusion.applyProjection(Document{{"_id", "ID"_sd}, {"a", 1}, {"b", 2}, {"c", 3}});
@@ -355,8 +363,7 @@ TEST(InclusionProjectionExecutionTest, ShouldIncludeIdIfExplicitlyIncluded) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldExcludeIdIfExplicitlyExcluded) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "_id" << false));
     auto result = inclusion.applyProjection(Document{{"a", 1}, {"b", 2}, {"_id", "ID"_sd}});
     auto expectedResult = Document{{"a", 1}};
@@ -364,8 +371,7 @@ TEST(InclusionProjectionExecutionTest, ShouldExcludeIdIfExplicitlyExcluded) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldReplaceIdWithComputedId) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("_id" << wrapInLiteral("newId")));
     auto result = inclusion.applyProjection(Document{{"a", 1}, {"b", 2}, {"_id", "ID"_sd}});
     auto expectedResult = Document{{"_id", "newId"_sd}};
@@ -377,8 +383,7 @@ TEST(InclusionProjectionExecutionTest, ShouldReplaceIdWithComputedId) {
 //
 
 TEST(InclusionProjectionExecutionTest, ShouldIncludeSimpleDottedFieldFromSubDoc) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.b" << true));
 
     // More than one field in sub document.
@@ -403,8 +408,7 @@ TEST(InclusionProjectionExecutionTest, ShouldIncludeSimpleDottedFieldFromSubDoc)
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldNotCreateSubDocIfDottedIncludedFieldDoesNotExist) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("sub.target" << true));
 
     // Should not add the path if it doesn't exist.
@@ -419,8 +423,7 @@ TEST(InclusionProjectionExecutionTest, ShouldNotCreateSubDocIfDottedIncludedFiel
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldApplyDottedInclusionToEachElementInArray) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.b" << true));
 
     vector<Value> nestedValues = {Value(1),
@@ -444,8 +447,7 @@ TEST(InclusionProjectionExecutionTest, ShouldApplyDottedInclusionToEachElementIn
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldAddComputedDottedFieldToSubDocument) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("sub.target" << wrapInLiteral("computedVal")));
 
     // Other fields exist in sub document, one of which is the specified field.
@@ -465,8 +467,7 @@ TEST(InclusionProjectionExecutionTest, ShouldAddComputedDottedFieldToSubDocument
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldCreateSubDocIfDottedComputedFieldDoesntExist) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("sub.target" << wrapInLiteral("computedVal")));
 
     // Should add the path if it doesn't exist.
@@ -480,8 +481,7 @@ TEST(InclusionProjectionExecutionTest, ShouldCreateSubDocIfDottedComputedFieldDo
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldCreateNestedSubDocumentsAllTheWayToComputedField) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.b.c.d" << wrapInLiteral("computedVal")));
 
     // Should add the path if it doesn't exist.
@@ -496,8 +496,7 @@ TEST(InclusionProjectionExecutionTest, ShouldCreateNestedSubDocumentsAllTheWayTo
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldAddComputedDottedFieldToEachElementInArray) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.b" << wrapInLiteral("COMPUTED")));
 
     vector<Value> nestedValues = {Value(1),
@@ -520,8 +519,7 @@ TEST(InclusionProjectionExecutionTest, ShouldAddComputedDottedFieldToEachElement
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldApplyInclusionsAndAdditionsToEachElementInArray) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.inc" << true << "a.comp" << wrapInLiteral("COMPUTED")));
 
     vector<Value> nestedValues = {Value(1),
@@ -548,8 +546,7 @@ TEST(InclusionProjectionExecutionTest, ShouldApplyInclusionsAndAdditionsToEachEl
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldAddOrIncludeSubFieldsOfId) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("_id.X" << true << "_id.Z" << wrapInLiteral("NEW")));
     auto result = inclusion.applyProjection(Document{{"_id", Document{{"X", 1}, {"Y", 2}}}});
     auto expectedResult = Document{{"_id", Document{{"X", 1}, {"Z", "NEW"_sd}}}};
@@ -557,8 +554,7 @@ TEST(InclusionProjectionExecutionTest, ShouldAddOrIncludeSubFieldsOfId) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldAllowMixedNestedAndDottedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     // Include all of "a.b", "a.c", "a.d", and "a.e".
     // Add new computed fields "a.W", "a.X", "a.Y", and "a.Z".
     inclusion.parse(BSON(
@@ -582,8 +578,7 @@ TEST(InclusionProjectionExecutionTest, ShouldAllowMixedNestedAndDottedFields) {
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldApplyNestedComputedFieldsInOrderSpecified) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << wrapInLiteral("FIRST") << "b.c" << wrapInLiteral("SECOND")));
     auto result = inclusion.applyProjection(Document{});
     auto expectedResult = Document{{"a", "FIRST"_sd}, {"b", Document{{"c", "SECOND"_sd}}}};
@@ -591,8 +586,7 @@ TEST(InclusionProjectionExecutionTest, ShouldApplyNestedComputedFieldsInOrderSpe
 }
 
 TEST(InclusionProjectionExecutionTest, ShouldApplyComputedFieldsAfterAllInclusions) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("b.c" << wrapInLiteral("NEW") << "a" << true));
     auto result = inclusion.applyProjection(Document{{"a", 1}});
     auto expectedResult = Document{{"a", 1}, {"b", Document{{"c", "NEW"_sd}}}};
@@ -611,8 +605,7 @@ TEST(InclusionProjectionExecutionTest, ShouldApplyComputedFieldsAfterAllInclusio
 }
 
 TEST(InclusionProjectionExecutionTest, ComputedFieldReplacingExistingShouldAppearAfterInclusions) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("b" << wrapInLiteral("NEW") << "a" << true));
     auto result = inclusion.applyProjection(Document{{"b", 1}, {"a", 1}});
     auto expectedResult = Document{{"a", 1}, {"b", "NEW"_sd}};
@@ -623,12 +616,252 @@ TEST(InclusionProjectionExecutionTest, ComputedFieldReplacingExistingShouldAppea
 }
 
 //
+// _id inclusion policy.
+//
+
+TEST(InclusionProjectionExecutionTest, ShouldIncludeIdByDefault) {
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
+    inclusion.parse(BSON("a" << true));
+
+    auto result = inclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{{"_id", 2}, {"a", 3}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldIncludeIdWithIncludePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("a" << true));
+
+    auto result = inclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{{"_id", 2}, {"a", 3}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldExcludeIdWithExcludePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kExcludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("a" << true));
+
+    auto result = inclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{{"a", 3}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldOverrideIncludePolicyWithExplicitExcludeIdSpec) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("_id" << false << "a" << true));
+
+    auto result = inclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{{"a", 3}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldOverrideExcludePolicyWithExplicitIncludeIdSpec) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kExcludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("_id" << true << "a" << true));
+
+    auto result = inclusion.applyProjection(Document{{"_id", 2}, {"a", 3}});
+    auto expectedResult = Document{{"_id", 2}, {"a", 3}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldAllowInclusionOfIdSubfieldWithDefaultIncludePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("_id.id1" << true << "a" << true));
+
+    auto result = inclusion.applyProjection(
+        Document{{"_id", Document{{"id1", 1}, {"id2", 2}}}, {"a", 3}, {"b", 4}});
+    auto expectedResult = Document{{"_id", Document{{"id1", 1}}}, {"a", 3}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldAllowInclusionOfIdSubfieldWithDefaultExcludePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kExcludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("_id.id1" << true << "a" << true));
+
+    auto result = inclusion.applyProjection(
+        Document{{"_id", Document{{"id1", 1}, {"id2", 2}}}, {"a", 3}, {"b", 4}});
+    auto expectedResult = Document{{"_id", Document{{"id1", 1}}}, {"a", 3}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+//
+// Nested array recursion.
+//
+
+TEST(InclusionProjectionExecutionTest, ShouldRecurseNestedArraysByDefault) {
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
+    inclusion.parse(BSON("a.b" << true));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => {a: [{b: 2}, [{b: 4}], {}]}
+    auto result = inclusion.applyProjection(
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}});
+
+    auto expectedResult = Document{{"a",
+                                    vector<Value>{Value(),
+                                                  Value(Document{{"b", 2}}),
+                                                  Value(vector<Value>{Value(Document{{"b", 4}})}),
+                                                  Value(Document{})}}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldRecurseNestedArraysForExplicitProRecursePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("a.b" << true));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => {a: [{b: 2}, [{b: 4}], {}]}
+    auto result = inclusion.applyProjection(
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}});
+
+    auto expectedResult = Document{{"a",
+                                    vector<Value>{Value(),
+                                                  Value(Document{{"b", 2}}),
+                                                  Value(vector<Value>{Value(Document{{"b", 4}})}),
+                                                  Value(Document{})}}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldNotRecurseNestedArraysForNoRecursePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kDoNotRecurseNestedArrays);
+    inclusion.parse(BSON("a.b" << true));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => {a: [{b: 2}, {}]}
+    auto result = inclusion.applyProjection(
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}});
+
+    auto expectedResult = Document{
+        {"a", vector<Value>{Value(), Value(Document{{"b", 2}}), Value(), Value(Document{})}}};
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ShouldRetainNestedArraysIfNoRecursionNeeded) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kDoNotRecurseNestedArrays);
+    inclusion.parse(BSON("a" << true));
+
+    // {a: [1, {b: 2, c: 3}, [{b: 4, c: 5}], {d: 6}]} => [output doc identical to input]
+    const auto inputDoc =
+        Document{{"a",
+                  vector<Value>{Value(1),
+                                Value(Document{{"b", 2}, {"c", 3}}),
+                                Value(vector<Value>{Value(Document{{"b", 4}, {"c", 5}})}),
+                                Value(Document{{"d", 6}})}}};
+
+    auto result = inclusion.applyProjection(inputDoc);
+    const auto& expectedResult = inputDoc;
+
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ComputedFieldIsAddedToNestedArrayElementsForRecursePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kRecurseNestedArrays);
+    inclusion.parse(BSON("a.b" << wrapInLiteral("COMPUTED")));
+
+    vector<Value> nestedValues = {Value(1),
+                                  Value(Document{}),
+                                  Value(Document{{"b", 1}}),
+                                  Value(Document{{"b", 1}, {"c", 2}}),
+                                  Value(vector<Value>{}),
+                                  Value(vector<Value>{Value(1), Value(Document{{"c", 1}})})};
+    vector<Value> expectedNestedValues = {
+        Value(Document{{"b", "COMPUTED"_sd}}),
+        Value(Document{{"b", "COMPUTED"_sd}}),
+        Value(Document{{"b", "COMPUTED"_sd}}),
+        Value(Document{{"b", "COMPUTED"_sd}}),
+        Value(vector<Value>{}),
+        Value(vector<Value>{Value(Document{{"b", "COMPUTED"_sd}}),
+                            Value(Document{{"b", "COMPUTED"_sd}})})};
+    auto result = inclusion.applyProjection(Document{{"a", nestedValues}});
+    auto expectedResult = Document{{"a", expectedNestedValues}};
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+TEST(InclusionProjectionExecutionTest, ComputedFieldShouldReplaceNestedArrayForNoRecursePolicy) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx,
+                                        ProjectionDefaultIdPolicy::kIncludeId,
+                                        ProjectionArrayRecursionPolicy::kDoNotRecurseNestedArrays);
+    inclusion.parse(BSON("a.b" << wrapInLiteral("COMPUTED")));
+
+    // For kRecurseNestedArrays, the computed field (1) replaces any scalar values in the array with
+    // a subdocument containing the new field, and (2) is added to each element of the array and all
+    // nested arrays individually. With kDoNotRecurseNestedArrays, the nested arrays are replaced
+    // rather than being traversed, in exactly the same way as scalar values.
+    vector<Value> nestedValues = {Value(1),
+                                  Value(Document{}),
+                                  Value(Document{{"b", 1}}),
+                                  Value(Document{{"b", 1}, {"c", 2}}),
+                                  Value(vector<Value>{}),
+                                  Value(vector<Value>{Value(1), Value(Document{{"c", 1}})})};
+
+    vector<Value> expectedNestedValues = {Value(Document{{"b", "COMPUTED"_sd}}),
+                                          Value(Document{{"b", "COMPUTED"_sd}}),
+                                          Value(Document{{"b", "COMPUTED"_sd}}),
+                                          Value(Document{{"b", "COMPUTED"_sd}}),
+                                          Value(Document{{"b", "COMPUTED"_sd}}),
+                                          Value(Document{{"b", "COMPUTED"_sd}})};
+
+    auto result = inclusion.applyProjection(Document{{"a", nestedValues}});
+    auto expectedResult = Document{{"a", expectedNestedValues}};
+    ASSERT_DOCUMENT_EQ(result, expectedResult);
+}
+
+//
 // Misc.
 //
 
 TEST(InclusionProjectionExecutionTest, ShouldAlwaysKeepMetadataFromOriginalDoc) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true));
 
     MutableDocument inputDocBuilder(Document{{"a", 1}});
@@ -648,8 +881,7 @@ TEST(InclusionProjectionExecutionTest, ShouldAlwaysKeepMetadataFromOriginalDoc)
 //
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForIdenticalProjection) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "b" << true));
 
     auto proj = BSON("_id" << false << "a" << true << "b" << true);
@@ -658,8 +890,7 @@ TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForIdenticalProjection) {
 }
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForSupersetProjection) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "b" << true));
 
     auto proj = BSON("_id" << false << "a" << true << "b" << true << "c" << true);
@@ -668,8 +899,7 @@ TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForSupersetProjection) {
 }
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForIdenticalNestedProjection) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.b" << true));
 
     auto proj = BSON("_id" << false << "a.b" << true);
@@ -678,8 +908,7 @@ TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForIdenticalNestedProjecti
 }
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForSupersetProjectionWithNestedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "c" << BSON("d" << true)));
 
     auto proj = BSON("_id" << false << "a" << true << "b" << true << "c.d" << true);
@@ -688,8 +917,7 @@ TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForSupersetProjectionWithN
 }
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissingFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "b" << true));
 
     auto proj = BSON("_id" << false << "a" << true);
@@ -701,8 +929,7 @@ TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissin
 
 TEST(InclusionProjectionSubsetTest,
      ShouldDetectNonSubsetForSupersetProjectionWithoutComputedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true << "b" << true << "c" << BSON("$literal" << 1)));
 
     auto proj = BSON("_id" << false << "a" << true << "b" << true);
@@ -711,8 +938,7 @@ TEST(InclusionProjectionSubsetTest,
 }
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissingNestedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a.b" << true << "a.c" << true));
 
     auto proj = BSON("_id" << false << "a.b" << true);
@@ -721,8 +947,7 @@ TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissin
 }
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithRenamedFields) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a"
                          << "$b"));
 
@@ -732,8 +957,7 @@ TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithRename
 }
 
 TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissingIdField) {
-    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
-    ParsedInclusionProjection inclusion(expCtx);
+    auto inclusion = makeInclusionProjectionWithDefaultPolicies();
     inclusion.parse(BSON("a" << true));
 
     auto proj = BSON("a" << true);