SERVER-20092: Add ability for DocumentSourceCursor to coalesce with a stage if it is doing an equivalent projection

13 files changed, 382 insertions, 51 deletions

diff --git a/jstests/aggregation/sources/project/remove_redundant_projects.js b/jstests/aggregation/sources/project/remove_redundant_projects.js
new file mode 100644
index 00000000000..7ca7ce8e6a3
--- /dev/null
+++ b/jstests/aggregation/sources/project/remove_redundant_projects.js
@@ -0,0 +1,107 @@
+// Tests that the aggregation pipeline correctly coalesces a $project stage at the front of the
+// pipeline that can be covered by a normal query.
+// @tags: [do_not_wrap_aggregations_in_facets]
+(function() {
+    "use strict";
+
+    load("jstests/aggregation/extras/utils.js");  // For orderedArrayEq.
+    load('jstests/libs/analyze_plan.js');         // For planHasStage().
+
+    let coll = db.remove_redundant_projects;
+    coll.drop();
+
+    assert.writeOK(coll.insert({_id: {a: 1, b: 1}, a: 1, c: {d: 1}, e: ['elem1']}));
+
+    let indexSpec = {a: 1, 'c.d': 1, 'e.0': 1};
+
+    /**
+     * Helper to test that for a given pipeline, the same results are returned whether or not an
+     * index is present.  Also tests whether a projection is absorbed by the pipeline
+     * ('expectProjectToCoalesce') and the corresponding project stage ('removedProjectStage') does
+     * not exist in the explain output.
+     */
+    function assertResultsMatch(
+        pipeline, expectProjectToCoalesce, removedProjectStage = null, index = indexSpec) {
+        // Add a match stage to ensure index scans are considered for planning (workaround for
+        // SERVER-20066).
+        pipeline = [{$match: {a: {$gte: 0}}}].concat(pipeline);
+
+        // Once with an index.
+        assert.commandWorked(coll.createIndex(index));
+        let explain = coll.explain().aggregate(pipeline);
+        let resultsWithIndex = coll.aggregate(pipeline).toArray();
+
+        // Projection does not get pushed down when sharding filter is used.
+        if (!explain.hasOwnProperty("shards")) {
+            // Check that $project uses the query system.
+            assert.eq(
+                expectProjectToCoalesce,
+                planHasStage(explain.stages[0].$cursor.queryPlanner.winningPlan, "PROJECTION"));
+
+            // Check that $project was removed from pipeline and pushed to the query system.
+            explain.stages.forEach(function(stage) {
+                if (stage.hasOwnProperty("$project"))
+                    assert.neq(removedProjectStage, stage["$project"]);
+            });
+        }
+
+        // Again without an index.
+        assert.commandWorked(coll.dropIndex(index));
+        let resultsWithoutIndex = coll.aggregate(pipeline).toArray();
+
+        assert(orderedArrayEq(resultsWithIndex, resultsWithoutIndex));
+    }
+
+    // Test that covered projections correctly use the query system for projection and the $project
+    // stage is removed from the pipeline.
+    assertResultsMatch([{$project: {_id: 0, a: 1}}], true, {_id: 0, a: 1});
+    assertResultsMatch(
+        [{$project: {_id: 0, a: 1}}, {$group: {_id: null, a: {$sum: "$a"}}}], true, {_id: 0, a: 1});
+    assertResultsMatch([{$sort: {a: -1}}, {$project: {_id: 0, a: 1}}], true, {_id: 0, a: 1});
+    assertResultsMatch(
+        [
+          {$sort: {a: 1, 'c.d': 1}},
+          {$project: {_id: 0, a: 1}},
+          {$group: {_id: "$a", arr: {$push: "$a"}}}
+        ],
+        true,
+        {_id: 0, a: 1});
+    assertResultsMatch([{$project: {_id: 0, c: {d: 1}}}], true, {_id: 0, c: {d: 1}});
+
+    // Test that projections with renamed fields are not removed from the pipeline, however an
+    // inclusion projection is still pushed to the query system.
+    assertResultsMatch([{$project: {_id: 0, f: "$a"}}], true);
+    assertResultsMatch([{$project: {_id: 0, a: 1, f: "$a"}}], true);
+
+    // Test that uncovered projections include the $project stage in the pipeline.
+    assertResultsMatch([{$project: {_id: 1, a: 1}}], false);
+    assertResultsMatch([{$project: {_id: 0, a: 1, b: 1}}], false);
+    assertResultsMatch([{$sort: {a: 1}}, {$project: {_id: 1, b: 1}}], false);
+    assertResultsMatch(
+        [{$sort: {a: 1}}, {$group: {_id: "$_id", arr: {$push: "$a"}}}, {$project: {arr: 1}}],
+        false);
+
+    // Test that projections with computed fields are kept in the pipeline.
+    assertResultsMatch([{$project: {computedField: {$sum: "$a"}}}], false);
+    assertResultsMatch([{$project: {a: ["$a", "$b"]}}], false);
+    assertResultsMatch(
+        [{$project: {e: {$filter: {input: "$e", as: "item", cond: {"$eq": ["$$item", "elem0"]}}}}}],
+        false);
+
+    // Test that only the first projection is removed from the pipeline.
+    assertResultsMatch(
+        [
+          {$project: {_id: 0, a: 1}},
+          {$group: {_id: "$a", arr: {$push: "$a"}, a: {$sum: "$a"}}},
+          {$project: {_id: 0}}
+        ],
+        true,
+        {_id: 0, a: 1});
+
+    // Test that projections on _id with nested fields are not removed from pipeline. Due to
+    // SERVER-7502, the dependency analysis does not generate a covered projection for nested
+    // fields in _id and thus we cannot remove the stage.
+    indexSpec = {'_id.a': 1, a: 1};
+    assertResultsMatch([{$project: {'_id.a': 1}}], false, null, indexSpec);
+
+}());
diff --git a/src/mongo/db/pipeline/document_source_project.cpp b/src/mongo/db/pipeline/document_source_project.cpp
index a0ee096f36e..6cf11fc5a51 100644
--- a/src/mongo/db/pipeline/document_source_project.cpp
+++ b/src/mongo/db/pipeline/document_source_project.cpp
@@ -40,7 +40,6 @@ namespace mongo {
 
 using boost::intrusive_ptr;
 using parsed_aggregation_projection::ParsedAggregationProjection;
-using parsed_aggregation_projection::ProjectionType;
 
 REGISTER_DOCUMENT_SOURCE(project,
                          LiteParsedDocumentSourceDefault::parse,
diff --git a/src/mongo/db/pipeline/document_source_replace_root.cpp b/src/mongo/db/pipeline/document_source_replace_root.cpp
index 7524ea84572..b5def20495e 100644
--- a/src/mongo/db/pipeline/document_source_replace_root.cpp
+++ b/src/mongo/db/pipeline/document_source_replace_root.cpp
@@ -52,6 +52,10 @@ public:
     ReplaceRootTransformation(const boost::intrusive_ptr<ExpressionContext>& expCtx)
         : _expCtx(expCtx) {}
 
+    TransformerType getType() const final {
+        return TransformerType::kReplaceRoot;
+    }
+
     Document applyTransformation(const Document& input) final {
         // Extract subdocument in the form of a Value.
         Value newRoot = _newRoot->evaluate(input);
diff --git a/src/mongo/db/pipeline/document_source_single_document_transformation.h b/src/mongo/db/pipeline/document_source_single_document_transformation.h
index 3b91a979046..108a13aa161 100644
--- a/src/mongo/db/pipeline/document_source_single_document_transformation.h
+++ b/src/mongo/db/pipeline/document_source_single_document_transformation.h
@@ -52,12 +52,33 @@ public:
      */
     class TransformerInterface {
     public:
+        enum class TransformerType {
+            kExclusionProjection,
+            kInclusionProjection,
+            kComputedProjection,
+            kReplaceRoot,
+        };
         virtual ~TransformerInterface() = default;
         virtual Document applyTransformation(const Document& input) = 0;
+        virtual TransformerType getType() const = 0;
         virtual void optimize() = 0;
         virtual Document serialize(boost::optional<ExplainOptions::Verbosity> explain) const = 0;
         virtual DocumentSource::GetDepsReturn addDependencies(DepsTracker* deps) const = 0;
         virtual GetModPathsReturn getModifiedPaths() const = 0;
+
+        /**
+         * Returns true if this transformer is an inclusion projection and is a subset of
+         * 'proj', which must be a valid projection specification. For example, if this
+         * TransformerInterface represents the inclusion projection
+         *
+         *      {a: 1, b: 1, c: 1}
+         *
+         * then it is a subset of the projection {a: 1, c: 1}, and this function returns
+         * true.
+         */
+        virtual bool isSubsetOfProjection(const BSONObj& proj) const {
+            return false;
+        }
     };
 
     DocumentSourceSingleDocumentTransformation(
@@ -77,6 +98,14 @@ public:
         return true;
     }
 
+    TransformerInterface::TransformerType getType() const {
+        return _parsedTransform->getType();
+    }
+
+    bool isSubsetOfProjection(const BSONObj& proj) const {
+        return _parsedTransform->isSubsetOfProjection(proj);
+    }
+
 protected:
     void doDispose() final;
 
diff --git a/src/mongo/db/pipeline/parsed_add_fields.h b/src/mongo/db/pipeline/parsed_add_fields.h
index 7afa29d6801..e0d60822ba3 100644
--- a/src/mongo/db/pipeline/parsed_add_fields.h
+++ b/src/mongo/db/pipeline/parsed_add_fields.h
@@ -62,8 +62,8 @@ public:
     static std::unique_ptr<ParsedAddFields> create(
         const boost::intrusive_ptr<ExpressionContext>& expCtx, const BSONObj& spec);
 
-    ProjectionType getType() const final {
-        return ProjectionType::kComputed;
+    TransformerType getType() const final {
+        return TransformerType::kComputedProjection;
     }
 
     /**
diff --git a/src/mongo/db/pipeline/parsed_aggregation_projection.cpp b/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
index 26894147b17..cfa7d6da39c 100644
--- a/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
+++ b/src/mongo/db/pipeline/parsed_aggregation_projection.cpp
@@ -46,6 +46,9 @@
 namespace mongo {
 namespace parsed_aggregation_projection {
 
+using TransformerType =
+    DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType;
+
 //
 // ProjectionSpecValidator
 //
@@ -151,7 +154,7 @@ public:
      * fields (ones which are defined by an expression or a literal) are treated as inclusion
      * projections for in this context of the$project stage.
      */
-    static ProjectionType parse(const BSONObj& spec) {
+    static TransformerType parse(const BSONObj& spec) {
         ProjectTypeParser parser(spec);
         parser.parse();
         invariant(parser._parsedType);
@@ -177,13 +180,14 @@ private:
             BSONElement elem = _rawObj.firstElement();
             if (elem.fieldNameStringData() == "_id" && (elem.isBoolean() || elem.isNumber()) &&
                 !elem.trueValue()) {
-                _parsedType = ProjectionType::kExclusion;
+                _parsedType = TransformerType::kExclusionProjection;
             }
         }
 
         // Default to inclusion if nothing (except maybe '_id') is explicitly included or excluded.
         if (!_parsedType) {
-            _parsedType = ProjectionType::kInclusion;
+            _parsedType = DocumentSourceSingleDocumentTransformation::TransformerInterface::
+                TransformerType::kInclusionProjection;
         }
     }
 
@@ -208,8 +212,12 @@ private:
                         str::stream() << "Bad projection specification, cannot exclude fields "
                                          "other than '_id' in an inclusion projection: "
                                       << _rawObj.toString(),
-                        !_parsedType || (*_parsedType == ProjectionType::kExclusion));
-                _parsedType = ProjectionType::kExclusion;
+                        !_parsedType ||
+                            (*_parsedType ==
+                             DocumentSourceSingleDocumentTransformation::TransformerInterface::
+                                 TransformerType::kExclusionProjection));
+                _parsedType = DocumentSourceSingleDocumentTransformation::TransformerInterface::
+                    TransformerType::kExclusionProjection;
             }
         } else {
             // A boolean true, a truthy numeric value, or any expression can only be used with an
@@ -219,8 +227,12 @@ private:
                     str::stream() << "Bad projection specification, cannot include fields or "
                                      "add computed fields during an exclusion projection: "
                                   << _rawObj.toString(),
-                    !_parsedType || (*_parsedType == ProjectionType::kInclusion));
-            _parsedType = ProjectionType::kInclusion;
+                    !_parsedType ||
+                        (*_parsedType ==
+                         DocumentSourceSingleDocumentTransformation::TransformerInterface::
+                             TransformerType::kInclusionProjection));
+            _parsedType = DocumentSourceSingleDocumentTransformation::TransformerInterface::
+                TransformerType::kInclusionProjection;
         }
     }
 
@@ -241,8 +253,12 @@ private:
                         str::stream() << "Bad projection specification, cannot include fields or "
                                          "add computed fields during an exclusion projection: "
                                       << _rawObj.toString(),
-                        !_parsedType || _parsedType == ProjectionType::kInclusion);
-                _parsedType = ProjectionType::kInclusion;
+                        !_parsedType ||
+                            _parsedType ==
+                                DocumentSourceSingleDocumentTransformation::TransformerInterface::
+                                    TransformerType::kInclusionProjection);
+                _parsedType = DocumentSourceSingleDocumentTransformation::TransformerInterface::
+                    TransformerType::kInclusionProjection;
                 continue;
             }
             parseElement(elem, FieldPath::getFullyQualifiedPath(prefix.fullPath(), fieldName));
@@ -253,7 +269,7 @@ private:
     const BSONObj& _rawObj;
 
     // This will be populated during parse().
-    boost::optional<ProjectionType> _parsedType;
+    boost::optional<TransformerType> _parsedType;
 };
 
 }  // namespace
@@ -273,11 +289,11 @@ std::unique_ptr<ParsedAggregationProjection> ParsedAggregationProjection::create
     auto projectionType = ProjectTypeParser::parse(spec);
     // kComputed is a projection type reserved for $addFields, and should never be detected by the
     // ProjectTypeParser.
-    invariant(projectionType != ProjectionType::kComputed);
+    invariant(projectionType != TransformerType::kComputedProjection);
 
     // We can't use make_unique() here, since the branches have different types.
     std::unique_ptr<ParsedAggregationProjection> parsedProject(
-        projectionType == ProjectionType::kInclusion
+        projectionType == TransformerType::kInclusionProjection
             ? static_cast<ParsedAggregationProjection*>(new ParsedInclusionProjection(expCtx))
             : static_cast<ParsedAggregationProjection*>(new ParsedExclusionProjection(expCtx)));
 
diff --git a/src/mongo/db/pipeline/parsed_aggregation_projection.h b/src/mongo/db/pipeline/parsed_aggregation_projection.h
index dfda0f66756..721c086f6e1 100644
--- a/src/mongo/db/pipeline/parsed_aggregation_projection.h
+++ b/src/mongo/db/pipeline/parsed_aggregation_projection.h
@@ -45,8 +45,6 @@ class ExpressionContext;
 
 namespace parsed_aggregation_projection {
 
-enum class ProjectionType { kExclusion, kInclusion, kComputed };
-
 /**
  * This class ensures that the specification was valid: that none of the paths specified conflict
  * with one another, that there is at least one field, etc. Here "projection" includes both
@@ -123,11 +121,6 @@ public:
     virtual ~ParsedAggregationProjection() = default;
 
     /**
-     * Returns the type of projection represented by this ParsedAggregationProjection.
-     */
-    virtual ProjectionType getType() const = 0;
-
-    /**
      * Parse the user-specified BSON object 'spec'. By the time this is called, 'spec' has already
      * been verified to not have any conflicting path specifications, and not to mix and match
      * inclusions and exclusions. 'variablesParseState' is used by any contained expressions to
diff --git a/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp b/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp
index 71dd1c2378f..2ee53275e6a 100644
--- a/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp
+++ b/src/mongo/db/pipeline/parsed_aggregation_projection_test.cpp
@@ -353,94 +353,140 @@ TEST(ParsedAggregationProjectionErrors, ShouldNotErrorOnTwoNestedFields) {
 TEST(ParsedAggregationProjectionType, ShouldDefaultToInclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << true));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << wrapInLiteral(1)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << wrapInLiteral(1)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldDetectExclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << false));
-    ASSERT(parsedProject->getType() == ProjectionType::kExclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kExclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id.x" << false));
-    ASSERT(parsedProject->getType() == ProjectionType::kExclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kExclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << BSON("x" << false)));
-    ASSERT(parsedProject->getType() == ProjectionType::kExclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kExclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("x" << BSON("_id" << false)));
-    ASSERT(parsedProject->getType() == ProjectionType::kExclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kExclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << false));
-    ASSERT(parsedProject->getType() == ProjectionType::kExclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kExclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldDetectInclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << true));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject =
         ParsedAggregationProjection::create(expCtx, BSON("_id" << false << "a" << true));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject =
         ParsedAggregationProjection::create(expCtx, BSON("_id" << false << "a.b.c" << true));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id.x" << true));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id" << BSON("x" << true)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("x" << BSON("_id" << true)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldTreatOnlyComputedFieldsAsAnInclusionProjection) {
     const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
     auto parsedProject = ParsedAggregationProjection::create(expCtx, BSON("a" << wrapInLiteral(1)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(
         expCtx, BSON("_id" << false << "a" << wrapInLiteral(1)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(
         expCtx, BSON("_id" << false << "a.b.c" << wrapInLiteral(1)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(expCtx, BSON("_id.x" << wrapInLiteral(1)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject =
         ParsedAggregationProjection::create(expCtx, BSON("_id" << BSON("x" << wrapInLiteral(1))));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject =
         ParsedAggregationProjection::create(expCtx, BSON("x" << BSON("_id" << wrapInLiteral(1))));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::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)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(
         expCtx, BSON("a.b" << true << "a.c" << wrapInLiteral(1)));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 
     parsedProject = ParsedAggregationProjection::create(
         expCtx, BSON("a" << BSON("b" << true << "c" << wrapInLiteral(1))));
-    ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+    ASSERT(parsedProject->getType() ==
+           DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+               kInclusionProjection);
 }
 
 TEST(ParsedAggregationProjectionType, ShouldCoerceNumericsToBools) {
@@ -449,7 +495,9 @@ TEST(ParsedAggregationProjectionType, ShouldCoerceNumericsToBools) {
     for (auto&& zero : zeros) {
         auto parsedProject =
             ParsedAggregationProjection::create(expCtx, Document{{"a", zero}}.toBson());
-        ASSERT(parsedProject->getType() == ProjectionType::kExclusion);
+        ASSERT(parsedProject->getType() ==
+               DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+                   kExclusionProjection);
     }
 
     std::vector<Value> nonZeroes = {
@@ -457,7 +505,9 @@ TEST(ParsedAggregationProjectionType, ShouldCoerceNumericsToBools) {
     for (auto&& nonZero : nonZeroes) {
         auto parsedProject =
             ParsedAggregationProjection::create(expCtx, Document{{"a", nonZero}}.toBson());
-        ASSERT(parsedProject->getType() == ProjectionType::kInclusion);
+        ASSERT(parsedProject->getType() ==
+               DocumentSourceSingleDocumentTransformation::TransformerInterface::TransformerType::
+                   kInclusionProjection);
     }
 }
 
diff --git a/src/mongo/db/pipeline/parsed_exclusion_projection.h b/src/mongo/db/pipeline/parsed_exclusion_projection.h
index d9c495a9cfb..de371df0ccd 100644
--- a/src/mongo/db/pipeline/parsed_exclusion_projection.h
+++ b/src/mongo/db/pipeline/parsed_exclusion_projection.h
@@ -100,8 +100,8 @@ public:
     ParsedExclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx)
         : ParsedAggregationProjection(expCtx), _root(new ExclusionNode()) {}
 
-    ProjectionType getType() const final {
-        return ProjectionType::kExclusion;
+    TransformerType getType() const final {
+        return TransformerType::kExclusionProjection;
     }
 
     Document serialize(boost::optional<ExplainOptions::Verbosity> explain) const final;
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection.cpp b/src/mongo/db/pipeline/parsed_inclusion_projection.cpp
index 300623b632b..6c552f2aacb 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection.cpp
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection.cpp
@@ -418,5 +418,21 @@ void ParsedInclusionProjection::parseSubObject(const BSONObj& subObj,
         }
     }
 }
+
+bool ParsedInclusionProjection::isSubsetOfProjection(const BSONObj& proj) const {
+    std::set<std::string> preservedPaths;
+    _root->addPreservedPaths(&preservedPaths);
+    for (auto&& includedField : preservedPaths) {
+        if (!proj.hasField(includedField))
+            return false;
+    }
+
+    // If the inclusion has any computed fields or renamed fields, then it's not a subset.
+    std::set<std::string> computedPaths;
+    StringMap<std::string> renamedPaths;
+    _root->addComputedPaths(&computedPaths, &renamedPaths);
+    return computedPaths.empty() && renamedPaths.empty();
+}
+
 }  // namespace parsed_aggregation_projection
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection.h b/src/mongo/db/pipeline/parsed_inclusion_projection.h
index e27ed5763aa..c42acc75e85 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection.h
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection.h
@@ -187,8 +187,8 @@ public:
     ParsedInclusionProjection(const boost::intrusive_ptr<ExpressionContext>& expCtx)
         : ParsedAggregationProjection(expCtx), _root(new InclusionNode()) {}
 
-    ProjectionType getType() const final {
-        return ProjectionType::kInclusion;
+    TransformerType getType() const final {
+        return TransformerType::kInclusionProjection;
     }
 
     /**
@@ -244,6 +244,13 @@ public:
      */
     Document applyProjection(const Document& inputDoc) const final;
 
+    /*
+     * Checks whether the inclusion projection represented by the InclusionNode
+     * tree is a subset of the object passed in. Projections that have any
+     * computed or renamed fields are not considered a subset.
+     */
+    bool isSubsetOfProjection(const BSONObj& proj) const final;
+
 private:
     /**
      * Attempts to parse 'objSpec' as an expression like {$add: [...]}. Adds a computed field to
diff --git a/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp b/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp
index add0c9e7bb3..4c1395504cd 100644
--- a/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp
+++ b/src/mongo/db/pipeline/parsed_inclusion_projection_test.cpp
@@ -643,6 +643,104 @@ TEST(InclusionProjectionExecutionTest, ShouldAlwaysKeepMetadataFromOriginalDoc)
     ASSERT_DOCUMENT_EQ(result, expectedDoc.freeze());
 }
 
+//
+// Detection of subset projection.
+//
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForIdenticalProjection) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a" << true << "b" << true));
+
+    auto proj = BSON("_id" << false << "a" << true << "b" << true);
+
+    ASSERT_TRUE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForSupersetProjection) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a" << true << "b" << true));
+
+    auto proj = BSON("_id" << false << "a" << true << "b" << true << "c" << true);
+
+    ASSERT_TRUE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForIdenticalNestedProjection) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a.b" << true));
+
+    auto proj = BSON("_id" << false << "a.b" << true);
+
+    ASSERT_TRUE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectSubsetForSupersetProjectionWithNestedFields) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a" << true << "c" << BSON("d" << true)));
+
+    auto proj = BSON("_id" << false << "a" << true << "b" << true << "c.d" << true);
+
+    ASSERT_TRUE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissingFields) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a" << true << "b" << true));
+
+    auto proj = BSON("_id" << false << "a" << true);
+    ASSERT_FALSE(inclusion.isSubsetOfProjection(proj));
+
+    proj = BSON("_id" << false << "a" << true << "c" << true);
+    ASSERT_FALSE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest,
+     ShouldDetectNonSubsetForSupersetProjectionWithoutComputedFields) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a" << true << "b" << true << "c" << BSON("$literal" << 1)));
+
+    auto proj = BSON("_id" << false << "a" << true << "b" << true);
+
+    ASSERT_FALSE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissingNestedFields) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a.b" << true << "a.c" << true));
+
+    auto proj = BSON("_id" << false << "a.b" << true);
+
+    ASSERT_FALSE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithRenamedFields) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a"
+                         << "$b"));
+
+    auto proj = BSON("_id" << false << "b" << true);
+
+    ASSERT_FALSE(inclusion.isSubsetOfProjection(proj));
+}
+
+TEST(InclusionProjectionSubsetTest, ShouldDetectNonSubsetForProjectionWithMissingIdField) {
+    const boost::intrusive_ptr<ExpressionContextForTest> expCtx(new ExpressionContextForTest());
+    ParsedInclusionProjection inclusion(expCtx);
+    inclusion.parse(BSON("a" << true));
+
+    auto proj = BSON("a" << true);
+
+    ASSERT_FALSE(inclusion.isSubsetOfProjection(proj));
+}
+
 }  // namespace
 }  // namespace parsed_aggregation_projection
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index 236e747a463..b26baeb7bb4 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -55,6 +55,7 @@
 #include "mongo/db/pipeline/document_source_merge_cursors.h"
 #include "mongo/db/pipeline/document_source_sample.h"
 #include "mongo/db/pipeline/document_source_sample_from_random_cursor.h"
+#include "mongo/db/pipeline/document_source_single_document_transformation.h"
 #include "mongo/db/pipeline/document_source_sort.h"
 #include "mongo/db/pipeline/pipeline.h"
 #include "mongo/db/query/collation/collator_interface.h"
@@ -435,6 +436,17 @@ void PipelineD::prepareCursorSource(Collection* collection,
                                                 &sortObj,
                                                 &projForQuery));
 
+
+    if (!projForQuery.isEmpty() && !sources.empty()) {
+        // Check for redundant $project in query with the same specification as the inclusion
+        // projection generated by the dependency optimization.
+        auto proj =
+            dynamic_cast<DocumentSourceSingleDocumentTransformation*>(sources.front().get());
+        if (proj && proj->isSubsetOfProjection(projForQuery)) {
+            sources.pop_front();
+        }
+    }
+
     addCursorSource(
         collection, pipeline, expCtx, std::move(exec), deps, queryObj, sortObj, projForQuery);
 }