SERVER-61422 Update SBE filter stage builder to use parameter markers

27 files changed, 519 insertions, 354 deletions

diff --git a/src/mongo/db/commands/cqf/cqf_aggregate.cpp b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
index 01d05a99293..9f7b5a8c243 100644
--- a/src/mongo/db/commands/cqf/cqf_aggregate.cpp
+++ b/src/mongo/db/commands/cqf/cqf_aggregate.cpp
@@ -263,6 +263,7 @@ static std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> optimizeAndCreateExe
                                              std::make_unique<YieldPolicyCallbacksImpl>(nss),
                                              false /*useExperimentalCommitTxnBehavior*/);
 
+    sbePlan->prepare(data.ctx);
     auto planExec = uassertStatusOK(plan_executor_factory::make(
         opCtx,
         nullptr /*cq*/,
diff --git a/src/mongo/db/exec/plan_cache_util.cpp b/src/mongo/db/exec/plan_cache_util.cpp
index 9763422cda6..c08097c312d 100644
--- a/src/mongo/db/exec/plan_cache_util.cpp
+++ b/src/mongo/db/exec/plan_cache_util.cpp
@@ -83,7 +83,6 @@ void updatePlanCache(OperationContext* opCtx,
         feature_flags::gFeatureFlagSbePlanCache.isEnabledAndIgnoreFCV()) {
         auto key = plan_cache_key_factory::make<sbe::PlanCacheKey>(query, collection);
         auto plan = std::make_unique<sbe::CachedSbePlan>(root.clone(), data);
-        resetRuntimeEnvironmentBeforeCaching(&plan->planStageData);
         sbe::getPlanCache(opCtx).setPinned(
             std::move(key),
             std::move(plan),
@@ -182,21 +181,5 @@ plan_cache_debug_info::DebugInfoSBE buildDebugInfo(const QuerySolution* solution
 
     return debugInfo;
 }
-
-void resetRuntimeEnvironmentBeforeCaching(stage_builder::PlanStageData* data) {
-    tassert(6183501, "PlanStageData should not be null", data);
-
-    // Manually reset "shardFilterer" to "Nothing" because we should not store
-    // "shardFilterer" which holds a "ScopedCollectionFilter" preventing data that
-    // may have been migrated from being deleted.
-    if (auto shardFiltererSlot = data->env->getSlotIfExists("shardFilterer"_sd)) {
-        data->env->resetSlot(*shardFiltererSlot, sbe::value::TypeTags::Nothing, 0, true);
-    }
-
-    // Reset all the parameterized slots.
-    for (auto [paramId, slotId] : data->inputParamToSlotMap) {
-        data->env->resetSlot(slotId, sbe::value::TypeTags::Nothing, 0, true);
-    }
-}
 }  // namespace plan_cache_util
 }  // namespace mongo
diff --git a/src/mongo/db/exec/plan_cache_util.h b/src/mongo/db/exec/plan_cache_util.h
index 6be4751aa64..9b3d4b540cb 100644
--- a/src/mongo/db/exec/plan_cache_util.h
+++ b/src/mongo/db/exec/plan_cache_util.h
@@ -87,11 +87,6 @@ plan_cache_debug_info::DebugInfo buildDebugInfo(
 plan_cache_debug_info::DebugInfoSBE buildDebugInfo(const QuerySolution* solution);
 
 /**
- * Resets all the parameterized slots in the RuntimeEnvironment of 'data'.
- */
-void resetRuntimeEnvironmentBeforeCaching(stage_builder::PlanStageData* data);
-
-/**
  * Caches the best candidate plan, chosen from the given 'candidates' based on the 'ranking'
  * decision, if the 'query' is of a type that can be cached. Otherwise, does nothing.
  *
@@ -107,8 +102,7 @@ void updatePlanCache(
     PlanCachingMode cachingMode,
     const CanonicalQuery& query,
     std::unique_ptr<plan_ranker::PlanRankingDecision> ranking,
-    const std::vector<plan_ranker::BaseCandidatePlan<PlanStageType, ResultType, Data>>&
-        candidates) {
+    std::vector<plan_ranker::BaseCandidatePlan<PlanStageType, ResultType, Data>>& candidates) {
     auto winnerIdx = ranking->candidateOrder[0];
     invariant(winnerIdx >= 0 && winnerIdx < candidates.size());
     auto& winningPlan = candidates[winnerIdx];
@@ -205,11 +199,13 @@ void updatePlanCache(
         if (winningPlan.solution->cacheData != nullptr) {
             if constexpr (std::is_same_v<PlanStageType, std::unique_ptr<sbe::PlanStage>>) {
                 if (feature_flags::gFeatureFlagSbePlanCache.isEnabledAndIgnoreFCV()) {
+                    tassert(6142201,
+                            "The winning CandidatePlan should contain the original plan",
+                            winningPlan.clonedPlan);
                     // Clone the winning SBE plan and its auxiliary data.
                     auto cachedPlan = std::make_unique<sbe::CachedSbePlan>(
-                        winningPlan.root->clone(), winningPlan.data);
-
-                    resetRuntimeEnvironmentBeforeCaching(&cachedPlan->planStageData);
+                        std::move(winningPlan.clonedPlan->first),
+                        std::move(winningPlan.clonedPlan->second));
 
                     PlanCacheLoggingCallbacks<sbe::PlanCacheKey,
                                               sbe::CachedSbePlan,
diff --git a/src/mongo/db/exec/sbe/abt/abt_lower.cpp b/src/mongo/db/exec/sbe/abt/abt_lower.cpp
index af3a763f7e3..7da6a8e2cef 100644
--- a/src/mongo/db/exec/sbe/abt/abt_lower.cpp
+++ b/src/mongo/db/exec/sbe/abt/abt_lower.cpp
@@ -254,7 +254,12 @@ std::unique_ptr<sbe::EExpression> SBEExpressionLowering::transport(
                 "Second argument to typeMatch() must be a 32-bit integer constant",
                 constPtr != nullptr && constPtr->isValueInt32());
 
-        return sbe::makeE<sbe::ETypeMatch>(std::move(args.at(0)), constPtr->getValueInt32());
+        return sbe::makeE<sbe::EFunction>(
+            "typeMatch",
+            sbe::makeEs(std::move(args.at(0)),
+                        sbe::makeE<sbe::EConstant>(
+                            sbe::value::TypeTags::NumberInt64,
+                            sbe::value::bitcastFrom<int64_t>(constPtr->getValueInt32()))));
     }
 
     // TODO - this is an open question how to do the name mappings.
diff --git a/src/mongo/db/exec/sbe/expressions/expression.cpp b/src/mongo/db/exec/sbe/expressions/expression.cpp
index fb7cc19fb2f..98cf5e3f1c1 100644
--- a/src/mongo/db/exec/sbe/expressions/expression.cpp
+++ b/src/mongo/db/exec/sbe/expressions/expression.cpp
@@ -499,6 +499,7 @@ static stdx::unordered_map<std::string, BuiltinFn> kBuiltinFunctions = {
      BuiltinFn{[](size_t n) { return n == 2; }, vm::Builtin::generateSortKey, false}},
     {"tsSecond", BuiltinFn{[](size_t n) { return n == 1; }, vm::Builtin::tsSecond, false}},
     {"tsIncrement", BuiltinFn{[](size_t n) { return n == 1; }, vm::Builtin::tsIncrement, false}},
+    {"typeMatch", BuiltinFn{[](size_t n) { return n == 2; }, vm::Builtin::typeMatch, false}},
 };
 
 /**
@@ -887,40 +888,6 @@ size_t ENumericConvert::estimateSize() const {
     return sizeof(*this) + size_estimator::estimate(_nodes);
 }
 
-std::unique_ptr<EExpression> ETypeMatch::clone() const {
-    return std::make_unique<ETypeMatch>(_nodes[0]->clone(), _typeMask);
-}
-
-vm::CodeFragment ETypeMatch::compileDirect(CompileCtx& ctx) const {
-    auto code = _nodes[0]->compileDirect(ctx);
-    code.appendTypeMatch(_typeMask);
-
-    return code;
-}
-
-std::vector<DebugPrinter::Block> ETypeMatch::debugPrint() const {
-    std::vector<DebugPrinter::Block> ret;
-
-    DebugPrinter::addKeyword(ret, "typeMatch");
-
-    ret.emplace_back("(`");
-
-    DebugPrinter::addBlocks(ret, _nodes[0]->debugPrint());
-    ret.emplace_back("`,");
-    std::stringstream ss;
-    ss << "0x" << std::setfill('0') << std::uppercase << std::setw(8) << std::hex << _typeMask;
-    ret.emplace_back(ss.str());
-
-    ret.emplace_back("`)");
-
-    return ret;
-}
-
-size_t ETypeMatch::estimateSize() const {
-    return sizeof(*this) + size_estimator::estimate(_nodes);
-}
-
-
 RuntimeEnvironment::RuntimeEnvironment(const RuntimeEnvironment& other)
     : _state{other._state}, _isSmp{other._isSmp} {
     for (auto&& [slotId, index] : _state->slots) {
diff --git a/src/mongo/db/exec/sbe/expressions/expression.h b/src/mongo/db/exec/sbe/expressions/expression.h
index ebe33abf8ae..67d95d0980c 100644
--- a/src/mongo/db/exec/sbe/expressions/expression.h
+++ b/src/mongo/db/exec/sbe/expressions/expression.h
@@ -689,30 +689,6 @@ private:
 };
 
 /**
- * This is a type match expression. It checks if a variable's BSONType is present within a given
- * set of BSONTypes encoded as a bitmask (_typeMask). If the variable's BSONType is in the set,
- * this expression returns true, otherwise it returns false.
- */
-class ETypeMatch final : public EExpression {
-public:
-    ETypeMatch(std::unique_ptr<EExpression> variable, uint32_t typeMask) : _typeMask(typeMask) {
-        _nodes.emplace_back(std::move(variable));
-        validateNodes();
-    }
-
-    std::unique_ptr<EExpression> clone() const override;
-
-    vm::CodeFragment compileDirect(CompileCtx& ctx) const override;
-
-    std::vector<DebugPrinter::Block> debugPrint() const override;
-
-    size_t estimateSize() const final;
-
-private:
-    uint32_t _typeMask;
-};
-
-/**
  * Behavior variants for bit tests supported by match expressions $bitsAllClear, $bitsAllSet,
  * $bitsAnyClear, $bitsAnySet.
  */
diff --git a/src/mongo/db/exec/sbe/stages/stages.h b/src/mongo/db/exec/sbe/stages/stages.h
index 9afd8f7de64..c44af3a26eb 100644
--- a/src/mongo/db/exec/sbe/stages/stages.h
+++ b/src/mongo/db/exec/sbe/stages/stages.h
@@ -396,7 +396,9 @@ public:
             child->attachNewYieldPolicy(yieldPolicy);
         }
 
-        _yieldPolicy = yieldPolicy;
+        if (_yieldPolicy) {
+            _yieldPolicy = yieldPolicy;
+        }
     }
 
 protected:
diff --git a/src/mongo/db/exec/sbe/vm/vm.cpp b/src/mongo/db/exec/sbe/vm/vm.cpp
index 5d28cec261d..ac9ba54ebfd 100644
--- a/src/mongo/db/exec/sbe/vm/vm.cpp
+++ b/src/mongo/db/exec/sbe/vm/vm.cpp
@@ -136,7 +136,6 @@ int Instruction::stackOffset[Instruction::Tags::lastInstruction] = {
     0,  // isMinKey
     0,  // isMaxKey
     0,  // isTimestamp
-    0,  // typeMatch
 
     0,  // function is special, the stack offset is encoded in the instruction itself
     0,  // functionSmall is special, the stack offset is encoded in the instruction itself
@@ -277,12 +276,6 @@ std::string CodeFragment::toString() const {
                 ss << "tag: " << tag;
                 break;
             }
-            case Instruction::typeMatch: {
-                auto typeMask = readFromMemory<uint32_t>(pcPointer);
-                pcPointer += sizeof(typeMask);
-                ss << "typeMask: " << typeMask;
-                break;
-            }
             case Instruction::function:
             case Instruction::functionSmall: {
                 auto f = readFromMemory<Builtin>(pcPointer);
@@ -572,17 +565,6 @@ void CodeFragment::appendIsRecordId() {
     appendSimpleInstruction(Instruction::isRecordId);
 }
 
-void CodeFragment::appendTypeMatch(uint32_t typeMask) {
-    Instruction i;
-    i.tag = Instruction::typeMatch;
-    adjustStackSimple(i);
-
-    auto offset = allocateSpace(sizeof(Instruction) + sizeof(typeMask));
-
-    offset += writeToMemory(offset, i);
-    offset += writeToMemory(offset, typeMask);
-}
-
 void CodeFragment::appendFunction(Builtin f, ArityType arity) {
     Instruction i;
     const bool isSmallArity = (arity <= std::numeric_limits<SmallArityType>::max());
@@ -3948,6 +3930,21 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinHash(ArityType
     return {false, value::TypeTags::NumberInt64, value::bitcastFrom<decltype(hashVal)>(hashVal)};
 }
 
+std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinTypeMatch(ArityType arity) {
+    invariant(arity == 2);
+
+    auto [inputOwn, inputTag, inputVal] = getFromStack(0);
+    auto [typeMaskOwn, typeMaskTag, typeMaskVal] = getFromStack(1);
+
+    if (inputTag != value::TypeTags::Nothing && typeMaskTag == value::TypeTags::NumberInt64) {
+        bool matches =
+            static_cast<bool>(getBSONTypeMask(inputTag) & value::bitcastTo<int64_t>(typeMaskVal));
+        return {false, value::TypeTags::Boolean, value::bitcastFrom<bool>(matches)};
+    }
+
+    return {false, value::TypeTags::Nothing, 0};
+}
+
 std::tuple<bool, value::TypeTags, value::Value> ByteCode::dispatchBuiltin(Builtin f,
                                                                           ArityType arity) {
     switch (f) {
@@ -4139,6 +4136,8 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::dispatchBuiltin(Builti
             return builtinTsSecond(arity);
         case Builtin::tsIncrement:
             return builtinTsIncrement(arity);
+        case Builtin::typeMatch:
+            return builtinTypeMatch(arity);
     }
 
     MONGO_UNREACHABLE;
@@ -5166,23 +5165,6 @@ void ByteCode::runInternal(const CodeFragment* code, int64_t position) {
                     }
                     break;
                 }
-                case Instruction::typeMatch: {
-                    auto typeMask = readFromMemory<uint32_t>(pcPointer);
-                    pcPointer += sizeof(typeMask);
-
-                    auto [owned, tag, val] = getFromStack(0);
-
-                    if (tag != value::TypeTags::Nothing) {
-                        bool matches = static_cast<bool>(getBSONTypeMask(tag) & typeMask);
-                        topStack(
-                            false, value::TypeTags::Boolean, value::bitcastFrom<bool>(matches));
-                    }
-
-                    if (owned) {
-                        value::releaseValue(tag, val);
-                    }
-                    break;
-                }
                 case Instruction::function:
                 case Instruction::functionSmall: {
                     auto f = readFromMemory<Builtin>(pcPointer);
diff --git a/src/mongo/db/exec/sbe/vm/vm.h b/src/mongo/db/exec/sbe/vm/vm.h
index 11e23dafd0c..5f3ca99d0a3 100644
--- a/src/mongo/db/exec/sbe/vm/vm.h
+++ b/src/mongo/db/exec/sbe/vm/vm.h
@@ -306,7 +306,6 @@ struct Instruction {
         isMinKey,
         isMaxKey,
         isTimestamp,
-        typeMatch,
 
         function,
         functionSmall,
@@ -450,8 +449,6 @@ struct Instruction {
                 return "isMaxKey";
             case isTimestamp:
                 return "isTimestamp";
-            case typeMatch:
-                return "typeMatch";
             case function:
                 return "function";
             case functionSmall:
@@ -567,6 +564,7 @@ enum class Builtin : uint8_t {
     generateSortKey,
     tsSecond,
     tsIncrement,
+    typeMatch,
 };
 
 /**
@@ -747,7 +745,6 @@ public:
     void appendIsTimestamp() {
         appendSimpleInstruction(Instruction::isTimestamp);
     }
-    void appendTypeMatch(uint32_t typeMask);
     void appendFunction(Builtin f, ArityType arity);
     void appendJump(int jumpOffset);
     void appendJumpTrue(int jumpOffset);
@@ -1165,6 +1162,7 @@ private:
     std::tuple<bool, value::TypeTags, value::Value> builtinGenerateSortKey(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinTsSecond(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinTsIncrement(ArityType arity);
+    std::tuple<bool, value::TypeTags, value::Value> builtinTypeMatch(ArityType arity);
 
     std::tuple<bool, value::TypeTags, value::Value> dispatchBuiltin(Builtin f, ArityType arity);
 
diff --git a/src/mongo/db/exec/sbe_cmd.cpp b/src/mongo/db/exec/sbe_cmd.cpp
index de2d5131932..b191553a85f 100644
--- a/src/mongo/db/exec/sbe_cmd.cpp
+++ b/src/mongo/db/exec/sbe_cmd.cpp
@@ -120,6 +120,7 @@ public:
         }
 
         root->attachToOperationContext(opCtx);
+        root->prepare(data.ctx);
         exec = uassertStatusOK(
             plan_executor_factory::make(opCtx,
                                         std::move(cq),
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index 741521b99d7..c9436df3505 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -91,7 +91,6 @@
 #include "mongo/db/query/sbe_multi_planner.h"
 #include "mongo/db/query/sbe_sub_planner.h"
 #include "mongo/db/query/sbe_utils.h"
-#include "mongo/db/query/shard_filterer_factory_impl.h"
 #include "mongo/db/query/stage_builder_util.h"
 #include "mongo/db/query/util/make_data_structure.h"
 #include "mongo/db/query/wildcard_multikey_paths.h"
@@ -450,33 +449,6 @@ bool shouldWaitForOplogVisibility(OperationContext* opCtx,
     return repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesForDatabase(opCtx, "admin");
 }
 
-void prepareExecutionTree(OperationContext* opCtx,
-                          const CollectionPtr& collection,
-                          const CanonicalQuery& cq,
-                          stage_builder::PlanStageData* stageData) {
-    tassert(6183502, "PlanStageData should not be null", stageData);
-    // Populate/renew "shardFilterer" if there exists a "shardFilterer" slot. The slot value
-    // should have been reset to "Nothing" on caching.
-    //
-    // TODO SERVER-61422: Populate the "shardFilterer" when preparing SBE plan.
-    if (auto shardFiltererSlot = stageData->env->getSlotIfExists("shardFilterer"_sd)) {
-        tassert(6108307,
-                "Setting shard filterer slot on un-sharded collection",
-                collection.isSharded());
-
-        auto shardFiltererFactory = std::make_unique<ShardFiltererFactoryImpl>(collection);
-        auto shardFilterer = shardFiltererFactory->makeShardFilterer(opCtx);
-        stageData->env->resetSlot(*shardFiltererSlot,
-                                  sbe::value::TypeTags::shardFilterer,
-                                  sbe::value::bitcastFrom<ShardFilterer*>(shardFilterer.release()),
-                                  true);
-    }
-
-    // If the cached plan is parameterized, bind new values for the parameters into the runtime
-    // environment.
-    input_params::bind(cq, stageData->inputParamToSlotMap, stageData->env);
-}
-
 namespace {
 /**
  * A class to hold the result of preparation of the query to be executed using classic engine. This
@@ -1098,22 +1070,6 @@ protected:
         auto stageData = std::move(cachedPlan->planStageData);
         stageData.debugInfo = std::move(cacheEntry->debugInfo);
 
-        root->attachToOperationContext(_opCtx);
-        root->attachNewYieldPolicy(_yieldPolicy);
-
-        auto expCtx = _cq->getExpCtxRaw();
-        tassert(5968200, "No expression context", expCtx);
-        if (expCtx->explain || expCtx->mayDbProfile) {
-            root->markShouldCollectTimingInfo();
-        }
-
-        // Register this plan to yield according to the configured policy.
-        auto sbeYieldPolicy = dynamic_cast<PlanYieldPolicySBE*>(_yieldPolicy);
-        invariant(sbeYieldPolicy);
-        sbeYieldPolicy->registerPlan(root.get());
-
-        prepareExecutionTree(_opCtx, _collections.getMainCollection(), *_cq, &stageData);
-
         auto result = makeResult();
         result->setDecisionWorks(cacheEntry->decisionWorks);
         result->setRecoveredPinnedCacheEntry(cacheEntry->isPinned());
@@ -1358,11 +1314,16 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
             fillOutSecondaryCollectionsInformation(opCtx, collections, cq.get()));
         roots[0] = helper.buildExecutableTree(*(solutions[0]));
     }
+    auto&& [root, data] = roots[0];
     if (!planningResult->recoveredPinnedCacheEntry()) {
-        auto&& [root, data] = roots[0];
         plan_cache_util::updatePlanCache(
             opCtx, collections.getMainCollection(), *cq, *solutions[0], *root, data);
     }
+
+    // Prepare the SBE tree for execution.
+    stage_builder::prepareSlotBasedExecutableTree(
+        opCtx, root.get(), &data, *cq, *mainColl, yieldPolicy.get());
+
     return plan_executor_factory::make(opCtx,
                                        std::move(cq),
                                        std::move(solutions[0]),
diff --git a/src/mongo/db/query/get_executor.h b/src/mongo/db/query/get_executor.h
index 89818bca9e5..a3d49f83dca 100644
--- a/src/mongo/db/query/get_executor.h
+++ b/src/mongo/db/query/get_executor.h
@@ -111,15 +111,6 @@ void fillOutPlannerParams(OperationContext* opCtx,
                           QueryPlannerParams* plannerParams);
 
 /**
- * Prepare the "PlanStageData" for execution by populating necessary value to "RuntimeEnvironment"
- * and binding parameterized values with corresponding slots.
- */
-void prepareExecutionTree(OperationContext* opCtx,
-                          const CollectionPtr& collection,
-                          const CanonicalQuery& cq,
-                          stage_builder::PlanStageData* data);
-
-/**
  * Return whether or not any component of the path 'path' is multikey given an index key pattern
  * and multikeypaths. If no multikey metdata is available for the index, and the index is marked
  * multikey, conservatively assumes that a component of 'path' _is_ multikey. The 'isMultikey'
diff --git a/src/mongo/db/query/plan_executor_factory.cpp b/src/mongo/db/query/plan_executor_factory.cpp
index 288b4a85571..9746fe72375 100644
--- a/src/mongo/db/query/plan_executor_factory.cpp
+++ b/src/mongo/db/query/plan_executor_factory.cpp
@@ -135,8 +135,6 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> make(
                 "slots"_attr = data.debugString(),
                 "stages"_attr = sbe::DebugPrinter{}.print(*rootStage));
 
-    rootStage->prepare(data.ctx);
-
     return {{new PlanExecutorSBE(
                  opCtx,
                  std::move(cq),
diff --git a/src/mongo/db/query/plan_ranker.h b/src/mongo/db/query/plan_ranker.h
index 4389b0b1fec..1d47608f802 100644
--- a/src/mongo/db/query/plan_ranker.h
+++ b/src/mongo/db/query/plan_ranker.h
@@ -191,6 +191,11 @@ struct BaseCandidatePlan {
     Status status{Status::OK()};
     // Any results produced during the plan's execution prior to scoring are retained here.
     std::queue<ResultType> results;
+    // This is used to track the original plan with clean PlanStage tree and the auxiliary data.
+    // The 'root' and 'data' in this struct could be used to execute trials in multi-planner before
+    // caching the winning plan, which requires necessary values bound to 'data'. These values
+    // should not be stored in the plan cache.
+    boost::optional<std::pair<PlanStageType, Data>> clonedPlan;
 };
 
 using CandidatePlan = BaseCandidatePlan<PlanStage*, WorkingSetID, WorkingSet*>;
diff --git a/src/mongo/db/query/sbe_multi_planner.cpp b/src/mongo/db/query/sbe_multi_planner.cpp
index 7e7cab6ec32..8a53a3fb952 100644
--- a/src/mongo/db/query/sbe_multi_planner.cpp
+++ b/src/mongo/db/query/sbe_multi_planner.cpp
@@ -100,20 +100,36 @@ CandidatePlans MultiPlanner::finalizeExecutionPlans(
         candidates[planIdx].root->close();
     }
 
-    // If the winning stage has exited early but has not fetched all results, clear the results
-    // queue and reopen the plan stage tree, as we cannot resume such execution tree from where
-    // the trial run has stopped, and, as a result, we cannot stash the results returned so far
-    // in the plan executor.
-    if (!winner.root->getCommonStats()->isEOF && winner.exitedEarly) {
+    // An SBE tree that exited early by throwing an exception cannot be reused by design. To work
+    // around this limitation, we clone the tree from the original tree. If there is a pipeline in
+    // "_cq" the winning candidate will be extended by building a new SBE tree below, so we don't
+    // need to clone a new copy here if the winner exited early.
+    if (winner.exitedEarly && _cq.pipeline().empty()) {
+        // Remove all the registered plans from _yieldPolicy's list of trees.
+        _yieldPolicy->clearRegisteredPlans();
+
+        tassert(6142204,
+                "The winning CandidatePlan should contain the original plan",
+                winner.clonedPlan);
+        // Clone a new copy of the original plan to use for execution so that the 'clonedPlan' in
+        // 'winner' can be inserted into the plan cache while in a clean state.
+        winner.data = stage_builder::PlanStageData(winner.clonedPlan->second);
+        // When we clone the tree below, the new tree's stats will be zeroed out. If this is an
+        // explain operation, save the stats from the old tree before we discard it.
         if (_cq.getExplain()) {
-            // We save the stats on early exit if it's either an explain operation, as closing and
-            // re-opening the winning plan (below) changes the stats.
             winner.data.savedStatsOnEarlyExit = winner.root->getStats(true /* includeDebugInfo  */);
         }
-        winner.root->close();
-        winner.root->open(false);
+        winner.root = winner.clonedPlan->first->clone();
+
+        stage_builder::prepareSlotBasedExecutableTree(_opCtx,
+                                                      winner.root.get(),
+                                                      &winner.data,
+                                                      _cq,
+                                                      _collections.getMainCollection(),
+                                                      _yieldPolicy);
         // Clear the results queue.
-        winner.results = decltype(winner.results){};
+        winner.results = {};
+        winner.root->open(false);
     }
 
     // Writes a cache entry for the winning plan to the plan cache if possible.
@@ -135,7 +151,8 @@ CandidatePlans MultiPlanner::finalizeExecutionPlans(
             _cq, std::move(winner.solution), _queryParams.secondaryCollectionsInfo);
         auto [rootStage, data] = stage_builder::buildSlotBasedExecutableTree(
             _opCtx, _collections, _cq, *solution, _yieldPolicy);
-        rootStage->prepare(data.ctx);
+        stage_builder::prepareSlotBasedExecutableTree(
+            _opCtx, rootStage.get(), &data, _cq, _collections.getMainCollection(), _yieldPolicy);
         candidates[winnerIdx] = sbe::plan_ranker::CandidatePlan{
             std::move(solution), std::move(rootStage), std::move(data)};
         candidates[winnerIdx].root->open(false);
diff --git a/src/mongo/db/query/sbe_runtime_planner.cpp b/src/mongo/db/query/sbe_runtime_planner.cpp
index 09e70caf434..f92a7a0dcda 100644
--- a/src/mongo/db/query/sbe_runtime_planner.cpp
+++ b/src/mongo/db/query/sbe_runtime_planner.cpp
@@ -91,7 +91,8 @@ BaseRuntimePlanner::prepareExecutionPlan(PlanStage* root,
     invariant(root);
     invariant(data);
 
-    root->prepare(data->ctx);
+    stage_builder::prepareSlotBasedExecutableTree(
+        _opCtx, root, data, _cq, _collections.getMainCollection(), _yieldPolicy);
 
     value::SlotAccessor* resultSlot{nullptr};
     if (auto slot = data->outputs.getIfExists(stage_builder::PlanStageSlots::kResult); slot) {
@@ -187,6 +188,11 @@ std::vector<plan_ranker::CandidatePlan> BaseRuntimePlanner::collectExecutionStat
         for (auto planIndex : planIndexes) {
             // Prepare the plan.
             auto&& [root, data] = roots[planIndex];
+            // Make a copy of the original plan. This pristine copy will be inserted into the plan
+            // cache if this candidate becomes the winner.
+            auto origPlan =
+                std::make_pair<std::unique_ptr<PlanStage>, stage_builder::PlanStageData>(
+                    root->clone(), stage_builder::PlanStageData(data));
 
             // Attach a unique TrialRunTracker to the plan, which is configured to use at most
             // 'maxNumReads' reads.
@@ -200,6 +206,8 @@ std::vector<plan_ranker::CandidatePlan> BaseRuntimePlanner::collectExecutionStat
                                   false /* exitedEarly */,
                                   Status::OK()});
             auto& currentCandidate = candidates.back();
+            // Store the original plan in the CandidatePlan.
+            currentCandidate.clonedPlan.emplace(std::move(origPlan));
             executeCandidateTrial(&currentCandidate, maxNumResults);
 
             // Reduce the number of reads the next candidates are allocated if this candidate is
diff --git a/src/mongo/db/query/sbe_stage_builder.cpp b/src/mongo/db/query/sbe_stage_builder.cpp
index 1eb9cb674b1..4080bedda08 100644
--- a/src/mongo/db/query/sbe_stage_builder.cpp
+++ b/src/mongo/db/query/sbe_stage_builder.cpp
@@ -62,6 +62,7 @@
 #include "mongo/db/pipeline/abt/field_map_builder.h"
 #include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/expression_visitor.h"
+#include "mongo/db/query/bind_input_params.h"
 #include "mongo/db/query/expression_walker.h"
 #include "mongo/db/query/optimizer/rewrites/const_eval.h"
 #include "mongo/db/query/optimizer/rewrites/path_lower.h"
@@ -71,6 +72,7 @@
 #include "mongo/db/query/sbe_stage_builder_filter.h"
 #include "mongo/db/query/sbe_stage_builder_index_scan.h"
 #include "mongo/db/query/sbe_stage_builder_projection.h"
+#include "mongo/db/query/shard_filterer_factory_impl.h"
 #include "mongo/db/query/util/make_data_structure.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/storage/execution_context.h"
@@ -391,6 +393,51 @@ std::unique_ptr<sbe::RuntimeEnvironment> makeRuntimeEnvironment(
     return env;
 }
 
+void prepareSlotBasedExecutableTree(OperationContext* opCtx,
+                                    sbe::PlanStage* root,
+                                    PlanStageData* data,
+                                    const CanonicalQuery& cq,
+                                    const CollectionPtr& collection,
+                                    PlanYieldPolicySBE* yieldPolicy) {
+    tassert(6183502, "PlanStage cannot be null", root);
+    tassert(6142205, "PlanStageData cannot be null", data);
+    tassert(6142206, "yieldPolicy cannot be null", yieldPolicy);
+
+    root->attachToOperationContext(opCtx);
+    root->attachNewYieldPolicy(yieldPolicy);
+
+    // Call markShouldCollectTimingInfo() if appropriate.
+    auto expCtx = cq.getExpCtxRaw();
+    tassert(6142207, "No expression context", expCtx);
+    if (expCtx->explain || expCtx->mayDbProfile) {
+        root->markShouldCollectTimingInfo();
+    }
+
+    // Register this plan to yield according to the configured policy.
+    yieldPolicy->registerPlan(root);
+
+    root->prepare(data->ctx);
+
+    // Populate/renew "shardFilterer" if there exists a "shardFilterer" slot. The slot value should
+    // be set to Nothing in the plan cache to avoid extending the lifetime of the ownership filter.
+    if (auto shardFiltererSlot = data->env->getSlotIfExists("shardFilterer"_sd)) {
+        tassert(6108307,
+                "Setting shard filterer slot on un-sharded collection",
+                collection.isSharded());
+
+        auto shardFiltererFactory = std::make_unique<ShardFiltererFactoryImpl>(collection);
+        auto shardFilterer = shardFiltererFactory->makeShardFilterer(opCtx);
+        data->env->resetSlot(*shardFiltererSlot,
+                             sbe::value::TypeTags::shardFilterer,
+                             sbe::value::bitcastFrom<ShardFilterer*>(shardFilterer.release()),
+                             true);
+    }
+
+    // If the cached plan is parameterized, bind new values for the parameters into the runtime
+    // environment.
+    input_params::bind(cq, data->inputParamToSlotMap, data->env);
+}
+
 PlanStageSlots::PlanStageSlots(const PlanStageReqs& reqs,
                                sbe::value::SlotIdGenerator* slotIdGenerator) {
     for (auto&& [slotName, isRequired] : reqs._slots) {
@@ -652,16 +699,14 @@ SlotBasedStageBuilder::SlotBasedStageBuilder(OperationContext* opCtx,
                                              const MultipleCollectionAccessor& collections,
                                              const CanonicalQuery& cq,
                                              const QuerySolution& solution,
-                                             PlanYieldPolicySBE* yieldPolicy,
-                                             ShardFiltererFactoryInterface* shardFiltererFactory)
+                                             PlanYieldPolicySBE* yieldPolicy)
     : StageBuilder(opCtx, cq, solution),
       _collections(collections),
       _mainNss(cq.nss()),
       _yieldPolicy(yieldPolicy),
       _data(makeRuntimeEnvironment(_cq, _opCtx, &_slotIdGenerator)),
-      _shardFiltererFactory(shardFiltererFactory),
       _state(_opCtx,
-             _data.env,
+             &_data,
              _cq.getExpCtxRaw()->variables,
              &_slotIdGenerator,
              &_frameIdGenerator,
@@ -2633,7 +2678,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     auto groupEvalStage = makeHashAgg(std::move(accProjEvalStage),
                                       dedupedGroupBySlots,
                                       std::move(accSlotToExprMap),
-                                      _state.env->getSlotIfExists("collator"_sd),
+                                      _state.data->env->getSlotIfExists("collator"_sd),
                                       _cq.getExpCtx()->allowDiskUse,
                                       nodeId);
 
@@ -2926,15 +2971,12 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     // there are orphaned documents from aborted migrations. To check if the document is owned by
     // the shard, we need to own a 'ShardFilterer', and extract the document's shard key as a
     // BSONObj.
-    // TODO SERVER-61422: Should not construct the "ShardFilterer" here.
-    auto shardFilterer = _shardFiltererFactory->makeShardFilterer(_opCtx);
-    auto shardKeyPattern = shardFilterer->getKeyPattern().toBSON();
-    auto shardFiltererSlot =
-        _data.env->registerSlot("shardFilterer"_sd,
-                                sbe::value::TypeTags::shardFilterer,
-                                sbe::value::bitcastFrom<ShardFilterer*>(shardFilterer.release()),
-                                true,
-                                &_slotIdGenerator);
+    auto shardKeyPattern = _collections.getMainCollection().getShardKeyPattern();
+    // We register the "shardFilterer" slot but not construct the ShardFilterer here is because once
+    // constructed the ShardFilterer will prevent orphaned documents from being deleted. We will
+    // construct the 'ShardFiltered' later while preparing the SBE tree for execution.
+    auto shardFiltererSlot = _data.env->registerSlot(
+        "shardFilterer"_sd, sbe::value::TypeTags::Nothing, 0, false, &_slotIdGenerator);
 
     // Determine if our child is an index scan and extract it's key pattern, or empty BSONObj if our
     // child is not an IXSCAN node.
diff --git a/src/mongo/db/query/sbe_stage_builder.h b/src/mongo/db/query/sbe_stage_builder.h
index 07487ebe991..a7539495982 100644
--- a/src/mongo/db/query/sbe_stage_builder.h
+++ b/src/mongo/db/query/sbe_stage_builder.h
@@ -50,6 +50,18 @@ std::unique_ptr<sbe::RuntimeEnvironment> makeRuntimeEnvironment(
     OperationContext* opCtx,
     sbe::value::SlotIdGenerator* slotIdGenerator);
 
+/**
+ * This function prepares the SBE tree for execution, such as attaching the OperationContext,
+ * ensuring that the SBE tree is registered with the PlanYieldPolicySBE and populating the
+ * "RuntimeEnvironment".
+ */
+void prepareSlotBasedExecutableTree(OperationContext* opCtx,
+                                    sbe::PlanStage* root,
+                                    PlanStageData* data,
+                                    const CanonicalQuery& cq,
+                                    const CollectionPtr& collection,
+                                    PlanYieldPolicySBE* yieldPolicy);
+
 class PlanStageReqs;
 
 /**
@@ -342,8 +354,7 @@ public:
                           const MultipleCollectionAccessor& collections,
                           const CanonicalQuery& cq,
                           const QuerySolution& solution,
-                          PlanYieldPolicySBE* yieldPolicy,
-                          ShardFiltererFactoryInterface* shardFilterer);
+                          PlanYieldPolicySBE* yieldPolicy);
 
     std::unique_ptr<sbe::PlanStage> build(const QuerySolutionNode* root) final;
 
@@ -476,9 +487,6 @@ private:
     bool _buildHasStarted{false};
     bool _shouldProduceRecordIdSlot{true};
 
-    // A factory to construct shard filters.
-    ShardFiltererFactoryInterface* _shardFiltererFactory;
-
     // Common parameters to SBE stage builder functions.
     StageBuilderState _state;
 };
diff --git a/src/mongo/db/query/sbe_stage_builder_accumulator.cpp b/src/mongo/db/query/sbe_stage_builder_accumulator.cpp
index f24668fb1f5..2cb8afed181 100644
--- a/src/mongo/db/query/sbe_stage_builder_accumulator.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_accumulator.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/pipeline/accumulator.h"
 #include "mongo/db/pipeline/accumulator_for_window_functions.h"
 #include "mongo/db/pipeline/accumulator_js_reduce.h"
+#include "mongo/db/query/sbe_stage_builder.h"
 #include "mongo/db/query/sbe_stage_builder_accumulator.h"
 #include "mongo/db/query/sbe_stage_builder_expression.h"
 #include "mongo/db/query/sbe_stage_builder_helpers.h"
@@ -59,7 +60,7 @@ std::pair<std::vector<std::unique_ptr<sbe::EExpression>>, EvalStage> buildAccumu
     EvalStage inputStage,
     PlanNodeId planNodeId) {
     std::vector<std::unique_ptr<sbe::EExpression>> aggs;
-    auto collatorSlot = state.env->getSlotIfExists("collator"_sd);
+    auto collatorSlot = state.data->env->getSlotIfExists("collator"_sd);
     if (collatorSlot) {
         aggs.push_back(makeFunction("collMin"_sd,
                                     sbe::makeE<sbe::EVariable>(*collatorSlot),
@@ -93,7 +94,7 @@ std::pair<std::vector<std::unique_ptr<sbe::EExpression>>, EvalStage> buildAccumu
     EvalStage inputStage,
     PlanNodeId planNodeId) {
     std::vector<std::unique_ptr<sbe::EExpression>> aggs;
-    auto collatorSlot = state.env->getSlotIfExists("collator"_sd);
+    auto collatorSlot = state.data->env->getSlotIfExists("collator"_sd);
     if (collatorSlot) {
         aggs.push_back(makeFunction("collMax"_sd,
                                     sbe::makeE<sbe::EVariable>(*collatorSlot),
@@ -309,7 +310,7 @@ std::pair<std::vector<std::unique_ptr<sbe::EExpression>>, EvalStage> buildAccumu
     EvalStage inputStage,
     PlanNodeId planNodeId) {
     std::vector<std::unique_ptr<sbe::EExpression>> aggs;
-    auto collatorSlot = state.env->getSlotIfExists("collator"_sd);
+    auto collatorSlot = state.data->env->getSlotIfExists("collator"_sd);
     if (collatorSlot) {
         aggs.push_back(makeFunction(
             "collAddToSet"_sd, sbe::makeE<sbe::EVariable>(*collatorSlot), std::move(arg)));
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 9ec9bb1ace3..18f4857bbaf 100644
--- a/src/mongo/db/query/sbe_stage_builder_coll_scan.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_coll_scan.cpp
@@ -266,7 +266,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateOptimizedOplo
     auto [seekRecordIdSlot, seekRecordIdExpression] =
         [&]() -> std::pair<boost::optional<sbe::value::SlotId>, std::unique_ptr<sbe::EExpression>> {
         if (isTailableResumeBranch) {
-            auto resumeRecordIdSlot = state.env->getSlot("resumeRecordId"_sd);
+            auto resumeRecordIdSlot = state.data->env->getSlot("resumeRecordId"_sd);
             return {resumeRecordIdSlot, makeVariable(resumeRecordIdSlot)};
         } else if (csn->resumeAfterRecordId) {
             auto [tag, val] = sbe::value::makeCopyRecordId(*csn->resumeAfterRecordId);
@@ -289,7 +289,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateOptimizedOplo
     const auto shouldTrackLatestOplogTimestamp =
         (csn->maxRecord || csn->shouldTrackLatestOplogTimestamp);
     auto&& [fields, slots, tsSlot] = makeOplogTimestampSlotsIfNeeded(
-        state.env, state.slotIdGenerator, shouldTrackLatestOplogTimestamp);
+        state.data->env, state.slotIdGenerator, shouldTrackLatestOplogTimestamp);
 
     sbe::ScanCallbacks callbacks({}, {}, makeOpenCallbackIfNeeded(collection, csn));
     auto stage = sbe::makeS<sbe::ScanStage>(collection->uuid(),
@@ -490,7 +490,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateOptimizedOplo
             recordIdSlot = state.slotId();
 
             std::tie(fields, slots, tsSlot) = makeOplogTimestampSlotsIfNeeded(
-                state.env, state.slotIdGenerator, shouldTrackLatestOplogTimestamp);
+                state.data->env, state.slotIdGenerator, shouldTrackLatestOplogTimestamp);
 
             stage = sbe::makeS<sbe::LoopJoinStage>(
                 sbe::makeS<sbe::LimitSkipStage>(std::move(stage), 1, boost::none, csn->nodeId()),
@@ -554,7 +554,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollSc
             auto [tag, val] = sbe::value::makeCopyRecordId(*csn->resumeAfterRecordId);
             return {state.slotId(), makeConstant(tag, val)};
         } else if (isTailableResumeBranch) {
-            auto resumeRecordIdSlot = state.env->getSlot("resumeRecordId"_sd);
+            auto resumeRecordIdSlot = state.data->env->getSlot("resumeRecordId"_sd);
             return {resumeRecordIdSlot, makeVariable(resumeRecordIdSlot)};
         }
         return {};
@@ -562,7 +562,7 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateGenericCollSc
 
     // See if we need to project out an oplog latest timestamp.
     auto&& [fields, slots, tsSlot] = makeOplogTimestampSlotsIfNeeded(
-        state.env, state.slotIdGenerator, csn->shouldTrackLatestOplogTimestamp);
+        state.data->env, state.slotIdGenerator, csn->shouldTrackLatestOplogTimestamp);
 
     sbe::ScanCallbacks callbacks({}, {}, makeOpenCallbackIfNeeded(collection, csn));
     auto stage = sbe::makeS<sbe::ScanStage>(collection->uuid(),
diff --git a/src/mongo/db/query/sbe_stage_builder_expression.cpp b/src/mongo/db/query/sbe_stage_builder_expression.cpp
index 81a7da56bb2..93fc8eacb72 100644
--- a/src/mongo/db/query/sbe_stage_builder_expression.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_expression.cpp
@@ -50,6 +50,7 @@
 #include "mongo/db/pipeline/expression_walker.h"
 #include "mongo/db/query/expression_walker.h"
 #include "mongo/db/query/projection_parser.h"
+#include "mongo/db/query/sbe_stage_builder.h"
 #include "mongo/db/query/sbe_stage_builder_eval_frame.h"
 #include "mongo/util/str.h"
 
@@ -251,7 +252,11 @@ void generateStringCaseConversionExpression(ExpressionVisitorContext* _context,
                          getBSONTypeMask(sbe::value::TypeTags::NumberDecimal) |
                          getBSONTypeMask(sbe::value::TypeTags::Date) |
                          getBSONTypeMask(sbe::value::TypeTags::Timestamp));
-    auto checkValidTypeExpr = sbe::makeE<sbe::ETypeMatch>(inputRef.clone(), typeMask);
+    auto checkValidTypeExpr =
+        makeFunction("typeMatch",
+                     inputRef.clone(),
+                     makeConstant(sbe::value::TypeTags::NumberInt64,
+                                  sbe::value::bitcastFrom<int64_t>(typeMask)));
     auto checkNullorMissing = generateNullOrMissing(inputRef);
     auto [emptyStrTag, emptyStrVal] = sbe::value::makeNewString("");
 
@@ -1071,7 +1076,7 @@ public:
         // comparisons (for example, a number will always compare as less than a string). The other
         // comparison primitives are designed for comparing values of the same type.
         auto cmp3w = makeBinaryOp(
-            sbe::EPrimBinary::cmp3w, lhsRef.clone(), rhsRef.clone(), _context->state.env);
+            sbe::EPrimBinary::cmp3w, lhsRef.clone(), rhsRef.clone(), _context->state.data->env);
         auto cmp = (comparisonOperator == sbe::EPrimBinary::cmp3w)
             ? std::move(cmp3w)
             : makeBinaryOp(comparisonOperator,
@@ -1086,10 +1091,14 @@ public:
         // We also need to explicitly check for 'bsonUndefined' type because it is considered equal
         // to "Nothing" according to MQL semantics.
         auto generateExists = [&](const sbe::EVariable& var) {
+            auto undefinedTypeMask = static_cast<int64_t>(getBSONTypeMask(BSONType::Undefined));
             return makeBinaryOp(
                 sbe::EPrimBinary::logicAnd,
                 makeFunction("exists", var.clone()),
-                sbe::makeE<sbe::ETypeMatch>(var.clone(), ~getBSONTypeMask(BSONType::Undefined)));
+                makeFunction("typeMatch",
+                             var.clone(),
+                             makeConstant(sbe::value::TypeTags::NumberInt64,
+                                          sbe::value::bitcastFrom<int64_t>(~undefinedTypeMask))));
         };
 
         auto nothingFallbackCmp =
@@ -1197,7 +1206,7 @@ public:
                                     sbe::makeSV(unionOutputSlot),
                                     _context->planNodeId);
 
-        auto collatorSlot = _context->state.env->getSlotIfExists("collator"_sd);
+        auto collatorSlot = _context->state.data->env->getSlotIfExists("collator"_sd);
 
         // Build a filter that will throw an 'EFail' if any element coming from the union is NOT
         // an array.
@@ -1297,7 +1306,7 @@ public:
         auto endDateExpression = _context->popExpr();
         auto startDateExpression = _context->popExpr();
 
-        auto timezoneDBSlot = _context->state.env->getSlot("timeZoneDB"_sd);
+        auto timezoneDBSlot = _context->state.data->env->getSlot("timeZoneDB"_sd);
 
         //  Set parameters for an invocation of built-in "dateDiff" function.
         arguments.push_back(sbe::makeE<sbe::EVariable>(timezoneDBSlot));
@@ -1666,7 +1675,7 @@ public:
         // for datetime computation. This global object is registered as an unowned value in the
         // runtime environment so we pass the corresponding slot to the datePartsWeekYear and
         // dateParts functions as a variable.
-        auto timeZoneDBSlot = _context->state.env->getSlot("timeZoneDB"_sd);
+        auto timeZoneDBSlot = _context->state.data->env->getSlot("timeZoneDB"_sd);
         auto computeDate = makeFunction(isIsoWeekYear ? "datePartsWeekYear" : "dateParts",
                                         sbe::makeE<sbe::EVariable>(timeZoneDBSlot),
                                         yearRef.clone(),
@@ -1734,9 +1743,10 @@ public:
         }
 
         // Add timezoneDB to arguments.
-        args.push_back(sbe::makeE<sbe::EVariable>(_context->state.env->getSlot("timeZoneDB"_sd)));
+        args.push_back(
+            sbe::makeE<sbe::EVariable>(_context->state.data->env->getSlot("timeZoneDB"_sd)));
         isoargs.push_back(
-            sbe::makeE<sbe::EVariable>(_context->state.env->getSlot("timeZoneDB"_sd)));
+            sbe::makeE<sbe::EVariable>(_context->state.data->env->getSlot("timeZoneDB"_sd)));
 
         // Add date to arguments.
         operands.push_back(std::move(date));
@@ -1770,21 +1780,30 @@ public:
             CaseValuePair{makeNot(makeFunction("isString", timezoneRef.clone())),
                           sbe::makeE<sbe::EFail>(ErrorCodes::Error{4997701},
                                                  "$dateToParts timezone must be a string")},
-            CaseValuePair{makeNot(makeFunction("isTimezone",
-                                               sbe::makeE<sbe::EVariable>(
-                                                   _context->state.env->getSlot("timeZoneDB"_sd)),
-                                               timezoneRef.clone())),
-                          sbe::makeE<sbe::EFail>(ErrorCodes::Error{4997704},
-                                                 "$dateToParts timezone must be a valid timezone")},
+            CaseValuePair{
+                makeNot(makeFunction(
+                    "isTimezone",
+                    sbe::makeE<sbe::EVariable>(_context->state.data->env->getSlot("timeZoneDB"_sd)),
+                    timezoneRef.clone())),
+                sbe::makeE<sbe::EFail>(ErrorCodes::Error{4997704},
+                                       "$dateToParts timezone must be a valid timezone")},
             CaseValuePair{generateNullOrMissing(frameId, 2),
                           sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::Null, 0)},
-            CaseValuePair{makeNot(sbe::makeE<sbe::ETypeMatch>(isoflagRef.clone(), isoTypeMask)),
-                          sbe::makeE<sbe::EFail>(ErrorCodes::Error{4997702},
-                                                 "$dateToParts iso8601 must be a boolean")},
+            CaseValuePair{
+                makeNot(makeFunction("typeMatch",
+                                     isoflagRef.clone(),
+                                     makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                  sbe::value::bitcastFrom<int64_t>(isoTypeMask)))),
+                sbe::makeE<sbe::EFail>(ErrorCodes::Error{4997702},
+                                       "$dateToParts iso8601 must be a boolean")},
             CaseValuePair{generateNullOrMissing(frameId, 0),
                           sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::Null, 0)},
             CaseValuePair{
-                makeNot(sbe::makeE<sbe::ETypeMatch>(dateRef.clone(), dateTypeMask())),
+                makeNot(makeFunction(
+                    "typeMatch",
+                    dateRef.clone(),
+                    makeConstant(sbe::value::TypeTags::NumberInt64,
+                                 sbe::value::bitcastFrom<int64_t>(dateTypeMask())))),
                 sbe::makeE<sbe::EFail>(ErrorCodes::Error{4997703},
                                        "$dateToParts date must have the format of a date")},
             std::move(checkIsoflagValue));
@@ -1879,7 +1898,7 @@ public:
                         "Encountered unexpected system variable ID",
                         it != Variables::kIdToBuiltinVarName.end());
 
-                auto variableSlot = _context->state.env->getSlotIfExists(it->second);
+                auto variableSlot = _context->state.data->env->getSlotIfExists(it->second);
                 uassert(5611301,
                         str::stream()
                             << "Builtin variable '$$" << it->second << "' is not available",
@@ -2316,10 +2335,17 @@ public:
             CaseValuePair{
                 makeBinaryOp(
                     sbe::EPrimBinary::logicAnd,
-                    sbe::makeE<sbe::ETypeMatch>(
-                        rhsVar.clone(), getBSONTypeMask(sbe::value::TypeTags::NumberDouble)),
-                    makeNot(sbe::makeE<sbe::ETypeMatch>(
-                        lhsVar.clone(), getBSONTypeMask(sbe::value::TypeTags::NumberDouble)))),
+                    makeFunction("typeMatch",
+                                 rhsVar.clone(),
+                                 makeConstant(sbe::value::TypeTags::NumberInt64,
+                                              sbe::value::bitcastFrom<int64_t>(getBSONTypeMask(
+                                                  sbe::value::TypeTags::NumberDouble)))),
+                    makeNot(
+                        makeFunction("typeMatch",
+                                     lhsVar.clone(),
+                                     makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                  sbe::value::bitcastFrom<int64_t>(getBSONTypeMask(
+                                                      sbe::value::TypeTags::NumberDouble)))))),
                 makeFunction("fillEmpty", std::move(numericConvert32), rhsVar.clone())},
             rhsVar.clone());
 
@@ -2582,7 +2608,7 @@ public:
         auto isEmptyFindStr = makeBinaryOp(sbe::EPrimBinary::eq,
                                            findRef.clone(),
                                            sbe::makeE<sbe::EConstant>(emptyStrTag, emptyStrVal),
-                                           _context->state.env);
+                                           _context->state.data->env);
 
         auto replaceOrReturnInputExpr = sbe::makeE<sbe::EIf>(
             std::move(isEmptyFindStr),
@@ -2660,7 +2686,7 @@ public:
         auto [specTag, specVal] = makeValue(expr->getSortPattern());
         auto specConstant = makeConstant(specTag, specVal);
 
-        auto collatorSlot = _context->state.env->getSlotIfExists("collator"_sd);
+        auto collatorSlot = _context->state.data->env->getSlotIfExists("collator"_sd);
 
         auto argumentIsNotArray = makeNot(makeFunction("isArray", inputRef.clone()));
         auto exprSortArr = buildMultiBranchConditional(
@@ -2730,7 +2756,7 @@ public:
             return makeBinaryOp(sbe::EPrimBinary::eq,
                                 var.clone(),
                                 sbe::makeE<sbe::EConstant>(emptyStrTag, emptyStrVal),
-                                _context->state.env);
+                                _context->state.data->env);
         };
 
         auto checkIsNullOrMissing = makeBinaryOp(sbe::EPrimBinary::logicOr,
@@ -3204,7 +3230,7 @@ private:
         }();
         auto date = _context->popExpr();
 
-        auto timeZoneDBSlot = _context->state.env->getSlot("timeZoneDB"_sd);
+        auto timeZoneDBSlot = _context->state.data->env->getSlot("timeZoneDB"_sd);
         args.push_back(sbe::makeE<sbe::EVariable>(timeZoneDBSlot));
 
         // Add date to arguments.
@@ -3231,10 +3257,14 @@ private:
                                                      << " timezone must be a valid timezone")},
             CaseValuePair{generateNullOrMissing(dateRef),
                           sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::Null, 0)},
-            CaseValuePair{makeNot(sbe::makeE<sbe::ETypeMatch>(dateRef.clone(), dateTypeMask())),
-                          sbe::makeE<sbe::EFail>(ErrorCodes::Error{4998202},
-                                                 str::stream()
-                                                     << "$" << exprName.toString()
+            CaseValuePair{
+                makeNot(
+                    makeFunction("typeMatch",
+                                 dateRef.clone(),
+                                 makeConstant(sbe::value::TypeTags::NumberInt64,
+                                              sbe::value::bitcastFrom<int64_t>(dateTypeMask())))),
+                sbe::makeE<sbe::EFail>(ErrorCodes::Error{4998202},
+                                       str::stream() << "$" << exprName.toString()
                                                      << " date must have a format of a date")},
             sbe::makeE<sbe::EFunction>(exprName.toString(), std::move(args)));
         _context->pushExpr(
@@ -3254,11 +3284,15 @@ private:
                                                           ErrorCodes::Error errorCode,
                                                           StringData expressionName,
                                                           StringData parameterName) {
-        return {makeNot(sbe::makeE<sbe::ETypeMatch>(dateRef.clone(), dateTypeMask())),
-                sbe::makeE<sbe::EFail>(errorCode,
-                                       str::stream()
-                                           << expressionName << " parameter '" << parameterName
-                                           << "' must be coercible to date")};
+        return {
+            makeNot(makeFunction("typeMatch",
+                                 dateRef.clone(),
+                                 makeConstant(sbe::value::TypeTags::NumberInt64,
+                                              sbe::value::bitcastFrom<int64_t>(dateTypeMask())))),
+            sbe::makeE<sbe::EFail>(errorCode,
+                                   str::stream()
+                                       << expressionName << " parameter '" << parameterName
+                                       << "' must be coercible to date")};
     }
 
     /**
@@ -3522,7 +3556,7 @@ private:
         checkExprsNull.reserve(arity);
         checkExprsNotArray.reserve(arity);
 
-        auto collatorSlot = _context->state.env->getSlotIfExists("collator"_sd);
+        auto collatorSlot = _context->state.data->env->getSlotIfExists("collator"_sd);
 
         auto [operatorName, setFunctionName] = [setOp, collatorSlot]() {
             switch (setOp) {
@@ -3665,18 +3699,24 @@ private:
             auto patternArgument = buildMultiBranchConditional(
                 CaseValuePair{makeFunction("isString", patternVar.clone()),
                               std::move(patternNullBytesCheck)},
-                CaseValuePair{sbe::makeE<sbe::ETypeMatch>(patternVar.clone(),
-                                                          getBSONTypeMask(BSONType::RegEx)),
+                CaseValuePair{makeFunction("typeMatch",
+                                           patternVar.clone(),
+                                           makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                        sbe::value::bitcastFrom<int64_t>(
+                                                            getBSONTypeMask(BSONType::RegEx)))),
                               makeFunction("getRegexPattern", patternVar.clone())},
                 makeError(5126601, "regex pattern must have either string or BSON RegEx type"));
 
             if (!optionsVar) {
                 // If no options are passed to the expression, try to extract them from the pattern.
-                auto optionsArgument =
-                    sbe::makeE<sbe::EIf>(sbe::makeE<sbe::ETypeMatch>(
-                                             patternVar.clone(), getBSONTypeMask(BSONType::RegEx)),
-                                         makeFunction("getRegexFlags", patternVar.clone()),
-                                         makeConstant(""));
+                auto optionsArgument = sbe::makeE<sbe::EIf>(
+                    makeFunction("typeMatch",
+                                 patternVar.clone(),
+                                 makeConstant(sbe::value::TypeTags::NumberInt64,
+                                              sbe::value::bitcastFrom<int64_t>(
+                                                  getBSONTypeMask(BSONType::RegEx)))),
+                    makeFunction("getRegexFlags", patternVar.clone()),
+                    makeConstant(""));
                 auto compiledRegex = makeFunction(
                     "regexCompile", std::move(patternArgument), std::move(optionsArgument));
                 return sbe::makeE<sbe::EIf>(makeFunction("isNull", patternVar.clone()),
@@ -3745,8 +3785,11 @@ private:
                             makeFunction("getRegexFlags", patternVar.clone()));
 
                         return sbe::makeE<sbe::EIf>(
-                            sbe::makeE<sbe::ETypeMatch>(patternVar.clone(),
-                                                        getBSONTypeMask(BSONType::RegEx)),
+                            makeFunction("typeMatch",
+                                         patternVar.clone(),
+                                         makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                      sbe::value::bitcastFrom<int64_t>(
+                                                          getBSONTypeMask(BSONType::RegEx)))),
                             std::move(checkBsonRegexOptions),
                             stringOptions.clone());
                     },
@@ -3825,7 +3868,7 @@ private:
         binds.push_back(std::move(convertedAmountInt64));
 
         sbe::EExpression::Vector args;
-        auto timeZoneDBSlot = _context->state.env->getSlot("timeZoneDB"_sd);
+        auto timeZoneDBSlot = _context->state.data->env->getSlot("timeZoneDB"_sd);
         args.push_back(sbe::makeE<sbe::EVariable>(timeZoneDBSlot));
         args.push_back(startDateRef.clone());
         args.push_back(unitRef.clone());
@@ -3862,15 +3905,19 @@ private:
                                                  str::stream() << "$" << dateExprName
                                                                << " expects a valid timezone")},
             CaseValuePair{
-                makeNot(sbe::makeE<sbe::ETypeMatch>(startDateRef.clone(), dateTypeMask())),
+                makeNot(
+                    makeFunction("typeMatch",
+                                 startDateRef.clone(),
+                                 makeConstant(sbe::value::TypeTags::NumberInt64,
+                                              sbe::value::bitcastFrom<int64_t>(dateTypeMask())))),
                 sbe::makeE<sbe::EFail>(ErrorCodes::Error{5166603},
                                        str::stream()
                                            << "$" << dateExprName
                                            << " must have startDate argument convertable to date")},
-            CaseValuePair{generateNonStringCheck(unitRef),
-                          sbe::makeE<sbe::EFail>(ErrorCodes::Error{5166604},
-                                                 str::stream()
-                                                     << "$" << dateExprName
+            CaseValuePair{
+                generateNonStringCheck(unitRef),
+                sbe::makeE<sbe::EFail>(ErrorCodes::Error{5166604},
+                                       str::stream() << "$" << dateExprName
                                                      << " expects unit argument of type string")},
             CaseValuePair{makeNot(makeFunction("isTimeUnit", unitRef.clone())),
                           sbe::makeE<sbe::EFail>(ErrorCodes::Error{5166605},
@@ -3899,9 +3946,12 @@ private:
 
 std::unique_ptr<sbe::EExpression> generateCoerceToBoolExpression(sbe::EVariable branchRef) {
     auto makeNotNullOrUndefinedCheck = [&branchRef]() {
-        return makeNot(sbe::makeE<sbe::ETypeMatch>(branchRef.clone(),
-                                                   getBSONTypeMask(BSONType::jstNULL) |
-                                                       getBSONTypeMask(BSONType::Undefined)));
+        return makeNot(makeFunction(
+            "typeMatch",
+            branchRef.clone(),
+            makeConstant(sbe::value::TypeTags::NumberInt64,
+                         sbe::value::bitcastFrom<int64_t>(getBSONTypeMask(BSONType::jstNULL) |
+                                                          getBSONTypeMask(BSONType::Undefined)))));
     };
 
     auto makeNeqFalseCheck = [&branchRef]() {
diff --git a/src/mongo/db/query/sbe_stage_builder_filter.cpp b/src/mongo/db/query/sbe_stage_builder_filter.cpp
index 3fe718c2943..c06f48781d3 100644
--- a/src/mongo/db/query/sbe_stage_builder_filter.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_filter.cpp
@@ -70,6 +70,7 @@
 #include "mongo/db/matcher/schema/expression_internal_schema_root_doc_eq.h"
 #include "mongo/db/matcher/schema/expression_internal_schema_unique_items.h"
 #include "mongo/db/matcher/schema/expression_internal_schema_xor.h"
+#include "mongo/db/query/sbe_stage_builder.h"
 #include "mongo/db/query/sbe_stage_builder_eval_frame.h"
 #include "mongo/db/query/sbe_stage_builder_expression.h"
 #include "mongo/db/query/util/make_data_structure.h"
@@ -624,6 +625,12 @@ void generatePredicate(MatchExpressionVisitorContext* context,
 void generateArraySize(MatchExpressionVisitorContext* context,
                        const SizeMatchExpression* matchExpr) {
     int size = matchExpr->getData();
+    // If there's an "inputParamId" in 'matchExpr' meaning this expr got parameterized, we can
+    // register a SlotId for it and use the slot directly.
+    boost::optional<sbe::value::SlotId> inputParamSlotId;
+    if (auto inputParam = matchExpr->getInputParamId()) {
+        inputParamSlotId = context->state.registerInputParamSlot(*inputParam);
+    }
 
     auto makePredicate = [&](sbe::value::SlotId inputSlot,
                              EvalStage inputStage) -> EvalExprStagePair {
@@ -650,13 +657,15 @@ void generateArraySize(MatchExpressionVisitorContext* context,
                                           context->planNodeId,
                                           1);
 
+        auto sizeExpr = inputParamSlotId ? makeVariable(*inputParamSlotId)
+                                         : makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                        sbe::value::bitcastFrom<int64_t>(size));
         // If the traversal result was not Nothing, compare it to the user provided value. If the
         // traversal result was Nothing, that means the array was empty, so replace Nothing with 0
         // and compare it to the user provided value.
         auto sizeOutput = makeBinaryOp(
             sbe::EPrimBinary::eq,
-            sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::NumberInt64,
-                                       sbe::value::bitcastFrom<int64_t>(size)),
+            std::move(sizeExpr),
             sbe::makeE<sbe::EIf>(makeFunction("exists", sbe::makeE<sbe::EVariable>(traverseSlot)),
                                  sbe::makeE<sbe::EVariable>(traverseSlot),
                                  sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::NumberInt64,
@@ -764,7 +773,7 @@ void generateComparison(MatchExpressionVisitorContext* context,
             return {makeFillEmptyFalse(makeBinaryOp(binaryOp,
                                                     std::move(inputExpr),
                                                     makeConstant(sbe::value::TypeTags::Null, 0),
-                                                    context->state.env)),
+                                                    context->state.data->env)),
                     std::move(inputStage)};
         } else if (sbe::value::isNaN(tagView, valView)) {
             // Construct an expression to perform a NaN check.
@@ -788,12 +797,19 @@ void generateComparison(MatchExpressionVisitorContext* context,
             }
         }
 
-        // SBE EConstant assumes ownership of the value so we have to make a copy here.
-        auto [tag, val] = sbe::value::copyValue(tagView, valView);
-
-        return {makeFillEmptyFalse(makeBinaryOp(
-                    binaryOp, makeVariable(inputSlot), makeConstant(tag, val), context->state.env)),
-                std::move(inputStage)};
+        auto valExpr = [&](sbe::value::TypeTags typeTag,
+                           sbe::value::Value value) -> std::unique_ptr<sbe::EExpression> {
+            if (auto inputParam = expr->getInputParamId()) {
+                return makeVariable(context->state.registerInputParamSlot(*inputParam));
+            }
+            auto [tag, val] = sbe::value::copyValue(typeTag, value);
+            return makeConstant(tag, val);
+        }(tagView, valView);
+
+        return {
+            makeFillEmptyFalse(makeBinaryOp(
+                binaryOp, makeVariable(inputSlot), std::move(valExpr), context->state.data->env)),
+            std::move(inputStage)};
     };
 
     // A 'kArrayAndItsElements' traversal mode matches the following semantics: when the path we are
@@ -831,16 +847,26 @@ void generateAlwaysBoolean(MatchExpressionVisitorContext* context, bool value) {
 void generateBitTest(MatchExpressionVisitorContext* context,
                      const BitTestMatchExpression* expr,
                      const sbe::BitTestBehavior& bitOp) {
-    auto makePredicate = [expr, bitOp](sbe::value::SlotId inputSlot,
-                                       EvalStage inputStage) -> EvalExprStagePair {
-        auto [bitPosTag, bitPosVal] = convertBitTestBitPositions(expr);
+    auto makePredicate = [expr, bitOp, context](sbe::value::SlotId inputSlot,
+                                                EvalStage inputStage) -> EvalExprStagePair {
+        // If there's an "inputParamId" in this expr meaning this expr got parameterized, we can
+        // register a SlotId for it and use the slot directly.
+        std::unique_ptr<sbe::EExpression> bitPosExpr = [&]() -> std::unique_ptr<sbe::EExpression> {
+            if (auto bitPosParamId = expr->getBitPositionsParamId()) {
+                auto bitPosSlotId = context->state.registerInputParamSlot(*bitPosParamId);
+                return makeVariable(bitPosSlotId);
+            } else {
+                auto [bitPosTag, bitPosVal] = convertBitTestBitPositions(expr);
+                return makeConstant(bitPosTag, bitPosVal);
+            }
+        }();
 
         // An EExpression for the BinData and position list for the binary case of
         // BitTestMatchExpressions. This function will be applied to values carrying BinData
         // elements.
         auto binaryBitTestExpr = makeFunction(
             "bitTestPosition"_sd,
-            sbe::makeE<sbe::EConstant>(bitPosTag, bitPosVal),
+            std::move(bitPosExpr),
             makeVariable(inputSlot),
             makeConstant(sbe::value::TypeTags::NumberInt32, static_cast<int32_t>(bitOp)));
 
@@ -862,17 +888,28 @@ void generateBitTest(MatchExpressionVisitorContext* context,
         // documentation. At some point, we should consider removing this call to round() to make
         // SBE's behavior consistent with MongoDB's documentation.
         auto numericBitTestInputExpr = sbe::makeE<sbe::EIf>(
-            sbe::makeE<sbe::ETypeMatch>(makeVariable(inputSlot),
-                                        getBSONTypeMask(sbe::value::TypeTags::NumberDecimal)),
+            makeFunction("typeMatch",
+                         makeVariable(inputSlot),
+                         makeConstant(sbe::value::TypeTags::NumberInt64,
+                                      sbe::value::bitcastFrom<int64_t>(
+                                          getBSONTypeMask(sbe::value::TypeTags::NumberDecimal)))),
             makeFunction("round"_sd, makeVariable(inputSlot)),
             makeVariable(inputSlot));
 
+        std::unique_ptr<sbe::EExpression> bitMaskExpr = [&]() -> std::unique_ptr<sbe::EExpression> {
+            if (auto bitMaskParamId = expr->getBitMaskParamId()) {
+                auto bitMaskSlotId = context->state.registerInputParamSlot(*bitMaskParamId);
+                return makeVariable(bitMaskSlotId);
+            } else {
+                return makeConstant(sbe::value::TypeTags::NumberInt64, expr->getBitMask());
+            }
+        }();
         // Convert the value to a 64-bit integer, and then pass the converted value along with the
         // mask to the appropriate bit-test function. If the value cannot be losslessly converted
         // to a 64-bit integer, this expression will return Nothing.
         auto numericBitTestExpr =
             makeFunction(numericBitTestFnName,
-                         makeConstant(sbe::value::TypeTags::NumberInt64, expr->getBitMask()),
+                         std::move(bitMaskExpr),
                          sbe::makeE<sbe::ENumericConvert>(std::move(numericBitTestInputExpr),
                                                           sbe::value::TypeTags::NumberInt64));
 
@@ -1416,6 +1453,27 @@ public:
     void visit(const GeoNearMatchExpression* expr) final {}
 
     void visit(const InMatchExpression* expr) final {
+        // If there's an "inputParamId" in this expr meaning this expr got parameterized, we can
+        // register a SlotId for it and use the slot directly. Note that we don't auto-parameterize
+        // $in if it contains null, regexes, or nested arrays.
+        if (auto inputParam = expr->getInputParamId()) {
+            auto inputParamSlotId =
+                _context->state.registerInputParamSlot(*expr->getInputParamId());
+            auto makePredicate = [&](sbe::value::SlotId inputSlot,
+                                     EvalStage inputStage) -> EvalExprStagePair {
+                return {makeIsMember(makeVariable(inputSlot),
+                                     makeVariable(inputParamSlotId),
+                                     _context->state.data->env),
+                        std::move(inputStage)};
+            };
+
+            generatePredicate(_context,
+                              expr->fieldRef(),
+                              std::move(makePredicate),
+                              LeafTraversalMode::kArrayElementsOnly);
+            return;
+        }
+
         auto&& [arrSetTag, arrSetVal, hasArray, hasNull] = convertInExpressionEqualities(expr);
         sbe::value::ValueGuard arrSetGuard{arrSetTag, arrSetVal};
 
@@ -1438,7 +1496,7 @@ public:
                 arrSetGuard.reset();
                 return {makeIsMember(std::move(inputExpr),
                                      sbe::makeE<sbe::EConstant>(arrSetTag, arrSetVal),
-                                     _context->state.env),
+                                     _context->state.data->env),
                         std::move(inputStage)};
             };
 
@@ -1531,7 +1589,7 @@ public:
                     branches.emplace_back(
                         makeIsMember(std::move(inputExpr),
                                      sbe::makeE<sbe::EConstant>(arrSetTag, arrSetVal),
-                                     _context->state.env),
+                                     _context->state.data->env),
                         EvalStage{});
                     branches.emplace_back(regexOutputSlot, std::move(regexStage));
 
@@ -1616,19 +1674,41 @@ public:
             sbe::EVariable dividendConvertedToNumberInt64{frameId, 0};
             auto truncatedArgument = sbe::makeE<sbe::ENumericConvert>(
                 makeFunction("trunc"_sd, dividend.clone()), sbe::value::TypeTags::NumberInt64);
+            tassert(6142202,
+                    "Either both divisor and remainer are parameterized or none",
+                    (expr->getDivisorInputParamId() && expr->getRemainderInputParamId()) ||
+                        (!expr->getDivisorInputParamId() && !expr->getRemainderInputParamId()));
+            // If there's related input param ids in this expr, we can register SlotIds for them,
+            // and use generated slots directly.
+            std::unique_ptr<sbe::EExpression> divisorExpr =
+                [&]() -> std::unique_ptr<sbe::EExpression> {
+                if (auto divisorParam = expr->getDivisorInputParamId()) {
+                    auto divisorSlotId = context->state.registerInputParamSlot(*divisorParam);
+                    return makeVariable(divisorSlotId);
+                } else {
+                    return makeConstant(sbe::value::TypeTags::NumberInt64,
+                                        sbe::value::bitcastFrom<int64_t>(expr->getDivisor()));
+                }
+            }();
+            std::unique_ptr<sbe::EExpression> remainderExpr =
+                [&]() -> std::unique_ptr<sbe::EExpression> {
+                if (auto remainderParam = expr->getRemainderInputParamId()) {
+                    auto remainderSlotId = context->state.registerInputParamSlot(*remainderParam);
+                    return makeVariable(remainderSlotId);
+                } else {
+                    return makeConstant(sbe::value::TypeTags::NumberInt64,
+                                        sbe::value::bitcastFrom<int64_t>(expr->getRemainder()));
+                }
+            }();
             auto modExpression = makeBinaryOp(
                 sbe::EPrimBinary::logicAnd,
                 // Return false if the dividend cannot be represented as a 64 bit integer.
                 makeNot(generateNullOrMissing(dividendConvertedToNumberInt64)),
-                makeFillEmptyFalse(makeBinaryOp(
-                    sbe::EPrimBinary::eq,
-                    makeFunction(
-                        "mod"_sd,
-                        dividendConvertedToNumberInt64.clone(),
-                        makeConstant(sbe::value::TypeTags::NumberInt64,
-                                     sbe::value::bitcastFrom<int64_t>(expr->getDivisor()))),
-                    makeConstant(sbe::value::TypeTags::NumberInt64,
-                                 sbe::value::bitcastFrom<int64_t>(expr->getRemainder())))));
+                makeFillEmptyFalse(makeBinaryOp(sbe::EPrimBinary::eq,
+                                                makeFunction("mod"_sd,
+                                                             dividendConvertedToNumberInt64.clone(),
+                                                             std::move(divisorExpr)),
+                                                std::move(remainderExpr))));
             return {
                 makeBinaryOp(sbe::EPrimBinary::logicAnd,
                              makeNot(makeBinaryOp(sbe::EPrimBinary::logicOr,
@@ -1678,15 +1758,38 @@ public:
     }
 
     void visit(const RegexMatchExpression* expr) final {
-        auto makePredicate = [expr](sbe::value::SlotId inputSlot,
-                                    EvalStage inputStage) -> EvalExprStagePair {
-            auto [bsonRegexTag, bsonRegexVal] =
-                sbe::value::makeNewBsonRegex(expr->getString(), expr->getFlags());
-            auto bsonRegexExpr = makeConstant(bsonRegexTag, bsonRegexVal);
-
-            auto [compiledRegexTag, compiledRegexVal] =
-                sbe::value::makeNewPcreRegex(expr->getString(), expr->getFlags());
-            auto compiledRegexExpr = makeConstant(compiledRegexTag, compiledRegexVal);
+        auto makePredicate = [expr, context = _context](sbe::value::SlotId inputSlot,
+                                                        EvalStage inputStage) -> EvalExprStagePair {
+            tassert(
+                6142203,
+                "Either both sourceRegex and compiledRegex are parameterized or none",
+                (expr->getSourceRegexInputParamId() && expr->getCompiledRegexInputParamId()) ||
+                    (!expr->getSourceRegexInputParamId() && !expr->getCompiledRegexInputParamId()));
+            std::unique_ptr<sbe::EExpression> bsonRegexExpr =
+                [&]() -> std::unique_ptr<sbe::EExpression> {
+                if (auto sourceRegexParam = expr->getSourceRegexInputParamId()) {
+                    auto sourceRegexSlotId =
+                        context->state.registerInputParamSlot(*sourceRegexParam);
+                    return makeVariable(sourceRegexSlotId);
+                } else {
+                    auto [bsonRegexTag, bsonRegexVal] =
+                        sbe::value::makeNewBsonRegex(expr->getString(), expr->getFlags());
+                    return makeConstant(bsonRegexTag, bsonRegexVal);
+                }
+            }();
+
+            std::unique_ptr<sbe::EExpression> compiledRegexExpr =
+                [&]() -> std::unique_ptr<sbe::EExpression> {
+                if (auto compiledRegexParam = expr->getCompiledRegexInputParamId()) {
+                    auto compiledRegexSlotId =
+                        context->state.registerInputParamSlot(*compiledRegexParam);
+                    return makeVariable(compiledRegexSlotId);
+                } else {
+                    auto [compiledRegexTag, compiledRegexVal] =
+                        sbe::value::makeNewPcreRegex(expr->getString(), expr->getFlags());
+                    return makeConstant(compiledRegexTag, compiledRegexVal);
+                }
+            }();
 
             auto resultExpr = makeBinaryOp(
                 sbe::EPrimBinary::logicOr,
@@ -1713,6 +1816,27 @@ public:
     void visit(const TwoDPtInAnnulusExpression* expr) final {}
 
     void visit(const TypeMatchExpression* expr) final {
+        // If there's an "inputParamId" in this expr meaning this expr got parameterized, we can
+        // register a SlotId for it and use the slot directly. Note that we don't auto-parameterize
+        // if the type set contains 'BSONType::Array'.
+        if (auto typeMaskParam = expr->getInputParamId()) {
+            auto typeMaskSlotId = _context->state.registerInputParamSlot(*typeMaskParam);
+            auto makePredicate = [typeMaskSlotId](sbe::value::SlotId inputSlot,
+                                                  EvalStage inputStage) -> EvalExprStagePair {
+                auto resultExpr = makeFillEmptyFalse(makeFunction(
+                    "typeMatch", makeVariable(inputSlot), makeVariable(typeMaskSlotId)));
+
+                return {std::move(resultExpr), std::move(inputStage)};
+            };
+
+            generatePredicate(_context,
+                              expr->fieldRef(),
+                              std::move(makePredicate),
+                              LeafTraversalMode::kArrayElementsOnly);
+
+            return;
+        }
+
         const auto traversalMode = expr->typeSet().hasType(BSONType::Array)
             ? LeafTraversalMode::kDoNotTraverseLeaf
             : LeafTraversalMode::kArrayElementsOnly;
@@ -1722,7 +1846,10 @@ public:
                                                       EvalStage inputStage) -> EvalExprStagePair {
             const MatcherTypeSet& ts = expr->typeSet();
             auto resultExpr = makeFillEmptyFalse(
-                sbe::makeE<sbe::ETypeMatch>(makeVariable(inputSlot), ts.getBSONTypeMask()));
+                makeFunction("typeMatch",
+                             makeVariable(inputSlot),
+                             makeConstant(sbe::value::TypeTags::NumberInt64,
+                                          sbe::value::bitcastFrom<int64_t>(ts.getBSONTypeMask()))));
 
             // $type is always applied to the leaf of the field path. For kDoNotTraverseLeaf mode,
             // generatePredicate() does not convert the predicate value to state when generating
@@ -1741,8 +1868,9 @@ public:
     }
 
     void visit(const WhereMatchExpression* expr) final {
-        auto makePredicate = [expr](sbe::value::SlotId inputSlot,
-                                    EvalStage inputStage) -> EvalExprStagePair {
+        auto makePredicate = [expr,
+                              ctx = this->_context](sbe::value::SlotId inputSlot,
+                                                    EvalStage inputStage) -> EvalExprStagePair {
             // Generally speaking, this visitor is non-destructive and does not mutate the
             // MatchExpression tree. However, in order to apply an optimization to avoid making a
             // copy of the 'JsFunction' object stored within 'WhereMatchExpression', we can transfer
@@ -1755,8 +1883,17 @@ public:
                 sbe::value::bitcastFrom<JsFunction*>(
                     const_cast<WhereMatchExpression*>(expr)->extractPredicate().release()));
 
-            auto whereExpr =
-                makeFunction("runJsPredicate", std::move(predicate), makeVariable(inputSlot));
+            std::unique_ptr<sbe::EExpression> whereExpr;
+            // If there's an "inputParamId" in this expr meaning this expr got parameterized, we can
+            // register a SlotId for it and use the slot directly.
+            if (auto inputParam = expr->getInputParamId()) {
+                auto inputParamSlotId = ctx->state.registerInputParamSlot(*inputParam);
+                whereExpr = makeFunction(
+                    "runJsPredicate", makeVariable(inputParamSlotId), makeVariable(inputSlot));
+            } else {
+                whereExpr =
+                    makeFunction("runJsPredicate", std::move(predicate), makeVariable(inputSlot));
+            }
             return {std::move(whereExpr), std::move(inputStage)};
         };
 
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.cpp b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
index 531e0d66bd2..d7185650509 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
@@ -31,6 +31,9 @@
 
 #include "mongo/db/query/sbe_stage_builder_helpers.h"
 
+#include <iterator>
+#include <numeric>
+
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/stages/branch.h"
 #include "mongo/db/exec/sbe/stages/co_scan.h"
@@ -43,8 +46,7 @@
 #include "mongo/db/exec/sbe/stages/unwind.h"
 #include "mongo/db/exec/sbe/values/bson.h"
 #include "mongo/db/matcher/matcher_type_set.h"
-#include <iterator>
-#include <numeric>
+#include "mongo/db/query/sbe_stage_builder.h"
 
 namespace mongo::stage_builder {
 
@@ -107,9 +109,12 @@ std::unique_ptr<sbe::EExpression> makeIsMember(std::unique_ptr<sbe::EExpression>
 std::unique_ptr<sbe::EExpression> generateNullOrMissing(const sbe::EVariable& var) {
     return makeBinaryOp(sbe::EPrimBinary::logicOr,
                         makeNot(makeFunction("exists", var.clone())),
-                        sbe::makeE<sbe::ETypeMatch>(var.clone(),
-                                                    getBSONTypeMask(BSONType::jstNULL) |
-                                                        getBSONTypeMask(BSONType::Undefined)));
+                        makeFunction("typeMatch",
+                                     var.clone(),
+                                     makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                  sbe::value::bitcastFrom<int64_t>(
+                                                      getBSONTypeMask(BSONType::jstNULL) |
+                                                      getBSONTypeMask(BSONType::Undefined)))));
 }
 
 std::unique_ptr<sbe::EExpression> generateNullOrMissing(const sbe::FrameId frameId,
@@ -121,9 +126,12 @@ std::unique_ptr<sbe::EExpression> generateNullOrMissing(const sbe::FrameId frame
 std::unique_ptr<sbe::EExpression> generateNullOrMissing(std::unique_ptr<sbe::EExpression> arg) {
     return makeBinaryOp(sbe::EPrimBinary::logicOr,
                         makeNot(makeFunction("exists", arg->clone())),
-                        sbe::makeE<sbe::ETypeMatch>(arg->clone(),
-                                                    getBSONTypeMask(BSONType::jstNULL) |
-                                                        getBSONTypeMask(BSONType::Undefined)));
+                        makeFunction("typeMatch",
+                                     arg->clone(),
+                                     makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                  sbe::value::bitcastFrom<int64_t>(
+                                                      getBSONTypeMask(BSONType::jstNULL) |
+                                                      getBSONTypeMask(BSONType::Undefined)))));
 }
 
 std::unique_ptr<sbe::EExpression> generateNonNumericCheck(const sbe::EVariable& var) {
@@ -133,8 +141,11 @@ std::unique_ptr<sbe::EExpression> generateNonNumericCheck(const sbe::EVariable&
 std::unique_ptr<sbe::EExpression> generateLongLongMinCheck(const sbe::EVariable& var) {
     return makeBinaryOp(
         sbe::EPrimBinary::logicAnd,
-        sbe::makeE<sbe::ETypeMatch>(var.clone(),
-                                    MatcherTypeSet{BSONType::NumberLong}.getBSONTypeMask()),
+        makeFunction("typeMatch",
+                     var.clone(),
+                     makeConstant(sbe::value::TypeTags::NumberInt64,
+                                  sbe::value::bitcastFrom<int64_t>(
+                                      MatcherTypeSet{BSONType::NumberLong}.getBSONTypeMask()))),
         makeBinaryOp(sbe::EPrimBinary::eq,
                      var.clone(),
                      sbe::makeE<sbe::EConstant>(
@@ -779,8 +790,17 @@ sbe::value::SlotId StageBuilderState::getGlobalVariableSlot(Variables::Id variab
     // Convert value of variable into SBE value.
     auto [tag, val] = makeValue(variables.getValue(variableId));
 
-    auto slotId = env->registerSlot(tag, val, true, slotIdGenerator);
+    auto slotId = data->env->registerSlot(tag, val, true, slotIdGenerator);
     globalVariables.emplace(variableId, slotId);
     return slotId;
 }
+
+sbe::value::SlotId StageBuilderState::registerInputParamSlot(
+    MatchExpression::InputParamId paramId) {
+    auto slotId = data->env->registerSlot(
+        sbe::value::TypeTags::Nothing, 0, false /* owned */, slotIdGenerator);
+    data->inputParamToSlotMap.emplace(paramId, slotId);
+    return slotId;
+}
+
 }  // namespace mongo::stage_builder
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.h b/src/mongo/db/query/sbe_stage_builder_helpers.h
index f1eb8958d0c..6a0d21f0834 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.h
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.h
@@ -865,13 +865,15 @@ std::pair<sbe::IndexKeysInclusionSet, std::vector<std::string>> makeIndexKeyIncl
     return {std::move(indexKeyBitset), std::move(keyFieldNames)};
 }
 
+struct PlanStageData;
+
 /**
  * Common parameters to SBE stage builder functions extracted into separate class to simplify
  * argument passing. Also contains a mapping of global variable ids to slot ids.
  */
 struct StageBuilderState {
     StageBuilderState(OperationContext* opCtx,
-                      sbe::RuntimeEnvironment* env,
+                      PlanStageData* data,
                       const Variables& variables,
                       sbe::value::SlotIdGenerator* slotIdGenerator,
                       sbe::value::FrameIdGenerator* frameIdGenerator,
@@ -882,7 +884,7 @@ struct StageBuilderState {
           frameIdGenerator{frameIdGenerator},
           spoolIdGenerator{spoolIdGenerator},
           opCtx{opCtx},
-          env{env},
+          data{data},
           variables{variables},
           needsMerge{needsMerge},
           allowDiskUse{allowDiskUse} {}
@@ -903,12 +905,20 @@ struct StageBuilderState {
         return spoolIdGenerator->generate();
     }
 
+    /**
+     * Register a Slot in the 'RuntimeEnvironment'. The newly registered Slot should be associated
+     * with 'paramId' and tracked in the 'InputParamToSlotMap' for auto-parameterization use. The
+     * slot is set to 'Nothing' on registration and will be populated with the real value when
+     * preparing the SBE plan for execution.
+     */
+    sbe::value::SlotId registerInputParamSlot(MatchExpression::InputParamId paramId);
+
     sbe::value::SlotIdGenerator* const slotIdGenerator;
     sbe::value::FrameIdGenerator* const frameIdGenerator;
     sbe::value::SpoolIdGenerator* const spoolIdGenerator;
 
     OperationContext* const opCtx;
-    sbe::RuntimeEnvironment* const env;
+    PlanStageData* const data;
 
     const Variables& variables;
     stdx::unordered_map<Variables::Id, sbe::value::SlotId> globalVariables;
diff --git a/src/mongo/db/query/sbe_stage_builder_lookup.cpp b/src/mongo/db/query/sbe_stage_builder_lookup.cpp
index 4794adbdbc1..223b7ef082d 100644
--- a/src/mongo/db/query/sbe_stage_builder_lookup.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_lookup.cpp
@@ -56,7 +56,11 @@ using namespace sbe::value;
 
 std::unique_ptr<EExpression> replaceUndefinedWithNullOrPassthrough(SlotId slot) {
     return makeE<EIf>(
-        makeE<ETypeMatch>(makeVariable(slot), getBSONTypeMask(TypeTags::bsonUndefined)),
+        makeFunction("typeMatch",
+                     makeVariable(slot),
+                     sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::NumberInt64,
+                                                sbe::value::bitcastFrom<int64_t>(
+                                                    getBSONTypeMask(TypeTags::bsonUndefined)))),
         makeConstant(TypeTags::Null, 0),
         makeVariable(slot));
 }
diff --git a/src/mongo/db/query/sbe_stage_builder_test_fixture.cpp b/src/mongo/db/query/sbe_stage_builder_test_fixture.cpp
index 2dc6ce23c9c..ff0207a7f5c 100644
--- a/src/mongo/db/query/sbe_stage_builder_test_fixture.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_test_fixture.cpp
@@ -30,6 +30,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/matcher/extensions_callback_noop.h"
 #include "mongo/db/pipeline/expression_context_for_test.h"
 #include "mongo/db/query/mock_yield_policies.h"
@@ -60,16 +61,31 @@ SbeStageBuilderTestFixture::buildPlanStage(
         CanonicalQuery::canonicalize(opCtx(), std::move(findCommand), false, expCtx);
     ASSERT_OK(statusWithCQ.getStatus());
 
-    stage_builder::SlotBasedStageBuilder builder{opCtx(),
-                                                 _collections,
-                                                 *statusWithCQ.getValue(),
-                                                 *querySolution,
-                                                 nullptr /* YieldPolicy */,
-                                                 shardFiltererInterface.get()};
+    CollectionMock coll(TenantNamespace(boost::none, _nss));
+    CollectionPtr collPtr(&coll);
+    MultipleCollectionAccessor& colls = _collections;
+    if (shardFiltererInterface) {
+        auto shardFilterer = shardFiltererInterface->makeShardFilterer(opCtx());
+        collPtr.setShardKeyPattern(shardFilterer->getKeyPattern().toBSON());
+        colls = MultipleCollectionAccessor(collPtr);
+    }
+
+    stage_builder::SlotBasedStageBuilder builder{
+        opCtx(), colls, *statusWithCQ.getValue(), *querySolution, nullptr /* YieldPolicy */};
 
     auto stage = builder.build(querySolution->root());
     auto data = builder.getPlanStageData();
 
+    // Reset "shardFilterer".
+    if (auto shardFiltererSlot = data.env->getSlotIfExists("shardFilterer"_sd);
+        shardFiltererSlot && shardFiltererInterface) {
+        auto shardFilterer = shardFiltererInterface->makeShardFilterer(opCtx());
+        data.env->resetSlot(*shardFiltererSlot,
+                            sbe::value::TypeTags::shardFilterer,
+                            sbe::value::bitcastFrom<ShardFilterer*>(shardFilterer.release()),
+                            true);
+    }
+
     auto slots = sbe::makeSV();
     if (hasRecordId) {
         slots.push_back(data.outputs.get(stage_builder::PlanStageSlots::kRecordId));
diff --git a/src/mongo/db/query/stage_builder_util.cpp b/src/mongo/db/query/stage_builder_util.cpp
index 645f2686084..96becfae481 100644
--- a/src/mongo/db/query/stage_builder_util.cpp
+++ b/src/mongo/db/query/stage_builder_util.cpp
@@ -69,25 +69,11 @@ buildSlotBasedExecutableTree(OperationContext* opCtx,
     auto sbeYieldPolicy = dynamic_cast<PlanYieldPolicySBE*>(yieldPolicy);
     invariant(sbeYieldPolicy);
 
-    auto shardFilterer =
-        std::make_unique<ShardFiltererFactoryImpl>(collections.getMainCollection());
-
-    auto builder = std::make_unique<SlotBasedStageBuilder>(
-        opCtx, collections, cq, solution, sbeYieldPolicy, shardFilterer.get());
+    auto builder =
+        std::make_unique<SlotBasedStageBuilder>(opCtx, collections, cq, solution, sbeYieldPolicy);
     auto root = builder->build(solution.root());
     auto data = builder->getPlanStageData();
 
-    root->attachToOperationContext(opCtx);
-
-    auto expCtx = cq.getExpCtxRaw();
-    tassert(5327100, "No expression context", expCtx);
-    if (expCtx->explain || expCtx->mayDbProfile) {
-        root->markShouldCollectTimingInfo();
-    }
-
-    // Register this plan to yield according to the configured policy.
-    sbeYieldPolicy->registerPlan(root.get());
-
     return {std::move(root), std::move(data)};
 }
 }  // namespace mongo::stage_builder