SERVER-71587 Allow SBE*Lowering access to named slots to support collation

38 files changed, 790 insertions, 443 deletions

diff --git a/src/mongo/db/exec/sbe/SConscript b/src/mongo/db/exec/sbe/SConscript
index 2660bb29659..947fb945d5e 100644
--- a/src/mongo/db/exec/sbe/SConscript
+++ b/src/mongo/db/exec/sbe/SConscript
@@ -38,7 +38,9 @@ sbeEnv.InjectThirdParty(libraries=['snappy'])
 sbeEnv.Library(
     target='query_sbe',
     source=[
+        'expressions/compile_ctx.cpp',
         'expressions/expression.cpp',
+        'expressions/runtime_environment.cpp',
         'size_estimator.cpp',
         'util/debug_print.cpp',
         'util/spilling.cpp',
@@ -50,8 +52,8 @@ sbeEnv.Library(
         'values/slot_printer.cpp',
         'vm/arith.cpp',
         'vm/datetime.cpp',
-        'vm/vm_printer.cpp',
         'vm/vm.cpp',
+        'vm/vm_printer.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
diff --git a/src/mongo/db/exec/sbe/abt/abt_lower.cpp b/src/mongo/db/exec/sbe/abt/abt_lower.cpp
index 55f0a1382b6..d935dd6f725 100644
--- a/src/mongo/db/exec/sbe/abt/abt_lower.cpp
+++ b/src/mongo/db/exec/sbe/abt/abt_lower.cpp
@@ -28,6 +28,8 @@
  */
 
 #include "mongo/db/exec/sbe/abt/abt_lower.h"
+#include "mongo/db/exec/sbe/abt/named_slots.h"
+#include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/stages/co_scan.h"
 #include "mongo/db/exec/sbe/stages/exchange.h"
 #include "mongo/db/exec/sbe/stages/filter.h"
@@ -47,7 +49,6 @@
 #include "mongo/db/exec/sbe/stages/unwind.h"
 #include "mongo/db/query/optimizer/utils/utils.h"
 
-
 namespace mongo::optimizer {
 
 static sbe::EExpression::Vector toInlinedVector(
@@ -144,6 +145,15 @@ std::unique_ptr<sbe::EExpression> SBEExpressionLowering::transport(
         }
     }(op.op());
 
+    if (sbe::EPrimBinary::isComparisonOp(sbeOp)) {
+        boost::optional<sbe::value::SlotId> collatorSlot =
+            _namedSlots.getSlotIfExists("collator"_sd);
+        if (collatorSlot) {
+            return sbe::makeE<sbe::EPrimBinary>(
+                sbeOp, std::move(lhs), std::move(rhs), sbe::makeE<sbe::EVariable>(*collatorSlot));
+        }
+    }
+
     return sbe::makeE<sbe::EPrimBinary>(sbeOp, std::move(lhs), std::move(rhs));
 }
 
@@ -394,7 +404,7 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const EvaluationNode& n,
     sbe::value::SlotMap<std::unique_ptr<sbe::EExpression>> projects;
 
     for (size_t idx = 0; idx < exprs.size(); ++idx) {
-        auto expr = SBEExpressionLowering{_env, _slotMap}.optimize(exprs[idx]);
+        auto expr = SBEExpressionLowering{_env, _slotMap, _namedSlots}.optimize(exprs[idx]);
         auto slot = _slotIdGenerator.generate();
 
         mapProjToSlot(names[idx], slot);
@@ -409,7 +419,7 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const FilterNode& n,
                                                       const ABT& child,
                                                       const ABT& filter) {
     auto input = generateInternal(child);
-    auto expr = SBEExpressionLowering{_env, _slotMap}.optimize(filter);
+    auto expr = SBEExpressionLowering{_env, _slotMap, _namedSlots}.optimize(filter);
     const PlanNodeId planNodeId = _nodeToGroupPropsMap.at(&n)._planNodeId;
 
     // Check if the filter expression is 'constant' (i.e., does not depend on any variables); then
@@ -607,7 +617,7 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const SpoolProducerNode& n
                 std::move(input), n.getSpoolId(), std::move(vals), planNodeId);
 
         case SpoolProducerType::Lazy: {
-            auto expr = SBEExpressionLowering{_env, _slotMap}.optimize(filter);
+            auto expr = SBEExpressionLowering{_env, _slotMap, _namedSlots}.optimize(filter);
             return sbe::makeS<sbe::SpoolLazyProducerStage>(
                 std::move(input), n.getSpoolId(), std::move(vals), std::move(expr), planNodeId);
         }
@@ -675,15 +685,14 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const GroupByNode& n,
     sbe::value::SlotMap<std::unique_ptr<sbe::EExpression>> aggs;
 
     for (size_t idx = 0; idx < exprs.size(); ++idx) {
-        auto expr = SBEExpressionLowering{_env, _slotMap}.optimize(exprs[idx]);
+        auto expr = SBEExpressionLowering{_env, _slotMap, _namedSlots}.optimize(exprs[idx]);
         auto slot = _slotIdGenerator.generate();
 
         mapProjToSlot(names[idx], slot);
         aggs.emplace(slot, std::move(expr));
     }
 
-    // TODO: use collator slot.
-    boost::optional<sbe::value::SlotId> collatorSlot;
+    boost::optional<sbe::value::SlotId> collatorSlot = _namedSlots.getSlotIfExists("collator"_sd);
     // Unused
     sbe::value::SlotVector seekKeysSlots;
 
@@ -711,7 +720,7 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const NestedLoopJoinNode&
         correlatedSlots.push_back(_slotMap.at(projectionName));
     }
 
-    auto expr = SBEExpressionLowering{_env, _slotMap}.optimize(filter);
+    auto expr = SBEExpressionLowering{_env, _slotMap, _namedSlots}.optimize(filter);
 
     const auto& leftChildProps = _nodeToGroupPropsMap.at(n.getLeftChild().cast<Node>());
     auto outerProjects = convertRequiredProjectionsToSlots(leftChildProps);
@@ -762,8 +771,7 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const HashJoinNode& n,
     auto outerKeys = convertProjectionsToSlots(n.getRightKeys());
     auto outerProjects = convertRequiredProjectionsToSlots(rightProps, outerKeys);
 
-    // TODO: use collator slot.
-    boost::optional<sbe::value::SlotId> collatorSlot;
+    boost::optional<sbe::value::SlotId> collatorSlot = _namedSlots.getSlotIfExists("collator"_sd);
     const PlanNodeId planNodeId = _nodeToGroupPropsMap.at(&n)._planNodeId;
     return sbe::makeS<sbe::HashJoinStage>(std::move(outerStage),
                                           std::move(innerStage),
@@ -833,6 +841,7 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const SortedMergeNode& n,
         boost::optional<sbe::value::SlotId> localRIDSlot;
         SBENodeLowering localLowering(_env,
                                       localMap,
+                                      _namedSlots,
                                       localRIDSlot,
                                       _slotIdGenerator,
                                       _metadata,
@@ -893,6 +902,7 @@ std::unique_ptr<sbe::PlanStage> SBENodeLowering::walk(const UnionNode& n,
         boost::optional<sbe::value::SlotId> localRIDSlot;
         SBENodeLowering localLowering(_env,
                                       localMap,
+                                      _namedSlots,
                                       localRIDSlot,
                                       _slotIdGenerator,
                                       _metadata,
@@ -1058,7 +1068,7 @@ std::unique_ptr<sbe::EExpression> SBENodeLowering::convertBoundsToExpr(
         sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::NumberInt32,
                                    sbe::value::bitcastFrom<uint32_t>(indexDef.getOrdering())));
 
-    auto exprLower = SBEExpressionLowering{_env, _slotMap};
+    auto exprLower = SBEExpressionLowering{_env, _slotMap, _namedSlots};
     bool inclusive = true;
     bool fullyInfinite = true;
     for (const auto& entry : interval) {
diff --git a/src/mongo/db/exec/sbe/abt/abt_lower.h b/src/mongo/db/exec/sbe/abt/abt_lower.h
index 062f6794930..747fa395687 100644
--- a/src/mongo/db/exec/sbe/abt/abt_lower.h
+++ b/src/mongo/db/exec/sbe/abt/abt_lower.h
@@ -35,13 +35,16 @@
 #include "mongo/db/query/optimizer/node_defs.h"
 #include "mongo/db/query/optimizer/reference_tracker.h"
 #include "mongo/db/query/optimizer/utils/utils.h"
+#include "mongo/stdx/unordered_map.h"
 
 namespace mongo::optimizer {
 
 class SBEExpressionLowering {
 public:
-    SBEExpressionLowering(const VariableEnvironment& env, SlotVarMap& slotMap)
-        : _env(env), _slotMap(slotMap) {}
+    SBEExpressionLowering(const VariableEnvironment& env,
+                          SlotVarMap& slotMap,
+                          const NamedSlotsProvider& namedSlots)
+        : _env(env), _slotMap(slotMap), _namedSlots(namedSlots) {}
 
     // The default noop transport.
     template <typename T, typename... Ts>
@@ -83,6 +86,7 @@ public:
 private:
     const VariableEnvironment& _env;
     SlotVarMap& _slotMap;
+    const NamedSlotsProvider& _namedSlots;
 
     sbe::FrameId _frameCounter{100};
     stdx::unordered_map<const Let*, sbe::FrameId> _letMap;
@@ -94,6 +98,7 @@ public:
     // TODO: SERVER-69540. Consider avoiding a mutable slotMap argument here.
     SBENodeLowering(const VariableEnvironment& env,
                     SlotVarMap& slotMap,
+                    const NamedSlotsProvider& namedSlots,
                     boost::optional<sbe::value::SlotId>& ridSlot,
                     sbe::value::SlotIdGenerator& ids,
                     const Metadata& metadata,
@@ -101,6 +106,7 @@ public:
                     const bool randomScan)
         : _env(env),
           _slotMap(slotMap),
+          _namedSlots(namedSlots),
           _ridSlot(ridSlot),
           _slotIdGenerator(ids),
           _metadata(metadata),
@@ -214,6 +220,7 @@ private:
 
     const VariableEnvironment& _env;
     SlotVarMap& _slotMap;
+    const NamedSlotsProvider& _namedSlots;
     boost::optional<sbe::value::SlotId>& _ridSlot;
     sbe::value::SlotIdGenerator& _slotIdGenerator;
 
diff --git a/src/mongo/db/exec/sbe/abt/abt_lower_bm.cpp b/src/mongo/db/exec/sbe/abt/abt_lower_bm.cpp
index 4209ad8b203..38ff3347b61 100644
--- a/src/mongo/db/exec/sbe/abt/abt_lower_bm.cpp
+++ b/src/mongo/db/exec/sbe/abt/abt_lower_bm.cpp
@@ -31,6 +31,7 @@
 #include <string>
 
 #include "mongo/db/exec/sbe/abt/abt_lower.h"
+#include "mongo/db/exec/sbe/abt/named_slots_mock.h"
 #include "mongo/db/query/optimizer/defs.h"
 #include "mongo/db/query/optimizer/metadata.h"
 #include "mongo/db/query/optimizer/metadata_factory.h"
@@ -81,11 +82,13 @@ protected:
 
             auto env = VariableEnvironment::build(n);
             SlotVarMap map;
+            MockEmptyNamedSlotsProvider namedSlots;
             boost::optional<sbe::value::SlotId> ridSlot;
             sbe::value::SlotIdGenerator ids;
 
             benchmark::DoNotOptimize(
-                SBENodeLowering{env, map, ridSlot, ids, m, _nodeMap, false}.optimize(n));
+                SBENodeLowering{env, map, namedSlots, ridSlot, ids, m, _nodeMap, false}.optimize(
+                    n));
             benchmark::ClobberMemory();
         }
     }
@@ -256,7 +259,8 @@ void BM_LowerABTLetExpr(benchmark::State& state) {
     for (auto keepRunning : state) {
         auto env = VariableEnvironment::build(n);
         SlotVarMap map;
-        benchmark::DoNotOptimize(SBEExpressionLowering{env, map}.optimize(n));
+        MockEmptyNamedSlotsProvider namedSlots;
+        benchmark::DoNotOptimize(SBEExpressionLowering{env, map, namedSlots}.optimize(n));
         benchmark::ClobberMemory();
     }
 }
diff --git a/src/mongo/db/exec/sbe/abt/abt_lower_defs.h b/src/mongo/db/exec/sbe/abt/abt_lower_defs.h
index ddffead4b10..20d2d192995 100644
--- a/src/mongo/db/exec/sbe/abt/abt_lower_defs.h
+++ b/src/mongo/db/exec/sbe/abt/abt_lower_defs.h
@@ -35,6 +35,7 @@
 
 namespace mongo::optimizer {
 
+class NamedSlotsProvider;
 using SlotVarMap = stdx::unordered_map<ProjectionName, sbe::value::SlotId, ProjectionName::Hasher>;
 
-}
+}  // namespace mongo::optimizer
diff --git a/src/mongo/db/exec/sbe/abt/abt_lower_test.cpp b/src/mongo/db/exec/sbe/abt/abt_lower_test.cpp
index bfeca39db82..d67e8938d42 100644
--- a/src/mongo/db/exec/sbe/abt/abt_lower_test.cpp
+++ b/src/mongo/db/exec/sbe/abt/abt_lower_test.cpp
@@ -30,6 +30,7 @@
 #include "mongo/base/string_data.h"
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/exec/sbe/abt/abt_lower.h"
+#include "mongo/db/exec/sbe/abt/named_slots_mock.h"
 #include "mongo/db/query/optimizer/defs.h"
 #include "mongo/db/query/optimizer/explain.h"
 #include "mongo/db/query/optimizer/metadata.h"
@@ -69,7 +70,8 @@ protected:
         stream << "-- OUTPUT:" << std::endl;
         auto env = VariableEnvironment::build(n);
         SlotVarMap map;
-        auto expr = SBEExpressionLowering{env, map}.optimize(n);
+        MockEmptyNamedSlotsProvider namedSlots;
+        auto expr = SBEExpressionLowering{env, map, namedSlots}.optimize(n);
         stream << expr->toString() << std::endl;
     }
 
@@ -106,6 +108,7 @@ protected:
         stream << "-- OUTPUT:" << std::endl;
         auto env = VariableEnvironment::build(n);
         SlotVarMap map;
+        MockEmptyNamedSlotsProvider namedSlots;
         boost::optional<sbe::value::SlotId> ridSlot;
         sbe::value::SlotIdGenerator ids;
         opt::unordered_map<std::string, ScanDefinition> scanDefs;
@@ -116,7 +119,8 @@ protected:
         scanDefs.insert({"otherColl", buildScanDefinition()});
 
         Metadata md(scanDefs);
-        auto planStage = SBENodeLowering{env, map, ridSlot, ids, md, _nodeMap, false}.optimize(n);
+        auto planStage =
+            SBENodeLowering{env, map, namedSlots, ridSlot, ids, md, _nodeMap, false}.optimize(n);
         sbe::DebugPrinter printer;
         stream << stripUUIDs(printer.print(*planStage)) << std::endl;
 
diff --git a/src/mongo/db/exec/sbe/abt/named_slots.h b/src/mongo/db/exec/sbe/abt/named_slots.h
new file mode 100644
index 00000000000..5f4b05408b9
--- /dev/null
+++ b/src/mongo/db/exec/sbe/abt/named_slots.h
@@ -0,0 +1,41 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/sbe/values/slot.h"
+
+namespace mongo::optimizer {
+
+class NamedSlotsProvider {
+public:
+    virtual boost::optional<sbe::value::SlotId> getSlotIfExists(StringData name) const = 0;
+};
+
+}  // namespace mongo::optimizer
diff --git a/src/mongo/db/exec/sbe/abt/named_slots_mock.h b/src/mongo/db/exec/sbe/abt/named_slots_mock.h
new file mode 100644
index 00000000000..97e766e76b3
--- /dev/null
+++ b/src/mongo/db/exec/sbe/abt/named_slots_mock.h
@@ -0,0 +1,43 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/sbe/abt/named_slots.h"
+
+namespace mongo::optimizer {
+
+class MockEmptyNamedSlotsProvider final : public NamedSlotsProvider {
+public:
+    boost::optional<sbe::value::SlotId> getSlotIfExists(StringData name) const final {
+        return boost::none;
+    }
+};
+
+}  // namespace mongo::optimizer
diff --git a/src/mongo/db/exec/sbe/abt/sbe_abt_test.cpp b/src/mongo/db/exec/sbe/abt/sbe_abt_test.cpp
index 2db794fbfe1..da96a3f9767 100644
--- a/src/mongo/db/exec/sbe/abt/sbe_abt_test.cpp
+++ b/src/mongo/db/exec/sbe/abt/sbe_abt_test.cpp
@@ -30,6 +30,7 @@
 #include "mongo/db/exec/sbe/abt/abt_lower.h"
 #include "mongo/db/exec/sbe/abt/sbe_abt_test_util.h"
 #include "mongo/db/pipeline/abt/document_source_visitor.h"
+#include "mongo/db/query/collation/collator_interface_mock.h"
 #include "mongo/db/query/optimizer/explain.h"
 #include "mongo/db/query/optimizer/metadata_factory.h"
 #include "mongo/db/query/optimizer/opt_phase_manager.h"
@@ -50,7 +51,7 @@ TEST_F(ABTSBE, Lower1) {
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
 
     ASSERT(expr);
 
@@ -70,7 +71,7 @@ TEST_F(ABTSBE, Lower2) {
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
 
     ASSERT(expr);
 
@@ -86,7 +87,7 @@ TEST_F(ABTSBE, Lower3) {
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
 
     ASSERT(expr);
 
@@ -117,7 +118,7 @@ TEST_F(ABTSBE, Lower4) {
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
 
     ASSERT(expr);
 
@@ -135,7 +136,7 @@ TEST_F(ABTSBE, Lower5) {
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
 
     ASSERT(expr);
 
@@ -185,7 +186,7 @@ TEST_F(ABTSBE, Lower6) {
         }
     } while (changed);
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
 
     ASSERT(expr);
 
@@ -235,7 +236,7 @@ TEST_F(ABTSBE, Lower7) {
         }
     } while (changed);
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
 
     ASSERT(expr);
     auto compiledExpr = compileExpression(*expr);
@@ -254,7 +255,7 @@ TEST_F(ABTSBE, LowerFunctionCallFail) {
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
     ASSERT(expr);
 
     auto compiledExpr = compileExpression(*expr);
@@ -284,7 +285,7 @@ TEST_F(ABTSBE, LowerFunctionCallConvert) {
                 Constant::int32(static_cast<int32_t>(sbe::value::TypeTags::NumberInt64))));
     auto env = VariableEnvironment::build(tree);
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
     ASSERT(expr);
 
     auto compiledExpr = compileExpression(*expr);
@@ -323,7 +324,7 @@ TEST_F(ABTSBE, LowerFunctionCallTypeMatch) {
                                 getBSONTypeMask(sbe::value::TypeTags::NumberDecimal))));
     auto env = VariableEnvironment::build(tree);
 
-    auto expr = SBEExpressionLowering{env, map}.optimize(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
     ASSERT(expr);
 
     auto compiledExpr = compileExpression(*expr);
@@ -343,6 +344,47 @@ TEST_F(ABTSBE, LowerFunctionCallTypeMatch) {
     }
 }
 
+TEST_F(ABTSBE, LowerComparisonCollation) {
+    sbe::value::OwnedValueAccessor lhsAccessor;
+    sbe::value::OwnedValueAccessor rhsAccessor;
+    auto lhsSlotId = bindAccessor(&lhsAccessor);
+    auto rhsSlotId = bindAccessor(&rhsAccessor);
+    SlotVarMap map;
+    map["lhs"] = lhsSlotId;
+    map["rhs"] = rhsSlotId;
+
+    CollatorInterfaceMock collator(CollatorInterfaceMock::MockType::kToLowerString);
+    registerSlot("collator"_sd,
+                 sbe::value::TypeTags::collator,
+                 sbe::value::bitcastFrom<const CollatorInterface*>(&collator),
+                 false);
+
+    auto tree = make<BinaryOp>(Operations::Cmp3w, make<Variable>("lhs"), make<Variable>("rhs"));
+
+    auto env = VariableEnvironment::build(tree);
+    auto expr = SBEExpressionLowering{env, map, *runtimeEnv()}.optimize(tree);
+    ASSERT(expr);
+    auto compiledExpr = compileExpression(*expr);
+
+    auto checkCmp3w = [&](StringData lhs, StringData rhs, int result) {
+        auto [lhsTag, lhsValue] = sbe::value::makeNewString(lhs);
+        lhsAccessor.reset(true, lhsTag, lhsValue);
+        auto [rhsTag, rhsValue] = sbe::value::makeNewString(rhs);
+        rhsAccessor.reset(true, rhsTag, rhsValue);
+
+        auto [tag, value] = runCompiledExpression(compiledExpr.get());
+        sbe::value::ValueGuard guard(tag, value);
+
+        ASSERT_EQ(sbe::value::TypeTags::NumberInt32, tag);
+        ASSERT_EQ(result, sbe::value::bitcastTo<int32_t>(value))
+            << "comparing string '" << lhs << "' and '" << rhs << "'";
+    };
+
+    checkCmp3w("ABC", "abc", 0);
+    checkCmp3w("aCC", "abb", 1);
+    checkCmp3w("AbX", "aBy", -1);
+}
+
 TEST_F(NodeSBE, Lower1) {
     auto prefixId = PrefixId::createForTests();
     Metadata metadata{{}};
@@ -388,11 +430,13 @@ TEST_F(NodeSBE, Lower1) {
     phaseManager.optimize(tree);
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
+    auto runtimeEnv = std::make_unique<sbe::RuntimeEnvironment>();
     boost::optional<sbe::value::SlotId> ridSlot;
     sbe::value::SlotIdGenerator ids;
 
     SBENodeLowering g{env,
                       map,
+                      *runtimeEnv,
                       ridSlot,
                       ids,
                       phaseManager.getMetadata(),
@@ -402,7 +446,7 @@ TEST_F(NodeSBE, Lower1) {
     ASSERT_EQ(1, map.size());
     ASSERT_FALSE(ridSlot);
 
-    sbe::CompileCtx ctx(std::make_unique<sbe::RuntimeEnvironment>());
+    sbe::CompileCtx ctx(std::move(runtimeEnv));
     sbePlan->prepare(ctx);
 
     std::vector<sbe::value::SlotAccessor*> accessors;
@@ -635,11 +679,13 @@ TEST_F(NodeSBE, RequireRID) {
     auto env = VariableEnvironment::build(tree);
 
     SlotVarMap map;
+    auto runtimeEnv = std::make_unique<sbe::RuntimeEnvironment>();
     boost::optional<sbe::value::SlotId> ridSlot;
     sbe::value::SlotIdGenerator ids;
 
     SBENodeLowering g{env,
                       map,
+                      *runtimeEnv,
                       ridSlot,
                       ids,
                       phaseManager.getMetadata(),
@@ -649,7 +695,7 @@ TEST_F(NodeSBE, RequireRID) {
     ASSERT_EQ(1, map.size());
     ASSERT_TRUE(ridSlot);
 
-    sbe::CompileCtx ctx(std::make_unique<sbe::RuntimeEnvironment>());
+    sbe::CompileCtx ctx(std::move(runtimeEnv));
     sbePlan->prepare(ctx);
 
     std::vector<sbe::value::SlotAccessor*> accessors;
@@ -816,14 +862,15 @@ TEST_F(NodeSBE, SpoolFibonacci) {
 
     auto env = VariableEnvironment::build(tree);
     SlotVarMap map;
+    auto runtimeEnv = std::make_unique<sbe::RuntimeEnvironment>();
     boost::optional<sbe::value::SlotId> ridSlot;
     sbe::value::SlotIdGenerator ids;
-    SBENodeLowering g{env, map, ridSlot, ids, metadata, props, false /*randomScan*/};
+    SBENodeLowering g{env, map, *runtimeEnv, ridSlot, ids, metadata, props, false /*randomScan*/};
     auto sbePlan = g.optimize(tree);
     ASSERT_EQ(1, map.size());
 
     auto opCtx = makeOperationContext();
-    sbe::CompileCtx ctx(std::make_unique<sbe::RuntimeEnvironment>());
+    sbe::CompileCtx ctx(std::move(runtimeEnv));
     sbePlan->prepare(ctx);
 
     std::vector<sbe::value::SlotAccessor*> accessors;
diff --git a/src/mongo/db/exec/sbe/abt/sbe_abt_test_util.cpp b/src/mongo/db/exec/sbe/abt/sbe_abt_test_util.cpp
index e84b7e948d7..795f6ad08d0 100644
--- a/src/mongo/db/exec/sbe/abt/sbe_abt_test_util.cpp
+++ b/src/mongo/db/exec/sbe/abt/sbe_abt_test_util.cpp
@@ -128,11 +128,13 @@ std::vector<BSONObj> runSBEAST(OperationContext* opCtx,
 
     SlotVarMap map;
     boost::optional<sbe::value::SlotId> ridSlot;
+    auto runtimeEnv = std::make_unique<sbe::RuntimeEnvironment>();
     sbe::value::SlotIdGenerator ids;
 
     auto env = VariableEnvironment::build(tree);
     SBENodeLowering g{env,
                       map,
+                      *runtimeEnv,
                       ridSlot,
                       ids,
                       phaseManager.getMetadata(),
@@ -143,7 +145,7 @@ std::vector<BSONObj> runSBEAST(OperationContext* opCtx,
     tassert(6624260, "Unexpected rid slot", !ridSlot);
     uassert(6624249, "Cannot optimize SBE plan", sbePlan != nullptr);
 
-    sbe::CompileCtx ctx(std::make_unique<sbe::RuntimeEnvironment>());
+    sbe::CompileCtx ctx(std::move(runtimeEnv));
     sbePlan->prepare(ctx);
 
     std::vector<sbe::value::SlotAccessor*> accessors;
diff --git a/src/mongo/db/exec/sbe/expression_test_base.h b/src/mongo/db/exec/sbe/expression_test_base.h
index b170e146754..08506ddae0b 100644
--- a/src/mongo/db/exec/sbe/expression_test_base.h
+++ b/src/mongo/db/exec/sbe/expression_test_base.h
@@ -30,6 +30,7 @@
 
 #include "mongo/unittest/unittest.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/sbe_unittest.h"
 #include "mongo/db/exec/sbe/stages/co_scan.h"
@@ -59,10 +60,14 @@ namespace mongo::sbe {
  */
 class EExpressionTestFixture : public virtual SBETestFixture {
 protected:
-    EExpressionTestFixture() : _ctx{std::make_unique<sbe::RuntimeEnvironment>()} {
+    EExpressionTestFixture(std::unique_ptr<sbe::RuntimeEnvironment> runtimeEnv)
+        : _runtimeEnv(runtimeEnv.get()), _ctx(std::move(runtimeEnv)) {
         _ctx.root = &_emptyStage;
     }
 
+    EExpressionTestFixture()
+        : EExpressionTestFixture(std::make_unique<sbe::RuntimeEnvironment>()) {}
+
     value::SlotId bindAccessor(value::SlotAccessor* accessor) {
         auto slot = _slotIdGenerator.generate();
         _ctx.pushCorrelated(slot, accessor);
@@ -70,6 +75,17 @@ protected:
         return slot;
     }
 
+    const RuntimeEnvironment* runtimeEnv() const {
+        return _runtimeEnv;
+    }
+
+    sbe::value::SlotId registerSlot(StringData name,
+                                    value::TypeTags tag,
+                                    value::Value val,
+                                    bool owned) {
+        return _runtimeEnv->registerSlot(name, tag, val, owned, &_slotIdGenerator);
+    }
+
     std::unique_ptr<vm::CodeFragment> compileExpression(const EExpression& expr) {
         return expr.compile(_ctx);
     }
@@ -283,6 +299,7 @@ protected:
 protected:
     value::SlotIdGenerator _slotIdGenerator;
     CoScanStage _emptyStage{kEmptyPlanNodeId};
+    RuntimeEnvironment* _runtimeEnv;
     CompileCtx _ctx;
     vm::ByteCode _vm;
     std::vector<std::pair<value::SlotId, value::SlotAccessor*>> boundAccessors;
diff --git a/src/mongo/db/exec/sbe/expressions/compile_ctx.cpp b/src/mongo/db/exec/sbe/expressions/compile_ctx.cpp
new file mode 100644
index 00000000000..0c4ba18f071
--- /dev/null
+++ b/src/mongo/db/exec/sbe/expressions/compile_ctx.cpp
@@ -0,0 +1,66 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
+
+namespace mongo::sbe {
+
+value::SlotAccessor* CompileCtx::getAccessor(value::SlotId slot) {
+    for (auto it = correlated.rbegin(); it != correlated.rend(); ++it) {
+        if (it->first == slot) {
+            return it->second;
+        }
+    }
+
+    return _env->getAccessor(slot);
+}
+
+std::shared_ptr<SpoolBuffer> CompileCtx::getSpoolBuffer(SpoolId spool) {
+    if (spoolBuffers.find(spool) == spoolBuffers.end()) {
+        spoolBuffers.emplace(spool, std::make_shared<SpoolBuffer>());
+    }
+    return spoolBuffers[spool];
+}
+
+void CompileCtx::pushCorrelated(value::SlotId slot, value::SlotAccessor* accessor) {
+    correlated.emplace_back(slot, accessor);
+}
+
+void CompileCtx::popCorrelated() {
+    correlated.pop_back();
+}
+
+CompileCtx CompileCtx::makeCopyForParallelUse() {
+    return {_env->makeCopyForParallelUse()};
+}
+
+CompileCtx CompileCtx::makeCopy() const {
+    return {_env->makeCopy()};
+}
+}  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/expressions/compile_ctx.h b/src/mongo/db/exec/sbe/expressions/compile_ctx.h
new file mode 100644
index 00000000000..15142a4adbc
--- /dev/null
+++ b/src/mongo/db/exec/sbe/expressions/compile_ctx.h
@@ -0,0 +1,83 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/sbe/expressions/runtime_environment.h"
+#include "mongo/stdx/unordered_map.h"
+
+namespace mongo::sbe {
+
+using SpoolBuffer = std::vector<value::MaterializedRow>;
+class PlanStage;
+
+struct CompileCtx {
+    CompileCtx(std::unique_ptr<RuntimeEnvironment> env) : _env{std::move(env)} {}
+
+    value::SlotAccessor* getAccessor(value::SlotId slot);
+
+    RuntimeEnvironment::Accessor* getRuntimeEnvAccessor(value::SlotId slotId) {
+        return _env->getAccessor(slotId);
+    }
+
+    std::shared_ptr<SpoolBuffer> getSpoolBuffer(SpoolId spool);
+
+    void pushCorrelated(value::SlotId slot, value::SlotAccessor* accessor);
+    void popCorrelated();
+
+    /**
+     * Make a copy of this CompileCtx. The underlying RuntimeEnvironment will also be copied.
+     *
+     * To create a copy of the underlying runtime environment for a parallel execution plan, please
+     * use makeCopyForParallelUse() method. This will result in the environment in this CompileCtx
+     * being converted to a parallel environment, as well as the newly created copy.
+     */
+    CompileCtx makeCopyForParallelUse();
+    CompileCtx makeCopy() const;
+
+    /**
+     * Root plan stage is used to resolve slot accessors introduced by PlanStage (optional).
+     * - if specified, the root plan stage will be used to resolve the slot accessor.
+     * - otherwise, if null, default context accessor resolution rules will be used.
+     */
+    PlanStage* root{nullptr};
+
+    value::SlotAccessor* accumulator{nullptr};
+    std::vector<std::pair<value::SlotId, value::SlotAccessor*>> correlated;
+    stdx::unordered_map<SpoolId, std::shared_ptr<SpoolBuffer>> spoolBuffers;
+    bool aggExpression{false};
+
+private:
+    // Any data that a PlanStage needs from the RuntimeEnvironment should not be accessed directly
+    // but insteady by looking up the corresponding slots. These slots are set up during the process
+    // of building PlanStages, so the PlanStages themselves should never need to add new slots to
+    // the RuntimeEnvironment.
+    std::unique_ptr<RuntimeEnvironment> _env;
+};
+}  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/expressions/expression.cpp b/src/mongo/db/exec/sbe/expressions/expression.cpp
index 91cd5e4f3ba..3ce4951cb56 100644
--- a/src/mongo/db/exec/sbe/expressions/expression.cpp
+++ b/src/mongo/db/exec/sbe/expressions/expression.cpp
@@ -36,6 +36,7 @@
 #include <stack>
 #include <vector>
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/stages/spool.h"
 #include "mongo/db/exec/sbe/stages/stages.h"
@@ -1390,171 +1391,5 @@ std::vector<DebugPrinter::Block> ENumericConvert::debugPrint() const {
 size_t ENumericConvert::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) {
-        emplaceAccessor(slotId, index);
-    }
-}
-
-RuntimeEnvironment::~RuntimeEnvironment() {
-    if (_state.use_count() == 1) {
-        for (size_t idx = 0; idx < _state->vals.size(); ++idx) {
-            if (_state->owned[idx]) {
-                releaseValue(_state->typeTags[idx], _state->vals[idx]);
-            }
-        }
-    }
-}
-
-value::SlotId RuntimeEnvironment::registerSlot(StringData name,
-                                               value::TypeTags tag,
-                                               value::Value val,
-                                               bool owned,
-                                               value::SlotIdGenerator* slotIdGenerator) {
-    auto slot = registerSlot(tag, val, owned, slotIdGenerator);
-    _state->nameSlot(name, slot);
-    return slot;
-}
-
-value::SlotId RuntimeEnvironment::registerSlot(value::TypeTags tag,
-                                               value::Value val,
-                                               bool owned,
-                                               value::SlotIdGenerator* slotIdGenerator) {
-    tassert(5645903, "Slot Id generator is null", slotIdGenerator);
-    auto slot = slotIdGenerator->generate();
-    emplaceAccessor(slot, _state->pushSlot(slot));
-    _accessors.at(slot).reset(owned, tag, val);
-    return slot;
-}
-
-value::SlotId RuntimeEnvironment::getSlot(StringData name) {
-    auto slot = getSlotIfExists(name);
-    uassert(4946305, str::stream() << "environment slot is not registered: " << name, slot);
-    return *slot;
-}
-
-boost::optional<value::SlotId> RuntimeEnvironment::getSlotIfExists(StringData name) {
-    if (auto it = _state->namedSlots.find(name); it != _state->namedSlots.end()) {
-        return it->second;
-    }
-
-    return boost::none;
-}
-
-void RuntimeEnvironment::resetSlot(value::SlotId slot,
-                                   value::TypeTags tag,
-                                   value::Value val,
-                                   bool owned) {
-    // With intra-query parallelism enabled the global environment can hold only read-only values.
-    invariant(!_isSmp);
-
-    if (auto it = _accessors.find(slot); it != _accessors.end()) {
-        it->second.reset(owned, tag, val);
-        return;
-    }
-
-    uasserted(4946300, str::stream() << "undefined slot accessor:" << slot);
-}
-
-RuntimeEnvironment::Accessor* RuntimeEnvironment::getAccessor(value::SlotId slot) {
-    if (auto it = _accessors.find(slot); it != _accessors.end()) {
-        return &it->second;
-    }
-
-    uasserted(4946301, str::stream() << "undefined slot accessor:" << slot);
-}
-
-
-std::unique_ptr<RuntimeEnvironment> RuntimeEnvironment::makeCopy() const {
-    return std::unique_ptr<RuntimeEnvironment>(new RuntimeEnvironment(*this));
-}
-
-std::unique_ptr<RuntimeEnvironment> RuntimeEnvironment::makeDeepCopy() const {
-    auto env = std::make_unique<RuntimeEnvironment>();
-
-    env->_state = _state->makeCopyWithoutValues();
-    for (auto&& [slotId, index] : _state->slots) {
-        // Copy the slot value.
-        auto [tag, val] = _accessors.at(slotId).copyOrMoveValue();
-
-        env->emplaceAccessor(slotId, index);
-        env->resetSlot(slotId, tag, val, true /* owned */);
-    }
-    env->_isSmp = _isSmp;
-
-    return env;
-}
-
-std::unique_ptr<RuntimeEnvironment> RuntimeEnvironment::makeCopyForParallelUse() {
-    // Once this environment is used to create a copy for a parallel plan execution, it becomes
-    // a parallel environment itself.
-    _isSmp = true;
-
-    return makeCopy();
-}
-
-void RuntimeEnvironment::debugString(StringBuilder* builder) {
-    using namespace std::literals;
-
-    value::SlotMap<StringData> slotName;
-    for (const auto& [name, slot] : _state->namedSlots) {
-        slotName[slot] = name;
-    }
-
-    *builder << "env: { ";
-    bool first = true;
-    for (auto&& [slot, _] : _state->slots) {
-        if (first) {
-            first = false;
-        } else {
-            *builder << ", ";
-        }
-
-        std::stringstream ss;
-        ss << _accessors.at(slot).getViewOfValue();
-
-        *builder << "s" << slot << " = " << ss.str();
-
-        if (auto it = slotName.find(slot); it != slotName.end()) {
-            *builder << " (" << it->second << ")";
-        }
-    }
-    *builder << " }";
-}
-
-value::SlotAccessor* CompileCtx::getAccessor(value::SlotId slot) {
-    for (auto it = correlated.rbegin(); it != correlated.rend(); ++it) {
-        if (it->first == slot) {
-            return it->second;
-        }
-    }
-
-    return env->getAccessor(slot);
-}
-
-std::shared_ptr<SpoolBuffer> CompileCtx::getSpoolBuffer(SpoolId spool) {
-    if (spoolBuffers.find(spool) == spoolBuffers.end()) {
-        spoolBuffers.emplace(spool, std::make_shared<SpoolBuffer>());
-    }
-    return spoolBuffers[spool];
-}
-
-void CompileCtx::pushCorrelated(value::SlotId slot, value::SlotAccessor* accessor) {
-    correlated.emplace_back(slot, accessor);
-}
-
-void CompileCtx::popCorrelated() {
-    correlated.pop_back();
-}
-
-CompileCtx CompileCtx::makeCopyForParallelUse() {
-    return {env->makeCopyForParallelUse()};
-}
-
-CompileCtx CompileCtx::makeCopy() const {
-    return {env->makeCopy()};
-}
 }  // namespace sbe
 }  // namespace mongo
diff --git a/src/mongo/db/exec/sbe/expressions/expression.h b/src/mongo/db/exec/sbe/expressions/expression.h
index 69ac98d5abf..ee0f997445f 100644
--- a/src/mongo/db/exec/sbe/expressions/expression.h
+++ b/src/mongo/db/exec/sbe/expressions/expression.h
@@ -33,255 +33,16 @@
 #include <string>
 #include <vector>
 
+#include "mongo/db/exec/sbe/abt/named_slots.h"
 #include "mongo/db/exec/sbe/util/debug_print.h"
 #include "mongo/db/exec/sbe/values/slot.h"
 #include "mongo/db/exec/sbe/values/value.h"
 #include "mongo/db/exec/sbe/vm/vm.h"
-#include "mongo/stdx/unordered_map.h"
-#include "mongo/util/string_map.h"
 
 namespace mongo {
 namespace sbe {
-using SpoolBuffer = std::vector<value::MaterializedRow>;
 
-/**
- * A holder for slots and accessors which are used in a PlanStage tree but:
- *  - Cannot be made constants due to restrictions on the lifetime of such values (e.g., they're
- *    singleton instances owned somewhere else).
- *  - Can be changed in runtime outside of the PlanStage tree (e.g., a resume recordId changed by a
- *    PlanExecutor).
- *
- * A RuntimeEnvironment object is created once per an execution thread. That means that each
- * producer and consumer in a parallel plan will have their own compilation environment, with their
- * own slot accessors. However, slot accessors in each of such environment will access shared data,
- * which is the same across all environments.
- *
- * To avoid data races, the values stored in the runtime environment are considered read-only when
- * used with a parallel plan. An attempt to change any slot with 'resetValue' will result in a user
- * exception.
- *
- * If the runtime environment is used in a serial plan, modifications of the slots is allowed.
- */
-class RuntimeEnvironment {
-public:
-    RuntimeEnvironment() = default;
-    RuntimeEnvironment(RuntimeEnvironment&&) = delete;
-    RuntimeEnvironment& operator=(const RuntimeEnvironment&) = delete;
-    RuntimeEnvironment& operator=(const RuntimeEnvironment&&) = delete;
-    ~RuntimeEnvironment();
-
-    class Accessor final : public value::SlotAccessor {
-    public:
-        Accessor(RuntimeEnvironment* env, size_t index) : _env{env}, _index{index} {}
-
-        std::pair<value::TypeTags, value::Value> getViewOfValue() const override {
-            return {_env->_state->typeTags[_index], _env->_state->vals[_index]};
-        }
-
-        std::pair<value::TypeTags, value::Value> copyOrMoveValue() override {
-            // Always make a copy.
-            return copyValue(_env->_state->typeTags[_index], _env->_state->vals[_index]);
-        }
-
-        std::pair<value::TypeTags, value::Value> copyOrMoveValue() const {
-            // Always make a copy.
-            return copyValue(_env->_state->typeTags[_index], _env->_state->vals[_index]);
-        }
-
-        void reset(bool owned, value::TypeTags tag, value::Value val) {
-            release();
-
-            _env->_state->typeTags[_index] = tag;
-            _env->_state->vals[_index] = val;
-            _env->_state->owned[_index] = owned;
-        }
-
-    private:
-        void release() {
-            if (_env->_state->owned[_index]) {
-                releaseValue(_env->_state->typeTags[_index], _env->_state->vals[_index]);
-                _env->_state->owned[_index] = false;
-            }
-        }
-
-        RuntimeEnvironment* const _env;
-        const size_t _index;
-    };
-
-    /**
-     * Registers and returns a SlotId for the given slot 'name'. The 'slotIdGenerator' is used
-     * to generate a new SlotId for the given slot 'name', which is then registered with this
-     * environment by creating a new SlotAccessor. The value 'val' is then stored within the
-     * SlotAccessor and the newly generated SlotId is returned.
-     *
-     * Both owned and unowned values can be stored in the runtime environment.
-     *
-     * A user exception is raised if this slot 'name' has been already registered.
-     */
-    value::SlotId registerSlot(StringData name,
-                               value::TypeTags tag,
-                               value::Value val,
-                               bool owned,
-                               value::SlotIdGenerator* slotIdGenerator);
-
-    /**
-     * Same as above, but allows to register an unnamed slot.
-     */
-    value::SlotId registerSlot(value::TypeTags tag,
-                               value::Value val,
-                               bool owned,
-                               value::SlotIdGenerator* slotIdGenerator);
-
-    /**
-     * Returns a SlotId registered for the given slot 'name'. If the slot with the specified name
-     * hasn't been registered, a user exception is raised.
-     */
-    value::SlotId getSlot(StringData name);
-
-    /**
-     * Returns a SlotId registered for the given slot 'name'. If the slot with the specified name
-     * hasn't been registered, boost::none is returned.
-     */
-    boost::optional<value::SlotId> getSlotIfExists(StringData name);
-
-    /**
-     * Store the given value in the specified slot within this runtime environment instance.
-     *
-     * A user exception is raised if the SlotId is not registered within this environment, or
-     * if this environment is used with a parallel plan.
-     */
-    void resetSlot(value::SlotId slot, value::TypeTags tag, value::Value val, bool owned);
-
-    /**
-     * Returns a SlotAccessor for the given SlotId which must be previously registered within this
-     * Environment by invoking 'registerSlot' method.
-     *
-     * A user exception is raised if the SlotId is not registered within this environment.
-     */
-    Accessor* getAccessor(value::SlotId slot);
-
-    /**
-     * Make a copy of this environment. The new environment will have its own set of SlotAccessors
-     * pointing to the same shared data holding slot values.
-     *
-     * To create a copy of the runtime environment for a parallel execution plan, please use
-     * makeCopyForParallelUse() method. This will result in this environment being converted to a
-     * parallel environment, as well as the newly created copy.
-     */
-    std::unique_ptr<RuntimeEnvironment> makeCopyForParallelUse();
-    std::unique_ptr<RuntimeEnvironment> makeCopy() const;
-
-    /**
-     * Make a "deep" copy of this environment. The new environment will have its own set of
-     * SlotAccessors pointing to data copied from this RuntimeEnvironment. All the slot values are
-     * made owned by the new environment as much as possible. There could be some uncopyable types
-     * which can not be owned by the new environment, e.g. TimeZoneDatabase.
-     */
-    std::unique_ptr<RuntimeEnvironment> makeDeepCopy() const;
-
-    /**
-     * Dumps all the slots currently defined in this environment into the given string builder.
-     */
-    void debugString(StringBuilder* builder);
-
-private:
-    RuntimeEnvironment(const RuntimeEnvironment&);
-
-    struct State {
-        size_t pushSlot(value::SlotId slot) {
-            auto index = vals.size();
-
-            typeTags.push_back(value::TypeTags::Nothing);
-            vals.push_back(0);
-            owned.push_back(false);
-
-            auto [_, inserted] = slots.emplace(slot, index);
-            uassert(4946302, str::stream() << "duplicate environment slot: " << slot, inserted);
-            return index;
-        }
-
-        void nameSlot(StringData name, value::SlotId slot) {
-            uassert(5645901, str::stream() << "undefined slot: " << slot, slots.count(slot));
-            auto [_, inserted] = namedSlots.emplace(name, slot);
-            uassert(5645902, str::stream() << "duplicate named slot: " << name, inserted);
-        }
-
-        std::unique_ptr<State> makeCopyWithoutValues() {
-            auto state = std::make_unique<State>();
-            state->namedSlots = namedSlots;
-            state->slots = slots;
-
-            // Populate slot values with default value.
-            state->typeTags.resize(typeTags.size(), value::TypeTags::Nothing);
-            state->vals.resize(vals.size(), 0);
-            state->owned.resize(owned.size(), false);
-
-            return state;
-        }
-
-        StringMap<value::SlotId> namedSlots;
-        value::SlotMap<size_t> slots;
-
-        std::vector<value::TypeTags> typeTags;
-        std::vector<value::Value> vals;
-        std::vector<bool> owned;
-    };
-
-    void emplaceAccessor(value::SlotId slot, size_t index) {
-        _accessors.emplace(slot, Accessor{this, index});
-    }
-
-    std::shared_ptr<State> _state{std::make_shared<State>()};
-    value::SlotMap<Accessor> _accessors;
-    bool _isSmp{false};
-
-    friend class Accessor;
-};
-
-class PlanStage;
-struct CompileCtx {
-    CompileCtx(std::unique_ptr<RuntimeEnvironment> env) : env{std::move(env)} {}
-
-    value::SlotAccessor* getAccessor(value::SlotId slot);
-
-    RuntimeEnvironment::Accessor* getRuntimeEnvAccessor(value::SlotId slotId) {
-        return env->getAccessor(slotId);
-    }
-
-    std::shared_ptr<SpoolBuffer> getSpoolBuffer(SpoolId spool);
-
-    void pushCorrelated(value::SlotId slot, value::SlotAccessor* accessor);
-    void popCorrelated();
-
-    /**
-     * Make a copy of this CompileCtx. The underlying RuntimeEnvironment will also be copied.
-     *
-     * To create a copy of the underlying runtime environment for a parallel execution plan, please
-     * use makeCopyForParallelUse() method. This will result in the environment in this CompileCtx
-     * being converted to a parallel environment, as well as the newly created copy.
-     */
-    CompileCtx makeCopyForParallelUse();
-    CompileCtx makeCopy() const;
-
-    /**
-     * Root plan stage is used to resolve slot accessors introduced by PlanStage (optional).
-     * - if specified, the root plan stage will be used to resolve the slot accessor.
-     * - otherwise, if null, default context accessor resolution rules will be used.
-     */
-    PlanStage* root{nullptr};
-
-    value::SlotAccessor* accumulator{nullptr};
-    std::vector<std::pair<value::SlotId, value::SlotAccessor*>> correlated;
-    stdx::unordered_map<SpoolId, std::shared_ptr<SpoolBuffer>> spoolBuffers;
-    bool aggExpression{false};
-
-private:
-    // Any data that a PlanStage needs from the RuntimeEnvironment should not be accessed directly
-    // but insteady by looking up the corresponding slots. These slots are set up during the process
-    // of building PlanStages, so the PlanStages themselves should never need to add new slots to
-    // the RuntimeEnvironment.
-    std::unique_ptr<RuntimeEnvironment> env;
-};
+struct CompileCtx;
 
 /**
  * This is an abstract base class of all expression types in SBE. The expression types derived form
diff --git a/src/mongo/db/exec/sbe/expressions/runtime_environment.cpp b/src/mongo/db/exec/sbe/expressions/runtime_environment.cpp
new file mode 100644
index 00000000000..088a3f7ff5a
--- /dev/null
+++ b/src/mongo/db/exec/sbe/expressions/runtime_environment.cpp
@@ -0,0 +1,173 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/exec/sbe/expressions/runtime_environment.h"
+
+namespace mongo::sbe {
+RuntimeEnvironment::RuntimeEnvironment(const RuntimeEnvironment& other)
+    : _state{other._state}, _isSmp{other._isSmp} {
+    for (auto&& [slotId, index] : _state->slots) {
+        emplaceAccessor(slotId, index);
+    }
+}
+
+RuntimeEnvironment::~RuntimeEnvironment() {
+    if (_state.use_count() == 1) {
+        for (size_t idx = 0; idx < _state->vals.size(); ++idx) {
+            if (_state->owned[idx]) {
+                releaseValue(_state->typeTags[idx], _state->vals[idx]);
+            }
+        }
+    }
+}
+
+value::SlotId RuntimeEnvironment::registerSlot(StringData name,
+                                               value::TypeTags tag,
+                                               value::Value val,
+                                               bool owned,
+                                               value::SlotIdGenerator* slotIdGenerator) {
+    auto slot = registerSlot(tag, val, owned, slotIdGenerator);
+    _state->nameSlot(name, slot);
+    return slot;
+}
+
+value::SlotId RuntimeEnvironment::registerSlot(value::TypeTags tag,
+                                               value::Value val,
+                                               bool owned,
+                                               value::SlotIdGenerator* slotIdGenerator) {
+    tassert(5645903, "Slot Id generator is null", slotIdGenerator);
+    auto slot = slotIdGenerator->generate();
+    emplaceAccessor(slot, _state->pushSlot(slot));
+    _accessors.at(slot).reset(owned, tag, val);
+    return slot;
+}
+
+value::SlotId RuntimeEnvironment::getSlot(StringData name) const {
+    auto slot = getSlotIfExists(name);
+    uassert(4946305, str::stream() << "environment slot is not registered: " << name, slot);
+    return *slot;
+}
+
+boost::optional<value::SlotId> RuntimeEnvironment::getSlotIfExists(StringData name) const {
+    if (auto it = _state->namedSlots.find(name); it != _state->namedSlots.end()) {
+        return it->second;
+    }
+
+    return boost::none;
+}
+
+void RuntimeEnvironment::resetSlot(value::SlotId slot,
+                                   value::TypeTags tag,
+                                   value::Value val,
+                                   bool owned) {
+    // With intra-query parallelism enabled the global environment can hold only read-only values.
+    invariant(!_isSmp);
+
+    if (auto it = _accessors.find(slot); it != _accessors.end()) {
+        it->second.reset(owned, tag, val);
+        return;
+    }
+
+    uasserted(4946300, str::stream() << "undefined slot accessor:" << slot);
+}
+
+RuntimeEnvironment::Accessor* RuntimeEnvironment::getAccessor(value::SlotId slot) {
+    if (auto it = _accessors.find(slot); it != _accessors.end()) {
+        return &it->second;
+    }
+
+    uasserted(4946301, str::stream() << "undefined slot accessor:" << slot);
+}
+
+const RuntimeEnvironment::Accessor* RuntimeEnvironment::getAccessor(value::SlotId slot) const {
+    if (auto it = _accessors.find(slot); it != _accessors.end()) {
+        return &it->second;
+    }
+
+    uasserted(4946303, str::stream() << "undefined slot accessor:" << slot);
+}
+
+std::unique_ptr<RuntimeEnvironment> RuntimeEnvironment::makeCopy() const {
+    return std::unique_ptr<RuntimeEnvironment>(new RuntimeEnvironment(*this));
+}
+
+std::unique_ptr<RuntimeEnvironment> RuntimeEnvironment::makeDeepCopy() const {
+    auto env = std::make_unique<RuntimeEnvironment>();
+
+    env->_state = _state->makeCopyWithoutValues();
+    for (auto&& [slotId, index] : _state->slots) {
+        // Copy the slot value.
+        auto [tag, val] = _accessors.at(slotId).copyOrMoveValue();
+
+        env->emplaceAccessor(slotId, index);
+        env->resetSlot(slotId, tag, val, true /* owned */);
+    }
+    env->_isSmp = _isSmp;
+
+    return env;
+}
+
+std::unique_ptr<RuntimeEnvironment> RuntimeEnvironment::makeCopyForParallelUse() {
+    // Once this environment is used to create a copy for a parallel plan execution, it becomes
+    // a parallel environment itself.
+    _isSmp = true;
+
+    return makeCopy();
+}
+
+void RuntimeEnvironment::debugString(StringBuilder* builder) {
+    using namespace std::literals;
+
+    value::SlotMap<StringData> slotName;
+    for (const auto& [name, slot] : _state->namedSlots) {
+        slotName[slot] = name;
+    }
+
+    *builder << "env: { ";
+    bool first = true;
+    for (auto&& [slot, _] : _state->slots) {
+        if (first) {
+            first = false;
+        } else {
+            *builder << ", ";
+        }
+
+        std::stringstream ss;
+        ss << _accessors.at(slot).getViewOfValue();
+
+        *builder << "s" << slot << " = " << ss.str();
+
+        if (auto it = slotName.find(slot); it != slotName.end()) {
+            *builder << " (" << it->second << ")";
+        }
+    }
+    *builder << " }";
+}
+
+}  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/expressions/runtime_environment.h b/src/mongo/db/exec/sbe/expressions/runtime_environment.h
new file mode 100644
index 00000000000..ef467c3c271
--- /dev/null
+++ b/src/mongo/db/exec/sbe/expressions/runtime_environment.h
@@ -0,0 +1,231 @@
+/**
+ *    Copyright (C) 2023-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/sbe/abt/named_slots.h"
+#include "mongo/db/exec/sbe/values/slot.h"
+#include "mongo/util/string_map.h"
+
+namespace mongo::sbe {
+/**
+ * A holder for slots and accessors which are used in a PlanStage tree but:
+ *  - Cannot be made constants due to restrictions on the lifetime of such values (e.g., they're
+ *    singleton instances owned somewhere else).
+ *  - Can be changed in runtime outside of the PlanStage tree (e.g., a resume recordId changed by a
+ *    PlanExecutor).
+ *
+ * A RuntimeEnvironment object is created once per an execution thread. That means that each
+ * producer and consumer in a parallel plan will have their own compilation environment, with their
+ * own slot accessors. However, slot accessors in each of such environment will access shared data,
+ * which is the same across all environments.
+ *
+ * To avoid data races, the values stored in the runtime environment are considered read-only when
+ * used with a parallel plan. An attempt to change any slot with 'resetValue' will result in a user
+ * exception.
+ *
+ * If the runtime environment is used in a serial plan, modifications of the slots is allowed.
+ */
+class RuntimeEnvironment final : public optimizer::NamedSlotsProvider {
+public:
+    RuntimeEnvironment() = default;
+    RuntimeEnvironment(RuntimeEnvironment&&) = delete;
+    RuntimeEnvironment& operator=(const RuntimeEnvironment&) = delete;
+    RuntimeEnvironment& operator=(const RuntimeEnvironment&&) = delete;
+    virtual ~RuntimeEnvironment();
+
+    class Accessor final : public value::SlotAccessor {
+    public:
+        Accessor(RuntimeEnvironment* env, size_t index) : _env{env}, _index{index} {}
+
+        std::pair<value::TypeTags, value::Value> getViewOfValue() const override {
+            return {_env->_state->typeTags[_index], _env->_state->vals[_index]};
+        }
+
+        std::pair<value::TypeTags, value::Value> copyOrMoveValue() override {
+            // Always make a copy.
+            return copyValue(_env->_state->typeTags[_index], _env->_state->vals[_index]);
+        }
+
+        std::pair<value::TypeTags, value::Value> copyOrMoveValue() const {
+            // Always make a copy.
+            return copyValue(_env->_state->typeTags[_index], _env->_state->vals[_index]);
+        }
+
+        void reset(bool owned, value::TypeTags tag, value::Value val) {
+            release();
+
+            _env->_state->typeTags[_index] = tag;
+            _env->_state->vals[_index] = val;
+            _env->_state->owned[_index] = owned;
+        }
+
+    private:
+        void release() {
+            if (_env->_state->owned[_index]) {
+                releaseValue(_env->_state->typeTags[_index], _env->_state->vals[_index]);
+                _env->_state->owned[_index] = false;
+            }
+        }
+
+        RuntimeEnvironment* const _env;
+        const size_t _index;
+    };
+
+    /**
+     * Registers and returns a SlotId for the given slot 'name'. The 'slotIdGenerator' is used
+     * to generate a new SlotId for the given slot 'name', which is then registered with this
+     * environment by creating a new SlotAccessor. The value 'val' is then stored within the
+     * SlotAccessor and the newly generated SlotId is returned.
+     *
+     * Both owned and unowned values can be stored in the runtime environment.
+     *
+     * A user exception is raised if this slot 'name' has been already registered.
+     */
+    value::SlotId registerSlot(StringData name,
+                               value::TypeTags tag,
+                               value::Value val,
+                               bool owned,
+                               value::SlotIdGenerator* slotIdGenerator);
+
+    /**
+     * Same as above, but allows to register an unnamed slot.
+     */
+    value::SlotId registerSlot(value::TypeTags tag,
+                               value::Value val,
+                               bool owned,
+                               value::SlotIdGenerator* slotIdGenerator);
+
+    /**
+     * Returns a SlotId registered for the given slot 'name'. If the slot with the specified name
+     * hasn't been registered, a user exception is raised.
+     */
+    value::SlotId getSlot(StringData name) const;
+
+    /**
+     * Returns a SlotId registered for the given slot 'name'. If the slot with the specified name
+     * hasn't been registered, boost::none is returned.
+     */
+    boost::optional<value::SlotId> getSlotIfExists(StringData name) const final;
+
+    /**
+     * Store the given value in the specified slot within this runtime environment instance.
+     *
+     * A user exception is raised if the SlotId is not registered within this environment, or
+     * if this environment is used with a parallel plan.
+     */
+    void resetSlot(value::SlotId slot, value::TypeTags tag, value::Value val, bool owned);
+
+    /**
+     * Returns a SlotAccessor for the given SlotId which must be previously registered within this
+     * Environment by invoking 'registerSlot' method.
+     *
+     * A user exception is raised if the SlotId is not registered within this environment.
+     */
+    Accessor* getAccessor(value::SlotId slot);
+    const Accessor* getAccessor(value::SlotId slot) const;
+
+    /**
+     * Make a copy of this environment. The new environment will have its own set of SlotAccessors
+     * pointing to the same shared data holding slot values.
+     *
+     * To create a copy of the runtime environment for a parallel execution plan, please use
+     * makeCopyForParallelUse() method. This will result in this environment being converted to a
+     * parallel environment, as well as the newly created copy.
+     */
+    std::unique_ptr<RuntimeEnvironment> makeCopyForParallelUse();
+    std::unique_ptr<RuntimeEnvironment> makeCopy() const;
+
+    /**
+     * Make a "deep" copy of this environment. The new environment will have its own set of
+     * SlotAccessors pointing to data copied from this RuntimeEnvironment. All the slot values are
+     * made owned by the new environment as much as possible. There could be some uncopyable types
+     * which can not be owned by the new environment, e.g. TimeZoneDatabase.
+     */
+    std::unique_ptr<RuntimeEnvironment> makeDeepCopy() const;
+
+    /**
+     * Dumps all the slots currently defined in this environment into the given string builder.
+     */
+    void debugString(StringBuilder* builder);
+
+private:
+    RuntimeEnvironment(const RuntimeEnvironment&);
+
+    struct State {
+        size_t pushSlot(value::SlotId slot) {
+            auto index = vals.size();
+
+            typeTags.push_back(value::TypeTags::Nothing);
+            vals.push_back(0);
+            owned.push_back(false);
+
+            auto [_, inserted] = slots.emplace(slot, index);
+            uassert(4946302, str::stream() << "duplicate environment slot: " << slot, inserted);
+            return index;
+        }
+
+        void nameSlot(StringData name, value::SlotId slot) {
+            uassert(5645901, str::stream() << "undefined slot: " << slot, slots.count(slot));
+            auto [_, inserted] = namedSlots.emplace(name, slot);
+            uassert(5645902, str::stream() << "duplicate named slot: " << name, inserted);
+        }
+
+        std::unique_ptr<State> makeCopyWithoutValues() {
+            auto state = std::make_unique<State>();
+            state->namedSlots = namedSlots;
+            state->slots = slots;
+
+            // Populate slot values with default value.
+            state->typeTags.resize(typeTags.size(), value::TypeTags::Nothing);
+            state->vals.resize(vals.size(), 0);
+            state->owned.resize(owned.size(), false);
+
+            return state;
+        }
+
+        StringMap<value::SlotId> namedSlots;
+        value::SlotMap<size_t> slots;
+
+        std::vector<value::TypeTags> typeTags;
+        std::vector<value::Value> vals;
+        std::vector<bool> owned;
+    };
+
+    void emplaceAccessor(value::SlotId slot, size_t index) {
+        _accessors.emplace(slot, Accessor{this, index});
+    }
+
+    std::shared_ptr<State> _state{std::make_shared<State>()};
+    value::SlotMap<Accessor> _accessors;
+    bool _isSmp{false};
+
+    friend class Accessor;
+};
+}  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/expressions/sbe_runtime_environment_test.cpp b/src/mongo/db/exec/sbe/expressions/sbe_runtime_environment_test.cpp
index f09a7ab264d..927b528d261 100644
--- a/src/mongo/db/exec/sbe/expressions/sbe_runtime_environment_test.cpp
+++ b/src/mongo/db/exec/sbe/expressions/sbe_runtime_environment_test.cpp
@@ -27,7 +27,7 @@
  *    it in the license file.
  */
 
-#include "mongo/db/exec/sbe/expressions/expression.h"
+#include "mongo/db/exec/sbe/expressions/runtime_environment.h"
 #include "mongo/unittest/unittest.h"
 
 namespace mongo::sbe {
diff --git a/src/mongo/db/exec/sbe/stages/branch.cpp b/src/mongo/db/exec/sbe/stages/branch.cpp
index adbbd533273..f42d1409d58 100644
--- a/src/mongo/db/exec/sbe/stages/branch.cpp
+++ b/src/mongo/db/exec/sbe/stages/branch.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/branch.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 
diff --git a/src/mongo/db/exec/sbe/stages/bson_scan.cpp b/src/mongo/db/exec/sbe/stages/bson_scan.cpp
index e63dd436115..6b2e2c22e12 100644
--- a/src/mongo/db/exec/sbe/stages/bson_scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/bson_scan.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/bson_scan.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/util/str.h"
diff --git a/src/mongo/db/exec/sbe/stages/co_scan.cpp b/src/mongo/db/exec/sbe/stages/co_scan.cpp
index 9666d03cf01..6656a6bf6dc 100644
--- a/src/mongo/db/exec/sbe/stages/co_scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/co_scan.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/co_scan.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 
 namespace mongo::sbe {
diff --git a/src/mongo/db/exec/sbe/stages/column_scan.cpp b/src/mongo/db/exec/sbe/stages/column_scan.cpp
index 55bb8cc36ea..7be25f90f74 100644
--- a/src/mongo/db/exec/sbe/stages/column_scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/column_scan.cpp
@@ -28,6 +28,7 @@
  */
 #include "mongo/db/exec/sbe/stages/column_scan.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/index/columns_access_method.h"
diff --git a/src/mongo/db/exec/sbe/stages/exchange.h b/src/mongo/db/exec/sbe/stages/exchange.h
index 15928cd50fb..2919f91955b 100644
--- a/src/mongo/db/exec/sbe/stages/exchange.h
+++ b/src/mongo/db/exec/sbe/stages/exchange.h
@@ -31,6 +31,7 @@
 
 #include <vector>
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/stages/stages.h"
 #include "mongo/stdx/condition_variable.h"
diff --git a/src/mongo/db/exec/sbe/stages/filter.h b/src/mongo/db/exec/sbe/stages/filter.h
index 059dd1c7ab4..0449f1abbb1 100644
--- a/src/mongo/db/exec/sbe/stages/filter.h
+++ b/src/mongo/db/exec/sbe/stages/filter.h
@@ -29,6 +29,7 @@
 
 #pragma once
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/stages/stages.h"
diff --git a/src/mongo/db/exec/sbe/stages/hash_agg.cpp b/src/mongo/db/exec/sbe/stages/hash_agg.cpp
index 4061c99152d..74b7625ede6 100644
--- a/src/mongo/db/exec/sbe/stages/hash_agg.cpp
+++ b/src/mongo/db/exec/sbe/stages/hash_agg.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/util/str.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 
 namespace mongo {
diff --git a/src/mongo/db/exec/sbe/stages/hash_join.cpp b/src/mongo/db/exec/sbe/stages/hash_join.cpp
index bad53262acb..41c512031e3 100644
--- a/src/mongo/db/exec/sbe/stages/hash_join.cpp
+++ b/src/mongo/db/exec/sbe/stages/hash_join.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/hash_join.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/util/str.h"
diff --git a/src/mongo/db/exec/sbe/stages/hash_lookup.cpp b/src/mongo/db/exec/sbe/stages/hash_lookup.cpp
index 69bd2a8b3b1..7341204fc07 100644
--- a/src/mongo/db/exec/sbe/stages/hash_lookup.cpp
+++ b/src/mongo/db/exec/sbe/stages/hash_lookup.cpp
@@ -33,6 +33,7 @@
 #include "mongo/db/exec/sbe/stages/hash_lookup.h"
 #include "mongo/db/exec/sbe/stages/stage_visitors.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/util/spilling.h"
diff --git a/src/mongo/db/exec/sbe/stages/ix_scan.cpp b/src/mongo/db/exec/sbe/stages/ix_scan.cpp
index c0ac103efdf..64b214cc956 100644
--- a/src/mongo/db/exec/sbe/stages/ix_scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/ix_scan.cpp
@@ -32,6 +32,7 @@
 #include "mongo/db/exec/sbe/stages/ix_scan.h"
 
 #include "mongo/db/catalog/index_catalog.h"
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/values/bson.h"
diff --git a/src/mongo/db/exec/sbe/stages/loop_join.cpp b/src/mongo/db/exec/sbe/stages/loop_join.cpp
index 72a2019a28b..8bb04581ff2 100644
--- a/src/mongo/db/exec/sbe/stages/loop_join.cpp
+++ b/src/mongo/db/exec/sbe/stages/loop_join.cpp
@@ -29,6 +29,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/stages/loop_join.h"
 #include "mongo/db/exec/sbe/stages/stage_visitors.h"
 
diff --git a/src/mongo/db/exec/sbe/stages/merge_join.cpp b/src/mongo/db/exec/sbe/stages/merge_join.cpp
index b85ea831d24..4908633d020 100644
--- a/src/mongo/db/exec/sbe/stages/merge_join.cpp
+++ b/src/mongo/db/exec/sbe/stages/merge_join.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/merge_join.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/util/str.h"
diff --git a/src/mongo/db/exec/sbe/stages/project.cpp b/src/mongo/db/exec/sbe/stages/project.cpp
index a645eb33249..326c0238798 100644
--- a/src/mongo/db/exec/sbe/stages/project.cpp
+++ b/src/mongo/db/exec/sbe/stages/project.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/project.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 
 namespace mongo {
diff --git a/src/mongo/db/exec/sbe/stages/scan.cpp b/src/mongo/db/exec/sbe/stages/scan.cpp
index 379fb9badc2..97c90334006 100644
--- a/src/mongo/db/exec/sbe/stages/scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/scan.cpp
@@ -32,6 +32,7 @@
 #include "mongo/db/exec/sbe/stages/scan.h"
 
 #include "mongo/config.h"
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/trial_run_tracker.h"
diff --git a/src/mongo/db/exec/sbe/stages/scan.h b/src/mongo/db/exec/sbe/stages/scan.h
index 7c70e50277d..1677ea0c38a 100644
--- a/src/mongo/db/exec/sbe/stages/scan.h
+++ b/src/mongo/db/exec/sbe/stages/scan.h
@@ -31,6 +31,7 @@
 
 #include "mongo/config.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
+#include "mongo/db/exec/sbe/expressions/runtime_environment.h"
 #include "mongo/db/exec/sbe/stages/collection_helpers.h"
 #include "mongo/db/exec/sbe/stages/stages.h"
 #include "mongo/db/exec/sbe/values/bson.h"
diff --git a/src/mongo/db/exec/sbe/stages/sort.cpp b/src/mongo/db/exec/sbe/stages/sort.cpp
index 9a1f90a95c8..f5afc97ad00 100644
--- a/src/mongo/db/exec/sbe/stages/sort.cpp
+++ b/src/mongo/db/exec/sbe/stages/sort.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/sort.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/trial_run_tracker.h"
diff --git a/src/mongo/db/exec/sbe/stages/sorted_merge.cpp b/src/mongo/db/exec/sbe/stages/sorted_merge.cpp
index 39cee407a00..201f6092dd4 100644
--- a/src/mongo/db/exec/sbe/stages/sorted_merge.cpp
+++ b/src/mongo/db/exec/sbe/stages/sorted_merge.cpp
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/sbe/stages/sorted_merge.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 
diff --git a/src/mongo/db/exec/sbe/stages/spool.cpp b/src/mongo/db/exec/sbe/stages/spool.cpp
index 91ec6321381..341dcc57d90 100644
--- a/src/mongo/db/exec/sbe/stages/spool.cpp
+++ b/src/mongo/db/exec/sbe/stages/spool.cpp
@@ -29,6 +29,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/stages/spool.h"
 
 namespace mongo::sbe {
diff --git a/src/mongo/db/exec/sbe/stages/traverse.cpp b/src/mongo/db/exec/sbe/stages/traverse.cpp
index 9caff25a2ec..d97e382023d 100644
--- a/src/mongo/db/exec/sbe/stages/traverse.cpp
+++ b/src/mongo/db/exec/sbe/stages/traverse.cpp
@@ -29,6 +29,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/size_estimator.h"
 #include "mongo/db/exec/sbe/stages/traverse.h"
 
diff --git a/src/mongo/db/exec/sbe/stages/union.cpp b/src/mongo/db/exec/sbe/stages/union.cpp
index 9d619f2f505..64c6d484ae5 100644
--- a/src/mongo/db/exec/sbe/stages/union.cpp
+++ b/src/mongo/db/exec/sbe/stages/union.cpp
@@ -31,6 +31,7 @@
 
 #include <fmt/format.h>
 
+#include "mongo/db/exec/sbe/expressions/compile_ctx.h"
 #include "mongo/db/exec/sbe/expressions/expression.h"
 #include "mongo/db/exec/sbe/size_estimator.h"