[NEW RBO] Make operator traversal lazy with explicit pre/post-order (#46092)

13 files changed, 356 insertions, 149 deletions

diff --git a/ydb/core/kqp/opt/rbo/analysis/logical_aliases.cpp b/ydb/core/kqp/opt/rbo/analysis/logical_aliases.cpp
index 100a29ab42f..3c01a70e9a9 100644
--- a/ydb/core/kqp/opt/rbo/analysis/logical_aliases.cpp
+++ b/ydb/core/kqp/opt/rbo/analysis/logical_aliases.cpp
@@ -291,7 +291,7 @@ TPlanAliases::TAliasMap TOpSort::ComputeAliases() {
 }
 
 void ComputePlanAliases(TOpRoot& root) {
-    const auto traversal = root.PostOrder();
+    const auto traversal = root.SnapshotTraversal();
 
     for (const auto& iter : traversal) {
         const auto& op = iter.Current;
diff --git a/ydb/core/kqp/opt/rbo/analysis/logical_name_constraints.cpp b/ydb/core/kqp/opt/rbo/analysis/logical_name_constraints.cpp
index 43148b6e331..3b3c6d25a2a 100644
--- a/ydb/core/kqp/opt/rbo/analysis/logical_name_constraints.cpp
+++ b/ydb/core/kqp/opt/rbo/analysis/logical_name_constraints.cpp
@@ -61,7 +61,7 @@ bool PropagateForbidden(const TIntrusivePtr<IOperator>& op, const TInfoUnitConst
 class TLogicalNameConstraints {
 public:
     explicit TLogicalNameConstraints(TOpRoot& root)
-        : Traversal(root.PostOrder()) {
+        : Traversal(root.SnapshotTraversal()) {
     }
 
     void Run() {
diff --git a/ydb/core/kqp/opt/rbo/kqp_operator.cpp b/ydb/core/kqp/opt/rbo/kqp_operator.cpp
index f04d7f1a097..8ff0044996d 100644
--- a/ydb/core/kqp/opt/rbo/kqp_operator.cpp
+++ b/ydb/core/kqp/opt/rbo/kqp_operator.cpp
@@ -74,11 +74,8 @@ const TVector<TInfoUnit>& IOperator::GetOutputIUs() {
 // (this is done because we need a smart pointer for iteration, but only have a raw this)
 void IOperator::ComputeOutputIUsSubtree() {
     for (auto& op : GetChildren()) {
-        TOpIterator begin(op, nullptr);
-        TOpIterator end(nullptr);
-
-        for (TOpIterator it = begin; it != end; it++) {
-            (*it).Current->ComputeOutputIUs();
+        for (const auto& item : IterateSubtree(op)) {
+            item.Current->ComputeOutputIUs();
         }
     }
 
@@ -1210,51 +1207,20 @@ void TOpRoot::ComputeOutputIUs() {
 // Need to override root recomputation of IUs, since it
 // needs to traverse all subplans as well.
 void TOpRoot::ComputeOutputIUsSubtree() {
-    for (auto it : *this) {
+    for (const auto& it : *this) {
         it.Current->ComputeOutputIUs();
     }
     ComputeOutputIUs();
 }
 
-void TOpRoot::ClearParentsRec(TIntrusivePtr<IOperator> op, std::unordered_set<IOperator*>& visited) const {
-    if (!op || !visited.insert(op.get()).second) {
-        return;
-    }
-
-    op->Parents.clear();
-    for (const auto& child : op->Children) {
-        ClearParentsRec(child, visited);
-    }
-}
-
-void TOpRoot::ComputeParentsRec(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent, ui32 parentChildIndex) const {
-    if (parent) {
-        const auto parentEntry = std::make_pair(parent.get(), parentChildIndex);
-        const auto it = std::find_if(op->Parents.begin(), op->Parents.end(), [&parentEntry](const std::pair<IOperator*, ui32>& opParent) {
-            return opParent.first == parentEntry.first && opParent.second == parentEntry.second;
-        });
-        if (it == op->Parents.end()) {
-            op->Parents.push_back(parentEntry);
-        }
-    }
-    for (size_t childIndex = 0; childIndex < op->Children.size(); ++childIndex) {
-        ComputeParentsRec(op->Children[childIndex], op, childIndex);
-    }
-}
-
 void TOpRoot::ComputeParents() {
-    std::unordered_set<IOperator*> visited;
-    ClearParentsRec(GetInput(), visited);
-    const auto subPlans = PlanProps.Subplans.Get();
-    for (const auto& subPlan : subPlans) {
-        ClearParentsRec(CastOperator<IOperator>(subPlan.Plan), visited);
-    }
-
-    TIntrusivePtr<TOpRoot> noParent = nullptr;
-    ComputeParentsRec(GetInput(), noParent, 0);
-
-    for (const auto& subPlan : subPlans) {
-        ComputeParentsRec(CastOperator<IOperator>(subPlan.Plan), noParent, 0);
+    // Root postorder visits active operators once and clears every child before its parents add edges.
+    for (const auto& item : *this) {
+        auto& op = item.Current;
+        op->Parents.clear();
+        for (ui32 childIndex = 0; childIndex < op->Children.size(); ++childIndex) {
+            op->Children[childIndex]->Parents.emplace_back(op.Get(), childIndex);
+        }
     }
 }
 
@@ -1301,79 +1267,120 @@ void TOpRoot::PlanToStringRec(TIntrusivePtr<IOperator> op, TExprContext& ctx, TS
     }
 }
 
-TOpIterator::TOpIterator(TOpRoot* ptr) {
-    if (!ptr) {
-        CurrElement = -1;
-        return;
-    }
-
-    std::unordered_set<IOperator*> visited;
-    auto child = ptr->GetInput();
-    BuildDfsList(DfsList, child, {}, size_t(0), visited, &ptr->PlanProps, nullptr, true);
-    CurrElement = 0;
+TOpIterator::TOpIterator(TIntrusivePtr<IOperator> op, TPlanProps* props, bool followSubplans, ETraversalOrder order)
+    : PlanProps(props)
+    , RecurseIntoSubplans(followSubplans)
+    , Order(order) {
+    Y_ENSURE(!followSubplans || props, "Following subplans requires plan properties");
+    // One allocation covers 98.8% of measured TPCH/TPCDS traversals.
+    Visited.reserve(96);
+    PushFrame(op, nullptr, size_t(0), nullptr);
+    Advance();
 }
 
-TOpIterator::TOpIterator(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent) {
-    std::unordered_set<IOperator*> visited;
-    BuildDfsList(DfsList, op, parent, size_t(0), visited, nullptr, nullptr);
-    CurrElement = 0;
+TOpIterator::TOpIterator(TOpIterator&& other)
+    : Stack(std::make_move_iterator(other.Stack.begin()), std::make_move_iterator(other.Stack.end()))
+    , Visited(std::move(other.Visited))
+    , Current(std::move(other.Current))
+    , PlanProps(other.PlanProps)
+    , RecurseIntoSubplans(other.RecurseIntoSubplans)
+    , Order(other.Order)
+    , AtEnd(other.AtEnd) {
+    other.Stack.clear();
+    other.Visited.clear();
+    other.Current = {};
+    other.PlanProps = nullptr;
+    other.RecurseIntoSubplans = false;
+    other.Order = ETraversalOrder::PostOrder;
+    other.AtEnd = true;
 }
 
-TOpIterator::TOpIterator(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent, TPlanProps* props) {
-    std::unordered_set<IOperator*> visited;
-    BuildDfsList(DfsList, op, parent, size_t(0), visited, props, nullptr, true);
-    CurrElement = 0;
+TOpIterator& TOpIterator::operator=(TOpIterator&& other) {
+    if (this != &other) {
+        Stack.assign(std::make_move_iterator(other.Stack.begin()), std::make_move_iterator(other.Stack.end()));
+        Visited = std::move(other.Visited);
+        Current = std::move(other.Current);
+        PlanProps = other.PlanProps;
+        RecurseIntoSubplans = other.RecurseIntoSubplans;
+        Order = other.Order;
+        AtEnd = other.AtEnd;
+
+        other.Stack.clear();
+        other.Visited.clear();
+        other.Current = {};
+        other.PlanProps = nullptr;
+        other.RecurseIntoSubplans = false;
+        other.Order = ETraversalOrder::PostOrder;
+        other.AtEnd = true;
+    }
+    return *this;
 }
 
-TOpIterator::TIteratorItem TOpIterator::operator*() const {
-    return DfsList[CurrElement];
+const TOpIterator::TIteratorItem& TOpIterator::operator*() const {
+    return Current;
 }
 
 TOpIterator& TOpIterator::operator++() {
-    if (CurrElement >= 0) {
-        CurrElement++;
-    }
-    if (CurrElement == DfsList.size()) {
-        CurrElement = -1;
-    }
+    Advance();
     return *this;
 }
 
-TOpIterator TOpIterator::operator++(int) {
-    TOpIterator tmp = *this;
+void TOpIterator::operator++(int) {
     ++(*this);
-    return tmp;
 }
 
-void TOpIterator::BuildDfsList(TVector<TIteratorItem>& dfsList, TIntrusivePtr<IOperator> current, TIntrusivePtr<IOperator> parent, size_t childIdx,
-                          std::unordered_set<IOperator*>& visited, TPlanProps* planProps, std::shared_ptr<TInfoUnit> subplanIU, bool recurseIntoSubplans) {
-
-    if(recurseIntoSubplans) {
-        Y_ENSURE(planProps);
-        auto subplanIUs = current->GetSubplanIUs(*planProps);
-        for (const auto & iu : subplanIUs) {
-            const auto & subplan = planProps->Subplans.PlanMap.at(iu);
-            BuildDfsList(dfsList, CastOperator<IOperator>(subplan.Plan), nullptr, 0, visited, planProps, std::make_shared<TInfoUnit>(iu), true);
-        }
+bool TOpIterator::PushFrame(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent, size_t childIdx, std::shared_ptr<TInfoUnit> subplanIU) {
+    if (!op || !Visited.insert(op.get()).second) {
+        return false;
     }
 
-    const auto& children = current->GetChildren();
-    for (size_t idx = 0, e = children.size(); idx < e; ++idx) {
-        BuildDfsList(dfsList, children[idx], current, idx, visited, planProps, subplanIU, recurseIntoSubplans);
-    }
-    if (!visited.contains(current.get())) {
-        dfsList.push_back(TOpIterator::TIteratorItem(current, parent, childIdx, subplanIU));
-    }
-    visited.insert(current.get());
+    Stack.emplace_back(op, parent, childIdx, subplanIU);
+    return true;
 }
 
-TOpTraversal::TOpTraversal(TOpRoot* root) {
-    if (!root) {
-        return;
+void TOpIterator::Advance() {
+    while (!Stack.empty()) {
+        auto& frame = Stack.back();
+
+        if (Order == ETraversalOrder::PreOrder && !frame.Emitted) {
+            frame.Emitted = true;
+            Current = TIteratorItem(frame.Current, frame.Parent, frame.ChildIndex, frame.SubplanIU);
+            AtEnd = false;
+            return;
+        }
+
+        if (RecurseIntoSubplans && !frame.SubplansLoaded) {
+            Y_ENSURE(PlanProps);
+            frame.SubplanIUs = frame.Current->GetSubplanIUs(*PlanProps);
+            frame.SubplansLoaded = true;
+        }
+
+        if (RecurseIntoSubplans && frame.NextSubplanIdx < frame.SubplanIUs.size()) {
+            const auto& iu = frame.SubplanIUs[frame.NextSubplanIdx++];
+            const auto& subplan = PlanProps->Subplans.PlanMap.at(iu);
+            PushFrame(CastOperator<IOperator>(subplan.Plan), nullptr, size_t(0), std::make_shared<TInfoUnit>(iu));
+            continue;
+        }
+
+        const auto& children = frame.Current->GetChildren();
+        if (frame.NextChildIdx < children.size()) {
+            const auto childIdx = frame.NextChildIdx++;
+            PushFrame(children[childIdx], frame.Current, childIdx, frame.SubplanIU);
+            continue;
+        }
+
+        if (Order == ETraversalOrder::PostOrder) {
+            Current = TIteratorItem(frame.Current, frame.Parent, frame.ChildIndex, frame.SubplanIU);
+            Stack.pop_back();
+            AtEnd = false;
+            return;
+        }
+
+        Stack.pop_back();
     }
 
-    std::unordered_set<IOperator*> visited;
-    TOpIterator::BuildDfsList(Items, root->GetInput(), {}, size_t(0), visited, &root->PlanProps, nullptr, true);
+    Current = TIteratorItem();
+    AtEnd = true;
 }
 
 TString ToStringPhase(EOpPhase phase) {
diff --git a/ydb/core/kqp/opt/rbo/kqp_operator.h b/ydb/core/kqp/opt/rbo/kqp_operator.h
index cb3d5839f4e..9301966145f 100644
--- a/ydb/core/kqp/opt/rbo/kqp_operator.h
+++ b/ydb/core/kqp/opt/rbo/kqp_operator.h
@@ -9,6 +9,8 @@
 #include <cstddef>
 #include <iterator>
 #include <optional>
+#include <library/cpp/containers/absl/flat_hash_set.h>
+#include <library/cpp/containers/stack_vector/stack_vec.h>
 #include <ydb/core/kqp/common/kqp_yql.h>
 #include <ydb/core/kqp/opt/kqp_opt.h>
 #include <yql/essentials/ast/yql_expr.h>
@@ -770,8 +772,32 @@ private:
     void RebuildChildren();
 };
 
+// End-of-traversal sentinel for TOpIterator
+struct TOpEnd {};
+
+// Order in which a traversal emits operators
+enum class ETraversalOrder {
+    // Referenced subplans and children before the node (dependencies first)
+    PostOrder,
+    // The node before its referenced subplans and children
+    PreOrder,
+};
+
+/**
+ * Lazy plan traversal with an explicit frame stack: O(depth) state plus a visited
+ * set for DAG dedup. Constructed through the range factories: TOpRoot::Iterate,
+ * IterateSubtree, IterateSubtreeWithSubplans.
+ *
+ * Mutation contract: the iterator walks live GetChildren() edges and keeps raw
+ * operator pointers in its visited set, so it must not be advanced after the plan
+ * is mutated. Mutate, then stop iterating and build a fresh iterator (as the rule
+ * engine does), or use a TOpTraversal snapshot when iteration must survive
+ * mutation.
+ */
 struct TOpIterator {
     struct TIteratorItem {
+        TIteratorItem() = default;
+
         TIteratorItem(TIntrusivePtr<IOperator> curr, TIntrusivePtr<IOperator> parent, size_t idx, std::shared_ptr<TInfoUnit> subplanIU)
             : Current(curr)
             , Parent(parent)
@@ -780,50 +806,143 @@ struct TOpIterator {
         }
 
         TIntrusivePtr<IOperator> Current;
+        // Parent/ChildIndex/SubplanIU describe the traversal edge along which the
+        // node was first reached — a property of this traversal, not of the graph.
+        // A shared (DAG) node may be reached via a different parent under a
+        // different traversal order; use IOperator::Parents for graph structure.
+        TIntrusivePtr<IOperator> Parent;
+        size_t ChildIndex = 0;
+        std::shared_ptr<TInfoUnit> SubplanIU;
+    };
+
+private:
+    struct TFrame {
+        TFrame(TIntrusivePtr<IOperator> current, TIntrusivePtr<IOperator> parent, size_t childIdx, std::shared_ptr<TInfoUnit> subplanIU)
+            : Current(current)
+            , Parent(parent)
+            , ChildIndex(childIdx)
+            , SubplanIU(subplanIU) {
+        }
+
+        TFrame(const TFrame&) = delete;
+        TFrame& operator=(const TFrame&) = delete;
+        TFrame(TFrame&&) noexcept = default;
+        TFrame& operator=(TFrame&&) noexcept = default;
+
+        TIntrusivePtr<IOperator> Current;
         TIntrusivePtr<IOperator> Parent;
-        size_t ChildIndex;
+        size_t ChildIndex = 0;
         std::shared_ptr<TInfoUnit> SubplanIU;
+        TVector<TInfoUnit> SubplanIUs;
+        size_t NextSubplanIdx = 0;
+        size_t NextChildIdx = 0;
+        bool SubplansLoaded = false;
+        // Pre-order only: the node was already emitted when its frame was entered
+        bool Emitted = false;
     };
 
+    // Only the range factories construct iterators
+    friend class TOpRange;
+
+    TOpIterator(TIntrusivePtr<IOperator> op, TPlanProps* props, bool followSubplans, ETraversalOrder order);
+
+public:
     using iterator_category = std::input_iterator_tag;
     using difference_type = std::ptrdiff_t;
+    using value_type = TIteratorItem;
+    using reference = const TIteratorItem&;
+    using pointer = const TIteratorItem*;
 
-    // Build a default iterator for the root of the plan, following subplans
-    // referenced by expressions in DFS order.
-    TOpIterator(TOpRoot* ptr);
+    // The iterator carries a traversal stack and a visited set, so copying is
+    // expensive and never needed. Moving relocates only the live stack frames.
+    TOpIterator(const TOpIterator&) = delete;
+    TOpIterator& operator=(const TOpIterator&) = delete;
+    TOpIterator(TOpIterator&& other);
+    TOpIterator& operator=(TOpIterator&& other);
 
-    // Build an iterator for traversing the children of specific operator
-    TOpIterator(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent);
+    const TIteratorItem& operator*() const;
 
-    // Build an iterator to travese the children of a specific iterator, recursing into
-    // subplans, as their UIs are encountered
-    TOpIterator(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent, TPlanProps* props);
-
-    TIteratorItem operator*() const;
+    const TIteratorItem* operator->() const {
+        return &Current;
+    }
 
     // Prefix increment
     TOpIterator& operator++();
 
-    // Postfix increment
-    TOpIterator operator++(int);
+    // Postfix increment returns void: returning the pre-increment iterator
+    // would require copying the traversal state
+    void operator++(int);
 
-    friend bool operator==(const TOpIterator& a, const TOpIterator& b) {
-        return a.CurrElement == b.CurrElement;
-    };
-    friend bool operator!=(const TOpIterator& a, const TOpIterator& b) {
-        return a.CurrElement != b.CurrElement;
-    };
+    friend bool operator==(const TOpIterator& a, TOpEnd) {
+        return a.AtEnd;
+    }
+    friend bool operator!=(const TOpIterator& a, TOpEnd) {
+        return !a.AtEnd;
+    }
 
 private:
-    friend class TOpTraversal;
+    bool PushFrame(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent, size_t childIdx, std::shared_ptr<TInfoUnit> subplanIU);
+    void Advance();
 
-    static void BuildDfsList(TVector<TIteratorItem>& dfsList, TIntrusivePtr<IOperator> current, TIntrusivePtr<IOperator> parent, size_t childIdx,
-        std::unordered_set<IOperator*>& visited, TPlanProps* planProps, std::shared_ptr<TInfoUnit> subplanIU, bool recurseIntoSubplans = false);
+    // Covers more than 90% of measured TPCH/TPCDS traversals without allocating frame storage.
+    TStackVec<TFrame, 24> Stack;
+    absl::flat_hash_set<IOperator*> Visited;
+    TIteratorItem Current;
+    TPlanProps* PlanProps = nullptr;
+    bool RecurseIntoSubplans = false;
+    ETraversalOrder Order = ETraversalOrder::PostOrder;
+    bool AtEnd = true;
+};
 
-    TVector<TIteratorItem> DfsList;
-    size_t CurrElement;
+/**
+ * Lazy traversal range: a cheap value describing what to traverse and in which
+ * order. begin() builds a fresh TOpIterator, end() is the TOpEnd sentinel.
+ * Obtain one from TOpRoot::Iterate, IterateSubtree or IterateSubtreeWithSubplans.
+ */
+class TOpRange {
+public:
+    TOpIterator begin() const {
+        return TOpIterator(Op, Props, FollowSubplans, Order);
+    }
+
+    TOpEnd end() const {
+        return {};
+    }
+
+private:
+    friend class TOpRoot;
+    friend TOpRange IterateSubtree(TIntrusivePtr<IOperator> op, ETraversalOrder order);
+    friend TOpRange IterateSubtreeWithSubplans(TIntrusivePtr<IOperator> op, TPlanProps& props, ETraversalOrder order);
+
+    TOpRange(TIntrusivePtr<IOperator> op, TPlanProps* props, bool followSubplans, ETraversalOrder order)
+        : Op(op)
+        , Props(props)
+        , FollowSubplans(followSubplans)
+        , Order(order) {
+    }
+
+    TIntrusivePtr<IOperator> Op;
+    TPlanProps* Props = nullptr;
+    bool FollowSubplans = false;
+    ETraversalOrder Order;
 };
 
+// Traverse the operators of a single plan, without following subplan references
+inline TOpRange IterateSubtree(TIntrusivePtr<IOperator> op, ETraversalOrder order = ETraversalOrder::PostOrder) {
+    return TOpRange(op, nullptr, false, order);
+}
+
+// Traverse an operator subtree, recursing into subplans referenced by expressions
+inline TOpRange IterateSubtreeWithSubplans(TIntrusivePtr<IOperator> op, TPlanProps& props, ETraversalOrder order = ETraversalOrder::PostOrder) {
+    return TOpRange(op, &props, true, order);
+}
+
+/**
+ * Traversal snapshot: drains a lazy iterator into a vector, so it is
+ * guaranteed to emit the same sequence as the lazy traversal it was built from.
+ * Buys what the lazy iterator cannot offer: reverse iteration, multiple passes,
+ * and validity across plan mutation. Obtain one from TOpRoot::SnapshotTraversal.
+ */
 class TOpTraversal {
 public:
     class TIterator {
@@ -873,7 +992,11 @@ public:
 
     using TReverseIterator = TVector<TOpIterator::TIteratorItem>::const_reverse_iterator;
 
-    explicit TOpTraversal(TOpRoot* root);
+    explicit TOpTraversal(TOpIterator it) {
+        for (; it != TOpEnd{}; ++it) {
+            Items.push_back(*it);
+        }
+    }
 
     TIterator begin() const {
         return TIterator(&Items, 0);
@@ -910,16 +1033,24 @@ public:
     void ComputePlanMetadata(TRBOContext& ctx);
     void ComputePlanStatistics(TRBOContext& ctx);
 
+    // Lazy traversal of the whole plan, following subplan references
+    TOpRange Iterate(ETraversalOrder order) {
+        return TOpRange(GetInput(), &PlanProps, true, order);
+    }
+
+    // Default root traversal preserves the historical dependency-first order.
     TOpIterator begin() {
-        return TOpIterator(this);
+        return Iterate(ETraversalOrder::PostOrder).begin();
     }
 
-    TOpIterator end() {
-        return TOpIterator(nullptr);
+    TOpEnd end() const {
+        return {};
     }
 
-    TOpTraversal PostOrder() {
-        return TOpTraversal(this);
+    // Snapshot of the whole-plan traversal, following subplan references;
+    // supports reverse iteration and stays valid across plan mutation
+    TOpTraversal SnapshotTraversal(ETraversalOrder order = ETraversalOrder::PostOrder) {
+        return TOpTraversal(Iterate(order).begin());
     }
 
     NJson::TJsonValue GetExecutionJson(ui64 & nodeCounter, THashMap<IOperator*, ui32>& operatorIds, ui32 explainFlags = 0x00);
@@ -934,11 +1065,6 @@ protected:
     void ComputeOutputIUsSubtree() override;
 
     friend void EnsureRequiredProps(TOpRoot& root, ui32 props, ui32& computedProps, TRBOContext& ctx, const TString& stageName);
-
-private:
-    void ClearParentsRec(TIntrusivePtr<IOperator> op, std::unordered_set<IOperator*>& visited) const;
-    void ComputeParentsRec(TIntrusivePtr<IOperator> op, TIntrusivePtr<IOperator> parent, ui32 parentChildIndex) const;
-
 };
 
 } // namespace NKqp
diff --git a/ydb/core/kqp/opt/rbo/kqp_plan_conversion_utils.cpp b/ydb/core/kqp/opt/rbo/kqp_plan_conversion_utils.cpp
index aa6779e7693..e5b6dbfb549 100644
--- a/ydb/core/kqp/opt/rbo/kqp_plan_conversion_utils.cpp
+++ b/ydb/core/kqp/opt/rbo/kqp_plan_conversion_utils.cpp
@@ -72,9 +72,8 @@ void AddVisibleDependencies(const TIntrusivePtr<IOperator>& op, TInfoUnitSet& de
 
     const auto visibleIUs = MakeInfoUnitSet(op->GetOutputIUs());
 
-    auto it = TOpIterator(op, nullptr);
-    for (; it != TOpIterator(nullptr); it++) {
-        auto currOp = (*it).Current;
+    for (const auto& item : IterateSubtree(op)) {
+        auto currOp = item.Current;
         if (currOp->Kind != EOperator::AddDependencies) {
             continue;
         }
@@ -95,10 +94,10 @@ TVector<DependencyPairType> ComputeDependentVariables(TIntrusivePtr<IOperator> o
 
     TVector<DependencyPairType> subplanDependencies;
 
-    // Iterate over just the operator of the current plan/subplan
-    auto it = TOpIterator(op, nullptr);
-    for(; it != TOpIterator(nullptr); it++) {
-        auto currOp = (*it).Current;
+    // This pass can insert operators, so preserve the original traversal while mutating the plan.
+    const TOpTraversal traversal(IterateSubtree(op).begin());
+    for (const auto& item : traversal) {
+        auto currOp = item.Current;
         auto subplanIUs = currOp->GetSubplanIUs(*props);
 
         // If the current operator contains references to subplans:
@@ -279,7 +278,7 @@ TIntrusivePtr<TOpRoot> PlanConverter::ConvertRoot(TExprNode::TPtr node) {
     opRoot->PlanProps = PlanProps;
  
     // We need to propagate plan properties reference into expressions in the plan
-    for (auto it : *opRoot) {
+    for (const auto& it : *opRoot) {
         for (auto exprRef : it.Current->GetExpressions()) {
             exprRef.get().PlanProps = &(opRoot->PlanProps);
         }
diff --git a/ydb/core/kqp/opt/rbo/kqp_plan_to_json.cpp b/ydb/core/kqp/opt/rbo/kqp_plan_to_json.cpp
index 80ec36309fd..706b7150176 100644
--- a/ydb/core/kqp/opt/rbo/kqp_plan_to_json.cpp
+++ b/ydb/core/kqp/opt/rbo/kqp_plan_to_json.cpp
@@ -385,7 +385,7 @@ NJson::TJsonValue TOpRoot::GetExecutionJson(ui64& nodeCounter, THashMap<IOperato
     std::set<int> stages;
     ui32 operatorId = 0;
 
-    for (auto it : *this) {
+    for (const auto& it : *this) {
         auto & currOp = it.Current;
         operatorIds.insert({currOp.Get(), operatorId++});
         int stageId = *currOp->Props.StageId;
diff --git a/ydb/core/kqp/opt/rbo/kqp_rbo.cpp b/ydb/core/kqp/opt/rbo/kqp_rbo.cpp
index 8f73b984a14..3f1a639a3e4 100644
--- a/ydb/core/kqp/opt/rbo/kqp_rbo.cpp
+++ b/ydb/core/kqp/opt/rbo/kqp_rbo.cpp
@@ -100,7 +100,7 @@ void TRuleBasedStage::RunStage(TOpRoot& root, TRBOContext& ctx) {
     while (fired && numMatches < maxNumOfMatches) {
         fired = false;
 
-        for (auto iter : root) {
+        for (const auto& iter : root) {
             for (const auto& rule : Rules) {
                 auto op = iter.Current;
                 if (!rule->QuickMatch(op)) {
diff --git a/ydb/core/kqp/opt/rbo/kqp_rbo_compute_statistics.cpp b/ydb/core/kqp/opt/rbo/kqp_rbo_compute_statistics.cpp
index 37882f8c711..1eda31c85b9 100644
--- a/ydb/core/kqp/opt/rbo/kqp_rbo_compute_statistics.cpp
+++ b/ydb/core/kqp/opt/rbo/kqp_rbo_compute_statistics.cpp
@@ -682,13 +682,13 @@ void TOpCBOTree::ComputeStatistics(TRBOContext& ctx, TPlanProps& planProps) {
 }
 
 void TOpRoot::ComputePlanMetadata(TRBOContext& ctx) {
-    for (auto it : *this) {
+    for (const auto& it : *this) {
         it.Current->ComputeMetadata(ctx, PlanProps);
     }
 }
 
 void TOpRoot::ComputePlanStatistics(TRBOContext& ctx) {
-    for (auto it : *this) {
+    for (const auto& it : *this) {
         it.Current->ComputeStatistics(ctx, PlanProps);
     }
 }
diff --git a/ydb/core/kqp/opt/rbo/kqp_rbo_transformer.cpp b/ydb/core/kqp/opt/rbo/kqp_rbo_transformer.cpp
index 4dff13a46c7..3b16b2b7f72 100644
--- a/ydb/core/kqp/opt/rbo/kqp_rbo_transformer.cpp
+++ b/ydb/core/kqp/opt/rbo/kqp_rbo_transformer.cpp
@@ -250,7 +250,7 @@ void TKqpNewRBOTransformer::CollectTablesAndColumnsNames(TExprContext& ctx) {
     Y_ENSURE(OpRoot);
     TRBOContext rboCtx(KqpCtx, ctx, TypeCtx, *RBOTypeAnnTransformer.Get(), FuncRegistry);
     OpRoot->ComputePlanMetadata(rboCtx);
-    for (auto it : *OpRoot) {
+    for (const auto& it : *OpRoot) {
         if (IsSuitableToCollectStatistics(it.Current)) {
             CollectTablesAndColumnsNames(it.Current);
         }
diff --git a/ydb/core/kqp/opt/rbo/kqp_rbo_type_ann.cpp b/ydb/core/kqp/opt/rbo/kqp_rbo_type_ann.cpp
index b39f4361923..899e4e6037e 100644
--- a/ydb/core/kqp/opt/rbo/kqp_rbo_type_ann.cpp
+++ b/ydb/core/kqp/opt/rbo/kqp_rbo_type_ann.cpp
@@ -555,8 +555,8 @@ TStatus ComputeTypes(TIntrusivePtr<IOperator> op, TRBOContext& ctx, TPlanProps&
 } // anonymous namespace
 
 TStatus TOpRoot::ComputeTypes(TRBOContext& ctx) {
-    for (auto it = begin(); it != end(); it++) {
-        auto status = ::NKikimr::NKqp::ComputeTypes((*it).Current, ctx, PlanProps);
+    for (const auto& item : *this) {
+        auto status = ::NKikimr::NKqp::ComputeTypes(item.Current, ctx, PlanProps);
         if (status != TStatus::Ok) {
             return status;
         }
diff --git a/ydb/core/kqp/opt/rbo/rules/constant_folding_stage.cpp b/ydb/core/kqp/opt/rbo/rules/constant_folding_stage.cpp
index 38928eab240..1a3cdeeba69 100644
--- a/ydb/core/kqp/opt/rbo/rules/constant_folding_stage.cpp
+++ b/ydb/core/kqp/opt/rbo/rules/constant_folding_stage.cpp
@@ -94,7 +94,7 @@ void TConstantFoldingStage::RunStage(TOpRoot &root, TRBOContext &ctx) {
     TVector<std::pair<TExprNode::TPtr, TExprNode::TPtr>> globalExtractedExprs;
     TVector<TIntrusivePtr<IOperator>> affectedOps;
 
-    for (auto it : root) {
+    for (const auto& it : root) {
         if (!it.Current->GetExpressions().empty()) {
             auto expressions = it.Current->GetExpressions();
             bool affected = false;
diff --git a/ydb/core/kqp/opt/rbo/ya.make b/ydb/core/kqp/opt/rbo/ya.make
index 3ad3f70329a..f2f08bccd5c 100644
--- a/ydb/core/kqp/opt/rbo/ya.make
+++ b/ydb/core/kqp/opt/rbo/ya.make
@@ -27,6 +27,8 @@ SRCS(
 )
 
 PEERDIR(
+    library/cpp/containers/absl
+    library/cpp/containers/stack_vector
     ydb/core/kqp/common
     ydb/core/kqp/opt/cbo
     ydb/core/kqp/opt/cbo/solver
diff --git a/ydb/core/kqp/ut/rbo/kqp_rbo_yql_ut.cpp b/ydb/core/kqp/ut/rbo/kqp_rbo_yql_ut.cpp
index 117500e6b21..9a2570a6633 100644
--- a/ydb/core/kqp/ut/rbo/kqp_rbo_yql_ut.cpp
+++ b/ydb/core/kqp/ut/rbo/kqp_rbo_yql_ut.cpp
@@ -1120,6 +1120,79 @@ Y_UNIT_TEST_SUITE(KqpRboYql) {
         UNIT_ASSERT_C(joinFilter.Contains("t1.b < t2.c"), joinFilter);
     }
 
+    Y_UNIT_TEST(OperatorIteratorMovePreservesDeepTraversal) {
+        const auto pos = NYql::TPositionHandle();
+        TIntrusivePtr<IOperator> op = MakeIntrusive<TOpEmptySource>(pos);
+        for (size_t i = 0; i < 30; ++i) {
+            op = MakeIntrusive<TOpJoin>(
+                op,
+                MakeIntrusive<TOpEmptySource>(pos),
+                pos,
+                "Cross",
+                TVector<std::pair<TInfoUnit, TInfoUnit>>{});
+        }
+        TOpRoot root(op, pos, TVector<TString>{});
+
+        auto source = root.begin();
+        TOpIterator moved(std::move(source));
+        UNIT_ASSERT(source == TOpEnd{});
+        ++source;
+        UNIT_ASSERT(source == TOpEnd{});
+
+        auto assigned = root.begin();
+        assigned = std::move(moved);
+        UNIT_ASSERT(moved == TOpEnd{});
+        ++moved;
+        UNIT_ASSERT(moved == TOpEnd{});
+
+        size_t count = 0;
+        IOperator* last = nullptr;
+        for (; assigned != TOpEnd{}; ++assigned) {
+            last = assigned->Current.Get();
+            ++count;
+        }
+
+        UNIT_ASSERT_VALUES_EQUAL(count, 61);
+        UNIT_ASSERT_VALUES_EQUAL(last, op.Get());
+    }
+
+    Y_UNIT_TEST(ComputeParentsHandlesSharedDagAndIgnoresInactiveSubplans) {
+        const auto pos = NYql::TPositionHandle();
+        auto shared = MakeIntrusive<TOpEmptySource>(pos);
+        auto join = MakeIntrusive<TOpJoin>(
+            shared,
+            shared,
+            pos,
+            "Cross",
+            TVector<std::pair<TInfoUnit, TInfoUnit>>{});
+        TOpRoot root(join, pos, TVector<TString>{});
+
+        auto inactiveChild = MakeIntrusive<TOpEmptySource>(pos);
+        auto inactiveSubplan = MakeIntrusive<TOpJoin>(
+            inactiveChild,
+            MakeIntrusive<TOpEmptySource>(pos),
+            pos,
+            "Cross",
+            TVector<std::pair<TInfoUnit, TInfoUnit>>{});
+        const TInfoUnit inactiveIU("inactive_subplan", true);
+        root.PlanProps.Subplans.Add(inactiveIU, TSubplanEntry{inactiveSubplan, {}, ESubplanType::EXPR, inactiveIU, {}});
+
+        root.ComputeParents();
+
+        UNIT_ASSERT(join->Parents.empty());
+        UNIT_ASSERT_VALUES_EQUAL(shared->Parents.size(), 2);
+        bool hasLeftEdge = false;
+        bool hasRightEdge = false;
+        for (const auto& [parent, childIndex] : shared->Parents) {
+            UNIT_ASSERT_VALUES_EQUAL(parent, join.Get());
+            hasLeftEdge |= childIndex == 0;
+            hasRightEdge |= childIndex == 1;
+        }
+        UNIT_ASSERT(hasLeftEdge);
+        UNIT_ASSERT(hasRightEdge);
+        UNIT_ASSERT(inactiveChild->Parents.empty());
+    }
+
     Y_UNIT_TEST(NameConstraintsPropagateThroughUnary) {
         NYql::TExprContext exprCtx;
         TPlanProps planProps;