deduplicate internal bindings

fix(peephole): deduplicate internal bindings

3 files changed, 25 insertions, 6 deletions

diff --git a/ydb/core/kqp/opt/peephole/kqp_opt_peephole.cpp b/ydb/core/kqp/opt/peephole/kqp_opt_peephole.cpp
index f44dbefc29..5c1f1cabc0 100644
--- a/ydb/core/kqp/opt/peephole/kqp_opt_peephole.cpp
+++ b/ydb/core/kqp/opt/peephole/kqp_opt_peephole.cpp
@@ -138,6 +138,7 @@ TMaybeNode<TKqpPhysicalTx> PeepholeOptimize(const TKqpPhysicalTx& tx, TExprConte
     stages.reserve(tx.Stages().Size());
     TNodeOnNodeOwnedMap stagesMap;
     TVector<TKqpParamBinding> bindings(tx.ParamBindings().begin(), tx.ParamBindings().end());
+    THashMap<TString, TKqpParamBinding> nonDetParamBindings;
 
     for (const auto& stage : tx.Stages()) {
         YQL_ENSURE(!optimizedStages.contains(stage.Ref().UniqueId()));
@@ -167,7 +168,7 @@ TMaybeNode<TKqpPhysicalTx> PeepholeOptimize(const TKqpPhysicalTx& tx, TExprConte
         }
 
         if (allowNonDeterministicFunctions) {
-            status = ReplaceNonDetFunctionsWithParams(newProgram, ctx, &bindings);
+            status = ReplaceNonDetFunctionsWithParams(newProgram, ctx, &nonDetParamBindings);
 
             if (status != TStatus::Ok) {
                 ctx.AddError(TIssue(ctx.GetPosition(stage.Pos()),
@@ -188,6 +189,10 @@ TMaybeNode<TKqpPhysicalTx> PeepholeOptimize(const TKqpPhysicalTx& tx, TExprConte
         optimizedStages.emplace(stage.Ref().UniqueId());
     }
 
+    for (const auto& [_, binding] : nonDetParamBindings) {
+        bindings.emplace_back(std::move(binding));
+    }
+
     return Build<TKqpPhysicalTx>(ctx, tx.Pos())
         .Stages()
             .Add(stages)
@@ -332,7 +337,7 @@ TAutoPtr<IGraphTransformer> CreateKqpTxsPeepholeTransformer(TAutoPtr<NYql::IGrap
     return new TKqpTxsPeepholeTransformer(std::move(typeAnnTransformer), typesCtx, config);
 }
 
-TStatus ReplaceNonDetFunctionsWithParams(TExprNode::TPtr& input, TExprContext& ctx, TVector<TKqpParamBinding>* paramBindings) {
+TStatus ReplaceNonDetFunctionsWithParams(TExprNode::TPtr& input, TExprContext& ctx, THashMap<TString, TKqpParamBinding>* paramBindings) {
     static const std::unordered_set<std::string_view> nonDeterministicFunctions = {
         "RandomNumber",
         "Random",
@@ -359,7 +364,7 @@ TStatus ReplaceNonDetFunctionsWithParams(TExprNode::TPtr& input, TExprContext& c
                     .Type(ExpandType(node->Pos(), *node->GetTypeAnn(), ctx))
                     .Done();
 
-                if (paramBindings) {
+                if (paramBindings && !paramBindings->contains(paramName)) {
                     auto binding = Build<TKqpTxInternalBinding>(ctx, node->Pos())
                         .Type(ExpandType(node->Pos(), *node->GetTypeAnn(), ctx))
                         .Kind().Build(callable.CallableName())
@@ -370,9 +375,8 @@ TStatus ReplaceNonDetFunctionsWithParams(TExprNode::TPtr& input, TExprContext& c
                         .Binding(binding)
                         .Done();
 
-                    paramBindings->emplace_back(std::move(paramBinding));
+                    paramBindings->insert({paramName, std::move(paramBinding)});
                 }
-
                 return param.Ptr();
             }
         }
diff --git a/ydb/core/kqp/opt/peephole/kqp_opt_peephole.h b/ydb/core/kqp/opt/peephole/kqp_opt_peephole.h
index a0602bb07d..64143adc39 100644
--- a/ydb/core/kqp/opt/peephole/kqp_opt_peephole.h
+++ b/ydb/core/kqp/opt/peephole/kqp_opt_peephole.h
@@ -12,6 +12,6 @@ TAutoPtr<NYql::IGraphTransformer> CreateKqpTxsPeepholeTransformer(TAutoPtr<NYql:
     NYql::TTypeAnnotationContext& typesCtx, const NYql::TKikimrConfiguration::TPtr& config);
 
 NYql::IGraphTransformer::TStatus ReplaceNonDetFunctionsWithParams(NYql::TExprNode::TPtr& input, NYql::TExprContext& ctx,
-    TVector<NYql::NNodes::TKqpParamBinding>* paramBindings = nullptr);
+     THashMap<TString, NYql::NNodes::TKqpParamBinding>* paramBindings = nullptr);
 
 } // namespace NKikimr::NKqp::NOpt
diff --git a/ydb/core/kqp/ut/kqp_query_ut.cpp b/ydb/core/kqp/ut/kqp_query_ut.cpp
index 40bf96098e..e59822a4ad 100644
--- a/ydb/core/kqp/ut/kqp_query_ut.cpp
+++ b/ydb/core/kqp/ut/kqp_query_ut.cpp
@@ -1200,6 +1200,21 @@ Y_UNIT_TEST_SUITE(KqpQuery) {
         }
     }
 
+    Y_UNIT_TEST_NEW_ENGINE(MultipleCurrentUtcTimestamp) {
+        TKikimrRunner kikimr;
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        auto query = Q1_(R"(
+            SELECT * FROM `/Root/Logs` WHERE Ts > Cast(CurrentUtcTimestamp() as Int64)
+            UNION ALL
+            SELECT * FROM `/Root/Logs` WHERE Ts < Cast(CurrentUtcTimestamp() as Int64);
+        )");
+
+        auto result = session.ExecuteDataQuery(query, TTxControl::BeginTx().CommitTx()).ExtractValueSync();
+        UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+    }
+
     Y_UNIT_TEST_NEW_ENGINE(SelectWhereInSubquery) {
         TKikimrRunner kikimr;
         auto db = kikimr.GetTableClient();