Use std::shared_ptr for NKqpProto::TKqpPhyQuery KIKIMR-14443

ref:20e961a2df26527b0d20c7f5f96ba13ead9e8799

8 files changed, 45 insertions, 44 deletions

diff --git a/ydb/core/kqp/common/kqp_transform.h b/ydb/core/kqp/common/kqp_transform.h
index 457c4248011..630d4c8890f 100644
--- a/ydb/core/kqp/common/kqp_transform.h
+++ b/ydb/core/kqp/common/kqp_transform.h
@@ -82,14 +82,14 @@ using TParamValueMap = THashMap<TString, NKikimrMiniKQL::TParams>;
 
 struct TDeferredEffect {
     NYql::NNodes::TMaybeNode<NYql::NNodes::TExprBase> Node;
-    NKqpProto::TKqpPhyTx PhysicalTx;
+    std::shared_ptr<const NKqpProto::TKqpPhyTx> PhysicalTx;
     TParamValueMap Params;
 
     explicit TDeferredEffect(const NYql::NNodes::TExprBase& node)
         : Node(node) {}
 
-    explicit TDeferredEffect(const NKqpProto::TKqpPhyTx& physicalTx)
-        : PhysicalTx(physicalTx) {}
+    explicit TDeferredEffect(std::shared_ptr<const NKqpProto::TKqpPhyTx>&& physicalTx)
+        : PhysicalTx(std::move(physicalTx)) {}
 };
 
 class TKqpTransactionContext;
@@ -128,12 +128,12 @@ private:
     }
 
     [[nodiscard]]
-    bool Add(const NKqpProto::TKqpPhyTx& physicalTx, TParamValueMap&& params) {
+    bool Add(std::shared_ptr<const NKqpProto::TKqpPhyTx>&& physicalTx, TParamValueMap&& params) {
         if (Engine.has_value() && *Engine != TKqpTransactionInfo::EEngine::NewEngine) {
             return false;
         }
         Engine.emplace(TKqpTransactionInfo::EEngine::NewEngine);
-        DeferredEffects.emplace_back(physicalTx);
+        DeferredEffects.emplace_back(std::move(physicalTx));
         DeferredEffects.back().Params = std::move(params);
         return true;
     }
@@ -174,8 +174,8 @@ public:
     }
 
     [[nodiscard]]
-    bool AddDeferredEffect(const NKqpProto::TKqpPhyTx& physicalTx, TParamValueMap&& params) {
-        return DeferredEffects.Add(physicalTx, std::move(params));
+    bool AddDeferredEffect(std::shared_ptr<const NKqpProto::TKqpPhyTx>&& physicalTx, TParamValueMap&& params) {
+        return DeferredEffects.Add(std::move(physicalTx), std::move(params));
     }
 
     const IKqpGateway::TKqpSnapshot& GetSnapshot() const {
diff --git a/ydb/core/kqp/host/kqp_host.cpp b/ydb/core/kqp/host/kqp_host.cpp
index b35a0772597..03539d916a2 100644
--- a/ydb/core/kqp/host/kqp_host.cpp
+++ b/ydb/core/kqp/host/kqp_host.cpp
@@ -656,30 +656,30 @@ public:
         output = input;
 
         if (!AsyncResult) {
-            const auto& query = *QueryCtx->PreparedQuery;
+            auto& query = QueryCtx->PreparedQuery;
             YQL_ENSURE(QueryCtx->Type != EKikimrQueryType::Unspecified);
 
-            if (query.GetVersion() == NKikimrKqp::TPreparedQuery::VERSION_PHYSICAL_V1) {
+            if (query->GetVersion() == NKikimrKqp::TPreparedQuery::VERSION_PHYSICAL_V1) {
                 if (CurrentKqlIndex) {
                     return TStatus::Ok;
                 }
 
-                const auto& phyQuery = query.GetPhysicalQuery();
+                std::shared_ptr<const NKqpProto::TKqpPhyQuery> phyQuery(query, &query->GetPhysicalQuery());
 
                 if (QueryCtx->Type == EKikimrQueryType::Scan) {
-                    AsyncResult = KqpRunner->ExecutePreparedScanQuery(Cluster, input.Get(), phyQuery, ctx,
-                        ExecuteCtx->ReplyTarget);
+                    AsyncResult = KqpRunner->ExecutePreparedScanQuery(Cluster, input.Get(), std::move(phyQuery),
+                        ctx, ExecuteCtx->ReplyTarget);
                 } else {
                     YQL_ENSURE(QueryCtx->Type == EKikimrQueryType::Dml);
-                    AsyncResult = KqpRunner->ExecutePreparedQueryNewEngine(Cluster, input.Get(), phyQuery, ctx,
-                        ExecuteCtx->Settings);
+                    AsyncResult = KqpRunner->ExecutePreparedQueryNewEngine(Cluster, input.Get(), std::move(phyQuery),
+                        ctx, ExecuteCtx->Settings);
                 }
             } else {
-                if (CurrentKqlIndex >= query.KqlsSize()) {
+                if (CurrentKqlIndex >= query->KqlsSize()) {
                     return TStatus::Ok;
                 }
 
-                const auto& kql = query.GetKqls(CurrentKqlIndex);
+                const auto& kql = query->GetKqls(CurrentKqlIndex);
 
                 YQL_ENSURE(QueryCtx->Type == EKikimrQueryType::Dml);
                 YQL_ENSURE(!kql.GetSettings().GetNewEngine());
diff --git a/ydb/core/kqp/host/kqp_host_impl.h b/ydb/core/kqp/host/kqp_host_impl.h
index fb1fb139cd5..889e8e176e5 100644
--- a/ydb/core/kqp/host/kqp_host_impl.h
+++ b/ydb/core/kqp/host/kqp_host_impl.h
@@ -262,12 +262,12 @@ public:
         const NYql::IKikimrQueryExecutor::TExecuteSettings& settings) = 0;
 
     virtual TIntrusivePtr<TAsyncQueryResult> ExecutePreparedQueryNewEngine(const TString& cluster,
-        const NYql::TExprNode::TPtr& world, const NKqpProto::TKqpPhyQuery& phyQuery, NYql::TExprContext& ctx,
-        const NYql::IKikimrQueryExecutor::TExecuteSettings& settings) = 0;
+        const NYql::TExprNode::TPtr& world, std::shared_ptr<const NKqpProto::TKqpPhyQuery>&& phyQuery,
+        NYql::TExprContext& ctx, const NYql::IKikimrQueryExecutor::TExecuteSettings& settings) = 0;
 
     virtual TIntrusivePtr<TAsyncQueryResult> ExecutePreparedScanQuery(const TString& cluster,
-        const NYql::TExprNode::TPtr& world, const NKqpProto::TKqpPhyQuery& phyQuery, NYql::TExprContext& ctx,
-        const NActors::TActorId& target) = 0;
+        const NYql::TExprNode::TPtr& world, std::shared_ptr<const NKqpProto::TKqpPhyQuery>&& phyQuery,
+        NYql::TExprContext& ctx, const NActors::TActorId& target) = 0;
 };
 
 TIntrusivePtr<IKqpRunner> CreateKqpRunner(TIntrusivePtr<IKqpGateway> gateway, const TString& cluster,
diff --git a/ydb/core/kqp/host/kqp_run_data.cpp b/ydb/core/kqp/host/kqp_run_data.cpp
index b2bc80d71be..98fc950e060 100644
--- a/ydb/core/kqp/host/kqp_run_data.cpp
+++ b/ydb/core/kqp/host/kqp_run_data.cpp
@@ -53,8 +53,9 @@ protected:
 
             for (const auto& effect : txState.DeferredEffects) {
                 YQL_ENSURE(!effect.Node);
-                YQL_ENSURE(effect.PhysicalTx.GetType() == NKqpProto::TKqpPhyTx::TYPE_DATA);
-                request.Transactions.emplace_back(effect.PhysicalTx, GetParamsRefMap(effect.Params));
+                YQL_ENSURE(effect.PhysicalTx->GetType() == NKqpProto::TKqpPhyTx::TYPE_DATA);
+                // TODO pass shared_ptr
+                request.Transactions.emplace_back(*effect.PhysicalTx, GetParamsRefMap(effect.Params));
             }
 
             if (!txState.DeferredEffects.Empty()) {
diff --git a/ydb/core/kqp/host/kqp_run_physical.cpp b/ydb/core/kqp/host/kqp_run_physical.cpp
index e6bd601eef6..7fe9e2f235f 100644
--- a/ydb/core/kqp/host/kqp_run_physical.cpp
+++ b/ydb/core/kqp/host/kqp_run_physical.cpp
@@ -17,12 +17,12 @@ IGraphTransformer::TStatus TKqpExecutePhysicalTransformerBase::DoTransform(TExpr
     YQL_ENSURE(TMaybeNode<TCoWorld>(input));
     YQL_ENSURE(TransformCtx->PhysicalQuery);
 
-    const auto& query = *TransformCtx->PhysicalQuery;
+    std::shared_ptr<const NKqpProto::TKqpPhyQuery> query = TransformCtx->PhysicalQuery;
 
     TStatus status = TStatus::Ok;
 
     auto& txState = TxState->Tx();
-    const ui64 txsCount = query.TransactionsSize();
+    const ui64 txsCount = query->TransactionsSize();
 
     if (settings.GetRollbackTx()) {
         if (ExecuteFlags.HasFlags(TKqpExecuteFlag::Rollback)) {
@@ -43,9 +43,9 @@ IGraphTransformer::TStatus TKqpExecutePhysicalTransformerBase::DoTransform(TExpr
                 status = Execute(nullptr, /* commit */ true, ctx);
             }
         } else {
-            const auto& tx = query.GetTransactions(CurrentTxIndex);
+            std::shared_ptr<const NKqpProto::TKqpPhyTx> tx(query, &query->GetTransactions(CurrentTxIndex));
 
-            if (tx.GetHasEffects()) {
+            if (tx->GetHasEffects()) {
                 if (!AddDeferredEffect(tx)) {
                     ctx.AddError(YqlIssue({}, TIssuesIds::KIKIMR_BAD_REQUEST,
                         "Failed to mix queries with old- and new- engines"));
@@ -63,12 +63,12 @@ IGraphTransformer::TStatus TKqpExecutePhysicalTransformerBase::DoTransform(TExpr
                 // last transaction and no effects (RO-query)
                 commit = true;
             }
-            status = Execute(&tx, commit, ctx);
+            status = Execute(tx.get(), commit, ctx);
         }
     }
 
     if (status == TStatus::Ok) {
-        for (const auto& resultBinding : query.GetResultBindings()) {
+        for (const auto& resultBinding : query->GetResultBindings()) {
             YQL_ENSURE(resultBinding.GetTypeCase() == NKqpProto::TKqpPhyResultBinding::kTxResultBinding);
             auto& txResultBinding = resultBinding.GetTxResultBinding();
             auto txIndex = txResultBinding.GetTxIndex();
@@ -171,11 +171,11 @@ IGraphTransformer::TStatus TKqpExecutePhysicalTransformerBase::Rollback() {
     return DoRollback();
 }
 
-bool TKqpExecutePhysicalTransformerBase::AddDeferredEffect(const NKqpProto::TKqpPhyTx& tx) {
+bool TKqpExecutePhysicalTransformerBase::AddDeferredEffect(std::shared_ptr<const NKqpProto::TKqpPhyTx> tx) {
     TParamValueMap params;
-    PreserveParams(tx, params);
+    PreserveParams(*tx, params);
 
-    return TxState->Tx().AddDeferredEffect(tx, std::move(params));
+    return TxState->Tx().AddDeferredEffect(std::move(tx), std::move(params));
 }
 
 NDq::TMkqlValueRef TKqpExecutePhysicalTransformerBase::GetParamValue(
diff --git a/ydb/core/kqp/host/kqp_run_physical.h b/ydb/core/kqp/host/kqp_run_physical.h
index c55fcc59a09..68af613c824 100644
--- a/ydb/core/kqp/host/kqp_run_physical.h
+++ b/ydb/core/kqp/host/kqp_run_physical.h
@@ -56,7 +56,7 @@ private:
     TStatus Execute(const NKqpProto::TKqpPhyTx* tx, bool commit, NYql::TExprContext& ctx);
     TStatus Rollback();
 
-    bool AddDeferredEffect(const NKqpProto::TKqpPhyTx& tx);
+    bool AddDeferredEffect(std::shared_ptr<const NKqpProto::TKqpPhyTx> tx);
 
     void PreserveParams(const NKqpProto::TKqpPhyTx& tx, TParamValueMap& paramsMap);
 
diff --git a/ydb/core/kqp/host/kqp_runner.cpp b/ydb/core/kqp/host/kqp_runner.cpp
index 42b4eca7c66..d34f86273eb 100644
--- a/ydb/core/kqp/host/kqp_runner.cpp
+++ b/ydb/core/kqp/host/kqp_runner.cpp
@@ -312,11 +312,11 @@ public:
     }
 
     TIntrusivePtr<TAsyncQueryResult> ExecutePreparedQueryNewEngine(const TString& cluster,
-        const NYql::TExprNode::TPtr& world, const NKqpProto::TKqpPhyQuery& phyQuery, TExprContext& ctx,
+        const NYql::TExprNode::TPtr& world, std::shared_ptr<const NKqpProto::TKqpPhyQuery>&& phyQuery, TExprContext& ctx,
         const IKikimrQueryExecutor::TExecuteSettings& settings) override
     {
         YQL_ENSURE(cluster == Cluster);
-        YQL_ENSURE(phyQuery.GetType() == NKqpProto::TKqpPhyQuery::TYPE_DATA);
+        YQL_ENSURE(phyQuery->GetType() == NKqpProto::TKqpPhyQuery::TYPE_DATA);
 
         if (!Config->HasAllowKqpNewEngine()) {
             ctx.AddError(TIssue(TPosition(), "NewEngine execution is not allowed on this cluster."));
@@ -324,17 +324,17 @@ public:
                 ctx.IssueManager.GetIssues()));
         }
 
-        return ExecutePhysicalDataQuery(world, phyQuery, ctx, settings);
+        return ExecutePhysicalDataQuery(world, std::move(phyQuery), ctx, settings);
     }
 
     TIntrusivePtr<TAsyncQueryResult> ExecutePreparedScanQuery(const TString& cluster,
-        const NYql::TExprNode::TPtr& world, const NKqpProto::TKqpPhyQuery& phyQuery, TExprContext& ctx,
+        const NYql::TExprNode::TPtr& world, std::shared_ptr<const NKqpProto::TKqpPhyQuery>&& phyQuery, TExprContext& ctx,
         const NActors::TActorId& target) override
     {
         YQL_ENSURE(cluster == Cluster);
-        YQL_ENSURE(phyQuery.GetType() == NKqpProto::TKqpPhyQuery::TYPE_SCAN);
+        YQL_ENSURE(phyQuery->GetType() == NKqpProto::TKqpPhyQuery::TYPE_SCAN);
 
-        return ExecutePhysicalScanQuery(world, phyQuery, ctx, target);
+        return ExecutePhysicalScanQuery(world, std::move(phyQuery), ctx, target);
     }
 
 private:
@@ -567,13 +567,13 @@ private:
     }
 
     TIntrusivePtr<TAsyncQueryResult> ExecutePhysicalDataQuery(const TExprNode::TPtr& world,
-        const NKqpProto::TKqpPhyQuery& phyQuery, TExprContext& ctx,
+        std::shared_ptr<const NKqpProto::TKqpPhyQuery>&& phyQuery, TExprContext& ctx,
         const IKikimrQueryExecutor::TExecuteSettings& settings)
     {
         PhysicalRunQueryTransformer->Rewind();
 
         TransformCtx->Reset();
-        TransformCtx->PhysicalQuery = &phyQuery;
+        TransformCtx->PhysicalQuery = phyQuery;
 
         YQL_ENSURE(TxState->Tx().EffectiveIsolationLevel);
         YQL_ENSURE(TransformCtx->Settings.GetIsolationLevel() == NKikimrKqp::ISOLATION_LEVEL_UNDEFINED);
@@ -583,19 +583,19 @@ private:
         TransformCtx->Settings.SetRollbackTx(settings.RollbackTx);
 
         bool strictDml = MergeFlagValue(Config->StrictDml.Get(Cluster), settings.StrictDml);
-        if (!ApplyTableOperations(phyQuery, strictDml, ctx)) {
+        if (!ApplyTableOperations(*phyQuery, strictDml, ctx)) {
             return MakeKikimrResultHolder(ResultFromErrors<IKqpHost::TQueryResult>(ctx.IssueManager.GetIssues()));
         }
         return MakeIntrusive<TPhysicalAsyncRunResult>(world, ctx, *PhysicalRunQueryTransformer, *TransformCtx);
     }
 
     TIntrusivePtr<TAsyncQueryResult> ExecutePhysicalScanQuery(const TExprNode::TPtr& world,
-        const NKqpProto::TKqpPhyQuery& phyQuery, TExprContext& ctx, const NActors::TActorId& target)
+        std::shared_ptr<const NKqpProto::TKqpPhyQuery>&& phyQuery, TExprContext& ctx, const NActors::TActorId& target)
     {
         ScanRunQueryTransformer->Rewind();
 
         TransformCtx->Reset();
-        TransformCtx->PhysicalQuery = &phyQuery;
+        TransformCtx->PhysicalQuery = phyQuery;
         TransformCtx->ReplyTarget = target;
 
         Y_ASSERT(!TxState->Tx().GetSnapshot().IsValid());
diff --git a/ydb/core/kqp/prepare/kqp_prepare.h b/ydb/core/kqp/prepare/kqp_prepare.h
index d114a449458..27c956ea148 100644
--- a/ydb/core/kqp/prepare/kqp_prepare.h
+++ b/ydb/core/kqp/prepare/kqp_prepare.h
@@ -91,7 +91,7 @@ struct TKqlTransformContext : TThrRefBase {
     NKikimrKqp::TPreparedKql* PreparingKql = nullptr;
     const NKikimrKqp::TPreparedKql* PreparedKql;
     NKqpProto::TKqpStatsQuery QueryStats;
-    const NKqpProto::TKqpPhyQuery* PhysicalQuery;
+    std::shared_ptr<const NKqpProto::TKqpPhyQuery> PhysicalQuery;
 
     TVector<TSimpleSharedPtr<NKikimrMiniKQL::TResult>> MkqlResults;
     TVector<NKikimrMiniKQL::TResult> PhysicalQueryResults;