SERVER-53080 generate optimized covered shard filtering plans in SBE

When the SHARDING_FILTER stage's child is an IXSCAN, we will
generate a plan where the slots built from pieces of the
IXSCAN's keys are used directly to determine the shard key.

17 files changed, 646 insertions, 173 deletions

diff --git a/jstests/libs/analyze_plan.js b/jstests/libs/analyze_plan.js
index 0fff85a7f80..fda996dfe56 100644
--- a/jstests/libs/analyze_plan.js
+++ b/jstests/libs/analyze_plan.js
@@ -59,11 +59,12 @@ function getPlanStages(root, stage) {
 
     if ("shards" in root) {
         if (Array.isArray(root.shards)) {
-            results = root.shards.reduce(
-                (res, shard) => res.concat(getPlanStages(
-                    shard.hasOwnProperty("winningPlan") ? shard.winningPlan : shard.executionStages,
-                    stage)),
-                results);
+            results =
+                root.shards.reduce((res, shard) => res.concat(getPlanStages(
+                                       shard.hasOwnProperty("winningPlan") ? getWinningPlan(shard)
+                                                                           : shard.executionStages,
+                                       stage)),
+                                   results);
         } else {
             const shards = Object.keys(root.shards);
             results = shards.reduce(
diff --git a/jstests/noPassthrough/shard_filtering.js b/jstests/noPassthrough/shard_filtering.js
new file mode 100644
index 00000000000..3117d4d3451
--- /dev/null
+++ b/jstests/noPassthrough/shard_filtering.js
@@ -0,0 +1,151 @@
+/**
+ * This test is intended to exercise shard filtering logic. This test works by sharding a
+ * collection, and then inserting orphaned documents directly into one of the shards. It then runs a
+ * find() and makes sure that orphaned documents are filtered out.
+ * @tags: [
+ *   requires_sharding,
+ * ]
+ */
+(function() {
+"use strict";
+
+load("jstests/libs/analyze_plan.js");
+
+// Deliberately inserts orphans outside of migration.
+TestData.skipCheckOrphans = true;
+const st = new ShardingTest({shards: 2});
+const collName = "test.shardfilter";
+const mongosDb = st.s.getDB("test");
+const mongosColl = st.s.getCollection(collName);
+
+assert.commandWorked(st.s.adminCommand({enableSharding: "test"}));
+st.ensurePrimaryShard("test", st.shard1.name);
+assert.commandWorked(
+    st.s.adminCommand({shardCollection: collName, key: {a: 1, "b.c": 1, "d.e.f": 1}}));
+
+// Put a chunk with no data onto shard0 in order to make sure that both shards get targeted.
+assert.commandWorked(st.s.adminCommand({split: collName, middle: {a: 20, "b.c": 0, "d.e.f": 0}}));
+assert.commandWorked(st.s.adminCommand({split: collName, middle: {a: 30, "b.c": 0, "d.e.f": 0}}));
+assert.commandWorked(st.s.adminCommand(
+    {moveChunk: collName, find: {a: 25, "b.c": 0, "d.e.f": 0}, to: st.shard0.shardName}));
+
+// Shard the collection and insert some docs.
+const docs = [
+    {_id: 0, a: 1, b: {c: 1}, d: {e: {f: 1}}, g: 100},
+    {_id: 1, a: 1, b: {c: 2}, d: {e: {f: 2}}, g: 100.9},
+    {_id: 2, a: 1, b: {c: 3}, d: {e: {f: 3}}, g: "a"},
+    {_id: 3, a: 1, b: {c: 3}, d: {e: {f: 3}}, g: [1, 2, 3]},
+    {_id: 4, a: "a", b: {c: "b"}, d: {e: {f: "c"}}, g: null},
+    {_id: 5, a: 1.0, b: {c: "b"}, d: {e: {f: Infinity}}, g: NaN}
+];
+assert.commandWorked(mongosColl.insert(docs));
+assert.eq(mongosColl.find().itcount(), 6);
+
+// Insert some documents with valid partial shard keys to both shards. The versions of these
+// documents on shard0 are orphans, since all of the data is owned by shard1.
+const docsWithMissingAndNullKeys = [
+    {_id: 6, a: "missingParts"},
+    {_id: 7, a: null, b: {c: 1}, d: {e: {f: 1}}},
+    {_id: 8, a: "null", b: {c: null}, d: {e: {f: 1}}},
+    {_id: 9, a: "deepNull", b: {c: 1}, d: {e: {f: null}}},
+];
+assert.commandWorked(st.shard0.getCollection(collName).insert(docsWithMissingAndNullKeys));
+assert.commandWorked(st.shard1.getCollection(collName).insert(docsWithMissingAndNullKeys));
+
+// Insert orphan docs without missing or null shard keys onto shard0 and test that they get filtered
+// out.
+const orphanDocs = [
+    {_id: 10, a: 100, b: {c: 10}, d: {e: {f: 999}}, g: "a"},
+    {_id: 11, a: 101, b: {c: 11}, d: {e: {f: 1000}}, g: "b"}
+];
+assert.commandWorked(st.shard0.getCollection(collName).insert(orphanDocs));
+assert.eq(mongosColl.find().itcount(), 10);
+
+// Insert docs directly into shard0 to test that regular (non-null, non-missing) shard keys get
+// filtered out.
+assert.commandWorked(st.shard0.getCollection(collName).insert(docs));
+assert.eq(mongosColl.find().itcount(), 10);
+
+// Ensure that shard filtering works correctly for a query that can use the index supporting the
+// shard key. In this case, shard filtering can occur before the FETCH stage, but the plan is not
+// covered.
+let explain = mongosColl.find({a: {$gte: 0}}).explain();
+assert.eq(explain.queryPlanner.winningPlan.stage, "SHARD_MERGE", explain);
+assert(planHasStage(mongosDb, explain.queryPlanner.winningPlan, "SHARDING_FILTER"), explain);
+assert(isIxscan(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert(!isIndexOnly(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert.sameMembers(mongosColl.find({a: {$gte: 0}}).toArray(), [
+    {_id: 0, a: 1, b: {c: 1}, d: {e: {f: 1}}, g: 100},
+    {_id: 1, a: 1, b: {c: 2}, d: {e: {f: 2}}, g: 100.9},
+    {_id: 2, a: 1, b: {c: 3}, d: {e: {f: 3}}, g: "a"},
+    {_id: 3, a: 1, b: {c: 3}, d: {e: {f: 3}}, g: [1, 2, 3]},
+    {_id: 5, a: 1, b: {c: "b"}, d: {e: {f: Infinity}}, g: NaN}
+]);
+
+// In this case, shard filtering is done as part of a covered plan.
+explain = mongosColl.find({a: {$gte: 0}}, {_id: 0, a: 1}).explain();
+assert.eq(explain.queryPlanner.winningPlan.stage, "SHARD_MERGE", explain);
+assert(planHasStage(mongosDb, explain.queryPlanner.winningPlan, "SHARDING_FILTER"), explain);
+assert(isIxscan(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert(isIndexOnly(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert.sameMembers(mongosColl.find({a: {$gte: 0}}, {_id: 0, a: 1}).toArray(), [
+    {a: 1},
+    {a: 1},
+    {a: 1},
+    {a: 1},
+    {a: 1},
+]);
+
+// Drop the collection and shard it by a new key that has no dotted fields. Again, make sure that
+// shard0 has an empty chunk.
+assert(mongosColl.drop());
+assert.commandWorked(st.s.adminCommand({shardCollection: collName, key: {a: 1, b: 1, c: 1, d: 1}}));
+assert.commandWorked(st.s.adminCommand({split: collName, middle: {a: 20, b: 0, c: 0, d: 0}}));
+assert.commandWorked(st.s.adminCommand({split: collName, middle: {a: 30, b: 0, c: 0, d: 0}}));
+assert.commandWorked(st.s.adminCommand(
+    {moveChunk: collName, find: {a: 25, b: 0, c: 0, d: 0}, to: st.shard0.shardName}));
+
+// Insert some data via mongos, and also insert some documents directly to shard0 to produce an
+// orphans.
+assert.commandWorked(mongosColl.insert([
+    {_id: 0, a: 0, b: 0, c: 0, d: 0},
+    {_id: 1, a: 1, b: 1, c: 1, d: 1},
+    {_id: 2, a: -1, b: -1, c: -1, d: -1},
+]));
+assert.commandWorked(st.shard0.getCollection(collName).insert({_id: 3, a: 0, b: 0, c: 0, d: 0}));
+assert.commandWorked(st.shard0.getCollection(collName).insert({_id: 4, a: 0, b: 99, c: 0, d: 99}));
+assert.commandWorked(st.shard0.getCollection(collName).insert({_id: 5, a: 0, b: 0, c: 99, d: 99}));
+assert.commandWorked(st.shard0.getCollection(collName).insert({_id: 6, a: 0, b: 99, c: 99, d: 99}));
+
+// Run a query that can use covered shard filtering where the projection involves more than one
+// field of the shard key.
+explain = mongosColl.find({a: {$gte: 0}}, {_id: 0, a: 1, b: 1, d: 1}).explain();
+assert.eq(explain.queryPlanner.winningPlan.stage, "SHARD_MERGE", explain);
+assert(planHasStage(mongosDb, explain.queryPlanner.winningPlan, "SHARDING_FILTER"), explain);
+assert(isIxscan(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert(isIndexOnly(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert.sameMembers(mongosColl.find({a: {$gte: 0}}, {_id: 0, a: 1, b: 1, d: 1}).toArray(),
+                   [{a: 0, b: 0, d: 0}, {a: 1, b: 1, d: 1}]);
+
+// Run a query that will use a covered OR plan.
+explain = mongosColl.find({$or: [{a: 0, c: 0}, {a: 25, c: 0}]}, {_id: 0, a: 1, c: 1}).explain();
+assert.eq(explain.queryPlanner.winningPlan.stage, "SHARD_MERGE", explain);
+assert(planHasStage(mongosDb, explain.queryPlanner.winningPlan, "SHARDING_FILTER"), explain);
+assert(planHasStage(mongosDb, explain.queryPlanner.winningPlan, "OR"), explain);
+assert(isIndexOnly(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert.sameMembers(
+    mongosColl.find({$or: [{a: 0, c: 0}, {a: 25, c: 0}]}, {_id: 0, a: 1, c: 1}).toArray(),
+    [{a: 0, c: 0}]);
+
+// Similar case to above, but here the index scans involve a single interval of the index.
+explain = mongosColl.find({$or: [{a: 0, b: 0}, {a: 25, b: 0}]}, {_id: 0, a: 1, b: 1}).explain();
+assert.eq(explain.queryPlanner.winningPlan.stage, "SHARD_MERGE", explain);
+assert(planHasStage(mongosDb, explain.queryPlanner.winningPlan, "SHARDING_FILTER"), explain);
+assert(planHasStage(mongosDb, explain.queryPlanner.winningPlan, "OR"), explain);
+assert(isIndexOnly(mongosDb, explain.queryPlanner.winningPlan), explain);
+assert.sameMembers(
+    mongosColl.find({$or: [{a: 0, b: 0}, {a: 25, b: 0}]}, {_id: 0, a: 1, b: 1}).toArray(),
+    [{a: 0, b: 0}]);
+
+st.stop();
+})();
diff --git a/jstests/noPassthrough/shard_filtering_sbe.js b/jstests/noPassthrough/shard_filtering_sbe.js
deleted file mode 100644
index 2c9d79a90a7..00000000000
--- a/jstests/noPassthrough/shard_filtering_sbe.js
+++ /dev/null
@@ -1,60 +0,0 @@
-/**
- * This test is intended to exercise shard filtering logic. This test works by sharding a
- * collection, and then inserting orphaned documents directly into one of the shards. It then runs a
- * find() and makes sure that orphaned documents are filtered out.
- * @tags: [
- *   requires_sharding,
- * ]
- */
-(function() {
-"use strict";
-
-// Deliberately inserts orphans outside of migration.
-TestData.skipCheckOrphans = true;
-const st = new ShardingTest({shards: 2});
-const collName = "test.shardfilter";
-const shard0Coll = st.s.getCollection(collName);
-
-assert.commandWorked(st.s.adminCommand({enableSharding: "test"}));
-st.ensurePrimaryShard("test", st.shard1.name);
-assert.commandWorked(
-    st.s.adminCommand({shardcollection: collName, key: {'a': 1, 'b.c': 1, 'd.e.f': 1}}));
-
-// Shard the collection and insert some docs.
-const docs = [
-    {a: 1, b: {c: 1}, d: {e: {f: 1}}, g: 100},
-    {a: 1, b: {c: 2}, d: {e: {f: 2}}, g: 100.9},
-    {a: 1, b: {c: 3}, d: {e: {f: 3}}, g: "a"},
-    {a: 1, b: {c: 3}, d: {e: {f: 3}}, g: [1, 2, 3]},
-    {a: "a", b: {c: "b"}, d: {e: {f: "c"}}, g: null},
-    {a: 1.0, b: {c: "b"}, d: {e: {f: Infinity}}, g: NaN}
-];
-assert.commandWorked(st.getDB("test").shardfilter.insert(docs));
-assert.eq(st.getDB("test").shardfilter.find().itcount(), 6);
-
-// Insert some documents with valid partial shard keys to both shards.
-const docsWithMissingAndNullKeys = [
-    {a: "missingParts"},
-    {a: null, b: {c: 1}, d: {e: {f: 1}}},
-    {a: "null", b: {c: null}, d: {e: {f: 1}}},
-    {a: "deepNull", b: {c: 1}, d: {e: {f: null}}},
-];
-assert.commandWorked(st.shard0.getCollection(collName).insert(docsWithMissingAndNullKeys));
-assert.commandWorked(st.shard1.getCollection(collName).insert(docsWithMissingAndNullKeys));
-
-// Insert docs without missing or null shard keys onto shard0 and test that they get filtered
-// out.
-const orphanDocs = [
-    {a: 100, b: {c: 10}, d: {e: {f: 999}}, g: "a"},
-    {a: 101, b: {c: 11}, d: {e: {f: 1000}}, g: "b"}
-];
-assert.commandWorked(st.shard0.getCollection(collName).insert(orphanDocs));
-assert.eq(st.getDB("test").shardfilter.find().itcount(), 10);
-
-// Insert docs directly into shard0 to test that regular (non-null, non-missing) shard keys get
-// filtered out.
-assert.commandWorked(st.shard0.getCollection(collName).insert(docs));
-assert.eq(st.getDB("test").shardfilter.find().itcount(), 10);
-
-st.stop();
-})();
diff --git a/src/mongo/db/exec/sbe/expressions/sbe_shard_filter_builtin_test.cpp b/src/mongo/db/exec/sbe/expressions/sbe_shard_filter_builtin_test.cpp
index 658ee8a852b..a94e1e48a8f 100644
--- a/src/mongo/db/exec/sbe/expressions/sbe_shard_filter_builtin_test.cpp
+++ b/src/mongo/db/exec/sbe/expressions/sbe_shard_filter_builtin_test.cpp
@@ -83,10 +83,10 @@ TEST_F(SbeShardFilterBuiltinTest, BasicFiltering) {
     value::ViewOfValueAccessor objAccessor;
     auto objSlot = bindAccessor(&objAccessor);
 
-    auto [tagEmptyObj, valEmptyObj] = value::makeNewObject();
-    value::ValueGuard objGuard{tagEmptyObj, valEmptyObj};
+    BSONObj emptyBson;
+    objAccessor.reset(value::TypeTags::bsonObject,
+                      value::bitcastFrom<const char*>(emptyBson.objdata()));
 
-    objAccessor.reset(tagEmptyObj, valEmptyObj);
     runAndAssertExpression(makeAlwaysPassShardFilter(), objSlot, true);
     runAndAssertExpression(makeAlwaysFailShardFilter(), objSlot, false);
     runAndAssertExpression(makeAlwaysPassShardFilter(), objSlot, true);
@@ -104,14 +104,26 @@ TEST_F(SbeShardFilterBuiltinTest, MissingShardKey) {
     runAndAssertExpression(makeAlwaysPassShardFilter(), inputSlot, false);
 };
 
+TEST_F(SbeShardFilterBuiltinTest, ShardKeyAsObjectNotBsonObjIsRejected) {
+    value::ViewOfValueAccessor inputSlotAccessor;
+    auto inputSlot = bindAccessor(&inputSlotAccessor);
+
+    auto [tagEmptyObj, valEmptyObj] = value::makeNewObject();
+    value::ValueGuard objGuard{tagEmptyObj, valEmptyObj};
+    inputSlotAccessor.reset(tagEmptyObj, valEmptyObj);
+
+    runAndAssertExpression(makeAlwaysPassShardFilter(), inputSlot, false);
+    runAndAssertExpression(makeAlwaysFailShardFilter(), inputSlot, false);
+};
+
 TEST_F(SbeShardFilterBuiltinTest, BadScopedCollFilterValue) {
     value::ViewOfValueAccessor objAccessor;
     auto objSlot = bindAccessor(&objAccessor);
 
-    auto [tagEmptyObj, valEmptyObj] = value::makeNewObject();
-    value::ValueGuard objGuard{tagEmptyObj, valEmptyObj};
+    BSONObj emptyBson;
+    objAccessor.reset(value::TypeTags::bsonObject,
+                      value::bitcastFrom<const char*>(emptyBson.objdata()));
 
-    objAccessor.reset(tagEmptyObj, valEmptyObj);
     runAndAssertExpression({value::TypeTags::Boolean, true}, objSlot, false);
     runAndAssertExpression({value::TypeTags::Boolean, true}, objSlot, false);
 };
diff --git a/src/mongo/db/exec/sbe/vm/vm.cpp b/src/mongo/db/exec/sbe/vm/vm.cpp
index 1f8d7a616bc..09cb54ffac8 100644
--- a/src/mongo/db/exec/sbe/vm/vm.cpp
+++ b/src/mongo/db/exec/sbe/vm/vm.cpp
@@ -2640,7 +2640,7 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinShardFilter(Ari
     auto [ownedFilter, filterTag, filterValue] = getFromStack(0);
     auto [ownedShardKey, shardKeyTag, shardKeyValue] = getFromStack(1);
 
-    if (filterTag != value::TypeTags::shardFilterer || shardKeyTag != value::TypeTags::Object) {
+    if (filterTag != value::TypeTags::shardFilterer || shardKeyTag != value::TypeTags::bsonObject) {
         if (filterTag == value::TypeTags::shardFilterer &&
             shardKeyTag == value::TypeTags::Nothing) {
             LOGV2_WARNING(5071200,
@@ -2652,13 +2652,11 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinShardFilter(Ari
         return {false, value::TypeTags::Nothing, 0};
     }
 
-    BSONObjBuilder bob;
-    bson::convertToBsonObj(bob, value::getObjectView(shardKeyValue));
-
+    BSONObj keyAsUnownedBson{sbe::value::bitcastTo<const char*>(shardKeyValue)};
     return {false,
             value::TypeTags::Boolean,
             value::bitcastFrom<bool>(
-                value::getShardFiltererView(filterValue)->keyBelongsToMe(bob.done()))};
+                value::getShardFiltererView(filterValue)->keyBelongsToMe(keyAsUnownedBson))};
 }
 
 std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinExtractSubArray(ArityType arity) {
diff --git a/src/mongo/db/exec/shard_filter.h b/src/mongo/db/exec/shard_filter.h
index 5f763852de6..38e23a41aea 100644
--- a/src/mongo/db/exec/shard_filter.h
+++ b/src/mongo/db/exec/shard_filter.h
@@ -35,39 +35,17 @@
 namespace mongo {
 
 /**
- * This stage drops documents that didn't belong to the shard we're executing on at the time of
- * construction. This matches the contract for sharded cursorids which guarantees that a
- * StaleConfigException will be thrown early or the cursorid for its entire lifetime will return
- * documents matching the shard version set on the connection at the time of cursorid creation.
+ * This stage drops documents (called "orphans") that don't logically belong to this shard according
+ * to the the provided 'collectionFilter'. No data should be returned from a query in ranges of
+ * migrations that committed after the query started, or from ranges not owned when the query began.
  *
- * A related system will ensure that the data migrated away from a shard will not be deleted as
- * long as there are active queries from before the migration. Currently, "active queries" is
- * defined by cursorids so it is important that the metadata used in this stage uses the same
- * version as the cursorid. Therefore, you must wrap any Runner using this Stage in a
- * ClientCursor during the same lock grab as constructing the Runner.
+ * A related system will ensure that the data migrated away from a shard will not be deleted as long
+ * as there are active queries from before the migration. By holding onto a copy of the provided
+ * 'collectionFilter', this stage signals to the sharding subsystem that the data required at the
+ * associated shard version cannot yet be deleted. In other words, no migrated data should be
+ * removed from a shard while there are queries that were active before the migration.
  *
- * BEGIN NOTE FROM GREG
- *
- * There are three sharded query contracts:
- *
- * 0) Migration commit takes the db lock - i.e. is serialized with writes and reads.
- * 1) No data should be returned from a query in ranges of migrations that committed after the
- * query started, or from ranges not owned when the query began.
- * 2) No migrated data should be removed from a shard while there are queries that were active
- * before the migration.
- *
- * As implementation details, collection metadata is used to determine the ranges of all data
- * not actively migrated (or orphaned).  CursorIds are currently used to establish "active"
- * queries before migration commit.
- *
- * Combining all this: if a query is started in a db lock and acquires in that (same) lock the
- * collection metadata and a cursorId, the query will return results for exactly the ranges in
- * the metadata (though of arbitrary staleness).  This is the sharded collection query contract.
- *
- * END NOTE FROM GREG
- *
- * Preconditions: Child must be fetched.  TODO: when covering analysis is in just build doc
- * and check that against shard key.  See SERVER-5022.
+ * Preconditions: Child must be fetched.
  */
 class ShardFilterStage final : public PlanStage {
 public:
@@ -93,7 +71,6 @@ public:
 private:
     WorkingSet* _ws;
 
-    // Stats
     ShardingFilterStats _specificStats;
 
     // Note: it is important that this owns the ScopedCollectionFilter from the time this stage
diff --git a/src/mongo/db/query/classic_stage_builder.cpp b/src/mongo/db/query/classic_stage_builder.cpp
index dfe01081c8f..a70ff51848f 100644
--- a/src/mongo/db/query/classic_stage_builder.cpp
+++ b/src/mongo/db/query/classic_stage_builder.cpp
@@ -357,7 +357,11 @@ std::unique_ptr<PlanStage> ClassicStageBuilder::build(const QuerySolutionNode* r
         }
         case STAGE_VIRTUAL_SCAN: {
             const auto* vsn = static_cast<const VirtualScanNode*>(root);
-            invariant(vsn->hasRecordId);
+
+            // The classic stage builder currently only supports VirtualScanNodes which represent
+            // collection scans that do not produce record ids.
+            invariant(!vsn->hasRecordId);
+            invariant(vsn->scanType == VirtualScanNode::ScanType::kCollScan);
 
             auto qds = std::make_unique<QueuedDataStage>(expCtx, _ws);
             for (auto&& arr : vsn->docs) {
diff --git a/src/mongo/db/query/classic_stage_builder_test.cpp b/src/mongo/db/query/classic_stage_builder_test.cpp
index 64f00ebf4c2..0fd2f44faa0 100644
--- a/src/mongo/db/query/classic_stage_builder_test.cpp
+++ b/src/mongo/db/query/classic_stage_builder_test.cpp
@@ -122,7 +122,8 @@ TEST_F(ClassicStageBuilderTest, VirtualScanTranslation) {
 
     // Construct a QuerySolution consisting of a single VirtualScanNode to test if a stream of
     // documents can be produced.
-    auto virtScan = std::make_unique<VirtualScanNode>(docs, true);
+    auto virtScan =
+        std::make_unique<VirtualScanNode>(docs, VirtualScanNode::ScanType::kCollScan, false);
 
     // Make a QuerySolution from the root virtual scan node.
     auto querySolution = makeQuerySolution(std::move(virtScan));
diff --git a/src/mongo/db/query/query_solution.cpp b/src/mongo/db/query/query_solution.cpp
index 2b00f582496..35b16f72588 100644
--- a/src/mongo/db/query/query_solution.cpp
+++ b/src/mongo/db/query/query_solution.cpp
@@ -241,8 +241,14 @@ QuerySolutionNode* CollectionScanNode::clone() const {
 // VirtualScanNode
 //
 
-VirtualScanNode::VirtualScanNode(std::vector<BSONArray> docs, bool hasRecordId)
-    : docs(std::move(docs)), hasRecordId(hasRecordId) {}
+VirtualScanNode::VirtualScanNode(std::vector<BSONArray> docs,
+                                 ScanType scanType,
+                                 bool hasRecordId,
+                                 BSONObj indexKeyPattern)
+    : docs(std::move(docs)),
+      scanType(scanType),
+      hasRecordId(hasRecordId),
+      indexKeyPattern(std::move(indexKeyPattern)) {}
 
 void VirtualScanNode::appendToString(str::stream* ss, int indent) const {
     addIndent(ss, indent);
@@ -252,10 +258,14 @@ void VirtualScanNode::appendToString(str::stream* ss, int indent) const {
     addIndent(ss, indent + 1);
     *ss << "hasRecordId = " << hasRecordId;
     addCommon(ss, indent);
+    *ss << "scanType = " << static_cast<size_t>(scanType);
+    addCommon(ss, indent);
+    *ss << "indexKeyPattern = " << indexKeyPattern;
+    addCommon(ss, indent);
 }
 
 QuerySolutionNode* VirtualScanNode::clone() const {
-    auto copy = new VirtualScanNode(docs, this->hasRecordId);
+    auto copy = new VirtualScanNode(docs, scanType, hasRecordId, indexKeyPattern);
     cloneBaseData(copy);
     return copy;
 }
diff --git a/src/mongo/db/query/query_solution.h b/src/mongo/db/query/query_solution.h
index 6445b7d8f2f..7e5d6632a74 100644
--- a/src/mongo/db/query/query_solution.h
+++ b/src/mongo/db/query/query_solution.h
@@ -515,8 +515,12 @@ struct CollectionScanNode : public QuerySolutionNodeWithSortSet {
  * collection or an index scan in memory by using a backing vector of BSONArray.
  */
 struct VirtualScanNode : public QuerySolutionNodeWithSortSet {
-    VirtualScanNode(std::vector<BSONArray> docs, bool hasRecordId);
-    virtual ~VirtualScanNode() {}
+    enum class ScanType { kCollScan, kIxscan };
+
+    VirtualScanNode(std::vector<BSONArray> docs,
+                    ScanType scanType,
+                    bool hasRecordId,
+                    BSONObj indexKeyPattern = {});
 
     virtual StageType getType() const {
         return STAGE_VIRTUAL_SCAN;
@@ -525,10 +529,15 @@ struct VirtualScanNode : public QuerySolutionNodeWithSortSet {
     virtual void appendToString(str::stream* ss, int indent) const;
 
     bool fetched() const {
-        return true;
+        return scanType == ScanType::kCollScan;
     }
     FieldAvailability getFieldAvailability(const std::string& field) const {
-        return FieldAvailability::kFullyProvided;
+        if (scanType == ScanType::kCollScan) {
+            return FieldAvailability::kFullyProvided;
+        } else {
+            return indexKeyPattern.hasField(field) ? FieldAvailability::kFullyProvided
+                                                   : FieldAvailability::kNotProvided;
+        }
     }
     bool sortedByDiskLoc() const {
         return false;
@@ -547,11 +556,17 @@ struct VirtualScanNode : public QuerySolutionNodeWithSortSet {
     // BSONObj in the first position of the array.
     std::vector<BSONArray> docs;
 
+    // Indicates whether the scan is mimicking a collection scan or index scan.
+    const ScanType scanType;
+
     // A flag to indicate the format of the BSONArray document payload in the above vector, docs. If
     // hasRecordId is set to true, then both a RecordId and a BSONObj document are stored in that
     // order for every BSONArray in docs. Otherwise, the RecordId is omitted and the BSONArray will
     // only carry a BSONObj document.
     bool hasRecordId;
+
+    // Set when 'scanType' is 'kIxscan'.
+    BSONObj indexKeyPattern;
 };
 
 struct AndHashNode : public QuerySolutionNode {
diff --git a/src/mongo/db/query/sbe_and_hash_test.cpp b/src/mongo/db/query/sbe_and_hash_test.cpp
index ea3f5f89621..fe74c34836e 100644
--- a/src/mongo/db/query/sbe_and_hash_test.cpp
+++ b/src/mongo/db/query/sbe_and_hash_test.cpp
@@ -49,7 +49,8 @@ protected:
         std::vector<std::vector<BSONArray>> docsVec) {
         auto andHashNode = std::make_unique<AndHashNode>();
         for (auto docs : docsVec) {
-            auto virtScan = std::make_unique<VirtualScanNode>(docs, true);
+            auto virtScan =
+                std::make_unique<VirtualScanNode>(docs, VirtualScanNode::ScanType::kCollScan, true);
             andHashNode->children.push_back(virtScan.release());
         }
         return std::move(andHashNode);
@@ -188,4 +189,4 @@ TEST_F(SbeAndHashTest, TestTwoIndexArrays) {
     runTest(docs, expected);
 }
 
-}  // namespace mongo
\ No newline at end of file
+}  // namespace mongo
diff --git a/src/mongo/db/query/sbe_shard_filter_test.cpp b/src/mongo/db/query/sbe_shard_filter_test.cpp
index 5f92418e9d4..28fd8d5615b 100644
--- a/src/mongo/db/query/sbe_shard_filter_test.cpp
+++ b/src/mongo/db/query/sbe_shard_filter_test.cpp
@@ -30,6 +30,8 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/exec/shard_filterer_mock.h"
+#include "mongo/db/pipeline/expression_context_for_test.h"
+#include "mongo/db/query/projection_parser.h"
 #include "mongo/db/query/query_solution.h"
 #include "mongo/db/query/sbe_stage_builder_test_fixture.h"
 #include "mongo/db/query/shard_filterer_factory_mock.h"
@@ -74,7 +76,8 @@ protected:
      * Makes a new QuerySolutionNode consisting of a ShardingFilterNode and a child VirtualScanNode.
      */
     std::unique_ptr<QuerySolutionNode> makeFilterVirtualScanTree(std::vector<BSONArray> docs) {
-        auto virtScan = std::make_unique<VirtualScanNode>(docs, false);
+        auto virtScan =
+            std::make_unique<VirtualScanNode>(docs, VirtualScanNode::ScanType::kCollScan, false);
         auto shardFilter = std::make_unique<ShardingFilterNode>();
         shardFilter->children.push_back(virtScan.release());
         return std::move(shardFilter);
@@ -90,7 +93,17 @@ protected:
         // Construct a QuerySolutionNode consisting of a ShardingFilterNode with a single child
         // VirtualScanNode.
         auto shardFilter = makeFilterVirtualScanTree(docs);
-        auto querySolution = makeQuerySolution(std::move(shardFilter));
+        runTest(std::move(shardFilter), expected, std::move(shardFiltererFactory));
+    }
+
+    /**
+     * Similar to the above, but rather than hardcoding a SHARDING_FILTER => VIRTUAL_SCAN query
+     * solution, uses the query solution node tree provided by the caller.
+     */
+    void runTest(std::unique_ptr<QuerySolutionNode> qsn,
+                 const BSONArray& expected,
+                 std::unique_ptr<ShardFiltererFactoryInterface> shardFiltererFactory) {
+        auto querySolution = makeQuerySolution(std::move(qsn));
 
         // Translate the QuerySolution to an sbe::PlanStage.
         auto [resultSlots, stage, data] =
@@ -218,4 +231,33 @@ TEST_F(SbeShardFilterTest, MissingFieldsAtBottomDottedPathFilledCorrectly) {
     auto expected = BSON_ARRAY(BSON("a" << BSON("b" << BSON("c" << BSON("d" << 1)))));
     runTest(docs, expected, makeAllNullShardKeyFiltererFactory(BSON("a.b.c.d" << 1)));
 }
+
+TEST_F(SbeShardFilterTest, CoveredShardFilterPlan) {
+    auto indexKeyPattern = BSON("a" << 1 << "b" << 1 << "c" << 1 << "d" << 1);
+    auto projection = BSON("a" << 1 << "c" << 1);
+    auto mockedIndexKeys =
+        std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 2 << "b" << 2 << "c" << 2 << "d" << 2)),
+                               BSON_ARRAY(BSON("a" << 3 << "b" << 3 << "c" << 3 << "d" << 3))};
+    auto expected = BSON_ARRAY(BSON("a" << 2 << "c" << 2) << BSON("a" << 3 << "c" << 3));
+
+    auto nss = NamespaceString{"db", "coll"};
+    auto expCtx = make_intrusive<ExpressionContextForTest>(nss);
+    auto emptyMatchExpression =
+        unittest::assertGet(MatchExpressionParser::parse(BSONObj{}, expCtx));
+    auto projectionAst = projection_ast::parse(expCtx, projection, ProjectionPolicies{});
+
+    // Construct a PROJECTION_COVERED => SHARDING_FILTER => VIRTUAL_SCAN query solution node tree
+    // where the virtual scan mocks an index scan with 'indexKeyPattern'.
+    auto virtScan = std::make_unique<VirtualScanNode>(
+        mockedIndexKeys, VirtualScanNode::ScanType::kIxscan, false, indexKeyPattern);
+    auto shardFilter = std::make_unique<ShardingFilterNode>();
+    shardFilter->children.push_back(virtScan.release());
+    auto projectNode = std::make_unique<ProjectionNodeCovered>(
+        std::move(shardFilter), *emptyMatchExpression, projectionAst, indexKeyPattern);
+
+    runTest(std::move(projectNode),
+            expected,
+            makeAlwaysPassShardFiltererFactory(BSON("a" << 1 << "c" << 1 << "d" << 1)));
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/query/sbe_stage_builder.cpp b/src/mongo/db/query/sbe_stage_builder.cpp
index 1fdf6fb72fb..8b358546f65 100644
--- a/src/mongo/db/query/sbe_stage_builder.cpp
+++ b/src/mongo/db/query/sbe_stage_builder.cpp
@@ -138,6 +138,35 @@ sbe::LockAcquisitionCallback makeLockAcquisitionCallback(bool checkNodeCanServeR
     };
 }
 
+/**
+ * Given an index key pattern, and a subset of the fields of the index key pattern that are depended
+ * on to compute the query, returns the corresponding 'IndexKeysInclusionSet' bit vector and field
+ * name vector.
+ *
+ * For example, suppose that we have an index key pattern {d: 1, c: 1, b: 1, a: 1}, and the caller
+ * depends on the set of 'requiredFields' {"b", "d"}. In this case, the pair of return values would
+ * be:
+ *  - 'IndexKeysInclusionSet' bit vector of 1010
+ *  - Field name vector of <"d", "b">
+ */
+template <typename T>
+std::pair<sbe::IndexKeysInclusionSet, std::vector<std::string>> makeIndexKeyInclusionSet(
+    const BSONObj& indexKeyPattern, const T& requiredFields) {
+    sbe::IndexKeysInclusionSet indexKeyBitset;
+    std::vector<std::string> keyFieldNames;
+    size_t i = 0;
+    for (auto&& elt : indexKeyPattern) {
+        if (requiredFields.count(elt.fieldNameStringData())) {
+            indexKeyBitset.set(i);
+            keyFieldNames.push_back(elt.fieldName());
+        }
+
+        ++i;
+    }
+
+    return {std::move(indexKeyBitset), std::move(keyFieldNames)};
+}
+
 }  // namespace
 
 SlotBasedStageBuilder::SlotBasedStageBuilder(OperationContext* opCtx,
@@ -229,8 +258,13 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
 
 std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder::buildVirtualScan(
     const QuerySolutionNode* root, const PlanStageReqs& reqs) {
+    using namespace std::literals;
     auto vsn = static_cast<const VirtualScanNode*>(root);
-    invariant(!reqs.getIndexKeyBitset());
+    // The caller should only have requested components of the index key if the virtual scan is
+    // mocking an index scan.
+    if (vsn->scanType == VirtualScanNode::ScanType::kCollScan) {
+        invariant(!reqs.getIndexKeyBitset());
+    }
 
     // Virtual scans cannot produce an oplogTsSlot, so assert that the caller doesn't need it.
     invariant(!reqs.has(kOplogTs));
@@ -245,22 +279,57 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     }
 
     inputGuard.reset();
-    auto [scanSlots, scanStage] =
+    auto [scanSlots, stage] =
         generateVirtualScanMulti(&_slotIdGenerator, vsn->hasRecordId ? 2 : 1, inputTag, inputVal);
 
-    PlanStageSlots outputs;
-
+    sbe::value::SlotId resultSlot;
     if (vsn->hasRecordId) {
         invariant(scanSlots.size() == 2);
-        outputs.set(kRecordId, scanSlots[0]);
-        outputs.set(kResult, scanSlots[1]);
+        resultSlot = scanSlots[1];
     } else {
         invariant(scanSlots.size() == 1);
-        invariant(!reqs.has(kRecordId));
-        outputs.set(kResult, scanSlots[0]);
+        resultSlot = scanSlots[0];
+    }
+
+    PlanStageSlots outputs;
+
+    if (reqs.has(kResult)) {
+        outputs.set(kResult, resultSlot);
+    } else if (reqs.getIndexKeyBitset()) {
+        // The caller wanted individual slots for certain components of a mock index scan. Use a
+        // project stage to produce those slots. Since the test will represent index keys as BSON
+        // objects, we use 'getField' expressions to extract the necessary fields.
+        invariant(!vsn->indexKeyPattern.isEmpty());
+
+        sbe::value::SlotVector indexKeySlots;
+        sbe::value::SlotMap<std::unique_ptr<sbe::EExpression>> projections;
+
+        size_t indexKeyPos = 0;
+        for (auto&& field : vsn->indexKeyPattern) {
+            if (reqs.getIndexKeyBitset()->test(indexKeyPos)) {
+                indexKeySlots.push_back(_slotIdGenerator.generate());
+                projections.emplace(
+                    indexKeySlots.back(),
+                    makeFunction("getField"sv,
+                                 sbe::makeE<sbe::EVariable>(resultSlot),
+                                 makeConstant(std::string_view{field.fieldName()})));
+            }
+            ++indexKeyPos;
+        }
+
+        stage =
+            sbe::makeS<sbe::ProjectStage>(std::move(stage), std::move(projections), root->nodeId());
+
+        outputs.setIndexKeySlots(indexKeySlots);
+    }
+
+    if (reqs.has(kRecordId)) {
+        invariant(vsn->hasRecordId);
+        invariant(scanSlots.size() == 2);
+        outputs.set(kRecordId, scanSlots[0]);
     }
 
-    return {std::move(scanStage), std::move(outputs)};
+    return {std::move(stage), std::move(outputs)};
 }
 
 std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder::buildIndexScan(
@@ -693,14 +762,13 @@ SlotBasedStageBuilder::buildProjectionCovered(const QuerySolutionNode* root,
     auto pn = static_cast<const ProjectionNodeCovered*>(root);
     invariant(pn->proj.isSimple());
 
-    // For now, we only support ProjectionNodeCovered when its child is an IndexScanNode.
-    uassert(5037301,
-            str::stream() << "Can't build exec tree for node: " << root->toString(),
-            pn->children[0]->getType() == STAGE_IXSCAN);
+    tassert(5037301,
+            str::stream() << "Can't build covered projection for fetched sub-plan: "
+                          << root->toString(),
+            !pn->children[0]->fetched());
 
     // This is a ProjectionCoveredNode, so we will be pulling all the data we need from one index.
     // Prepare a bitset to indicate which parts of the index key we need for the projection.
-    std::vector<std::string> keyFieldNames;
     StringSet requiredFields = {pn->proj.getRequiredFields().begin(),
                                 pn->proj.getRequiredFields().end()};
 
@@ -715,17 +783,8 @@ SlotBasedStageBuilder::buildProjectionCovered(const QuerySolutionNode* root,
     // all of the fields that the projection needs.
     auto childReqs = reqs.copy().clear(kResult);
 
-    sbe::IndexKeysInclusionSet indexKeyBitset;
-    size_t i = 0;
-    for (auto&& elt : pn->coveredKeyObj) {
-        if (requiredFields.count(elt.fieldNameStringData())) {
-            indexKeyBitset.set(i);
-            keyFieldNames.push_back(elt.fieldName());
-        }
-
-        ++i;
-    }
-
+    auto [indexKeyBitset, keyFieldNames] =
+        makeIndexKeyInclusionSet(pn->coveredKeyObj, requiredFields);
     childReqs.getIndexKeyBitset() = std::move(indexKeyBitset);
 
     auto [inputStage, outputs] = build(pn->children[0], childReqs);
@@ -1127,19 +1186,89 @@ SlotBasedStageBuilder::makeUnionForTailableCollScan(const QuerySolutionNode* roo
     return {std::move(unionStage), std::move(outputs)};
 }
 
-std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder::buildShardFilter(
-    const QuerySolutionNode* root, const PlanStageReqs& reqs) {
-    using namespace std::literals;
+namespace {
+/**
+ * Given an SBE subtree 'childStage' which computes the shard key and puts it into the given
+ * 'shardKeySlot', augments the SBE plan to actually perform shard filtering. Namely, a FilterStage
+ * is added at the root of the tree whose filter expression uses 'shardFilterer' to determine
+ * whether the shard key value in 'shardKeySlot' belongs to an owned range or not.
+ */
+auto buildShardFilterGivenShardKeySlot(sbe::value::SlotId shardKeySlot,
+                                       std::unique_ptr<sbe::PlanStage> childStage,
+                                       std::unique_ptr<ShardFilterer> shardFilterer,
+                                       PlanNodeId nodeId) {
+    auto shardFilterFn =
+        makeFunction("shardFilter",
+                     makeConstant(sbe::value::TypeTags::shardFilterer,
+                                  sbe::value::bitcastFrom<ShardFilterer*>(shardFilterer.release())),
+                     sbe::makeE<sbe::EVariable>(shardKeySlot));
+
+    return sbe::makeS<sbe::FilterStage<false>>(
+        std::move(childStage), std::move(shardFilterFn), nodeId);
+}
+}  // namespace
 
-    const auto filterNode = static_cast<const ShardingFilterNode*>(root);
+std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots>
+SlotBasedStageBuilder::buildShardFilterCovered(const ShardingFilterNode* filterNode,
+                                               std::unique_ptr<ShardFilterer> shardFilterer,
+                                               BSONObj shardKeyPattern,
+                                               BSONObj indexKeyPattern,
+                                               const QuerySolutionNode* child,
+                                               PlanStageReqs childReqs) {
+    StringDataSet shardKeyFields;
+    for (auto&& shardKeyElt : shardKeyPattern) {
+        shardKeyFields.insert(shardKeyElt.fieldNameStringData());
+    }
 
-    uassert(5071201,
-            "STAGE_SHARD_FILTER is curently only supported in SBE for collection scan plans",
-            filterNode->children[0]->getType() == StageType::STAGE_COLLSCAN ||
-                filterNode->children[0]->getType() == StageType::STAGE_VIRTUAL_SCAN);
+    // Save the bit vector describing the fields from the index that our parent requires. The shard
+    // filtering process may require additional fields that are not needed by the parent (for
+    // example, if the parent is projecting field "a" but the shard key is {a: 1, b: 1}). We will
+    // need the parent's reqs later on so that we can hand the correct slot vector for these fields
+    // back to our parent.
+    auto parentIndexKeyReqs = childReqs.getIndexKeyBitset();
+
+    // Determine the set of fields from the index required to obtain the shard key and union those
+    // with the set of fields from the index required by the parent stage.
+    auto [shardKeyIndexReqs, projectFields] =
+        makeIndexKeyInclusionSet(indexKeyPattern, shardKeyFields);
+    childReqs.getIndexKeyBitset() =
+        parentIndexKeyReqs.value_or(sbe::IndexKeysInclusionSet{}) | shardKeyIndexReqs;
+
+    auto [stage, outputs] = build(child, childReqs);
+
+    invariant(outputs.getIndexKeySlots());
+    auto indexKeySlots = *outputs.getIndexKeySlots();
+
+    auto shardKeySlot = _slotIdGenerator.generate();
+
+    auto mkObjStage = sbe::makeS<sbe::MakeBsonObjStage>(std::move(stage),
+                                                        shardKeySlot,
+                                                        boost::none,
+                                                        boost::none,
+                                                        std::vector<std::string>{},
+                                                        std::move(projectFields),
+                                                        indexKeySlots,
+                                                        true,
+                                                        false,
+                                                        filterNode->nodeId());
+
+    auto filterStage = buildShardFilterGivenShardKeySlot(
+        shardKeySlot, std::move(mkObjStage), std::move(shardFilterer), filterNode->nodeId());
+
+    outputs.setIndexKeySlots(!parentIndexKeyReqs ? boost::none
+                                                 : boost::optional<sbe::value::SlotVector>{
+                                                       makeIndexKeyOutputSlotsMatchingParentReqs(
+                                                           indexKeyPattern,
+                                                           *parentIndexKeyReqs,
+                                                           *childReqs.getIndexKeyBitset(),
+                                                           indexKeySlots)});
+
+    return {std::move(filterStage), std::move(outputs)};
+}
 
-    auto childReqs = reqs.copy().set(kResult);
-    auto [stage, outputs] = build(filterNode->children[0], childReqs);
+std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder::buildShardFilter(
+    const QuerySolutionNode* root, const PlanStageReqs& reqs) {
+    const auto filterNode = static_cast<const ShardingFilterNode*>(root);
 
     // If we're sharded make sure that we don't return data that isn't owned by the shard. This
     // situation can occur when pending documents from in-progress migrations are inserted and when
@@ -1147,6 +1276,41 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     // the shard, we need to own a 'ShardFilterer', and extract the document's shard key as a
     // BSONObj.
     auto shardFilterer = _shardFiltererFactory->makeShardFilterer(_opCtx);
+    auto shardKeyPattern = shardFilterer->getKeyPattern().toBSON();
+
+    // Determine if our child is an index scan and extract it's key pattern, or empty BSONObj if our
+    // child is not an IXSCAN node.
+    BSONObj indexKeyPattern = [&]() {
+        auto childNode = filterNode->children[0];
+        switch (childNode->getType()) {
+            case StageType::STAGE_IXSCAN:
+                return static_cast<const IndexScanNode*>(childNode)->index.keyPattern;
+            case StageType::STAGE_VIRTUAL_SCAN:
+                return static_cast<const VirtualScanNode*>(childNode)->indexKeyPattern;
+            default:
+                return BSONObj{};
+        }
+    }();
+
+    // If we're not required to fill out the 'kResult' slot, then instead we can request a slot from
+    // the child for each of the fields which constitute the shard key. This allows us to avoid
+    // materializing an intermediate object for plans where shard filtering can be performed based
+    // on the contents of index keys.
+    //
+    // We only apply this optimization in the special case that the child QSN is an IXSCAN, since in
+    // this case we can request exactly the fields we need according to their position in the index
+    // key pattern.
+    auto childReqs = reqs.copy().setIf(kResult, indexKeyPattern.isEmpty());
+    if (!childReqs.has(kResult)) {
+        return buildShardFilterCovered(filterNode,
+                                       std::move(shardFilterer),
+                                       std::move(shardKeyPattern),
+                                       std::move(indexKeyPattern),
+                                       filterNode->children[0],
+                                       std::move(childReqs));
+    }
+
+    auto [stage, outputs] = build(filterNode->children[0], childReqs);
 
     // Build an expression to extract the shard key from the document based on the shard key
     // pattern. To do this, we iterate over the shard key pattern parts and build nested 'getField'
@@ -1156,8 +1320,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     std::vector<std::string> projectFields;
     std::unique_ptr<sbe::EExpression> bindShardKeyPart;
 
-    BSONObjIterator keyPatternIter(shardFilterer->getKeyPattern().toBSON());
-    while (auto keyPatternElem = keyPatternIter.next()) {
+    for (auto&& keyPatternElem : shardKeyPattern) {
         auto fieldRef = FieldRef{keyPatternElem.fieldNameStringData()};
         fieldSlots.push_back(_slotIdGenerator.generate());
         projectFields.push_back(fieldRef.dottedField().toString());
@@ -1172,7 +1335,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     auto shardKeySlot{_slotIdGenerator.generate()};
 
     // Build an object which will hold a flattened shard key from the projections above.
-    auto shardKeyObjStage = sbe::makeS<sbe::MakeObjStage>(
+    auto shardKeyObjStage = sbe::makeS<sbe::MakeBsonObjStage>(
         sbe::makeS<sbe::ProjectStage>(std::move(stage), std::move(projections), root->nodeId()),
         shardKeySlot,
         boost::none,
@@ -1204,18 +1367,10 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     auto finalShardKeyObjStage = makeProjectStage(
         std::move(shardKeyObjStage), root->nodeId(), finalShardKeySlot, std::move(arrayChecks));
 
-    // Build a 'FilterStage' to skip over documents that don't belong to the shard. Shard membership
-    // of the document is checked by invoking 'shardFilter' with the owned 'ShardFilterer' along
-    // with the shard key that sits in the 'finalShardKeySlot' of 'MakeObjStage'.
-    auto shardFilterFn = sbe::makeE<sbe::EFunction>(
-        "shardFilter"sv,
-        sbe::makeEs(sbe::makeE<sbe::EConstant>(
-                        sbe::value::TypeTags::shardFilterer,
-                        sbe::value::bitcastFrom<ShardFilterer*>(shardFilterer.release())),
-                    sbe::makeE<sbe::EVariable>(finalShardKeySlot)));
-
-    return {sbe::makeS<sbe::FilterStage<false>>(
-                std::move(finalShardKeyObjStage), std::move(shardFilterFn), root->nodeId()),
+    return {buildShardFilterGivenShardKeySlot(finalShardKeySlot,
+                                              std::move(finalShardKeyObjStage),
+                                              std::move(shardFilterer),
+                                              root->nodeId()),
             std::move(outputs)};
 }
 
diff --git a/src/mongo/db/query/sbe_stage_builder.h b/src/mongo/db/query/sbe_stage_builder.h
index 4a9d277ab00..25c59bc4a4f 100644
--- a/src/mongo/db/query/sbe_stage_builder.h
+++ b/src/mongo/db/query/sbe_stage_builder.h
@@ -328,6 +328,21 @@ private:
     std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> buildShardFilter(
         const QuerySolutionNode* root, const PlanStageReqs& reqs);
 
+    /**
+     * Constructs an optimized SBE plan for 'filterNode' in the case that the fields of the
+     * 'shardKeyPattern' are provided by 'childIxscan'. In this case, the SBE plan for the child
+     * index scan node will fill out slots for the necessary components of the index key. These
+     * slots can be read directly in order to determine the shard key that should be passed to the
+     * 'shardFilterer'.
+     */
+    std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> buildShardFilterCovered(
+        const ShardingFilterNode* filterNode,
+        std::unique_ptr<ShardFilterer> shardFilterer,
+        BSONObj shardKeyPattern,
+        BSONObj indexKeyPattern,
+        const QuerySolutionNode* child,
+        PlanStageReqs childReqs);
+
     sbe::value::SlotIdGenerator _slotIdGenerator;
     sbe::value::FrameIdGenerator _frameIdGenerator;
     sbe::value::SpoolIdGenerator _spoolIdGenerator;
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.cpp b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
index 9b4899df1d2..b103ec7dc5b 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
@@ -607,4 +607,30 @@ std::unique_ptr<FilterStateHelper> makeFilterStateHelper(bool trackIndex) {
     }
     return std::make_unique<BooleanStateHelper>();
 }
+
+sbe::value::SlotVector makeIndexKeyOutputSlotsMatchingParentReqs(
+    const BSONObj& indexKeyPattern,
+    sbe::IndexKeysInclusionSet parentIndexKeyReqs,
+    sbe::IndexKeysInclusionSet childIndexKeyReqs,
+    sbe::value::SlotVector childOutputSlots) {
+    tassert(5308000,
+            "'childIndexKeyReqs' had fewer bits set than 'parentIndexKeyReqs'",
+            parentIndexKeyReqs.count() <= childIndexKeyReqs.count());
+    sbe::value::SlotVector newIndexKeySlots;
+
+    size_t slotIdx = 0;
+    for (size_t indexFieldNumber = 0;
+         indexFieldNumber < static_cast<size_t>(indexKeyPattern.nFields());
+         ++indexFieldNumber) {
+        if (parentIndexKeyReqs.test(indexFieldNumber)) {
+            newIndexKeySlots.push_back(childOutputSlots[slotIdx]);
+        }
+
+        if (childIndexKeyReqs.test(indexFieldNumber)) {
+            ++slotIdx;
+        }
+    }
+
+    return newIndexKeySlots;
+}
 }  // namespace mongo::stage_builder
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.h b/src/mongo/db/query/sbe_stage_builder_helpers.h
index 3f95b9c4c59..117de1dfcae 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.h
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.h
@@ -687,4 +687,32 @@ EvalExprStagePair generateShortCircuitingLogicalOp(sbe::EPrimBinary::Op logicOp,
                                                    sbe::value::SlotIdGenerator* slotIdGenerator,
                                                    const FilterStateHelper& stateHelper);
 
+/**
+ * Imagine that we have some parent QuerySolutionNode X and child QSN Y which both participate in a
+ * covered plan. Stage X requests some slots to be constructed out of the index keys using
+ * 'parentIndexKeyReqs'. Stage Y requests it's own slots, and adds those to the set requested by X,
+ * resulting in 'childIndexKeyReqs'. Note the invariant that 'childIndexKeyReqs' is a superset of
+ * 'parentIndexKeyReqs'. Let's notate the number of slots requested by 'childIndexKeyReqs' as |Y|
+ * and the set of slots requested by 'parentIndexKeyReqs' as |X|.
+ *
+ * The underlying SBE plan is constructed, and returns a vector of |Y| slots. However, the parent
+ * stage expects a vector of just |X| slots. The purpose of this function is to calculate and return
+ * the appropriate subset of the slot vector so that the parent stage X receives its expected |X|
+ * slots.
+ *
+ * As a concrete example, let's say the QSN tree is X => Y => IXSCAN and the index key pattern is
+ * {a: 1, b: 1, c: 1, d: 1}. X requests "a" and "d" using the bit vector 1001. Y additionally
+ * requires "c" so it requests three slots with the bit vector 1011. As a result, Y receives a
+ * 3-element slot vector, <s1, s2, s3>. Here, s1 will contain the value of "a", s2 contains "c", and
+ * s3 contain s"d".
+ *
+ * Parent QSN X expects just a two element slot vector where the first slot is for "a" and the
+ * second is for "d". This function would therefore return the slot vector <s1, s3>.
+ */
+sbe::value::SlotVector makeIndexKeyOutputSlotsMatchingParentReqs(
+    const BSONObj& indexKeyPattern,
+    sbe::IndexKeysInclusionSet parentIndexKeyReqs,
+    sbe::IndexKeysInclusionSet childIndexKeyReqs,
+    sbe::value::SlotVector childOutputSlots);
+
 }  // namespace mongo::stage_builder
diff --git a/src/mongo/db/query/sbe_stage_builder_test.cpp b/src/mongo/db/query/sbe_stage_builder_test.cpp
index a35066c6a28..b6c7ad5acd7 100644
--- a/src/mongo/db/query/sbe_stage_builder_test.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_test.cpp
@@ -52,7 +52,8 @@ TEST_F(SbeStageBuilderTest, TestVirtualScan) {
 
     // Construct a QuerySolution consisting of a single VirtualScanNode to test if a stream of
     // documents can be produced.
-    auto virtScan = std::make_unique<VirtualScanNode>(docs, true);
+    auto virtScan =
+        std::make_unique<VirtualScanNode>(docs, VirtualScanNode::ScanType::kCollScan, true);
 
     // Make a QuerySolution from the root virtual scan node.
     auto querySolution = makeQuerySolution(std::move(virtScan));
@@ -87,7 +88,8 @@ TEST_F(SbeStageBuilderTest, TestLimitOneVirtualScan) {
 
     // Construct a QuerySolution consisting of a root limit node that takes ownership of a
     // VirtualScanNode.
-    auto virtScan = std::make_unique<VirtualScanNode>(docs, true);
+    auto virtScan =
+        std::make_unique<VirtualScanNode>(docs, VirtualScanNode::ScanType::kCollScan, true);
     auto limitNode = std::make_unique<LimitNode>();
     limitNode->children.push_back(virtScan.release());
     limitNode->limit = 1;
@@ -119,4 +121,99 @@ TEST_F(SbeStageBuilderTest, TestLimitOneVirtualScan) {
     }
     ASSERT_EQ(index, 1);
 }
+
+TEST_F(SbeStageBuilderTest, VirtualCollScanWithoutRecordId) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 2)),
+                                       BSON_ARRAY(BSON("a" << 2 << "b" << 2)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << 2))};
+
+    // Construct a QuerySolution consisting of a root limit node that takes ownership of a
+    // VirtualScanNode.
+    auto virtScan =
+        std::make_unique<VirtualScanNode>(docs, VirtualScanNode::ScanType::kCollScan, false);
+    auto querySolution = makeQuerySolution(std::move(virtScan));
+
+    // Translate the QuerySolution tree to an sbe::PlanStage.
+    auto shardFiltererInterface = makeAlwaysPassShardFiltererInterface();
+    auto [resultSlots, stage, data] =
+        buildPlanStage(std::move(querySolution), false, std::move(shardFiltererInterface));
+
+    // Prepare the sbe::PlanStage for execution.
+    auto resultAccessors = prepareTree(&data.ctx, stage.get(), resultSlots);
+    ASSERT_EQ(resultAccessors.size(), 1u);
+
+    int64_t index = 0;
+    for (auto st = stage->getNext(); st == sbe::PlanState::ADVANCED; st = stage->getNext()) {
+        // Assert that the document produced from the stage is what we expect.
+        auto [tagDoc, valDoc] = resultAccessors[0]->getViewOfValue();
+        ASSERT_TRUE(tagDoc == sbe::value::TypeTags::bsonObject);
+        auto bo = BSONObj(sbe::value::bitcastTo<const char*>(valDoc));
+
+        ASSERT_BSONOBJ_EQ(bo, BSON("a" << ++index << "b" << 2));
+    }
+    ASSERT_EQ(index, 3);
+}
+
+TEST_F(SbeStageBuilderTest, VirtualIndexScan) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(int64_t{0} << BSON("a" << 1 << "b" << 2)),
+                                       BSON_ARRAY(int64_t{1} << BSON("a" << 2 << "b" << 2)),
+                                       BSON_ARRAY(int64_t{2} << BSON("a" << 3 << "b" << 2))};
+
+    // Construct a QuerySolution consisting of a single VirtualScanNode to test if a stream of
+    // documents can be produced.
+    auto virtScan = std::make_unique<VirtualScanNode>(
+        docs, VirtualScanNode::ScanType::kIxscan, true, BSON("a" << 1 << "b" << 1));
+    auto querySolution = makeQuerySolution(std::move(virtScan));
+
+    // Translate the QuerySolution tree to an sbe::PlanStage.
+    auto shardFiltererInterface = makeAlwaysPassShardFiltererInterface();
+    auto [resultSlots, stage, data] =
+        buildPlanStage(std::move(querySolution), true, std::move(shardFiltererInterface));
+    auto resultAccessors = prepareTree(&data.ctx, stage.get(), resultSlots);
+    ASSERT_EQ(resultAccessors.size(), 2u);
+
+    int64_t index = 0;
+    for (auto st = stage->getNext(); st == sbe::PlanState::ADVANCED; st = stage->getNext()) {
+        // Assert that the recordIDs are what we expect.
+        auto [tag, val] = resultAccessors[0]->getViewOfValue();
+        ASSERT_TRUE(tag == sbe::value::TypeTags::NumberInt64);
+        ASSERT_EQ(index, sbe::value::bitcastTo<int64_t>(val));
+
+        // Assert that the document produced from the stage is what we expect.
+        auto [tagDoc, valDoc] = resultAccessors[1]->getViewOfValue();
+        ASSERT_TRUE(tagDoc == sbe::value::TypeTags::bsonObject);
+        auto bo = BSONObj(sbe::value::bitcastTo<const char*>(valDoc));
+        ASSERT_BSONOBJ_EQ(bo, BSON("a" << ++index << "b" << 2));
+    }
+    ASSERT_EQ(index, 3);
+}
+
+TEST_F(SbeStageBuilderTest, VirtualIndexScanWithoutRecordId) {
+    auto docs = std::vector<BSONArray>{BSON_ARRAY(BSON("a" << 1 << "b" << 2)),
+                                       BSON_ARRAY(BSON("a" << 2 << "b" << 2)),
+                                       BSON_ARRAY(BSON("a" << 3 << "b" << 2))};
+
+    // Construct a QuerySolution consisting of a single VirtualScanNode to test if a stream of
+    // documents can be produced.
+    auto virtScan = std::make_unique<VirtualScanNode>(
+        docs, VirtualScanNode::ScanType::kIxscan, false, BSON("a" << 1 << "b" << 1));
+    auto querySolution = makeQuerySolution(std::move(virtScan));
+
+    // Translate the QuerySolution tree to an sbe::PlanStage.
+    auto shardFiltererInterface = makeAlwaysPassShardFiltererInterface();
+    auto [resultSlots, stage, data] =
+        buildPlanStage(std::move(querySolution), false, std::move(shardFiltererInterface));
+    auto resultAccessors = prepareTree(&data.ctx, stage.get(), resultSlots);
+    ASSERT_EQ(resultAccessors.size(), 1u);
+
+    int64_t index = 0;
+    for (auto st = stage->getNext(); st == sbe::PlanState::ADVANCED; st = stage->getNext()) {
+        // Assert that the document produced from the stage is what we expect.
+        auto [tagDoc, valDoc] = resultAccessors[0]->getViewOfValue();
+        ASSERT_TRUE(tagDoc == sbe::value::TypeTags::bsonObject);
+        auto bo = BSONObj(sbe::value::bitcastTo<const char*>(valDoc));
+        ASSERT_BSONOBJ_EQ(bo, BSON("a" << ++index << "b" << 2));
+    }
+    ASSERT_EQ(index, 3);
+}
 }  // namespace mongo