Revert "SERVER-71798 Expand the set of queries eligible for SBE in the 6.3 release"

15 files changed, 276 insertions, 186 deletions

diff --git a/src/mongo/db/query/canonical_query.cpp b/src/mongo/db/query/canonical_query.cpp
index 45fc3ef9c41..516e33fe15a 100644
--- a/src/mongo/db/query/canonical_query.cpp
+++ b/src/mongo/db/query/canonical_query.cpp
@@ -208,7 +208,8 @@ Status CanonicalQuery::init(OperationContext* opCtx,
     _root = MatchExpression::normalize(std::move(root));
 
     // If caching is disabled, do not perform any autoparameterization.
-    if (!internalQueryDisablePlanCache.load()) {
+    if (!internalQueryDisablePlanCache.load() &&
+        feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
         const bool hasNoTextNodes =
             !QueryPlannerCommon::hasNode(_root.get(), MatchExpression::TEXT);
         if (hasNoTextNodes) {
@@ -547,8 +548,10 @@ std::string CanonicalQuery::toStringShort() const {
 }
 
 CanonicalQuery::QueryShapeString CanonicalQuery::encodeKey() const {
-    return (!_forceClassicEngine && _sbeCompatible) ? canonical_query_encoder::encodeSBE(*this)
-                                                    : canonical_query_encoder::encode(*this);
+    return (feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV() && !_forceClassicEngine &&
+            _sbeCompatible)
+        ? canonical_query_encoder::encodeSBE(*this)
+        : canonical_query_encoder::encode(*this);
 }
 
 CanonicalQuery::QueryShapeString CanonicalQuery::encodeKeyForPlanCacheCommand() const {
diff --git a/src/mongo/db/query/canonical_query_encoder.cpp b/src/mongo/db/query/canonical_query_encoder.cpp
index 4313b6efc2b..93036c4feda 100644
--- a/src/mongo/db/query/canonical_query_encoder.cpp
+++ b/src/mongo/db/query/canonical_query_encoder.cpp
@@ -44,6 +44,7 @@
 #include "mongo/db/pipeline/document_source_lookup.h"
 #include "mongo/db/query/analyze_regex.h"
 #include "mongo/db/query/projection.h"
+#include "mongo/db/query/query_feature_flags_gen.h"
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/query/tree_walker.h"
 #include "mongo/logv2/log.h"
@@ -1088,6 +1089,9 @@ void encodeKeyForAutoParameterizedMatchSBE(MatchExpression* matchExpr, BufBuilde
 }  // namespace
 
 std::string encodeSBE(const CanonicalQuery& cq) {
+    tassert(6512900,
+            "using the SBE plan cache key encoding requires SBE to be fully enabled",
+            feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV());
     tassert(6142104,
             "attempting to encode SBE plan cache key for SBE-incompatible query",
             cq.isSbeCompatible());
diff --git a/src/mongo/db/query/canonical_query_encoder_test.cpp b/src/mongo/db/query/canonical_query_encoder_test.cpp
index 12593f56490..4987272cf80 100644
--- a/src/mongo/db/query/canonical_query_encoder_test.cpp
+++ b/src/mongo/db/query/canonical_query_encoder_test.cpp
@@ -427,6 +427,9 @@ TEST(CanonicalQueryEncoderTest, ComputeKeySBE) {
     // SBE must be enabled in order to generate SBE plan cache keys.
     RAIIServerParameterControllerForTest controllerSBE("internalQueryFrameworkControl",
                                                        "trySbeEngine");
+
+    RAIIServerParameterControllerForTest controllerSBEPlanCache("featureFlagSbeFull", true);
+
     testComputeSBEKey(gctx, "{}", "{}", "{}");
     testComputeSBEKey(gctx, "{$or: [{a: 1}, {b: 2}]}", "{}", "{}");
     testComputeSBEKey(gctx, "{a: 1}", "{}", "{}");
@@ -499,6 +502,7 @@ TEST(CanonicalQueryEncoderTest, ComputeKeySBEWithPipeline) {
     RAIIServerParameterControllerForTest controllerSBE("internalQueryFrameworkControl",
                                                        "trySbeEngine");
 
+    RAIIServerParameterControllerForTest controllerSBEPlanCache("featureFlagSbeFull", true);
 
     auto getLookupBson = [](StringData localField, StringData foreignField, StringData asField) {
         return BSON("$lookup" << BSON("from" << foreignNss.coll() << "localField" << localField
@@ -528,6 +532,7 @@ TEST(CanonicalQueryEncoderTest, ComputeKeySBEWithReadConcern) {
     // SBE must be enabled in order to generate SBE plan cache keys.
     RAIIServerParameterControllerForTest controllerSBE("internalQueryFrameworkControl",
                                                        "trySbeEngine");
+    RAIIServerParameterControllerForTest controllerSBEPlanCache("featureFlagSbeFull", true);
 
     // Find command without read concern.
     auto findCommand = std::make_unique<FindCommandRequest>(nss);
diff --git a/src/mongo/db/query/canonical_query_test.cpp b/src/mongo/db/query/canonical_query_test.cpp
index 2fb5614fd16..a15a3b918b0 100644
--- a/src/mongo/db/query/canonical_query_test.cpp
+++ b/src/mongo/db/query/canonical_query_test.cpp
@@ -456,6 +456,7 @@ TEST(CanonicalQueryTest, InvalidSortOrdersFailToCanonicalize) {
 }
 
 TEST(CanonicalQueryTest, DoNotParameterizeTextExpressions) {
+    RAIIServerParameterControllerForTest controllerSBEPlanCache("featureFlagSbeFull", true);
     auto cq =
         canonicalize("{$text: {$search: \"Hello World!\"}}",
                      MatchExpressionParser::kDefaultSpecialFeatures | MatchExpressionParser::kText);
@@ -463,6 +464,7 @@ TEST(CanonicalQueryTest, DoNotParameterizeTextExpressions) {
 }
 
 TEST(CanonicalQueryTest, DoParameterizeRegularExpressions) {
+    RAIIServerParameterControllerForTest controllerSBEPlanCache("featureFlagSbeFull", true);
     auto cq = canonicalize("{a: 1, b: {$lt: 5}}");
     ASSERT_TRUE(cq->isParameterized());
 }
diff --git a/src/mongo/db/query/classic_plan_cache.cpp b/src/mongo/db/query/classic_plan_cache.cpp
index 41874a76d8a..7789d894cb5 100644
--- a/src/mongo/db/query/classic_plan_cache.cpp
+++ b/src/mongo/db/query/classic_plan_cache.cpp
@@ -130,7 +130,8 @@ bool shouldCacheQuery(const CanonicalQuery& query) {
     const MatchExpression* expr = query.root();
 
     if (!query.getSortPattern() && expr->matchType() == MatchExpression::AND &&
-        expr->numChildren() == 0 && !query.isSbeCompatible()) {
+        expr->numChildren() == 0 &&
+        !(feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV() && query.isSbeCompatible())) {
         return false;
     }
 
diff --git a/src/mongo/db/query/explain.cpp b/src/mongo/db/query/explain.cpp
index b9b69eac18d..545ec8553f4 100644
--- a/src/mongo/db/query/explain.cpp
+++ b/src/mongo/db/query/explain.cpp
@@ -98,6 +98,7 @@ void generatePlannerInfo(PlanExecutor* exec,
         const QuerySettings* querySettings =
             QuerySettingsDecoration::get(mainCollection->getSharedDecorations());
         if (exec->getCanonicalQuery()->isSbeCompatible() &&
+            feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV() &&
             !exec->getCanonicalQuery()->getForceClassicEngine()) {
             const auto planCacheKeyInfo =
                 plan_cache_key_factory::make(*exec->getCanonicalQuery(), collections);
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index 157095d941e..b3a28aef907 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -666,7 +666,7 @@ public:
         _fromPlanCache = val;
     }
 
-    bool isRecoveredFromPlanCache() const {
+    bool isRecoveredFromPlanCache() {
         return _fromPlanCache;
     }
 
@@ -1148,25 +1148,64 @@ protected:
     std::unique_ptr<SlotBasedPrepareExecutionResult> buildCachedPlan(
         const sbe::PlanCacheKey& planCacheKey) final {
         if (shouldCacheQuery(*_cq)) {
-            getResult()->planCacheInfo().planCacheKey = planCacheKey.planCacheKeyHash();
+            if (!feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+                return buildCachedPlanFromClassicCache();
+            } else {
+                getResult()->planCacheInfo().planCacheKey = planCacheKey.planCacheKeyHash();
 
-            auto&& planCache = sbe::getPlanCache(_opCtx);
-            auto cacheEntry = planCache.getCacheEntryIfActive(planCacheKey);
-            if (!cacheEntry) {
-                return nullptr;
+                auto&& planCache = sbe::getPlanCache(_opCtx);
+                auto cacheEntry = planCache.getCacheEntryIfActive(planCacheKey);
+                if (!cacheEntry) {
+                    return nullptr;
+                }
+
+                auto&& cachedPlan = std::move(cacheEntry->cachedPlan);
+                auto root = std::move(cachedPlan->root);
+                auto stageData = std::move(cachedPlan->planStageData);
+                stageData.debugInfo = cacheEntry->debugInfo;
+
+                auto result = releaseResult();
+                result->setDecisionWorks(cacheEntry->decisionWorks);
+                result->setRecoveredPinnedCacheEntry(cacheEntry->isPinned());
+                result->emplace(std::make_pair(std::move(root), std::move(stageData)));
+                result->setRecoveredFromPlanCache(true);
+                return result;
             }
+        }
 
-            auto&& cachedPlan = std::move(cacheEntry->cachedPlan);
-            auto root = std::move(cachedPlan->root);
-            auto stageData = std::move(cachedPlan->planStageData);
-            stageData.debugInfo = cacheEntry->debugInfo;
+        return nullptr;
+    }
 
-            auto result = releaseResult();
-            result->setDecisionWorks(cacheEntry->decisionWorks);
-            result->setRecoveredPinnedCacheEntry(cacheEntry->isPinned());
-            result->emplace(std::make_pair(std::move(root), std::move(stageData)));
-            result->setRecoveredFromPlanCache(true);
-            return result;
+    // A temporary function to allow recovering SBE plans from the classic plan cache. When the
+    // feature flag for "SBE full" is disabled, we are still able to use the classic plan cache for
+    // queries that execute in SBE.
+    //
+    // TODO SERVER-64882: Remove this function when "featureFlagSbeFull" is removed.
+    std::unique_ptr<SlotBasedPrepareExecutionResult> buildCachedPlanFromClassicCache() {
+        const auto& mainColl = getMainCollection();
+        auto planCacheKey = plan_cache_key_factory::make<PlanCacheKey>(*_cq, mainColl);
+        getResult()->planCacheInfo().planCacheKey = planCacheKey.planCacheKeyHash();
+
+        // Try to look up a cached solution for the query.
+        if (auto cs = CollectionQueryInfo::get(mainColl).getPlanCache()->getCacheEntryIfActive(
+                planCacheKey)) {
+            initializePlannerParamsIfNeeded();
+            // We have a CachedSolution.  Have the planner turn it into a QuerySolution.
+            auto statusWithQs = QueryPlanner::planFromCache(*_cq, _plannerParams, *cs);
+
+            if (statusWithQs.isOK()) {
+                auto querySolution = std::move(statusWithQs.getValue());
+                if (_cq->isCountLike() && turnIxscanIntoCount(querySolution.get())) {
+                    LOGV2_DEBUG(
+                        20923, 2, "Using fast count", "query"_attr = redact(_cq->toStringShort()));
+                }
+
+                auto result = releaseResult();
+                addSolutionToResult(result.get(), std::move(querySolution));
+                result->setDecisionWorks(cs->decisionWorks);
+                result->setRecoveredFromPlanCache(true);
+                return result;
+            }
         }
 
         return nullptr;
@@ -1390,52 +1429,56 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
 }
 
 /**
- * Checks if the result of query planning is SBE compatible. In this function, 'sbeFull' indicates
- * whether the full set of features supported by SBE is enabled, while 'canUseRegularSbe' indicates
- * whether the query is compatible with the subset of SBE enabled by default.
+ * Checks if the result of query planning is SBE compatible.
  */
 bool shouldPlanningResultUseSbe(bool sbeFull,
-                                bool canUseRegularSbe,
                                 bool columnIndexPresent,
+                                bool aggSpecificStagesPushedDown,
                                 const SlotBasedPrepareExecutionResult& planningResult) {
-    // If we have an entry in the SBE plan cache, then we can use SBE.
-    if (planningResult.isRecoveredFromPlanCache()) {
-        return true;
-    }
-
     // For now this function assumes one of these is true. If all are false, we should not use
     // SBE.
     tassert(6164401,
-            "Expected sbeFull, or a regular SBE compatiable query, or a CSI present",
-            sbeFull || canUseRegularSbe || columnIndexPresent);
+            "Expected sbeFull, or a CSI present, or agg specific stages pushed down",
+            sbeFull || columnIndexPresent || aggSpecificStagesPushedDown);
 
     const auto& solutions = planningResult.solutions();
     if (solutions.empty()) {
         // Query needs subplanning (plans are generated later, we don't have access yet).
         invariant(planningResult.needsSubplanning());
 
-        // Use SBE for rooted $or queries if SBE is fully enabled or the query is SBE compatible to
-        // begin with.
-        return sbeFull || canUseRegularSbe;
+        // TODO: SERVER-71798 if the below conditions are not met, a column index will not be used
+        // even if it could be.
+        return sbeFull || aggSpecificStagesPushedDown;
     }
 
     // Check that the query solution is SBE compatible.
     const bool allStagesCompatible =
         std::all_of(solutions.begin(), solutions.end(), [](const auto& solution) {
-            // We must have a solution, otherwise we would have early exited.
-            invariant(solution->root());
-            return isQueryPlanSbeCompatible(solution.get());
+            return solution->root() ==
+                nullptr /* we won't have a query solution if we pulled it from the cache */
+                || isQueryPlanSbeCompatible(solution.get());
         });
 
     if (!allStagesCompatible) {
         return false;
     }
 
-    if (sbeFull || canUseRegularSbe) {
+    if (sbeFull || aggSpecificStagesPushedDown) {
         return true;
     }
 
-    // Return true if we have a column scan plan, and false otherwise.
+    // If no pipeline is pushed down and SBE full is off, the only other case we'll use SBE for
+    // is when a column index plan was constructed.
+    tassert(6164400, "Expected CSI to be present", columnIndexPresent);
+
+    // The only time a query solution is not available is when the plan comes from the SBE plan
+    // cache. The plan cache is gated by sbeFull, which was already checked earlier. So, at this
+    // point we're guaranteed sbeFull is off, and this further implies that the returned plan(s)
+    // did not come from the cache.
+    tassert(6164402,
+            "Did not expect a plan from the plan cache",
+            !sbeFull && solutions.front()->root());
+
     return solutions.size() == 1 &&
         solutions.front()->root()->hasNode(StageType::STAGE_COLUMN_SCAN);
 }
@@ -1475,38 +1518,6 @@ bool maybeQueryIsColumnScanEligible(OperationContext* opCtx,
 }
 
 /**
- * Function which returns true if 'cq' uses features that are currently supported in SBE without
- * 'featureFlagSbeFull' being set; false otherwise.
- */
-bool shouldUseRegularSbe(const CanonicalQuery& cq) {
-    const auto* proj = cq.getProj();
-
-    // Disallow projections which use expressions.
-    if (proj && proj->hasExpressions()) {
-        return false;
-    }
-
-    // Disallow projections which have dotted paths.
-    if (proj && proj->hasDottedPaths()) {
-        return false;
-    }
-
-    // Disallow filters which feature $expr.
-    if (cq.countNodes(cq.root(), MatchExpression::MatchType::EXPRESSION) > 0) {
-        return false;
-    }
-
-    const auto& sortPattern = cq.getSortPattern();
-
-    // Disallow sorts which have a common prefix.
-    if (sortPattern && sortPatternHasPartsWithCommonPrefix(*sortPattern)) {
-        return false;
-    }
-
-    return true;
-}
-
-/**
  * Attempts to create a slot-based executor for the query, if the query plan is eligible for SBE
  * execution. This function has three possible return values:
  *
@@ -1532,20 +1543,18 @@ attemptToGetSlotBasedExecutor(
     }
 
     const bool sbeFull = feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV();
-    const bool canUseRegularSbe = shouldUseRegularSbe(*canonicalQuery);
+    const bool aggSpecificStagesPushedDown = !canonicalQuery->pipeline().empty();
 
-    // Attempt to use SBE if the query may be eligible for column scan, if the currently supported
-    // subset of SBE is being used, or if SBE is fully enabled. Otherwise, fallback to the classic
-    // engine right away.
-    if (sbeFull || canUseRegularSbe ||
+    // Attempt to use SBE if we find any $group/$lookup stages eligible for execution in SBE, if the
+    // query may be eligible for column scan, or if SBE is fully enabled. Otherwise, fallback to the
+    // classic engine right away.
+    if (aggSpecificStagesPushedDown || sbeFull ||
         maybeQueryIsColumnScanEligible(opCtx, collections, canonicalQuery.get())) {
-        // Create the SBE prepare execution helper and initialize the params for the planner. Our
-        // decision about using SBE will depend on whether there is a column index present.
-
         auto sbeYieldPolicy = makeSbeYieldPolicy(
             opCtx, yieldPolicy, &collections.getMainCollection(), canonicalQuery->nss());
         SlotBasedPrepareExecutionHelper helper{
             opCtx, collections, canonicalQuery.get(), sbeYieldPolicy.get(), plannerParams.options};
+
         auto planningResultWithStatus = helper.prepare();
         if (!planningResultWithStatus.isOK()) {
             return planningResultWithStatus.getStatus();
@@ -1554,8 +1563,10 @@ attemptToGetSlotBasedExecutor(
         const bool csiPresent =
             helper.plannerParams() && !helper.plannerParams()->columnStoreIndexes.empty();
 
-        if (shouldPlanningResultUseSbe(
-                sbeFull, canUseRegularSbe, csiPresent, *planningResultWithStatus.getValue())) {
+        if (shouldPlanningResultUseSbe(sbeFull,
+                                       csiPresent,
+                                       aggSpecificStagesPushedDown,
+                                       *planningResultWithStatus.getValue())) {
             if (extractAndAttachPipelineStages) {
                 // We know now that we will use SBE, so we need to remove the pushed-down stages
                 // from the original pipeline object.
@@ -1635,8 +1646,6 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
                     std::move(stdx::get<std::unique_ptr<CanonicalQuery>>(maybeExecutor));
             }
         }
-        // Ensure that 'sbeCompatible' is set accordingly.
-        canonicalQuery->setSbeCompatible(false);
         return getClassicExecutor(
             opCtx, mainColl, std::move(canonicalQuery), yieldPolicy, plannerParams);
     }();
diff --git a/src/mongo/db/query/projection.h b/src/mongo/db/query/projection.h
index f920a778e9a..c093083c068 100644
--- a/src/mongo/db/query/projection.h
+++ b/src/mongo/db/query/projection.h
@@ -118,13 +118,6 @@ public:
      */
     bool isFieldRetainedExactly(StringData path) const;
 
-
-    /**
-     * Returns true if this projection has any dotted paths; false otherwise.
-     */
-    bool hasDottedPaths() const {
-        return _deps.hasDottedPath;
-    }
     /**
      * A projection is considered "simple" if it operates only on top-level fields,
      * has no positional projection or expressions, and doesn't require metadata.
diff --git a/src/mongo/db/query/query_feature_flags.idl b/src/mongo/db/query/query_feature_flags.idl
index d0289f1f761..55a3a24c408 100644
--- a/src/mongo/db/query/query_feature_flags.idl
+++ b/src/mongo/db/query/query_feature_flags.idl
@@ -77,7 +77,8 @@ feature_flags:
       default: false
 
     featureFlagSbeFull:
-      description: "Feature flag for enabling SBE for a much larger class of queries than what is exposed by default"
+      description: "Feature flag for enabling full SBE support. Enables SBE for a much larger class
+      of queries, including NLJ $lookup plans. Also enables the SBE plan cache."
       cpp_varname: gFeatureFlagSbeFull
       default: false
 
diff --git a/src/mongo/db/query/query_utils.cpp b/src/mongo/db/query/query_utils.cpp
index 71dd2acded6..917817c739e 100644
--- a/src/mongo/db/query/query_utils.cpp
+++ b/src/mongo/db/query/query_utils.cpp
@@ -34,21 +34,6 @@
 
 namespace mongo {
 
-bool sortPatternHasPartsWithCommonPrefix(const SortPattern& sortPattern) {
-    StringDataSet prefixSet;
-    for (const auto& part : sortPattern) {
-        // Ignore any $meta sorts that may be present.
-        if (!part.fieldPath) {
-            continue;
-        }
-        auto [_, inserted] = prefixSet.insert(part.fieldPath->getFieldName(0));
-        if (!inserted) {
-            return true;
-        }
-    }
-    return false;
-}
-
 bool isIdHackEligibleQuery(const CollectionPtr& collection, const CanonicalQuery& query) {
     const auto& findCommand = query.getFindCommandRequest();
     return !findCommand.getShowRecordId() && findCommand.getHint().isEmpty() &&
@@ -81,11 +66,10 @@ bool isQuerySbeCompatible(const CollectionPtr* collection, const CanonicalQuery*
     const bool isQueryNotAgainstClusteredCollection =
         !(collection->get() && collection->get()->isClustered());
 
-    const auto* proj = cq->getProj();
-
-    const bool doesNotRequireMatchDetails = !proj || !proj->requiresMatchDetails();
+    const bool doesNotRequireMatchDetails =
+        !cq->getProj() || !cq->getProj()->requiresMatchDetails();
 
-    const bool doesNotHaveElemMatchProject = !proj || !proj->containsElemMatch();
+    const bool doesNotHaveElemMatchProject = !cq->getProj() || !cq->getProj()->containsElemMatch();
 
     const bool isNotInnerSideOfLookup = !(expCtx && expCtx->inLookup);
 
diff --git a/src/mongo/db/query/query_utils.h b/src/mongo/db/query/query_utils.h
index 55a5e069ad3..97165860da1 100644
--- a/src/mongo/db/query/query_utils.h
+++ b/src/mongo/db/query/query_utils.h
@@ -34,12 +34,6 @@
 
 namespace mongo {
 /**
- * Returns 'true' if 'sortPattern' contains any sort pattern parts that share a common prefix, false
- * otherwise.
- */
-bool sortPatternHasPartsWithCommonPrefix(const SortPattern& sortPattern);
-
-/**
  * Returns 'true' if 'query' on the given 'collection' can be answered using a special IDHACK plan.
  */
 bool isIdHackEligibleQuery(const CollectionPtr& collection, const CanonicalQuery& query);
diff --git a/src/mongo/db/query/sbe_cached_solution_planner.cpp b/src/mongo/db/query/sbe_cached_solution_planner.cpp
index 5fbe8be2ec3..5927bf5722c 100644
--- a/src/mongo/db/query/sbe_cached_solution_planner.cpp
+++ b/src/mongo/db/query/sbe_cached_solution_planner.cpp
@@ -52,24 +52,45 @@ CandidatePlans CachedSolutionPlanner::plan(
     std::vector<std::unique_ptr<QuerySolution>> solutions,
     std::vector<std::pair<std::unique_ptr<PlanStage>, stage_builder::PlanStageData>> roots) {
     if (!_cq.pipeline().empty()) {
-        // We'd like to check if there is any foreign collection in the hash_lookup stage that is no
-        // longer eligible for using a hash_lookup plan. In this case we invalidate the cache and
-        // immediately replan without ever running a trial period.
+        // When "featureFlagSbeFull" is enabled we use the SBE plan cache. If the plan cache is
+        // enabled we'd like to check if there is any foreign collection in the hash_lookup stage
+        // that is no longer eligible for it. In this case we invalidate the cache and immediately
+        // replan without ever running a trial period.
         auto secondaryCollectionsInfo =
             fillOutSecondaryCollectionsInformation(_opCtx, _collections, &_cq);
 
-        for (const auto& foreignCollection : roots[0].second.foreignHashJoinCollections) {
-            const auto collectionInfo = secondaryCollectionsInfo.find(foreignCollection);
-            tassert(6693500,
-                    "Foreign collection must be present in the collections info",
-                    collectionInfo != secondaryCollectionsInfo.end());
-            tassert(6693501, "Foreign collection must exist", collectionInfo->second.exists);
+        if (feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+            for (const auto& foreignCollection : roots[0].second.foreignHashJoinCollections) {
+                const auto collectionInfo = secondaryCollectionsInfo.find(foreignCollection);
+                tassert(6693500,
+                        "Foreign collection must be present in the collections info",
+                        collectionInfo != secondaryCollectionsInfo.end());
+                tassert(6693501, "Foreign collection must exist", collectionInfo->second.exists);
 
-            if (!QueryPlannerAnalysis::isEligibleForHashJoin(collectionInfo->second)) {
-                return replan(/* shouldCache */ true,
-                              str::stream() << "Foreign collection " << foreignCollection
-                                            << " is not eligible for hash join anymore");
+                if (!QueryPlannerAnalysis::isEligibleForHashJoin(collectionInfo->second)) {
+                    return replan(/* shouldCache */ true,
+                                  str::stream() << "Foreign collection " << foreignCollection
+                                                << " is not eligible for hash join anymore");
+                }
             }
+        } else {
+            // The SBE plan cache is not enabled. If the cached plan is accepted we'd like to keep
+            // the results from the trials even if there are parts of agg pipelines being lowered
+            // into SBE, so we run the trial with the extended plan. This works because
+            // TrialRunTracker, attached to HashAgg stage in $group queries, tracks as "results" the
+            // results of its child stage. For $lookup queries, the TrialRunTracker will only track
+            // the number of reads from the local side. Thus, we can use the number of reads the
+            // plan was cached with during multiplanning even though multiplanning ran trials of
+            // pre-extended plans.
+            //
+            // The SBE plan cache stores the entire plan, including the part for any agg pipeline
+            // pushed down to SBE. Therefore, this logic is only necessary when "featureFlagSbeFull"
+            // is disabled.
+            _yieldPolicy->clearRegisteredPlans();
+            solutions[0] = QueryPlanner::extendWithAggPipeline(
+                _cq, std::move(solutions[0]), secondaryCollectionsInfo);
+            roots[0] = stage_builder::buildSlotBasedExecutableTree(
+                _opCtx, _collections, _cq, *solutions[0], _yieldPolicy);
         }
     }
     // If the '_decisionReads' is not present then we do not run a trial period, keeping the current
@@ -206,9 +227,18 @@ CandidatePlans CachedSolutionPlanner::replan(bool shouldCache, std::string reaso
     _yieldPolicy->clearRegisteredPlans();
 
     if (shouldCache) {
+        const auto& mainColl = _collections.getMainCollection();
         // Deactivate the current cache entry.
-        auto&& sbePlanCache = sbe::getPlanCache(_opCtx);
-        sbePlanCache.deactivate(plan_cache_key_factory::make(_cq, _collections));
+        //
+        // TODO SERVER-64882: We currently deactivate cache entries in both the classic and SBE plan
+        // caches. Once we always use the SBE plan cache for queries eligible for SBE, this code can
+        // be simplified to only deactivate the entry in the SBE plan cache.
+        auto cache = CollectionQueryInfo::get(mainColl).getPlanCache();
+        cache->deactivate(plan_cache_key_factory::make<mongo::PlanCacheKey>(_cq, mainColl));
+        if (feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+            auto&& sbePlanCache = sbe::getPlanCache(_opCtx);
+            sbePlanCache.deactivate(plan_cache_key_factory::make(_cq, _collections));
+        }
     }
 
     auto buildExecutableTree = [&](const QuerySolution& sol) {
diff --git a/src/mongo/db/query/sbe_plan_cache.cpp b/src/mongo/db/query/sbe_plan_cache.cpp
index 2129554a2d9..1498fa28932 100644
--- a/src/mongo/db/query/sbe_plan_cache.cpp
+++ b/src/mongo/db/query/sbe_plan_cache.cpp
@@ -48,23 +48,27 @@ const auto sbePlanCacheDecoration =
 class PlanCacheOnParamChangeUpdaterImpl final : public plan_cache_util::OnParamChangeUpdater {
 public:
     void updateCacheSize(ServiceContext* serviceCtx, memory_util::MemorySize memSize) final {
-        auto newSizeBytes = memory_util::getRequestedMemSizeInBytes(memSize);
-        auto cappedCacheSize = memory_util::capMemorySize(newSizeBytes /*requestedSizeBytes*/,
-                                                          500 /*maximumSizeGB*/,
-                                                          25 /*percentTotalSystemMemory*/);
-        if (cappedCacheSize < newSizeBytes) {
-            LOGV2_DEBUG(6007001,
-                        1,
-                        "The plan cache size has been capped",
-                        "cappedSize"_attr = cappedCacheSize);
+        if (feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+            auto newSizeBytes = memory_util::getRequestedMemSizeInBytes(memSize);
+            auto cappedCacheSize = memory_util::capMemorySize(newSizeBytes /*requestedSizeBytes*/,
+                                                              500 /*maximumSizeGB*/,
+                                                              25 /*percentTotalSystemMemory*/);
+            if (cappedCacheSize < newSizeBytes) {
+                LOGV2_DEBUG(6007001,
+                            1,
+                            "The plan cache size has been capped",
+                            "cappedSize"_attr = cappedCacheSize);
+            }
+            auto& globalPlanCache = sbePlanCacheDecoration(serviceCtx);
+            globalPlanCache->reset(cappedCacheSize);
         }
-        auto& globalPlanCache = sbePlanCacheDecoration(serviceCtx);
-        globalPlanCache->reset(cappedCacheSize);
     }
 
     void clearCache(ServiceContext* serviceCtx) final {
-        auto& globalPlanCache = sbePlanCacheDecoration(serviceCtx);
-        globalPlanCache->clear();
+        if (feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+            auto& globalPlanCache = sbePlanCacheDecoration(serviceCtx);
+            globalPlanCache->clear();
+        }
     }
 };
 
@@ -73,29 +77,38 @@ ServiceContext::ConstructorActionRegisterer planCacheRegisterer{
         plan_cache_util::sbePlanCacheOnParamChangeUpdater(serviceCtx) =
             std::make_unique<PlanCacheOnParamChangeUpdaterImpl>();
 
-        auto status = memory_util::MemorySize::parse(planCacheSize.get());
-        uassertStatusOK(status);
-        auto size = memory_util::getRequestedMemSizeInBytes(status.getValue());
-        auto cappedCacheSize = memory_util::capMemorySize(
-            size /*requestedSizeBytes*/, 500 /*maximumSizeGB*/, 25 /*percentTotalSystemMemory*/);
-        if (cappedCacheSize < size) {
-            LOGV2_DEBUG(6007000,
-                        1,
-                        "The plan cache size has been capped",
-                        "cappedSize"_attr = cappedCacheSize);
+        if (feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+            auto status = memory_util::MemorySize::parse(planCacheSize.get());
+            uassertStatusOK(status);
+            auto size = memory_util::getRequestedMemSizeInBytes(status.getValue());
+            auto cappedCacheSize = memory_util::capMemorySize(size /*requestedSizeBytes*/,
+                                                              500 /*maximumSizeGB*/,
+                                                              25 /*percentTotalSystemMemory*/);
+            if (cappedCacheSize < size) {
+                LOGV2_DEBUG(6007000,
+                            1,
+                            "The plan cache size has been capped",
+                            "cappedSize"_attr = cappedCacheSize);
+            }
+            auto& globalPlanCache = sbePlanCacheDecoration(serviceCtx);
+            globalPlanCache =
+                std::make_unique<sbe::PlanCache>(cappedCacheSize, ProcessInfo::getNumCores());
         }
-        auto& globalPlanCache = sbePlanCacheDecoration(serviceCtx);
-        globalPlanCache =
-            std::make_unique<sbe::PlanCache>(cappedCacheSize, ProcessInfo::getNumCores());
     }};
 
 }  // namespace
 
 sbe::PlanCache& getPlanCache(ServiceContext* serviceCtx) {
+    uassert(5933402,
+            "Cannot getPlanCache() if 'featureFlagSbeFull' is disabled",
+            feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV());
     return *sbePlanCacheDecoration(serviceCtx);
 }
 
 sbe::PlanCache& getPlanCache(OperationContext* opCtx) {
+    uassert(5933401,
+            "Cannot getPlanCache() if 'featureFlagSbeFull' is disabled",
+            feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV());
     tassert(5933400, "Cannot get the global SBE plan cache by a nullptr", opCtx);
     return getPlanCache(opCtx->getServiceContext());
 }
@@ -104,29 +117,32 @@ void clearPlanCacheEntriesWith(ServiceContext* serviceCtx,
                                UUID collectionUuid,
                                size_t collectionVersion,
                                bool matchSecondaryCollections) {
-    auto removed = sbe::getPlanCache(serviceCtx)
-                       .removeIf([&collectionUuid, collectionVersion, matchSecondaryCollections](
-                                     const PlanCacheKey& key, const sbe::PlanCacheEntry& entry) {
-                           if (key.getMainCollectionState().version == collectionVersion &&
-                               key.getMainCollectionState().uuid == collectionUuid) {
-                               return true;
-                           }
-                           if (matchSecondaryCollections) {
-                               for (auto& collectionState : key.getSecondaryCollectionStates()) {
-                                   if (collectionState.version == collectionVersion &&
-                                       collectionState.uuid == collectionUuid) {
-                                       return true;
-                                   }
-                               }
-                           }
-                           return false;
-                       });
-
-    LOGV2_DEBUG(6006600,
-                1,
-                "Clearing SBE Plan Cache",
-                "collectionUuid"_attr = collectionUuid,
-                "collectionVersion"_attr = collectionVersion,
-                "removedEntries"_attr = removed);
+    if (feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+        auto removed =
+            sbe::getPlanCache(serviceCtx)
+                .removeIf([&collectionUuid, collectionVersion, matchSecondaryCollections](
+                              const PlanCacheKey& key, const sbe::PlanCacheEntry& entry) {
+                    if (key.getMainCollectionState().version == collectionVersion &&
+                        key.getMainCollectionState().uuid == collectionUuid) {
+                        return true;
+                    }
+                    if (matchSecondaryCollections) {
+                        for (auto& collectionState : key.getSecondaryCollectionStates()) {
+                            if (collectionState.version == collectionVersion &&
+                                collectionState.uuid == collectionUuid) {
+                                return true;
+                            }
+                        }
+                    }
+                    return false;
+                });
+
+        LOGV2_DEBUG(6006600,
+                    1,
+                    "Clearing SBE Plan Cache",
+                    "collectionUuid"_attr = collectionUuid,
+                    "collectionVersion"_attr = collectionVersion,
+                    "removedEntries"_attr = removed);
+    }
 }
 }  // namespace mongo::sbe
diff --git a/src/mongo/db/query/sbe_stage_builder.cpp b/src/mongo/db/query/sbe_stage_builder.cpp
index 8fe87acf72b..9c354a469ae 100644
--- a/src/mongo/db/query/sbe_stage_builder.cpp
+++ b/src/mongo/db/query/sbe_stage_builder.cpp
@@ -60,8 +60,6 @@
 #include "mongo/db/query/bind_input_params.h"
 #include "mongo/db/query/expression_walker.h"
 #include "mongo/db/query/index_bounds_builder.h"
-#include "mongo/db/query/optimizer/rewrites/const_eval.h"
-#include "mongo/db/query/query_utils.h"
 #include "mongo/db/query/sbe_stage_builder_abt_helpers.h"
 #include "mongo/db/query/sbe_stage_builder_accumulator.h"
 #include "mongo/db/query/sbe_stage_builder_coll_scan.h"
@@ -1217,13 +1215,19 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
         return buildSortCovered(root, reqs);
     }
 
-    // getExecutor() should never call into buildSlotBasedExecutableTree() when the query
-    // contains $meta, so this assertion should always be true.
+    StringDataSet prefixSet;
+    bool hasPartsWithCommonPrefix = false;
     for (const auto& part : sortPattern) {
+        // getExecutor() should never call into buildSlotBasedExecutableTree() when the query
+        // contains $meta, so this assertion should always be true.
         tassert(5037002, "Sort with $meta is not supported in SBE", part.fieldPath);
+
+        if (!hasPartsWithCommonPrefix) {
+            auto [_, prefixWasNotPresent] = prefixSet.insert(part.fieldPath->getFieldName(0));
+            hasPartsWithCommonPrefix = !prefixWasNotPresent;
+        }
     }
 
-    const bool hasPartsWithCommonPrefix = sortPatternHasPartsWithCommonPrefix(sortPattern);
     auto fields = reqs.getFields();
 
     if (!hasPartsWithCommonPrefix) {
diff --git a/src/mongo/db/query/sbe_stage_builder_filter.cpp b/src/mongo/db/query/sbe_stage_builder_filter.cpp
index ffc9f38260b..57210c4b0c8 100644
--- a/src/mongo/db/query/sbe_stage_builder_filter.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_filter.cpp
@@ -1194,6 +1194,37 @@ public:
 private:
     MatchExpressionVisitorContext* _context;
 };
+
+EvalExpr applyClassicMatcher(const MatchExpression* root,
+                             EvalExpr inputExpr,
+                             StageBuilderState& state) {
+    return makeFunction("applyClassicMatcher",
+                        makeConstant(sbe::value::TypeTags::classicMatchExpresion,
+                                     sbe::value::bitcastFrom<const MatchExpression*>(
+                                         root->shallowClone().release())),
+                        inputExpr.extractExpr(state));
+}
+
+EvalExpr applyClassicMatcherOverIndexScan(const MatchExpression* root,
+                                          const PlanStageSlots* slots,
+                                          const std::vector<std::string>& keyFields) {
+    BSONObjBuilder keyPatternBuilder;
+    auto keySlots = sbe::makeSV();
+    for (const auto& field : keyFields) {
+        keyPatternBuilder.append(field, 1);
+        keySlots.emplace_back(
+            slots->get(std::make_pair(PlanStageSlots::kField, StringData(field))));
+    }
+
+    auto keyPatternTree = buildKeyPatternTree(keyPatternBuilder.obj(), keySlots);
+    auto mkObjExpr = buildNewObjExpr(keyPatternTree.get());
+
+    return makeFunction("applyClassicMatcher",
+                        makeConstant(sbe::value::TypeTags::classicMatchExpresion,
+                                     sbe::value::bitcastFrom<const MatchExpression*>(
+                                         root->shallowClone().release())),
+                        std::move(mkObjExpr));
+}
 }  // namespace
 
 EvalExpr generateFilter(StageBuilderState& state,
@@ -1208,6 +1239,18 @@ EvalExpr generateFilter(StageBuilderState& state,
         return EvalExpr{};
     }
 
+    // We only use the classic matcher path (aka "franken matcher") when SBE is not fully enabled.
+    // Fully enabling SBE turns on the SBE plan cache, and embedding the classic matcher into the
+    // query execution tree is not compatible with the plan cache's use of auto-parameterization.
+    // This is because when embedding the classic matcher all of the constants used in the filter
+    // are in the MatchExpression itself rather than in slots.
+    if (!feature_flags::gFeatureFlagSbeFull.isEnabledAndIgnoreFCV()) {
+        tassert(7097207, "Expected input slot to be defined", rootSlot || isFilterOverIxscan);
+
+        return isFilterOverIxscan ? applyClassicMatcherOverIndexScan(root, slots, keyFields)
+                                  : applyClassicMatcher(root, toEvalExpr(rootSlot), state);
+    }
+
     MatchExpressionVisitorContext context{state, rootSlot, root, slots, isFilterOverIxscan};
 
     MatchExpressionPreVisitor preVisitor{&context};