SERVER-72004 Fixed SBE slot disabling and prepare for yielding

17 files changed, 215 insertions, 75 deletions

diff --git a/src/mongo/db/exec/sbe/sbe_loop_join_test.cpp b/src/mongo/db/exec/sbe/sbe_loop_join_test.cpp
index b3d223aa24b..98226915454 100644
--- a/src/mongo/db/exec/sbe/sbe_loop_join_test.cpp
+++ b/src/mongo/db/exec/sbe/sbe_loop_join_test.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/exec/sbe/sbe_plan_stage_test.h"
 #include "mongo/db/exec/sbe/stages/loop_join.h"
+#include "mongo/db/exec/sbe/stages/spool.h"
 #include "mongo/util/assert_util.h"
 
 namespace mongo::sbe {
@@ -166,4 +167,43 @@ TEST_F(LoopJoinStageTest, LoopJoinEqualityPredicate) {
     }
     ASSERT_EQ(i, expected.size());
 }
+
+TEST_F(LoopJoinStageTest, LoopJoinInnerBlockingStage) {
+    auto ctx = makeCompileCtx();
+
+    // Build a scan for the outer loop.
+    auto [outerScanSlot, outerScanStage] = generateVirtualScan(BSON_ARRAY(1 << 2));
+
+    // Build a scan for the inner loop.
+    auto [innerScanSlot, innerScanStage] = generateVirtualScan(BSON_ARRAY(3 << 4 << 5));
+
+    auto spoolStage = makeS<SpoolEagerProducerStage>(
+        std::move(innerScanStage), generateSpoolId(), makeSV(innerScanSlot), kEmptyPlanNodeId);
+
+    // Build and prepare for execution loop join of the two scan stages.
+    auto loopJoin = makeS<LoopJoinStage>(std::move(outerScanStage),
+                                         std::move(spoolStage),
+                                         makeSV(outerScanSlot) /*outerProjects*/,
+                                         makeSV() /*outerCorrelated*/,
+                                         nullptr /*predicate*/,
+                                         kEmptyPlanNodeId);
+
+    prepareTree(ctx.get(), loopJoin.get());
+    auto outer = loopJoin->getAccessor(*ctx, outerScanSlot);
+    auto inner = loopJoin->getAccessor(*ctx, innerScanSlot);
+
+    // Expected output: cartesian product of the two scans.
+    std::vector<std::pair<int, int>> expected{{1, 3}, {1, 4}, {1, 5}, {2, 3}, {2, 4}, {2, 5}};
+    int i = 0;
+    for (auto st = loopJoin->getNext(); st == PlanState::ADVANCED; st = loopJoin->getNext(), i++) {
+        ASSERT_LT(i, expected.size());
+
+        auto [outerTag, outerVal] = outer->copyOrMoveValue();
+        assertValuesEqual(outerTag, outerVal, value::TypeTags::NumberInt32, expected[i].first);
+
+        auto [innerTag, innerVal] = inner->copyOrMoveValue();
+        assertValuesEqual(innerTag, innerVal, value::TypeTags::NumberInt32, expected[i].second);
+    }
+    ASSERT_EQ(i, expected.size());
+}
 }  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/sbe_plan_stage_test.cpp b/src/mongo/db/exec/sbe/sbe_plan_stage_test.cpp
index c2312e19fcd..51078c6d442 100644
--- a/src/mongo/db/exec/sbe/sbe_plan_stage_test.cpp
+++ b/src/mongo/db/exec/sbe/sbe_plan_stage_test.cpp
@@ -69,6 +69,10 @@ PlanStageTestFixture::generateVirtualScanMulti(int32_t numSlots, const BSONArray
 
 void PlanStageTestFixture::prepareTree(CompileCtx* ctx, PlanStage* root) {
     Lock::GlobalLock globalLock{operationContext(), MODE_IS};
+    if (_yieldPolicy) {
+        _yieldPolicy->clearRegisteredPlans();
+        _yieldPolicy->registerPlan(root);
+    }
     root->attachToOperationContext(operationContext());
     root->prepare(*ctx);
     root->open(false);
diff --git a/src/mongo/db/exec/sbe/sbe_plan_stage_test.h b/src/mongo/db/exec/sbe/sbe_plan_stage_test.h
index 36f9c8e85d7..8751e488bca 100644
--- a/src/mongo/db/exec/sbe/sbe_plan_stage_test.h
+++ b/src/mongo/db/exec/sbe/sbe_plan_stage_test.h
@@ -83,15 +83,19 @@ using MakeStageFn = std::function<std::pair<T, std::unique_ptr<PlanStage>>(
  */
 class PlanStageTestFixture : public CatalogTestFixture {
 public:
-    PlanStageTestFixture() = default;
+    PlanStageTestFixture(bool enableYield = true) : _enableYield(enableYield){};
 
     void setUp() override {
         CatalogTestFixture::setUp();
+        _yieldPolicy = _enableYield ? makeYieldPolicy() : nullptr;
         _slotIdGenerator.reset(new value::SlotIdGenerator());
+        _spoolIdGenerator.reset(new value::SpoolIdGenerator());
     }
 
     void tearDown() override {
-        _slotIdGenerator.reset();
+        _spoolIdGenerator.reset(nullptr);
+        _slotIdGenerator.reset(nullptr);
+        _yieldPolicy.reset(nullptr);
         CatalogTestFixture::tearDown();
     }
 
@@ -99,6 +103,14 @@ public:
         return _slotIdGenerator->generate();
     }
 
+    SpoolId generateSpoolId() {
+        return _spoolIdGenerator->generate();
+    }
+
+    PlanYieldPolicySBE* getYieldPolicy() const {
+        return _yieldPolicy.get();
+    }
+
     /**
      * Makes a new CompileCtx suitable for preparing an sbe::PlanStage tree.
      */
@@ -134,7 +146,8 @@ public:
      */
     std::pair<value::SlotId, std::unique_ptr<PlanStage>> generateVirtualScan(value::TypeTags arrTag,
                                                                              value::Value arrVal) {
-        return stage_builder::generateVirtualScan(_slotIdGenerator.get(), arrTag, arrVal);
+        return stage_builder::generateVirtualScan(
+            _slotIdGenerator.get(), arrTag, arrVal, _yieldPolicy.get());
     };
 
     /**
@@ -151,7 +164,7 @@ public:
     std::pair<value::SlotVector, std::unique_ptr<PlanStage>> generateVirtualScanMulti(
         int32_t numSlots, value::TypeTags arrTag, value::Value arrVal) {
         return stage_builder::generateVirtualScanMulti(
-            _slotIdGenerator.get(), numSlots, arrTag, arrVal);
+            _slotIdGenerator.get(), numSlots, arrTag, arrVal, _yieldPolicy.get());
     };
 
     /**
@@ -242,8 +255,28 @@ public:
                       value::Value expectedVal,
                       const MakeStageFn<value::SlotVector>& makeStageMulti);
 
+protected:
+    std::unique_ptr<PlanYieldPolicySBE> makeYieldPolicy() {
+        return std::make_unique<PlanYieldPolicySBE>(
+            PlanYieldPolicy::YieldPolicy::YIELD_AUTO,
+            operationContext()->getServiceContext()->getFastClockSource(),
+            0,
+            Milliseconds::zero(),
+            &_yieldable,
+            nullptr);
+    }
+
 private:
+    class MockYieldable : public Yieldable {
+        void yield() const override {}
+        void restore() const override {}
+    };
+
+    MockYieldable _yieldable;
+    bool _enableYield;
+    std::unique_ptr<PlanYieldPolicySBE> _yieldPolicy;
     std::unique_ptr<value::SlotIdGenerator> _slotIdGenerator;
+    std::unique_ptr<value::SpoolIdGenerator> _spoolIdGenerator;
 };
 
 }  // namespace mongo::sbe
diff --git a/src/mongo/db/exec/sbe/sbe_spool_test.cpp b/src/mongo/db/exec/sbe/sbe_spool_test.cpp
index f7f75188db3..8b879949aae 100644
--- a/src/mongo/db/exec/sbe/sbe_spool_test.cpp
+++ b/src/mongo/db/exec/sbe/sbe_spool_test.cpp
@@ -39,10 +39,6 @@ namespace mongo::sbe {
 
 class SbeSpoolTest : public PlanStageTestFixture {
 public:
-    SpoolId generateSpoolId() {
-        return _spoolIdGenerator.generate();
-    }
-
     /**
      * Given an input subtree 'outerBranch' and a 'spoolId', constructs a plan of the following
      * shape:
@@ -94,9 +90,6 @@ public:
 
         return makeSpoolConsumer<false>(std::move(outerBranch), spoolId);
     }
-
-private:
-    sbe::value::SpoolIdGenerator _spoolIdGenerator;
 };
 
 TEST_F(SbeSpoolTest, SpoolEagerProducerBasic) {
diff --git a/src/mongo/db/exec/sbe/stages/ix_scan.cpp b/src/mongo/db/exec/sbe/stages/ix_scan.cpp
index 91f47c7c1b1..c0ac103efdf 100644
--- a/src/mongo/db/exec/sbe/stages/ix_scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/ix_scan.cpp
@@ -127,16 +127,14 @@ value::SlotAccessor* IndexScanStageBase::getAccessor(CompileCtx& ctx, value::Slo
 
 void IndexScanStageBase::doSaveState(bool relinquishCursor) {
     if (relinquishCursor) {
-        if (slotsAccessible()) {
-            if (_recordAccessor) {
-                prepareForYielding(*_recordAccessor);
-            }
-            if (_recordIdAccessor) {
-                prepareForYielding(*_recordIdAccessor);
-            }
-            for (auto& accessor : _accessors) {
-                prepareForYielding(accessor);
-            }
+        if (_recordAccessor) {
+            prepareForYielding(*_recordAccessor, slotsAccessible());
+        }
+        if (_recordIdAccessor) {
+            prepareForYielding(*_recordIdAccessor, slotsAccessible());
+        }
+        for (auto& accessor : _accessors) {
+            prepareForYielding(accessor, slotsAccessible());
         }
 
         if (_cursor) {
@@ -463,10 +461,10 @@ void SimpleIndexScanStage::doSaveState(bool relinquishCursor) {
     // as the index scan is opened.
     if (_open && relinquishCursor) {
         if (_seekKeyLowHolder) {
-            prepareForYielding(*_seekKeyLowHolder);
+            prepareForYielding(*_seekKeyLowHolder, true);
         }
         if (_seekKeyHighHolder) {
-            prepareForYielding(*_seekKeyHighHolder);
+            prepareForYielding(*_seekKeyHighHolder, true);
         }
     }
 
diff --git a/src/mongo/db/exec/sbe/stages/loop_join.cpp b/src/mongo/db/exec/sbe/stages/loop_join.cpp
index 6f46c099f04..123faa7daf5 100644
--- a/src/mongo/db/exec/sbe/stages/loop_join.cpp
+++ b/src/mongo/db/exec/sbe/stages/loop_join.cpp
@@ -207,7 +207,7 @@ void LoopJoinStage::close() {
 }
 
 void LoopJoinStage::doSaveState(bool relinquishCursor) {
-    if (_isReadingLeftSide || _outerGetNext) {
+    if (_isReadingLeftSide) {
         // If we yield while reading the left side, there is no need to prepareForYielding() data
         // held in the right side, since we will have to re-open it anyway.
         const bool recursive = true;
diff --git a/src/mongo/db/exec/sbe/stages/makeobj.cpp b/src/mongo/db/exec/sbe/stages/makeobj.cpp
index b1d55813b84..a2cb193240b 100644
--- a/src/mongo/db/exec/sbe/stages/makeobj.cpp
+++ b/src/mongo/db/exec/sbe/stages/makeobj.cpp
@@ -405,11 +405,11 @@ size_t MakeObjStageBase<O>::estimateCompileTimeSize() const {
 
 template <MakeObjOutputType O>
 void MakeObjStageBase<O>::doSaveState(bool relinquishCursor) {
-    if (!slotsAccessible() || !relinquishCursor) {
+    if (!relinquishCursor) {
         return;
     }
 
-    prepareForYielding(_obj);
+    prepareForYielding(_obj, slotsAccessible());
 }
 
 // Explicit template instantiations.
diff --git a/src/mongo/db/exec/sbe/stages/merge_join.cpp b/src/mongo/db/exec/sbe/stages/merge_join.cpp
index d6f03af7502..b85ea831d24 100644
--- a/src/mongo/db/exec/sbe/stages/merge_join.cpp
+++ b/src/mongo/db/exec/sbe/stages/merge_join.cpp
@@ -325,13 +325,13 @@ void MergeJoinStage::close() {
 }
 
 void MergeJoinStage::doSaveState(bool relinquishCursor) {
-    if (!slotsAccessible() || !relinquishCursor) {
+    if (!relinquishCursor) {
         return;
     }
 
     // We only have to save shallow non-owning materialized rows.
-    prepareForYielding(_currentOuterKey);
-    prepareForYielding(_currentInnerKey);
+    prepareForYielding(_currentOuterKey, true);
+    prepareForYielding(_currentInnerKey, true);
 }
 
 std::unique_ptr<PlanStageStats> MergeJoinStage::getStats(bool includeDebugInfo) const {
diff --git a/src/mongo/db/exec/sbe/stages/project.cpp b/src/mongo/db/exec/sbe/stages/project.cpp
index c534c5c8cdc..a645eb33249 100644
--- a/src/mongo/db/exec/sbe/stages/project.cpp
+++ b/src/mongo/db/exec/sbe/stages/project.cpp
@@ -159,13 +159,13 @@ size_t ProjectStage::estimateCompileTimeSize() const {
 }
 
 void ProjectStage::doSaveState(bool relinquishCursor) {
-    if (!slotsAccessible() || !relinquishCursor) {
+    if (!relinquishCursor) {
         return;
     }
 
     for (auto& [slotId, codeAndAccessor] : _fields) {
         auto& [code, accessor] = codeAndAccessor;
-        prepareForYielding(accessor);
+        prepareForYielding(accessor, slotsAccessible());
     }
 }
 
diff --git a/src/mongo/db/exec/sbe/stages/scan.cpp b/src/mongo/db/exec/sbe/stages/scan.cpp
index 12581bf31d3..379fb9badc2 100644
--- a/src/mongo/db/exec/sbe/stages/scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/scan.cpp
@@ -183,8 +183,8 @@ value::SlotAccessor* ScanStage::getAccessor(CompileCtx& ctx, value::SlotId slot)
 }
 
 void ScanStage::doSaveState(bool relinquishCursor) {
-    if (slotsAccessible()) {
 #if defined(MONGO_CONFIG_DEBUG_BUILD)
+    if (slotsAccessible()) {
         if (_recordAccessor &&
             _recordAccessor->getViewOfValue().first != value::TypeTags::Nothing) {
             auto [tag, val] = _recordAccessor->getViewOfValue();
@@ -195,18 +195,22 @@ void ScanStage::doSaveState(bool relinquishCursor) {
             _lastReturned.clear();
             _lastReturned.assign(raw, raw + size);
         }
+    }
 #endif
 
-        if (relinquishCursor) {
-            if (_recordAccessor) {
-                prepareForYielding(*_recordAccessor);
-            }
-            if (_recordIdAccessor) {
-                prepareForYielding(*_recordIdAccessor);
-            }
-            for (auto& [fieldName, accessor] : _fieldAccessors) {
-                prepareForYielding(*accessor);
-            }
+    if (relinquishCursor) {
+        if (_recordAccessor) {
+            prepareForYielding(*_recordAccessor, slotsAccessible());
+        }
+        if (_recordIdAccessor) {
+            // TODO: SERVER-72054
+            // RecordId are currently (incorrectly) accessed after EOF, therefore
+            // we must treat them as always accessible ratther invalidate them when slots are
+            // disabled. We should use slotsAccessible() instead of true, once the bug is fixed.
+            prepareForYielding(*_recordIdAccessor, true);
+        }
+        for (auto& [fieldName, accessor] : _fieldAccessors) {
+            prepareForYielding(*accessor, slotsAccessible());
         }
     }
 
@@ -216,7 +220,6 @@ void ScanStage::doSaveState(bool relinquishCursor) {
     }
 #endif
 
-
     if (auto cursor = getActiveCursor(); cursor != nullptr && relinquishCursor) {
         cursor->save();
     }
@@ -765,8 +768,8 @@ value::SlotAccessor* ParallelScanStage::getAccessor(CompileCtx& ctx, value::Slot
 }
 
 void ParallelScanStage::doSaveState(bool relinquishCursor) {
-    if (slotsAccessible()) {
 #if defined(MONGO_CONFIG_DEBUG_BUILD)
+    if (slotsAccessible()) {
         if (_recordAccessor &&
             _recordAccessor->getViewOfValue().first != value::TypeTags::Nothing) {
             auto [tag, val] = _recordAccessor->getViewOfValue();
@@ -777,17 +780,21 @@ void ParallelScanStage::doSaveState(bool relinquishCursor) {
             _lastReturned.clear();
             _lastReturned.assign(raw, raw + size);
         }
+    }
 #endif
 
-        if (_recordAccessor) {
-            prepareForYielding(*_recordAccessor);
-        }
-        if (_recordIdAccessor) {
-            prepareForYielding(*_recordIdAccessor);
-        }
-        for (auto& [fieldName, accessor] : _fieldAccessors) {
-            prepareForYielding(*accessor);
-        }
+    if (_recordAccessor) {
+        prepareForYielding(*_recordAccessor, slotsAccessible());
+    }
+    if (_recordIdAccessor) {
+        // TODO: SERVER-72054
+        // RecordId are currently (incorrectly) accessed after EOF, therefore
+        // we must treat them as always accessible ratther invalidate them when slots are
+        // disabled. We should use slotsAccessible() instead of true, once the bug is fixed.
+        prepareForYielding(*_recordIdAccessor, true);
+    }
+    for (auto& [fieldName, accessor] : _fieldAccessors) {
+        prepareForYielding(*accessor, slotsAccessible());
     }
 
 #if defined(MONGO_CONFIG_DEBUG_BUILD)
diff --git a/src/mongo/db/exec/sbe/stages/spool.cpp b/src/mongo/db/exec/sbe/stages/spool.cpp
index 47ca744962c..91ec6321381 100644
--- a/src/mongo/db/exec/sbe/stages/spool.cpp
+++ b/src/mongo/db/exec/sbe/stages/spool.cpp
@@ -285,12 +285,12 @@ PlanState SpoolLazyProducerStage::getNext() {
 }
 
 void SpoolLazyProducerStage::doSaveState(bool relinquishCursor) {
-    if (!slotsAccessible() || !relinquishCursor) {
+    if (!relinquishCursor) {
         return;
     }
 
     for (auto& [slot, accessor] : _outAccessors) {
-        prepareForYielding(accessor);
+        prepareForYielding(accessor, slotsAccessible());
     }
 }
 
diff --git a/src/mongo/db/exec/sbe/stages/stages.h b/src/mongo/db/exec/sbe/stages/stages.h
index 4f3fc836ae4..c34477c56c0 100644
--- a/src/mongo/db/exec/sbe/stages/stages.h
+++ b/src/mongo/db/exec/sbe/stages/stages.h
@@ -124,19 +124,35 @@ public:
      * Ensures that accessor owns the underlying BSON value, which can be potentially owned by
      * storage.
      */
-    static void prepareForYielding(value::OwnedValueAccessor& accessor) {
-        auto [tag, value] = accessor.getViewOfValue();
-        if (shouldCopyValue(tag)) {
-            accessor.makeOwned();
+    void prepareForYielding(value::OwnedValueAccessor& accessor, bool isAccessible) {
+        if (isAccessible) {
+            auto [tag, value] = accessor.getViewOfValue();
+            if (shouldCopyValue(tag)) {
+                accessor.makeOwned();
+            }
+        } else {
+#if defined(MONGO_CONFIG_DEBUG_BUILD)
+            auto [tag, val] = value::getPoisonValue();
+            accessor.reset(false, tag, val);
+#endif
         }
     }
 
-    static void prepareForYielding(value::MaterializedRow& row) {
-        for (size_t idx = 0; idx < row.size(); idx++) {
-            auto [tag, value] = row.getViewOfValue(idx);
-            if (shouldCopyValue(tag)) {
-                row.makeOwned(idx);
+    void prepareForYielding(value::MaterializedRow& row, bool isAccessible) {
+        if (isAccessible) {
+            for (size_t idx = 0; idx < row.size(); idx++) {
+                auto [tag, value] = row.getViewOfValue(idx);
+                if (shouldCopyValue(tag)) {
+                    row.makeOwned(idx);
+                }
+            }
+        } else {
+#if defined(MONGO_CONFIG_DEBUG_BUILD)
+            auto [tag, val] = value::getPoisonValue();
+            for (size_t idx = 0; idx < row.size(); idx++) {
+                row.reset(idx, false, tag, val);
             }
+#endif
         }
     }
 
@@ -387,6 +403,10 @@ public:
         _participateInTrialRunTracking = false;
     }
 
+    bool slotsAccessible() const {
+        return _slotsAccessible;
+    }
+
 protected:
     PlanState trackPlanState(PlanState state) {
         if (state == PlanState::IS_EOF) {
@@ -404,10 +424,6 @@ protected:
         _slotsAccessible = false;
     }
 
-    bool slotsAccessible() const {
-        return _slotsAccessible;
-    }
-
     /**
      * Returns an optional timer which is used to collect time spent executing the current stage.
      * May return boost::none if it is not necessary to collect timing info.
diff --git a/src/mongo/db/exec/sbe/stages/traverse.cpp b/src/mongo/db/exec/sbe/stages/traverse.cpp
index a95613f7be6..9caff25a2ec 100644
--- a/src/mongo/db/exec/sbe/stages/traverse.cpp
+++ b/src/mongo/db/exec/sbe/stages/traverse.cpp
@@ -290,11 +290,11 @@ void TraverseStage::doSaveState(bool relinquishCursor) {
         _outFieldOutputAccessor.reset();
     }
 
-    if (!slotsAccessible() || !relinquishCursor) {
+    if (!relinquishCursor) {
         return;
     }
 
-    prepareForYielding(_outFieldOutputAccessor);
+    prepareForYielding(_outFieldOutputAccessor, slotsAccessible());
 }
 
 void TraverseStage::doRestoreState(bool relinquishCursor) {
diff --git a/src/mongo/db/exec/sbe/stages/unwind.cpp b/src/mongo/db/exec/sbe/stages/unwind.cpp
index 7ad10eecb23..6b19c49849d 100644
--- a/src/mongo/db/exec/sbe/stages/unwind.cpp
+++ b/src/mongo/db/exec/sbe/stages/unwind.cpp
@@ -27,6 +27,7 @@
  *    it in the license file.
  */
 
+#include "mongo/config.h"
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/exec/sbe/stages/unwind.h"
@@ -41,8 +42,9 @@ UnwindStage::UnwindStage(std::unique_ptr<PlanStage> input,
                          value::SlotId outIndex,
                          bool preserveNullAndEmptyArrays,
                          PlanNodeId planNodeId,
+                         PlanYieldPolicy* yieldPolicy,
                          bool participateInTrialRunTracking)
-    : PlanStage("unwind"_sd, planNodeId, participateInTrialRunTracking),
+    : PlanStage("unwind"_sd, yieldPolicy, planNodeId, participateInTrialRunTracking),
       _inField(inField),
       _outField(outField),
       _outIndex(outIndex),
@@ -61,6 +63,7 @@ std::unique_ptr<PlanStage> UnwindStage::clone() const {
                                          _outIndex,
                                          _preserveNullAndEmptyArrays,
                                          _commonStats.nodeId,
+                                         _yieldPolicy,
                                          _participateInTrialRunTracking);
 }
 
@@ -102,6 +105,17 @@ void UnwindStage::open(bool reOpen) {
 PlanState UnwindStage::getNext() {
     auto optTimer(getOptTimer(_opCtx));
 
+    checkForInterrupt(_opCtx);
+
+#if defined(MONGO_CONFIG_DEBUG_BUILD)
+    // If we are iterating over the array, then we are not going to advance the child.
+    // In such case, we expect that child slots are accessible, as otherwise our enumerator state is
+    // not valid.
+    tassert(7200405,
+            "Child stage slots must be accessible while iterating over array elements",
+            !_inArray || _children[0]->slotsAccessible());
+#endif
+
     if (!_inArray) {
         do {
             // We are about to call getNext() on our child so do not bother saving our internal
@@ -204,15 +218,14 @@ std::vector<DebugPrinter::Block> UnwindStage::debugPrint() const {
 }
 
 void UnwindStage::doSaveState(bool fullSave) {
-    if (!slotsAccessible() || !fullSave) {
+    if (!fullSave) {
         return;
     }
-
     if (_outFieldOutputAccessor) {
-        prepareForYielding(*_outFieldOutputAccessor);
+        prepareForYielding(*_outFieldOutputAccessor, slotsAccessible());
     }
     if (_outIndexOutputAccessor) {
-        prepareForYielding(*_outIndexOutputAccessor);
+        prepareForYielding(*_outIndexOutputAccessor, slotsAccessible());
     }
 }
 
diff --git a/src/mongo/db/exec/sbe/stages/unwind.h b/src/mongo/db/exec/sbe/stages/unwind.h
index 57b28d9c1cf..58c519d424f 100644
--- a/src/mongo/db/exec/sbe/stages/unwind.h
+++ b/src/mongo/db/exec/sbe/stages/unwind.h
@@ -53,6 +53,7 @@ public:
                 value::SlotId outIndex,
                 bool preserveNullAndEmptyArrays,
                 PlanNodeId planNodeId,
+                PlanYieldPolicy* yieldPolicy = nullptr,
                 bool participateInTrialRunTracking = true);
 
     std::unique_ptr<PlanStage> clone() const final;
diff --git a/src/mongo/db/exec/sbe/values/slot.h b/src/mongo/db/exec/sbe/values/slot.h
index f853f816d4d..efd92255c9f 100644
--- a/src/mongo/db/exec/sbe/values/slot.h
+++ b/src/mongo/db/exec/sbe/values/slot.h
@@ -31,6 +31,7 @@
 
 #include <type_traits>
 
+#include "mongo/config.h"
 #include "mongo/db/exec/sbe/values/value.h"
 #include "mongo/util/id_generator.h"
 
@@ -74,6 +75,21 @@ public:
     virtual std::pair<TypeTags, Value> copyOrMoveValue() = 0;
 };
 
+class SlotAccessorHelper {
+public:
+    /**
+     * Asserts that a value has not been invalidated.
+     */
+    MONGO_COMPILER_ALWAYS_INLINE static void dassertValidSlotValue(TypeTags tag, Value val) {
+#if defined(MONGO_CONFIG_DEBUG_BUILD)
+        tassert(7200401,
+                str::stream() << "Unexpected poison value. Likely attempted to access a slot that "
+                                 "was disabled and invalidated in save/restore cycle.",
+                !value::isPoisonValue(tag, val));
+#endif
+    }
+};
+
 /**
  * Accessor for a slot which provides a view of a value that is owned elsewhere.
  */
@@ -151,6 +167,7 @@ public:
      * Returns a non-owning view of the value.
      */
     std::pair<TypeTags, Value> getViewOfValue() const override {
+        SlotAccessorHelper::dassertValidSlotValue(_tag, _val);
         return {_tag, _val};
     }
 
@@ -160,6 +177,7 @@ public:
      * the caller owns the resulting value.
      */
     std::pair<TypeTags, Value> copyOrMoveValue() override {
+        SlotAccessorHelper::dassertValidSlotValue(_tag, _val);
         if (_owned) {
             _owned = false;
             return {_tag, _val};
@@ -416,10 +434,12 @@ public:
     }
 
     std::pair<value::TypeTags, value::Value> getViewOfValue(size_t idx) const {
+        SlotAccessorHelper::dassertValidSlotValue(tags()[idx], values()[idx]);
         return {tags()[idx], values()[idx]};
     }
 
     std::pair<value::TypeTags, value::Value> copyOrMoveValue(size_t idx) {
+        SlotAccessorHelper::dassertValidSlotValue(tags()[idx], values()[idx]);
         if (owned()[idx]) {
             owned()[idx] = false;
             return {tags()[idx], values()[idx]};
diff --git a/src/mongo/db/exec/sbe/values/value.h b/src/mongo/db/exec/sbe/values/value.h
index 1e245452f26..b647a27fb72 100644
--- a/src/mongo/db/exec/sbe/values/value.h
+++ b/src/mongo/db/exec/sbe/values/value.h
@@ -43,6 +43,7 @@
 #include "mongo/base/data_type_endian.h"
 #include "mongo/base/data_view.h"
 #include "mongo/bson/ordering.h"
+#include "mongo/config.h"
 #include "mongo/db/exec/shard_filterer.h"
 #include "mongo/db/fts/fts_matcher.h"
 #include "mongo/db/query/bson_typemask.h"
@@ -1388,6 +1389,20 @@ std::pair<TypeTags, Value> makeCopyCollator(const CollatorInterface& collator);
 
 std::pair<TypeTags, Value> makeCopyIndexBounds(const IndexBounds& collator);
 
+#if defined(MONGO_CONFIG_DEBUG_BUILD)
+/**
+ * Returns a poison value that should never be encountered in production.
+ * Used by asserts/invariants to invalidate values that should never be accessed.
+ */
+inline std::pair<TypeTags, Value> getPoisonValue() {
+    return {TypeTags::Nothing, (uint64_t)-1};
+}
+
+inline bool isPoisonValue(TypeTags tag, Value val) {
+    return tag == TypeTags::Nothing && val == (uint64_t)-1;
+}
+#endif
+
 inline std::pair<TypeTags, Value> copyValue(TypeTags tag, Value val) {
     switch (tag) {
         case TypeTags::RecordId: