SERVER-74521 Support clustered collections in SBE

15 files changed, 222 insertions, 339 deletions

diff --git a/src/mongo/db/query/plan_executor_sbe.cpp b/src/mongo/db/query/plan_executor_sbe.cpp
index 5f9c347f463..9d3f4595b48 100644
--- a/src/mongo/db/query/plan_executor_sbe.cpp
+++ b/src/mongo/db/query/plan_executor_sbe.cpp
@@ -81,13 +81,16 @@ PlanExecutorSBE::PlanExecutorSBE(OperationContext* opCtx,
         uassert(4822866, "Query does not have recordId slot.", _resultRecordId);
     }
 
+    sbe::RuntimeEnvironment* env = _rootData.env;
     if (_rootData.shouldTrackLatestOplogTimestamp) {
-        _oplogTs = _rootData.env->getAccessor(_rootData.env->getSlot("oplogTs"_sd));
+        _oplogTs = env->getAccessor(env->getSlot("oplogTs"_sd));
     }
 
     if (_rootData.shouldUseTailableScan) {
-        _resumeRecordIdSlot = _rootData.env->getSlot("resumeRecordId"_sd);
+        _resumeRecordIdSlot = env->getSlot("resumeRecordId"_sd);
     }
+    _minRecordIdSlot = env->getSlotIfExists("minRecordId"_sd);
+    _maxRecordIdSlot = env->getSlotIfExists("maxRecordId"_sd);
 
     if (!_stash.empty()) {
         // The PlanExecutor keeps an extra reference to the last object pulled out of the PlanStage
diff --git a/src/mongo/db/query/plan_executor_sbe.h b/src/mongo/db/query/plan_executor_sbe.h
index 8d21efc6aec..163c4c9a426 100644
--- a/src/mongo/db/query/plan_executor_sbe.h
+++ b/src/mongo/db/query/plan_executor_sbe.h
@@ -185,8 +185,16 @@ private:
     sbe::value::Value _valLastRecordId{0};
     sbe::RuntimeEnvironment::Accessor* _oplogTs{nullptr};
 
+    // Only for a resumed scan ("seek"). Slot holding the TypeTags::RecordId of the record to resume
+    // the scan from. '_seekRecordId' is the RecordId value, initialized from the slot at runtime.
     boost::optional<sbe::value::SlotId> _resumeRecordIdSlot;
 
+    // Only for clustered collection scans, holds the minimum record ID of the scan, if applicable.
+    boost::optional<sbe::value::SlotId> _minRecordIdSlot;
+
+    // Only for clustered collection scans, holds the maximum record ID of the scan, if applicable.
+    boost::optional<sbe::value::SlotId> _maxRecordIdSlot;
+
     // NOTE: '_stash' stores documents as BSON. Currently, one of the '_stash' is usages is to store
     // documents received from the plan during multiplanning. This means that the documents
     // generated during multiplanning cannot exceed maximum BSON size. $group and $lookup CAN
diff --git a/src/mongo/db/query/plan_explainer_sbe.cpp b/src/mongo/db/query/plan_explainer_sbe.cpp
index 18f42a417fa..2d5371804a2 100644
--- a/src/mongo/db/query/plan_explainer_sbe.cpp
+++ b/src/mongo/db/query/plan_explainer_sbe.cpp
@@ -56,7 +56,13 @@ void statsToBSON(const QuerySolutionNode* node,
         return;
     }
 
-    bob->append("stage", stageTypeToString(node->getType()));
+    StageType nodeType = node->getType();
+    if ((nodeType == STAGE_COLLSCAN) &&
+        static_cast<const CollectionScanNode*>(node)->doSbeClusteredCollectionScan()) {
+        bob->append("stage", sbeClusteredCollectionScanToString());
+    } else {
+        bob->append("stage", stageTypeToString(node->getType()));
+    }
     bob->appendNumber("planNodeId", static_cast<long long>(node->nodeId()));
 
     // Display the BSON representation of the filter, if there is one.
diff --git a/src/mongo/db/query/planner_access.cpp b/src/mongo/db/query/planner_access.cpp
index f086caf4173..2d748cb9ed1 100644
--- a/src/mongo/db/query/planner_access.cpp
+++ b/src/mongo/db/query/planner_access.cpp
@@ -446,9 +446,15 @@ void deprioritizeUnboundedIndexScan(IndexScanNode* solnRoot,
 
 std::unique_ptr<QuerySolutionNode> QueryPlannerAccess::makeCollectionScan(
     const CanonicalQuery& query, bool tailable, const QueryPlannerParams& params, int direction) {
+
+    // The following are expensive to look up, so only do it once for each.
+    const mongo::NamespaceString nss = query.nss();
+    const bool isOplog = nss.isOplog();
+    const bool isChangeCollection = nss.isChangeCollection();
+
     // Make the (only) node, a collection scan.
     auto csn = std::make_unique<CollectionScanNode>();
-    csn->name = query.ns().toString();
+    csn->name = nss.ns().toString();
     csn->filter = query.root()->clone();
     csn->tailable = tailable;
     csn->shouldTrackLatestOplogTimestamp =
@@ -456,6 +462,8 @@ std::unique_ptr<QuerySolutionNode> QueryPlannerAccess::makeCollectionScan(
     csn->shouldWaitForOplogVisibility =
         params.options & QueryPlannerParams::OPLOG_SCAN_WAIT_FOR_VISIBLE;
     csn->direction = direction;
+    csn->isOplog = isOplog;
+    csn->isClustered = params.clusteredInfo ? true : false;
 
     if (params.clusteredInfo) {
         csn->clusteredIndex = params.clusteredInfo->getIndexSpec();
@@ -476,8 +484,8 @@ std::unique_ptr<QuerySolutionNode> QueryPlannerAccess::makeCollectionScan(
     // the collection scan to return timestamp-based tokens. Otherwise, we should
     // return generic RecordId-based tokens.
     if (query.getFindCommandRequest().getRequestResumeToken()) {
-        csn->shouldTrackLatestOplogTimestamp = query.nss().isOplogOrChangeCollection();
-        csn->requestResumeToken = !query.nss().isOplogOrChangeCollection();
+        csn->shouldTrackLatestOplogTimestamp = (isOplog || isChangeCollection);
+        csn->requestResumeToken = !csn->shouldTrackLatestOplogTimestamp;
     }
 
     // Extract and assign the RecordId from the 'resumeAfter' token, if present.
@@ -489,13 +497,13 @@ std::unique_ptr<QuerySolutionNode> QueryPlannerAccess::makeCollectionScan(
 
     const bool assertMinTsHasNotFallenOffOplog =
         params.options & QueryPlannerParams::ASSERT_MIN_TS_HAS_NOT_FALLEN_OFF_OPLOG;
-    if (query.nss().isOplogOrChangeCollection() && csn->direction == 1) {
+    if ((isOplog || isChangeCollection) && csn->direction == 1) {
         // Takes Timestamp 'ts' as input, transforms it to the RecordIdBound and assigns it to the
         // output parameter 'recordId'. The RecordId format for the change collection is a string,
         // where as the RecordId format for the oplog is a long integer. The timestamp should be
         // converted to the required format before assigning it to the 'recordId'.
         auto assignRecordIdFromTimestamp = [&](auto& ts, auto* recordId) {
-            auto keyFormat = query.nss().isChangeCollection() ? KeyFormat::String : KeyFormat::Long;
+            auto keyFormat = isChangeCollection ? KeyFormat::String : KeyFormat::Long;
             auto status = record_id_helpers::keyForOptime(ts, keyFormat);
             if (status.isOK()) {
                 *recordId = RecordIdBound(status.getValue());
@@ -539,7 +547,7 @@ std::unique_ptr<QuerySolutionNode> QueryPlannerAccess::makeCollectionScan(
     auto collCollator = params.clusteredCollectionCollator;
     csn->hasCompatibleCollation = CollatorInterface::collatorsMatch(queryCollator, collCollator);
 
-    if (params.clusteredInfo && !csn->resumeAfterRecordId) {
+    if (csn->isClustered && !csn->resumeAfterRecordId) {
         // This is a clustered collection. Attempt to perform an efficient, bounded collection scan
         // via minRecord and maxRecord if applicable. During this process, we will check if the
         // query is guaranteed to exclude values of the cluster key which are affected by collation.
diff --git a/src/mongo/db/query/query_solution.cpp b/src/mongo/db/query/query_solution.cpp
index 67bc634ffb7..b1b97b3a609 100644
--- a/src/mongo/db/query/query_solution.cpp
+++ b/src/mongo/db/query/query_solution.cpp
@@ -352,6 +352,8 @@ std::unique_ptr<QuerySolutionNode> CollectionScanNode::clone() const {
     copy->name = this->name;
     copy->tailable = this->tailable;
     copy->direction = this->direction;
+    copy->isClustered = this->isClustered;
+    copy->isOplog = this->isOplog;
     copy->shouldTrackLatestOplogTimestamp = this->shouldTrackLatestOplogTimestamp;
     copy->assertTsHasNotFallenOff = this->assertTsHasNotFallenOff;
     copy->shouldWaitForOplogVisibility = this->shouldWaitForOplogVisibility;
diff --git a/src/mongo/db/query/query_solution.h b/src/mongo/db/query/query_solution.h
index 8beba56a658..c1aab09c38f 100644
--- a/src/mongo/db/query/query_solution.h
+++ b/src/mongo/db/query/query_solution.h
@@ -148,7 +148,7 @@ struct QuerySolutionNode {
     std::string toString() const;
 
     /**
-     * What stage should this be transcribed to?  See stage_types.h.
+     * What stage should this node be transcribed to?  See stage_types.h.
      */
     virtual StageType getType() const = 0;
 
@@ -453,6 +453,11 @@ struct CollectionScanNode : public QuerySolutionNodeWithSortSet {
         return false;
     }
 
+    // Tells whether this scan will be performed as a clustered collection scan in SBE.
+    bool doSbeClusteredCollectionScan() const {
+        return (isClustered && !isOplog && (minRecord || maxRecord || resumeAfterRecordId));
+    }
+
     std::unique_ptr<QuerySolutionNode> clone() const final;
 
     // Name of the namespace.
@@ -492,8 +497,15 @@ struct CollectionScanNode : public QuerySolutionNodeWithSortSet {
     // Assert that the specified timestamp has not fallen off the oplog or change collection.
     boost::optional<Timestamp> assertTsHasNotFallenOff = boost::none;
 
+    // Scan direction: 1 means forward; -1 means reverse.
     int direction{1};
 
+    // Tells whether the collection is clustered (which includes oplog collections).
+    bool isClustered = false;
+
+    // Tells whether the collection is an oplog.
+    bool isOplog = false;
+
     // By default, includes the minRecord and maxRecord when present.
     CollectionScanParams::ScanBoundInclusion boundInclusion =
         CollectionScanParams::ScanBoundInclusion::kIncludeBothStartAndEndRecords;
@@ -502,6 +514,7 @@ struct CollectionScanNode : public QuerySolutionNodeWithSortSet {
     bool shouldWaitForOplogVisibility = false;
 
     // Once the first matching document is found, assume that all documents after it must match.
+    // This is useful for oplog queries where we know we will see records ordered by the ts field.
     bool stopApplyingFilterAfterFirstMatch = false;
 
     // Whether the collection scan should have low storage admission priority.
@@ -823,7 +836,7 @@ struct ReturnKeyNode : public QuerySolutionNode {
                   std::vector<FieldPath> sortKeyMetaFields)
         : QuerySolutionNode(std::move(child)), sortKeyMetaFields(std::move(sortKeyMetaFields)) {}
 
-    StageType getType() const final {
+    virtual StageType getType() const {
         return STAGE_RETURN_KEY;
     }
 
@@ -913,7 +926,7 @@ public:
 struct ProjectionNodeDefault final : ProjectionNode {
     using ProjectionNode::ProjectionNode;
 
-    StageType getType() const final {
+    virtual StageType getType() const {
         return STAGE_PROJECTION_DEFAULT;
     }
 
@@ -935,7 +948,7 @@ struct ProjectionNodeCovered final : ProjectionNode {
         : ProjectionNode(std::move(child), fullExpression, std::move(proj)),
           coveredKeyObj(std::move(coveredKeyObj)) {}
 
-    StageType getType() const final {
+    virtual StageType getType() const {
         return STAGE_PROJECTION_COVERED;
     }
 
@@ -956,7 +969,7 @@ struct ProjectionNodeCovered final : ProjectionNode {
 struct ProjectionNodeSimple final : ProjectionNode {
     using ProjectionNode::ProjectionNode;
 
-    StageType getType() const final {
+    virtual StageType getType() const {
         return STAGE_PROJECTION_SIMPLE;
     }
 
@@ -968,7 +981,7 @@ struct ProjectionNodeSimple final : ProjectionNode {
 };
 
 struct SortKeyGeneratorNode : public QuerySolutionNode {
-    StageType getType() const final {
+    virtual StageType getType() const {
         return STAGE_SORT_KEY_GENERATOR;
     }
 
@@ -1043,7 +1056,7 @@ private:
  * Represents sort algorithm that can handle any kind of input data.
  */
 struct SortNodeDefault final : public SortNode {
-    virtual StageType getType() const override {
+    virtual StageType getType() const {
         return STAGE_SORT_DEFAULT;
     }
 
@@ -1332,7 +1345,7 @@ struct EofNode : public QuerySolutionNodeWithSortSet {
 struct TextOrNode : public OrNode {
     TextOrNode() {}
 
-    StageType getType() const override {
+    virtual StageType getType() const {
         return STAGE_TEXT_OR;
     }
 
@@ -1344,7 +1357,7 @@ struct TextMatchNode : public QuerySolutionNodeWithSortSet {
     TextMatchNode(IndexEntry index, std::unique_ptr<fts::FTSQuery> ftsQuery, bool wantTextScore)
         : index(std::move(index)), ftsQuery(std::move(ftsQuery)), wantTextScore(wantTextScore) {}
 
-    StageType getType() const override {
+    virtual StageType getType() const {
         return STAGE_TEXT_MATCH;
     }
 
@@ -1406,7 +1419,7 @@ struct GroupNode : public QuerySolutionNode {
         needsAnyMetadata = deps.getNeedsAnyMetadata();
     }
 
-    StageType getType() const override {
+    virtual StageType getType() const {
         return STAGE_GROUP;
     }
 
@@ -1505,7 +1518,7 @@ struct EqLookupNode : public QuerySolutionNode {
           idxEntry(std::move(idxEntry)),
           shouldProduceBson(shouldProduceBson) {}
 
-    StageType getType() const override {
+    virtual StageType getType() const {
         return STAGE_EQ_LOOKUP;
     }
 
@@ -1581,7 +1594,7 @@ struct SentinelNode : public QuerySolutionNode {
 
     SentinelNode() {}
 
-    StageType getType() const override {
+    virtual StageType getType() const {
         return STAGE_SENTINEL;
     }
 
diff --git a/src/mongo/db/query/query_utils.cpp b/src/mongo/db/query/query_utils.cpp
index 62a28f86290..c6f72e0afa3 100644
--- a/src/mongo/db/query/query_utils.cpp
+++ b/src/mongo/db/query/query_utils.cpp
@@ -75,13 +75,21 @@ bool isQuerySbeCompatible(const CollectionPtr* collection, const CanonicalQuery*
         return false;
     }
 
-    // Queries against the oplog, a change collection, a time-series collection, or a clustered
-    // collection are not supported. Also queries on the inner side of a $lookup are not considered
-    // for SBE.
+    // Queries against the oplog, a change collection, or a time-series collection are not
+    // supported. Also queries on the inner side of a $lookup are not considered for SBE.
     const auto& nss = cq->nss();
-    if (expCtx->inLookup || (*collection && collection->get()->isClustered()) || nss.isOplog() ||
-        nss.isChangeCollection() || nss.isTimeseriesBucketsCollection() ||
-        !cq->metadataDeps().none()) {
+    if (expCtx->inLookup || nss.isOplog() || nss.isChangeCollection() ||
+        nss.isTimeseriesBucketsCollection() || !cq->metadataDeps().none()) {
+        return false;
+    }
+
+    // TODO SERVER-75715: Remove this code block once SBE support for clustered collection scans is
+    // fully implemented.
+    // (Ignore FCV check): This is intentional because we always want to use this feature once the
+    // feature flag is enabled.
+    const bool sbeFull = feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCVUnsafe();
+    if (!sbeFull && (*collection && collection->get()->isClustered())) {
+        // Queries against a clustered collection are not currently supported by SBE.
         return false;
     }
 
diff --git a/src/mongo/db/query/sbe_multi_planner.cpp b/src/mongo/db/query/sbe_multi_planner.cpp
index 714fd4ae0d4..0a2d20e9d47 100644
--- a/src/mongo/db/query/sbe_multi_planner.cpp
+++ b/src/mongo/db/query/sbe_multi_planner.cpp
@@ -347,7 +347,7 @@ CandidatePlans MultiPlanner::finalizeExecutionPlans(
     }
 
     // Writes a cache entry for the winning plan to the plan cache if possible.
-    plan_cache_util::updatePlanCache(
+    plan_cache_util::updatePlanCacheFromCandidates(
         _opCtx, _collections, _cachingMode, _cq, std::move(decision), candidates);
 
     return {std::move(candidates), winnerIdx};
diff --git a/src/mongo/db/query/sbe_stage_builder_coll_scan.cpp b/src/mongo/db/query/sbe_stage_builder_coll_scan.cpp
index 442240c2ac4..509848baddc 100644
--- a/src/mongo/db/query/sbe_stage_builder_coll_scan.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_coll_scan.cpp
@@ -57,7 +57,7 @@ namespace {
 
 boost::optional<sbe::value::SlotId> registerOplogTs(sbe::RuntimeEnvironment* env,
                                                     sbe::value::SlotIdGenerator* slotIdGenerator) {
-    auto slotId = env->getSlotIfExists("oplogTs"_sd);
+    boost::optional<sbe::value::SlotId> slotId = env->getSlotIfExists("oplogTs"_sd);
     if (!slotId) {
         return env->registerSlot(
             "oplogTs"_sd, sbe::value::TypeTags::Nothing, 0, false, slotIdGenerator);
@@ -71,11 +71,11 @@ boost::optional<sbe::value::SlotId> registerOplogTs(sbe::RuntimeEnvironment* env
  * standalone value of the same SlotId (the latter is returned purely for convenience purposes).
  */
 std::tuple<std::vector<std::string>, sbe::value::SlotVector, boost::optional<sbe::value::SlotId>>
-makeOplogTimestampSlotsIfNeeded(sbe::RuntimeEnvironment* env,
-                                sbe::value::SlotIdGenerator* slotIdGenerator,
-                                bool shouldTrackLatestOplogTimestamp) {
+makeOplogTimestampSlotIfNeeded(sbe::RuntimeEnvironment* env,
+                               sbe::value::SlotIdGenerator* slotIdGenerator,
+                               bool shouldTrackLatestOplogTimestamp) {
     if (shouldTrackLatestOplogTimestamp) {
-        auto slotId = registerOplogTs(env, slotIdGenerator);
+        boost::optional<sbe::value::SlotId> slotId = registerOplogTs(env, slotIdGenerator);
         return {{repl::OpTime::kTimestampFieldName.toString()}, sbe::makeSV(*slotId), slotId};
     }
     return {};
@@ -172,6 +172,8 @@ std::unique_ptr<sbe::PlanStage> buildResumeFromRecordIdSubtree(
                                                                   std::vector<std::string>{},
                                                                   sbe::makeSV(),
                                                                   seekSlot,
+                                                                  boost::none /* minRecordIdSlot */,
+                                                                  boost::none /* maxRecordIdSlot */,
                                                                   forward,
                                                                   yieldPolicy,
                                                                   csn->nodeId(),
@@ -211,7 +213,7 @@ std::unique_ptr<sbe::PlanStage> buildResumeFromRecordIdSubtree(
         csn->nodeId());
 
     // Construct the final loop join. Note that for the resume branch of a tailable cursor case we
-    // use the 'seek' stage as an inner branch, since we need to produce all records starting  from
+    // use the 'seek' stage as an inner branch, since we need to produce all records starting from
     // the supplied position. For a resume token case we also inject a 'skip 1' stage on top of the
     // inner branch, as we need to start _after_ the resume RecordId. In both cases we inject a
     // 'limit 1' stage on top of the outer branch, as it should produce just a single seek recordId.
@@ -228,81 +230,82 @@ std::unique_ptr<sbe::PlanStage> buildResumeFromRecordIdSubtree(
 }
 
 /**
- * Creates a collection scan sub-tree optimized for oplog scans. We can built an optimized scan
- * when any of the following scenarios apply:
+ * Creates a collection scan sub-tree optimized for clustered collection scans. Should only be
+ * called on clustered collections. We can build an optimized scan when any of the following
+ * scenarios apply:
  *
- * 1. There is a predicate on the 'ts' field of the oplog collection.
- *    1.1 If a lower bound on 'ts' is present, the collection scan will seek directly to the
- *        RecordId of an oplog entry as close to this lower bound as possible without going higher.
- *    1.2 If the query is *only* a lower bound on 'ts' on a forward scan, every document in the
- *        collection after the first matching one must also match. To avoid wasting time running the
- *        filter on every document to be returned, we will stop applying the filter once it finds
- *        the first match.
- *    1.3 If an upper bound on 'ts' is present, the collection scan will stop and return EOF the
- *        first time it fetches a document that does not pass the filter and has 'ts' greater than
- *        the upper bound.
- * 2. The user request specified a $_resumeAfter recordId from which to begin the scan.
+ * 1. 'csn->minRecord' and/or 'csn->maxRecord' exist.
+ *    1.1 CollectionScanParams::FORWARD scan:
+ *        a. If 'csn->minRecord' is present, the collection scan will seek directly to the RecordId
+ *           of a record as close to this lower bound as possible without going higher.
+ *        b. If 'csn->maxRecord' is present, the collection scan will stop and return EOF the first
+ *           time it fetches a document greater than this upper bound.
+ *    1.2 CollectionScanParams::BACKWARD scan:
+ *        a. If 'csn->maxRecord' is present, the collection scan will seek directly to the RecordId
+ *           of a record as close to this upper bound as possible without going lower.
+ *        b. If 'csn->minRecord' is present, the collection scan will stop and return EOF the first
+ *           time it fetches a document less than this lower bound.
+ * 2. The user request specified a $_resumeAfter RecordId from which to begin the scan AND the scan
+ *    is forward AND neither 'csn->minRecord' nor 'csn->maxRecord' exist.
+ *    2a. The scan will continue with the next RecordId after $_resumeAfter.
  */
-std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateOptimizedOplogScan(
+std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateClusteredCollScan(
     StageBuilderState& state,
     const CollectionPtr& collection,
     const CollectionScanNode* csn,
-    std::vector<std::string> fields,
+    std::vector<std::string> scanFieldNames,
     PlanYieldPolicy* yieldPolicy,
     bool isTailableResumeBranch) {
-    invariant(collection->ns().isOplog());
-    // We can apply oplog scan optimizations only when at least one of the following was specified.
-    invariant(csn->resumeAfterRecordId || csn->minRecord || csn->maxRecord);
+    const bool forward = csn->direction == CollectionScanParams::FORWARD;
+    sbe::RuntimeEnvironment* env = state.data->env;
+
+    invariant(csn->doSbeClusteredCollectionScan());
+    invariant(!csn->resumeAfterRecordId || forward);
+    invariant(!csn->resumeAfterRecordId || !csn->tailable);
     // The minRecord and maxRecord optimizations are not compatible with resumeAfterRecordId.
     invariant(!(csn->resumeAfterRecordId && (csn->minRecord || csn->maxRecord)));
-    // Oplog scan optimizations can only be done for a forward scan.
-    invariant(csn->direction == CollectionScanParams::FORWARD);
+    // 'stopApplyingFilterAfterFirstMatch' is only for oplog scans; this method doesn't do them.
+    invariant(!csn->stopApplyingFilterAfterFirstMatch);
 
-    auto fieldSlots = state.slotIdGenerator->generateMultiple(fields.size());
+    auto scanFieldSlots = state.slotIdGenerator->generateMultiple(scanFieldNames.size());
 
-    auto resultSlot = state.slotId();
-    auto recordIdSlot = state.slotId();
+    sbe::value::SlotId resultSlot = state.slotId();
+    sbe::value::SlotId recordIdSlot = state.slotId();
 
-    // Start the scan from the RecordId stored in seekRecordId.
-    // Otherwise, if we're building a collection scan for a resume branch of a special union
-    // sub-tree implementing a tailable cursor scan, we can use the seekRecordIdSlot directly
-    // to access the recordId to resume the scan from.
+    // Iff this is a resume or fetch, set 'seekRecordIdSlot' and 'seekRecordIdExpression' to the
+    // RecordId resume point of the scan. If we're building a collection scan for a resume branch of
+    // a special union sub-tree implementing a tailable cursor scan, we can use the already existing
+    // 'resumeRecordIdSlot' directly as the 'seekRecordIdSlot' to access the recordId to resume the
+    // scan from. Otherwise we must create a slot for it.
     auto [seekRecordIdSlot, seekRecordIdExpression] =
         [&]() -> std::pair<boost::optional<sbe::value::SlotId>, std::unique_ptr<sbe::EExpression>> {
         if (isTailableResumeBranch) {
-            auto resumeRecordIdSlot = state.data->env->getSlot("resumeRecordId"_sd);
+            sbe::value::SlotId resumeRecordIdSlot = env->getSlot("resumeRecordId"_sd);
             return {resumeRecordIdSlot, makeVariable(resumeRecordIdSlot)};
         } else if (csn->resumeAfterRecordId) {
             auto [tag, val] = sbe::value::makeCopyRecordId(*csn->resumeAfterRecordId);
             return {state.slotId(), makeConstant(tag, val)};
-        } else if (csn->minRecord) {
-            auto cursor = collection->getRecordStore()->getCursor(state.opCtx);
-            auto startRec = cursor->seekNear(csn->minRecord->recordId());
-            if (startRec) {
-                LOGV2_DEBUG(205841, 3, "Using direct oplog seek");
-                auto [tag, val] = sbe::value::makeCopyRecordId(startRec->id);
-                return {state.slotId(), makeConstant(tag, val)};
-            }
         }
         return {};
-    }();
-
-    // Check if we need to project out an oplog 'ts' field as part of the collection scan. We will
-    // need it either when 'maxRecord' bound has been provided, so that we can apply an EOF filter,
-    // of if we need to track the latest oplog timestamp.
-    const auto shouldTrackLatestOplogTimestamp =
-        (csn->maxRecord || csn->shouldTrackLatestOplogTimestamp);
-    auto&& [scanFields, scanFieldSlots, tsSlot] = makeOplogTimestampSlotsIfNeeded(
-        state.data->env, state.slotIdGenerator, shouldTrackLatestOplogTimestamp);
-
-    bool createScanWithAndWithoutFilter = (csn->filter && csn->stopApplyingFilterAfterFirstMatch);
-
-    if (!createScanWithAndWithoutFilter) {
-        scanFields.insert(scanFields.end(), fields.begin(), fields.end());
-        scanFieldSlots.insert(scanFieldSlots.end(), fieldSlots.begin(), fieldSlots.end());
+    }();  // lambda end and call
+
+    // Create minRecordId and/or maxRecordId slots as needed.
+    boost::optional<sbe::value::SlotId> minRecordSlot;
+    boost::optional<sbe::value::SlotId> maxRecordSlot;
+    if (csn->minRecord) {
+        auto [tag, val] = sbe::value::makeCopyRecordId(csn->minRecord->recordId());
+        minRecordSlot = env->registerSlot("minRecordId"_sd, tag, val, true, state.slotIdGenerator);
+    }
+    if (csn->maxRecord) {
+        auto [tag, val] = sbe::value::makeCopyRecordId(csn->maxRecord->recordId());
+        maxRecordSlot = env->registerSlot("maxRecordId"_sd, tag, val, true, state.slotIdGenerator);
     }
 
-    sbe::ScanCallbacks callbacks({}, {}, makeOpenCallbackIfNeeded(collection, csn));
+    // Create the ScanStage.
+    bool excludeScanEndRecordId =
+        (csn->boundInclusion ==
+             CollectionScanParams::ScanBoundInclusion::kExcludeBothStartAndEndRecords ||
+         csn->boundInclusion == CollectionScanParams::ScanBoundInclusion::kIncludeStartRecordOnly);
     auto stage = sbe::makeS<sbe::ScanStage>(collection->uuid(),
                                             resultSlot,
                                             recordIdSlot,
@@ -310,16 +313,22 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateOptimizedOplo
                                             boost::none /* indexIdentSlot */,
                                             boost::none /* indexKeySlot */,
                                             boost::none /* keyPatternSlot */,
-                                            tsSlot,
-                                            std::move(scanFields),
-                                            std::move(scanFieldSlots),
+                                            boost::none /* oplogTsSlot */,
+                                            scanFieldNames,  // do not std::move - used later
+                                            scanFieldSlots,  // do not std::move - used later
                                             seekRecordIdSlot,
-                                            true /* forward */,
+                                            minRecordSlot,
+                                            maxRecordSlot,
+                                            forward,
                                             yieldPolicy,
                                             csn->nodeId(),
-                                            std::move(callbacks));
+                                            sbe::ScanCallbacks{},
+                                            false /* lowPriority default */,
+                                            false /* useRandomCursor default */,
+                                            true /* participateInTrialRunTracking default */,
+                                            excludeScanEndRecordId);
 
-    // Start the scan from the seekRecordId.
+    // Iff this is a resume or fetch, build the subtree to start the scan from the seekRecordId.
     if (seekRecordIdSlot) {
         stage = buildResumeFromRecordIdSubtree(state,
                                                collection,
@@ -332,230 +341,34 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateOptimizedOplo
                                                csn->resumeAfterRecordId.has_value());
     }
 
-    // Create a filter which checks the first document to ensure either that its 'ts' is less than
-    // or equal the minimum timestamp that should not have rolled off the oplog, or that it is a
-    // replica set initialization message. If this fails, then we throw
-    // ErrorCodes::OplogQueryMinTsMissing. We avoid doing this check on the resumable branch of a
-    // tailable scan; it only needs to be done once, when the initial branch is run.
-    if (csn->assertTsHasNotFallenOff && !isTailableResumeBranch) {
-        invariant(csn->shouldTrackLatestOplogTimestamp);
-
-        // There should always be a 'tsSlot' already allocated on the RuntimeEnvironment for the
-        // existing scan that we created previously.
-        invariant(tsSlot);
-
-        // We will be constructing a filter that needs to see the 'ts' field. We name it 'minTsSlot'
-        // here so that it does not shadow the 'tsSlot' which we allocated earlier. Our filter will
-        // also need to see the 'op' and 'o.msg' fields.
-        auto opTypeSlot = state.slotId();
-        auto oObjSlot = state.slotId();
-        auto minTsSlot = state.slotId();
-        sbe::value::SlotVector minTsSlots = {minTsSlot, opTypeSlot, oObjSlot};
-        std::vector<std::string> minTsFields = {
-            repl::OpTime::kTimestampFieldName.toString(), "op", "o"};
-
-        // If the first entry we see in the oplog is the replset initialization, then it doesn't
-        // matter if its timestamp is later than the specified minTs; no events earlier than the
-        // minTs can have fallen off this oplog. Otherwise, we must verify that the timestamp of the
-        // first observed oplog entry is earlier than or equal to the minTs time.
-        //
-        // To achieve this, we build a two-branch union subtree. The left branch is a scan with a
-        // filter that checks the first entry in the oplog for the above criteria, throws via EFail
-        // if they are not met, and EOFs otherwise. The right branch of the union plan is the tree
-        // that we originally built above.
-        //
-        // union [s9, s10, s11] [
-        //     [s6, s7, s8] efilter {if (ts <= minTs || op == "n" && isObject (o) &&
-        //                      getField (o, "msg") == "initiating set", false, fail ( 326 ))}
-        //     scan [s6 = ts, s7 = op, s8 = o] @oplog,
-        //     <stage>
-
-        // Set up the filter stage to be used in the left branch of the union. If the main body of
-        // the expression does not match the input document, it throws OplogQueryMinTsMissing. If
-        // the expression does match, then it returns 'false', which causes the filter (and as a
-        // result, the branch) to EOF immediately. Note that the resultSlot and recordIdSlot
-        // arguments to the ScanStage are boost::none, as we do not need them.
-        sbe::ScanCallbacks branchCallbacks{};
-        auto minTsBranch = sbe::makeS<sbe::FilterStage<false, true>>(
-            sbe::makeS<sbe::ScanStage>(collection->uuid(),
-                                       boost::none /* resultSlot */,
-                                       boost::none /* recordIdSlot */,
-                                       boost::none /* snapshotIdSlot */,
-                                       boost::none /* identIdentSlot */,
-                                       boost::none /* indexKeySlot */,
-                                       boost::none /* keyPatternSlot */,
-                                       boost::none /* oplogTsSlot*/,
-                                       std::move(minTsFields),
-                                       minTsSlots, /* don't move this */
-                                       boost::none,
-                                       true /* forward */,
-                                       yieldPolicy,
-                                       csn->nodeId(),
-                                       branchCallbacks),
-            sbe::makeE<sbe::EIf>(
-                makeBinaryOp(
-                    sbe::EPrimBinary::logicOr,
-                    makeBinaryOp(sbe::EPrimBinary::lessEq,
-                                 makeVariable(minTsSlot),
-                                 makeConstant(sbe::value::TypeTags::Timestamp,
-                                              csn->assertTsHasNotFallenOff->asULL())),
-                    makeBinaryOp(
-                        sbe::EPrimBinary::logicAnd,
-                        makeBinaryOp(sbe::EPrimBinary::eq,
-                                     makeVariable(opTypeSlot),
-                                     makeConstant("n")),
-                        makeBinaryOp(sbe::EPrimBinary::logicAnd,
-                                     makeFunction("isObject", makeVariable(oObjSlot)),
-                                     makeBinaryOp(sbe::EPrimBinary::eq,
-                                                  makeFunction("getField",
-                                                               makeVariable(oObjSlot),
-                                                               makeConstant("msg")),
-                                                  makeConstant(repl::kInitiatingSetMsg))))),
-                makeConstant(sbe::value::TypeTags::Boolean, false),
-                sbe::makeE<sbe::EFail>(ErrorCodes::OplogQueryMinTsMissing,
-                                       "Specified minTs has already fallen off the oplog")),
-            csn->nodeId());
-
-        // All branches of the UnionStage must have the same number of input and output slots, and
-        // we want to remap all slots from the basic scan we constructed earlier through the union
-        // stage to the output. We're lucky that the real scan happens to have the same number of
-        // slots (resultSlot, recordSlot, tsSlot) as the minTs check branch (minTsSlot, opTypeSlot,
-        // oObjSlot), so we don't have to compensate with any unused slots. Note that the minTsSlots
-        // will never be mapped to output in practice, since the minTs branch either throws or EOFs.
-        //
-        // We also need to update the local variables for each slot to their remapped values, so
-        // subsequent subtrees constructed by this function refer to the correct post-union slots.
-        auto realSlots = sbe::makeSV(resultSlot, recordIdSlot, *tsSlot);
-        resultSlot = state.slotId();
-        recordIdSlot = state.slotId();
-        tsSlot = state.slotId();
-        auto outputSlots = sbe::makeSV(resultSlot, recordIdSlot, *tsSlot);
-
-        if (!createScanWithAndWithoutFilter) {
-            auto unusedFieldSlots = state.slotIdGenerator->generateMultiple(fieldSlots.size());
-            minTsSlots.insert(minTsSlots.end(), unusedFieldSlots.begin(), unusedFieldSlots.end());
-
-            realSlots.insert(realSlots.end(), fieldSlots.begin(), fieldSlots.end());
-
-            size_t numFieldSlots = fieldSlots.size();
-            fieldSlots = state.slotIdGenerator->generateMultiple(numFieldSlots);
-
-            outputSlots.insert(outputSlots.end(), fieldSlots.begin(), fieldSlots.end());
-        }
-
-        // Create the union stage. The left branch, which runs first, is our resumability check.
-        stage = sbe::makeS<sbe::UnionStage>(
-            sbe::makeSs(std::move(minTsBranch), std::move(stage)),
-            makeVector<sbe::value::SlotVector>(std::move(minTsSlots), std::move(realSlots)),
-            std::move(outputSlots),
-            csn->nodeId());
-    }
-
-    // Add an EOF filter to stop the scan after we fetch the first document that has 'ts' greater
-    // than the upper bound.
-    if (csn->maxRecord) {
-        // The 'maxRecord' optimization is not compatible with 'stopApplyingFilterAfterFirstMatch'.
-        invariant(!csn->stopApplyingFilterAfterFirstMatch);
-        invariant(tsSlot);
-
-        stage = sbe::makeS<sbe::FilterStage<false, true>>(
-            std::move(stage),
-            makeBinaryOp(sbe::EPrimBinary::lessEq,
-                         makeVariable(*tsSlot),
-                         makeConstant(sbe::value::TypeTags::Timestamp,
-                                      csn->maxRecord->recordId().getLong())),
-            csn->nodeId());
+    // When the start and/or end scan bounds are from an expression, ScanStage::getNext() treats
+    // them both as inclusive, and 'csn->filter' will enforce any exclusions. If the bound(s) came
+    // from the "min" (always inclusive) and/or "max" (always exclusive) keywords, there may be no
+    // filter, so ScanStage->getNext() must directly enforce the bounds. min's inclusivity matches
+    // getNext()'s default behavior, but max's exclusivity does not and thus is enforced by the
+    // excludeScanEndRecordId argument to the ScanStage constructor above.
+    EvalExpr filterExpr = generateFilter(state, csn->filter.get(), resultSlot, nullptr);
+    if (!filterExpr.isNull()) {
+        stage = sbe::makeS<sbe::FilterStage<false>>(
+            std::move(stage), filterExpr.extractExpr(state), csn->nodeId());
     }
 
-    // If csn->stopApplyingFilterAfterFirstMatch is true, assert that csn has a filter.
-    invariant(!csn->stopApplyingFilterAfterFirstMatch || csn->filter);
-
-    if (csn->filter) {
-        auto filterExpr = generateFilter(state, csn->filter.get(), resultSlot, nullptr);
-        if (!filterExpr.isNull()) {
-            stage = sbe::makeS<sbe::FilterStage<false>>(
-                std::move(stage), filterExpr.extractExpr(state), csn->nodeId());
-        }
-
-        // We may be requested to stop applying the filter after the first match. This can happen
-        // if the query is just a lower bound on 'ts' on a forward scan. In this case every document
-        // in the collection after the first matching one must also match, so there is no need to
-        // run the filter on such elements.
-        //
-        // To apply this optimization we will construct the following sub-tree:
-        //
-        //       nlj [] [seekRecordIdSlot]
-        //           left
-        //              limit 1
-        //              filter <predicate>
-        //              <stage>
-        //           right
-        //              seek seekRecordIdSlot resultSlot recordIdSlot @coll
-        //
-        // Here, the nested loop join outer branch is the collection scan we constructed above, with
-        // a csn->filter predicate sitting on top. The 'limit 1' stage is to ensure this branch
-        // returns a single row. Once executed, this branch will filter out documents which doesn't
-        // satisfy the predicate, and will return the first document, along with a RecordId, that
-        // matches. This RecordId is then used as a starting point of the collection scan in the
-        // inner branch, and the execution will continue from this point further on, without
-        // applying the filter.
-        if (createScanWithAndWithoutFilter) {
-            invariant(!csn->maxRecord);
-            invariant(csn->direction == CollectionScanParams::FORWARD);
-
-            seekRecordIdSlot = recordIdSlot;
-            resultSlot = state.slotId();
-            recordIdSlot = state.slotId();
-
-            std::tie(scanFields, scanFieldSlots, tsSlot) = makeOplogTimestampSlotsIfNeeded(
-                state.data->env, state.slotIdGenerator, shouldTrackLatestOplogTimestamp);
-
-            scanFields.insert(scanFields.end(), fields.begin(), fields.end());
-            scanFieldSlots.insert(scanFieldSlots.end(), fieldSlots.begin(), fieldSlots.end());
-
-            stage = sbe::makeS<sbe::LoopJoinStage>(
-                sbe::makeS<sbe::LimitSkipStage>(std::move(stage), 1, boost::none, csn->nodeId()),
-                sbe::makeS<sbe::ScanStage>(collection->uuid(),
-                                           resultSlot,
-                                           recordIdSlot,
-                                           boost::none /* snapshotIdSlot */,
-                                           boost::none /* indexIdentSlot */,
-                                           boost::none /* indexKeySlot */,
-                                           boost::none /* keyPatternSlot */,
-                                           tsSlot,
-                                           std::move(scanFields),
-                                           std::move(scanFieldSlots),
-                                           seekRecordIdSlot,
-                                           true /* forward */,
-                                           yieldPolicy,
-                                           csn->nodeId(),
-                                           sbe::ScanCallbacks{}),
-                sbe::makeSV(),
-                sbe::makeSV(*seekRecordIdSlot),
-                nullptr,
-                csn->nodeId());
-        }
-    }
-
-    // If csn->shouldTrackLatestOplogTimestamp is true, assert that we generated tsSlot.
-    invariant(!csn->shouldTrackLatestOplogTimestamp || tsSlot);
-
     PlanStageSlots outputs;
     outputs.set(PlanStageSlots::kResult, resultSlot);
     outputs.set(PlanStageSlots::kRecordId, recordIdSlot);
-    for (size_t i = 0; i < fields.size(); ++i) {
-        outputs.set(std::make_pair(PlanStageSlots::kField, fields[i]), fieldSlots[i]);
+    for (size_t i = 0; i < scanFieldNames.size(); ++i) {
+        outputs.set(std::make_pair(PlanStageSlots::kField, scanFieldNames[i]), scanFieldSlots[i]);
     }
 
     return {std::move(stage), std::move(outputs)};
-}
+}  // generateClusteredCollScan
 
 /**
  * Generates a generic collection scan sub-tree.
  *  - If a resume token has been provided, the scan will start from a RecordId contained within this
- * token.
+ *    token.
  *  - Else if 'isTailableResumeBranch' is true, the scan will start from a RecordId contained in
- * slot "resumeRecordId".
+ *    slot "resumeRecordId".
  *  - Otherwise the scan will start from the beginning of the collection.
  */
 std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollScan(
@@ -565,7 +378,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollSc
     std::vector<std::string> fields,
     PlanYieldPolicy* yieldPolicy,
     bool isTailableResumeBranch) {
-    const auto forward = csn->direction == CollectionScanParams::FORWARD;
+    const bool forward = csn->direction == CollectionScanParams::FORWARD;
 
     invariant(!csn->shouldTrackLatestOplogTimestamp || collection->ns().isOplog());
     invariant(!csn->resumeAfterRecordId || forward);
@@ -599,7 +412,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollSc
     }();
 
     // See if we need to project out an oplog latest timestamp.
-    auto&& [scanFields, scanFieldSlots, tsSlot] = makeOplogTimestampSlotsIfNeeded(
+    auto&& [scanFields, scanFieldSlots, oplogTsSlot] = makeOplogTimestampSlotIfNeeded(
         state.data->env, state.slotIdGenerator, csn->shouldTrackLatestOplogTimestamp);
 
     scanFields.insert(scanFields.end(), fields.begin(), fields.end());
@@ -613,10 +426,12 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollSc
                                             boost::none /* indexIdentSlot */,
                                             boost::none /* indexKeySlot */,
                                             boost::none /* keyPatternSlot */,
-                                            tsSlot,
+                                            oplogTsSlot,
                                             std::move(scanFields),
                                             std::move(scanFieldSlots),
                                             seekRecordIdSlot,
+                                            boost::none /* minRecordIdSlot */,
+                                            boost::none /* maxRecordIdSlot */,
                                             forward,
                                             yieldPolicy,
                                             csn->nodeId(),
@@ -643,9 +458,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollSc
     }
 
     if (csn->filter) {
-        // The 'stopApplyingFilterAfterFirstMatch' optimization is only applicable when the 'ts'
-        // lower bound is also provided for an oplog scan, and is handled in
-        // 'generateOptimizedOplogScan()'.
+        // 'stopApplyingFilterAfterFirstMatch' is only for oplog scans; this method doesn't do them.
         invariant(!csn->stopApplyingFilterAfterFirstMatch);
 
         auto filterExpr = generateFilter(state, csn->filter.get(), resultSlot, &outputs);
@@ -656,7 +469,8 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollSc
     }
 
     return {std::move(stage), std::move(outputs)};
-}
+}  // generateGenericCollScan
+
 }  // namespace
 
 std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateCollScan(
@@ -666,8 +480,9 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateCollScan(
     std::vector<std::string> fields,
     PlanYieldPolicy* yieldPolicy,
     bool isTailableResumeBranch) {
-    if (csn->minRecord || csn->maxRecord || csn->stopApplyingFilterAfterFirstMatch) {
-        return generateOptimizedOplogScan(
+
+    if (csn->doSbeClusteredCollectionScan()) {
+        return generateClusteredCollScan(
             state, collection, csn, std::move(fields), yieldPolicy, isTailableResumeBranch);
     } else {
         return generateGenericCollScan(
diff --git a/src/mongo/db/query/sbe_stage_builder_coll_scan.h b/src/mongo/db/query/sbe_stage_builder_coll_scan.h
index c0eb9a098e7..0d1776712e0 100644
--- a/src/mongo/db/query/sbe_stage_builder_coll_scan.h
+++ b/src/mongo/db/query/sbe_stage_builder_coll_scan.h
@@ -41,16 +41,16 @@ namespace mongo::stage_builder {
 class PlanStageSlots;
 
 /**
- * Generates an SBE plan stage sub-tree implementing an collection scan. 'fields' can be used to
+ * Generates an SBE plan stage sub-tree implementing a collection scan. 'fields' can be used to
  * specify top-level fields that should be retrieved during the scan. For each name in 'fields',
  * there will be a corresponding kField slot in the PlanStageSlots object returned with the same
  * name.
  *
  * On success, a tuple containing the following data is returned:
  *   * A slot to access a fetched document (a resultSlot)
- *   * A slot to access a recordId (a recordIdSlot)
- *   * An optional slot to access a latest oplog timestamp (oplogTsSlot), if we scan the oplog and
- *     were requested to track this data.
+ *   * A slot to access the doc's RecordId (a recordIdSlot)
+ *   * An optional slot to access the latest oplog timestamp ("ts" field) for oplog scans that were
+ *     requested to track this data or that are clustered scans ("ts" is the oplog clustering key).
  *   * A generated PlanStage sub-tree.
  *
  * In cases of an error, throws.
@@ -59,7 +59,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateCollScan(
     StageBuilderState& state,
     const CollectionPtr& collection,
     const CollectionScanNode* csn,
-    std::vector<std::string> fields,
+    std::vector<std::string> scanFieldNames,
     PlanYieldPolicy* yieldPolicy,
     bool isTailableResumeBranch);
 
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.cpp b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
index 66494c2c514..5f9040974ca 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
@@ -827,7 +827,7 @@ std::unique_ptr<sbe::PlanStage> makeLoopJoinForFetch(std::unique_ptr<sbe::PlanSt
                                                      sbe::value::SlotId recordIdSlot,
                                                      std::vector<std::string> fields,
                                                      sbe::value::SlotVector fieldSlots,
-                                                     sbe::value::SlotId seekKeySlot,
+                                                     sbe::value::SlotId seekRecordIdSlot,
                                                      sbe::value::SlotId snapshotIdSlot,
                                                      sbe::value::SlotId indexIdentSlot,
                                                      sbe::value::SlotId indexKeySlot,
@@ -842,7 +842,7 @@ std::unique_ptr<sbe::PlanStage> makeLoopJoinForFetch(std::unique_ptr<sbe::PlanSt
 
     sbe::ScanCallbacks callbacks(indexKeyCorruptionCheckCallback, indexKeyConsistencyCheckCallback);
 
-    // Scan the collection in the range [seekKeySlot, Inf).
+    // Scan the collection in the range [seekRecordIdSlot, Inf).
     auto scanStage = sbe::makeS<sbe::ScanStage>(collToFetch->uuid(),
                                                 resultSlot,
                                                 recordIdSlot,
@@ -853,8 +853,10 @@ std::unique_ptr<sbe::PlanStage> makeLoopJoinForFetch(std::unique_ptr<sbe::PlanSt
                                                 boost::none,
                                                 std::move(fields),
                                                 std::move(fieldSlots),
-                                                seekKeySlot,
-                                                true,
+                                                seekRecordIdSlot,
+                                                boost::none /* minRecordIdSlot */,
+                                                boost::none /* maxRecordIdSlot */,
+                                                true /* forward */,
                                                 nullptr,
                                                 planNodeId,
                                                 std::move(callbacks));
@@ -865,7 +867,8 @@ std::unique_ptr<sbe::PlanStage> makeLoopJoinForFetch(std::unique_ptr<sbe::PlanSt
         std::move(inputStage),
         sbe::makeS<sbe::LimitSkipStage>(std::move(scanStage), 1, boost::none, planNodeId),
         std::move(slotsToForward),
-        sbe::makeSV(seekKeySlot, snapshotIdSlot, indexIdentSlot, indexKeySlot, indexKeyPatternSlot),
+        sbe::makeSV(
+            seekRecordIdSlot, snapshotIdSlot, indexIdentSlot, indexKeySlot, indexKeyPatternSlot),
         nullptr,
         planNodeId);
 }
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.h b/src/mongo/db/query/sbe_stage_builder_helpers.h
index e55c61d1bb0..c0279b790c1 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.h
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.h
@@ -518,7 +518,7 @@ std::unique_ptr<sbe::PlanStage> makeLoopJoinForFetch(std::unique_ptr<sbe::PlanSt
                                                      sbe::value::SlotId recordIdSlot,
                                                      std::vector<std::string> fields,
                                                      sbe::value::SlotVector fieldSlots,
-                                                     sbe::value::SlotId seekKeySlot,
+                                                     sbe::value::SlotId seekRecordIdSlot,
                                                      sbe::value::SlotId snapshotIdSlot,
                                                      sbe::value::SlotId indexIdentSlot,
                                                      sbe::value::SlotId indexKeySlot,
diff --git a/src/mongo/db/query/sbe_stage_builder_lookup.cpp b/src/mongo/db/query/sbe_stage_builder_lookup.cpp
index bc604952fb6..bb433c6ae68 100644
--- a/src/mongo/db/query/sbe_stage_builder_lookup.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_lookup.cpp
@@ -1203,21 +1203,24 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
                 auto foreignResultSlot = _slotIdGenerator.generate();
                 auto foreignRecordIdSlot = _slotIdGenerator.generate();
 
-                auto foreignStage = makeS<ScanStage>(foreignColl->uuid(),
-                                                     foreignResultSlot,
-                                                     foreignRecordIdSlot,
-                                                     boost::none /* snapshotIdSlot */,
-                                                     boost::none /* indexIdentSlot */,
-                                                     boost::none /* indexKeySlot */,
-                                                     boost::none /* indexKeyPatternSlot */,
-                                                     boost::none /* tsSlot */,
-                                                     std::vector<std::string>{} /* fields */,
-                                                     makeSV() /* vars */,
-                                                     boost::none /* seekKeySlot */,
-                                                     true /* forward */,
-                                                     _yieldPolicy,
-                                                     eqLookupNode->nodeId(),
-                                                     ScanCallbacks{});
+                auto foreignStage =
+                    makeS<sbe::ScanStage>(foreignColl->uuid(),
+                                          foreignResultSlot,
+                                          foreignRecordIdSlot,
+                                          boost::none /* snapshotIdSlot */,
+                                          boost::none /* indexIdentSlot */,
+                                          boost::none /* indexKeySlot */,
+                                          boost::none /* indexKeyPatternSlot */,
+                                          boost::none /* oplogTsSlot */,
+                                          std::vector<std::string>{} /* scanFieldNames */,
+                                          makeSV() /* scanFieldSlots */,
+                                          boost::none /* seekRecordIdSlot */,
+                                          boost::none /* minRecordIdSlot */,
+                                          boost::none /* maxRecordIdSlot */,
+                                          true /* forward */,
+                                          _yieldPolicy,
+                                          eqLookupNode->nodeId(),
+                                          ScanCallbacks{});
 
                 return buildLookupStage(_state,
                                         eqLookupNode->lookupStrategy,
diff --git a/src/mongo/db/query/stage_types.cpp b/src/mongo/db/query/stage_types.cpp
index aca0d7370cd..d37de3600bd 100644
--- a/src/mongo/db/query/stage_types.cpp
+++ b/src/mongo/db/query/stage_types.cpp
@@ -86,4 +86,9 @@ StringData stageTypeToString(StageType stageType) {
     }
     return kStageTypesMap.at(STAGE_UNKNOWN);
 }
+
+StringData sbeClusteredCollectionScanToString() {
+    static const StringData kClusteredIxscan = "CLUSTERED_IXSCAN"_sd;
+    return kClusteredIxscan;
+}
 }  // namespace mongo
diff --git a/src/mongo/db/query/stage_types.h b/src/mongo/db/query/stage_types.h
index a0f725dff0a..89d63066b56 100644
--- a/src/mongo/db/query/stage_types.h
+++ b/src/mongo/db/query/stage_types.h
@@ -44,14 +44,17 @@ using PlanNodeId = uint32_t;
 static constexpr PlanNodeId kEmptyPlanNodeId = 0u;
 
 /**
- * These map to implementations of the PlanStage interface, all of which live in db/exec/
+ * These map to implementations of the PlanStage interface, all of which live in db/exec/. These
+ * stage types are shared between Classic and SBE.
  */
 enum StageType {
     STAGE_AND_HASH,
     STAGE_AND_SORTED,
     STAGE_BATCHED_DELETE,
     STAGE_CACHED_PLAN,
+
     STAGE_COLLSCAN,
+
     STAGE_COLUMN_SCAN,
 
     // A virtual scan stage that simulates a collection scan and doesn't depend on underlying
@@ -160,4 +163,10 @@ inline bool isSortStageType(StageType stageType) {
 }
 
 StringData stageTypeToString(StageType stageType);
+
+/**
+ * Returns the explain() stage type string for a STAGE_COLLSCAN stage that is performing a clustered
+ * collection scan in SBE, to match Classic's explain() output.
+ */
+StringData sbeClusteredCollectionScanToString();
 }  // namespace mongo