Add config for retries in data_executer

9 files changed, 72 insertions, 34 deletions

diff --git a/ydb/core/kqp/executer_actor/kqp_data_executer.cpp b/ydb/core/kqp/executer_actor/kqp_data_executer.cpp
index cf123827a0..7267944eb8 100644
--- a/ydb/core/kqp/executer_actor/kqp_data_executer.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_data_executer.cpp
@@ -123,8 +123,9 @@ public:
     }
 
     TKqpDataExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database, const TMaybe<TString>& userToken,
-        TKqpRequestCounters::TPtr counters, bool streamResult, ui32 executerDelayToRetryMs)
-        : TBase(std::move(request), database, userToken, counters, executerDelayToRetryMs, TWilsonKqp::DataExecuter, "DataExecuter")
+        TKqpRequestCounters::TPtr counters, bool streamResult,
+        const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig)
+        : TBase(std::move(request), database, userToken, counters, executerRetriesConfig, TWilsonKqp::DataExecuter, "DataExecuter")
         , StreamResult(streamResult)
     {
         YQL_ENSURE(Request.IsolationLevel != NKikimrKqp::ISOLATION_LEVEL_UNDEFINED);
@@ -2018,7 +2019,8 @@ private:
 
         Planner = CreateKqpPlanner(TxId, SelfId(), {}, std::move(tasksPerNode), Request.Snapshot,
             Database, Nothing(), Deadline.GetOrElse(TInstant::Zero()), Request.StatsMode,
-            Request.DisableLlvmForUdfStages, Request.LlvmEnabled, false, Nothing(), ExecuterSpan, {});
+            Request.DisableLlvmForUdfStages, Request.LlvmEnabled, false, Nothing(), 
+            ExecuterSpan, {}, ExecuterRetriesConfig);
         Planner->ProcessTasksForDataExecuter();
 
         // then start data tasks with known actor ids of compute tasks
@@ -2327,9 +2329,9 @@ private:
 } // namespace
 
 IActor* CreateKqpDataExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database, const TMaybe<TString>& userToken,
-    TKqpRequestCounters::TPtr counters, bool streamResult, ui32 executerDelayToRetryMs)
+    TKqpRequestCounters::TPtr counters, bool streamResult, const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig)
 {
-    return new TKqpDataExecuter(std::move(request), database, userToken, counters, streamResult, executerDelayToRetryMs);
+    return new TKqpDataExecuter(std::move(request), database, userToken, counters, streamResult, executerRetriesConfig);
 }
 
 } // namespace NKqp
diff --git a/ydb/core/kqp/executer_actor/kqp_executer.h b/ydb/core/kqp/executer_actor/kqp_executer.h
index 1d6d70929e..275a0858d4 100644
--- a/ydb/core/kqp/executer_actor/kqp_executer.h
+++ b/ydb/core/kqp/executer_actor/kqp_executer.h
@@ -85,7 +85,8 @@ struct TEvKqpExecuter {
 
 IActor* CreateKqpExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
     const TMaybe<TString>& userToken, TKqpRequestCounters::TPtr counters, 
-    const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation, ui32 executerDelayToRetryMs);
+    const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig);
 
 std::unique_ptr<TEvKqpExecuter::TEvTxResponse> ExecutePure(
     IKqpGateway::TExecPhysicalRequest&& request, TKqpRequestCounters::TPtr counters, TActorId owner);
diff --git a/ydb/core/kqp/executer_actor/kqp_executer_impl.cpp b/ydb/core/kqp/executer_actor/kqp_executer_impl.cpp
index 9bc27a4a15..450b5eda1a 100644
--- a/ydb/core/kqp/executer_actor/kqp_executer_impl.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_executer_impl.cpp
@@ -146,12 +146,12 @@ TActorId ReportToRl(ui64 ru, const TString& database, const TString& userToken,
 
 IActor* CreateKqpExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
     const TMaybe<TString>& userToken, TKqpRequestCounters::TPtr counters, const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
-    ui32 executerDelayToRetryMs)
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig)
 {
     if (request.Transactions.empty()) {
         // commit-only or rollback-only data transaction
         YQL_ENSURE(request.EraseLocks);
-        return CreateKqpDataExecuter(std::move(request), database, userToken, counters, false, executerDelayToRetryMs);
+        return CreateKqpDataExecuter(std::move(request), database, userToken, counters, false, executerRetriesConfig);
     }
 
     TMaybe<NKqpProto::TKqpPhyTx::EType> txsType;
@@ -167,13 +167,13 @@ IActor* CreateKqpExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TSt
     switch (*txsType) {
         case NKqpProto::TKqpPhyTx::TYPE_COMPUTE:
         case NKqpProto::TKqpPhyTx::TYPE_DATA:
-            return CreateKqpDataExecuter(std::move(request), database, userToken, counters, false, executerDelayToRetryMs);
+            return CreateKqpDataExecuter(std::move(request), database, userToken, counters, false, executerRetriesConfig);
 
         case NKqpProto::TKqpPhyTx::TYPE_SCAN:
-            return CreateKqpScanExecuter(std::move(request), database, userToken, counters, aggregation, executerDelayToRetryMs);
+            return CreateKqpScanExecuter(std::move(request), database, userToken, counters, aggregation, executerRetriesConfig);
 
         case NKqpProto::TKqpPhyTx::TYPE_GENERIC:
-            return CreateKqpDataExecuter(std::move(request), database, userToken, counters, true, executerDelayToRetryMs);
+            return CreateKqpDataExecuter(std::move(request), database, userToken, counters, true, executerRetriesConfig);
 
         default:
             YQL_ENSURE(false, "Unsupported physical tx type: " << (ui32)*txsType);
diff --git a/ydb/core/kqp/executer_actor/kqp_executer_impl.h b/ydb/core/kqp/executer_actor/kqp_executer_impl.h
index 40c8f2a280..d417e0252e 100644
--- a/ydb/core/kqp/executer_actor/kqp_executer_impl.h
+++ b/ydb/core/kqp/executer_actor/kqp_executer_impl.h
@@ -110,15 +110,17 @@ protected:
     };
 
 public:
-    TKqpExecuterBase(IKqpGateway::TExecPhysicalRequest&& request, const TString& database, const TMaybe<TString>& userToken,
-        TKqpRequestCounters::TPtr counters, ui32 executerDelayToRetryMs, ui64 spanVerbosity = 0, TString spanName = "no_name")
+    TKqpExecuterBase(IKqpGateway::TExecPhysicalRequest&& request, const TString& database, const TMaybe<TString>& userToken, 
+        TKqpRequestCounters::TPtr counters, 
+        const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
+        ui64 spanVerbosity = 0, TString spanName = "no_name")
         : Request(std::move(request))
         , Database(database)
         , UserToken(userToken)
         , Counters(counters)
         , ExecuterSpan(spanVerbosity, std::move(Request.TraceId), spanName)
         , Planner(nullptr)
-        , ExecuterDelayToRetryMs(executerDelayToRetryMs)
+        , ExecuterRetriesConfig(executerRetriesConfig)
     {
         ResponseEv = std::make_unique<TEvKqpExecuter::TEvTxResponse>(Request.TxAlloc);
         ResponseEv->Orbit = std::move(Request.Orbit);
@@ -417,7 +419,7 @@ protected:
             case TEvKqpNode::TEvStartKqpTasksRequest::EventType: {
                 if (reason == TEvents::TEvUndelivered::EReason::ReasonActorUnknown) {
                     LOG_D("Schedule a retry by ActorUnknown reason, nodeId:" << ev->Sender.NodeId() << " requestId: " << ev->Cookie);
-                    this->Schedule(TDuration::MilliSeconds(ExecuterDelayToRetryMs), new typename TEvPrivate::TEvRetry(ev->Cookie, ev->Sender));
+                    this->Schedule(TDuration::MilliSeconds(Planner->GetCurrentRetryDelay(ev->Cookie)), new typename TEvPrivate::TEvRetry(ev->Cookie, ev->Sender));
                     return;
                 }
                 InvalidateNode(ev->Sender.NodeId());
@@ -1180,7 +1182,7 @@ protected:
     TString LastComputeActorId = "";
 
     std::unique_ptr<TKqpPlanner> Planner;
-    ui32 ExecuterDelayToRetryMs;
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig ExecuterRetriesConfig;
 
 private:
     static constexpr TDuration ResourceUsageUpdateInterval = TDuration::MilliSeconds(100);
@@ -1191,11 +1193,13 @@ private:
 IActor* CreateKqpLiteralExecuter(IKqpGateway::TExecPhysicalRequest&& request, TKqpRequestCounters::TPtr counters);
 
 IActor* CreateKqpDataExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
-    const TMaybe<TString>& userToken, TKqpRequestCounters::TPtr counters, bool streamResult, ui32 executerDelayToRetryMs);
+    const TMaybe<TString>& userToken, TKqpRequestCounters::TPtr counters, bool streamResult,
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig);
 
 IActor* CreateKqpScanExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
     const TMaybe<TString>& userToken, TKqpRequestCounters::TPtr counters, 
-    const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation, ui32 executerDelayToRetryMs);
+    const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig);
 
 } // namespace NKqp
 } // namespace NKikimr
diff --git a/ydb/core/kqp/executer_actor/kqp_planner.cpp b/ydb/core/kqp/executer_actor/kqp_planner.cpp
index 0b8fa4cefc..23f6cdfade 100644
--- a/ydb/core/kqp/executer_actor/kqp_planner.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_planner.cpp
@@ -28,7 +28,8 @@ TKqpPlanner::TKqpPlanner(ui64 txId, const TActorId& executer, TVector<NDqProto::
     const TString& database, const TMaybe<TString>& userToken, TInstant deadline,
     const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool disableLlvmForUdfStages, bool enableLlvm,
     bool withSpilling, const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& executerSpan,
-    TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot)
+    TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot,
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig)
     : TxId(txId)
     , ExecuterId(executer)
     , ComputeTasks(std::move(computeTasks))
@@ -44,6 +45,7 @@ TKqpPlanner::TKqpPlanner(ui64 txId, const TActorId& executer, TVector<NDqProto::
     , RlPath(rlPath)
     , ResourcesSnapshot(std::move(resourcesSnapshot))
     , ExecuterSpan(executerSpan)
+    , ExecuterRetriesConfig(executerRetriesConfig)
 {
     if (!Database) {
         // a piece of magic for tests
@@ -57,16 +59,15 @@ TKqpPlanner::TKqpPlanner(ui64 txId, const TActorId& executer, TVector<NDqProto::
 bool TKqpPlanner::SendStartKqpTasksRequest(ui32 requestId, const TActorId& target) {
     auto& requestData = Requests[requestId];
 
-    if (requestData.RetryNumber == 3) {
+    if (requestData.RetryNumber == ExecuterRetriesConfig.GetMaxRetryNumber() + 1) {
         return false;
     }
 
     auto ev = MakeHolder<TEvKqpNode::TEvStartKqpTasksRequest>();
     ev->Record = requestData.request;
 
-    if (requestData.RetryNumber == 1) {
-        LOG_D("Try to retry by ActorUnknown reason, nodeId: " << target.NodeId() << ", requestId: " << requestId);
-    } else if (requestData.RetryNumber == 2) {
+    if (requestData.RetryNumber == ExecuterRetriesConfig.GetMaxRetryNumber()) {
+        LOG_E("Retry failed by retries limit, requestId: " << requestId);
         TMaybe<ui32> targetNode;
         for (size_t i = 0; i < ResourcesSnapshot.size(); ++i) {
             if (!TrackingNodes.contains(ResourcesSnapshot[i].nodeid())) {
@@ -85,6 +86,11 @@ bool TKqpPlanner::SendStartKqpTasksRequest(ui32 requestId, const TActorId& targe
         LOG_E("Retry failed because all nodes are busy, requestId: " << requestId);
         return false;
     }
+
+    if (requestData.RetryNumber >= 1) {
+        LOG_D("Try to retry by ActorUnknown reason, nodeId: " << target.NodeId() << ", requestId: " << requestId);
+    }
+
     requestData.RetryNumber++;
 
     TlsActivationContext->Send(std::make_unique<NActors::IEventHandle>(target, ExecuterId, ev.Release(),
@@ -130,6 +136,18 @@ void TKqpPlanner::ProcessTasksForDataExecuter() {
     }
 }
 
+ui32 TKqpPlanner::GetCurrentRetryDelay(ui32 requestId) {
+    auto& requestData = Requests[requestId];
+    if (requestData.CurrentDelay == 0) {
+        requestData.CurrentDelay = ExecuterRetriesConfig.GetMinDelayToRetryMs();
+        return requestData.CurrentDelay;
+    }
+    requestData.CurrentDelay *= 2;
+    requestData.CurrentDelay = Min(requestData.CurrentDelay, ExecuterRetriesConfig.GetMaxDelayToRetryMs());
+    requestData.CurrentDelay = requestData.CurrentDelay * AppData()->RandomProvider->Uniform(100, 120) / 100;
+    return requestData.CurrentDelay;
+}
+
 void TKqpPlanner::ProcessTasksForScanExecuter() {
     PrepareToProcess();
 
@@ -434,10 +452,11 @@ std::unique_ptr<TKqpPlanner> CreateKqpPlanner(ui64 txId, const TActorId& execute
     const TString& database, const TMaybe<TString>& userToken, TInstant deadline,
     const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool disableLlvmForUdfStages, bool enableLlvm,
     bool withSpilling, const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& executerSpan,
-    TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot)
+    TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot, const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig)
 {
     return std::make_unique<TKqpPlanner>(txId, executer, std::move(tasks), std::move(mainTasksPerNode), snapshot,
-        database, userToken, deadline, statsMode, disableLlvmForUdfStages, enableLlvm, withSpilling, rlPath, executerSpan, std::move(resourcesSnapshot));
+        database, userToken, deadline, statsMode, disableLlvmForUdfStages, enableLlvm, withSpilling, rlPath, executerSpan, 
+        std::move(resourcesSnapshot), executerRetriesConfig);
 }
 
 } // namespace NKikimr::NKqp
diff --git a/ydb/core/kqp/executer_actor/kqp_planner.h b/ydb/core/kqp/executer_actor/kqp_planner.h
index def2a6f3a8..1fc08d0f26 100644
--- a/ydb/core/kqp/executer_actor/kqp_planner.h
+++ b/ydb/core/kqp/executer_actor/kqp_planner.h
@@ -22,6 +22,7 @@ class TKqpPlanner {
         NKikimrKqp::TEvStartKqpTasksRequest request;
         ui32 flag;
         ui32 RetryNumber = 0;
+        ui32 CurrentDelay = 0;
     };
 
 public:
@@ -30,7 +31,7 @@ public:
         const TString& database, const TMaybe<TString>& userToken, TInstant deadline,
         const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool disableLlvmForUdfStages,
         bool enableLlvm, bool withSpilling, const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& ExecuterSpan,
-        TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot);
+        TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot, const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig);
     bool SendStartKqpTasksRequest(ui32 requestId, const TActorId& target);
 
     void ProcessTasksForScanExecuter();
@@ -38,6 +39,8 @@ public:
 
     ui64 GetComputeTasksNumber() const;
     ui64 GetMainTasksNumber() const;
+
+    ui32 GetCurrentRetryDelay(ui32 requestId);
 private:
     void PrepareToProcess();
 
@@ -66,6 +69,7 @@ private:
     THashSet<ui32> TrackingNodes;
     const TVector<NKikimrKqp::TKqpNodeResources> ResourcesSnapshot;
     NWilson::TSpan& ExecuterSpan;
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& ExecuterRetriesConfig;
     ui64 LocalRunMemoryEst;
     TVector<TTaskResourceEstimation> ResourceEstimations;
     TVector<RequestData> Requests;
@@ -76,6 +80,7 @@ std::unique_ptr<TKqpPlanner> CreateKqpPlanner(ui64 txId, const TActorId& execute
     const TString& database, const TMaybe<TString>& userToken, TInstant deadline,
     const Ydb::Table::QueryStatsCollection::Mode& statsMode, bool disableLlvmForUdfStages, bool enableLlvm,
     bool withSpilling, const TMaybe<NKikimrKqp::TRlPath>& rlPath, NWilson::TSpan& executerSpan,
-    TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot);
+    TVector<NKikimrKqp::TKqpNodeResources>&& resourcesSnapshot,
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& ExecuterRetriesConfig);
 
 } // namespace NKikimr::NKqp
diff --git a/ydb/core/kqp/executer_actor/kqp_scan_executer.cpp b/ydb/core/kqp/executer_actor/kqp_scan_executer.cpp
index 200a53a8a2..f61622abbc 100644
--- a/ydb/core/kqp/executer_actor/kqp_scan_executer.cpp
+++ b/ydb/core/kqp/executer_actor/kqp_scan_executer.cpp
@@ -69,8 +69,8 @@ public:
     TKqpScanExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
         const TMaybe<TString>& userToken, TKqpRequestCounters::TPtr counters,
         const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
-        ui32 executerDelayToRetryMs)
-        : TBase(std::move(request), database, userToken, counters, executerDelayToRetryMs, TWilsonKqp::ScanExecuter, "ScanExecuter")
+        const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig)
+        : TBase(std::move(request), database, userToken, counters, executerRetriesConfig, TWilsonKqp::ScanExecuter, "ScanExecuter")
         , AggregationSettings(aggregation)
     {
         YQL_ENSURE(Request.Transactions.size() == 1);
@@ -746,7 +746,8 @@ private:
         Planner = CreateKqpPlanner(TxId, SelfId(), std::move(computeTasks),
             std::move(scanTasks), Request.Snapshot,
             Database, UserToken, Deadline.GetOrElse(TInstant::Zero()), Request.StatsMode,
-            Request.DisableLlvmForUdfStages, Request.LlvmEnabled, AppData()->EnableKqpSpilling, Request.RlPath, ExecuterSpan, std::move(snapshot));
+            Request.DisableLlvmForUdfStages, Request.LlvmEnabled, AppData()->EnableKqpSpilling,
+            Request.RlPath, ExecuterSpan, std::move(snapshot), ExecuterRetriesConfig);
         LOG_D("Execute scan tx, computeTasks: " << Planner->GetComputeTasksNumber() << ", scanTasks: " << Planner->GetMainTasksNumber());
 
         Planner->ProcessTasksForScanExecuter();
@@ -822,9 +823,9 @@ private:
 IActor* CreateKqpScanExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
     const TMaybe<TString>& userToken, TKqpRequestCounters::TPtr counters,
     const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
-    ui32 executerDelayToRetryMs)
+    const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig)
 {
-    return new TKqpScanExecuter(std::move(request), database, userToken, counters, aggregation, executerDelayToRetryMs);
+    return new TKqpScanExecuter(std::move(request), database, userToken, counters, aggregation, executerRetriesConfig);
 }
 
 } // namespace NKqp
diff --git a/ydb/core/kqp/session_actor/kqp_session_actor.cpp b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
index 119735f03e..9044d8f85e 100644
--- a/ydb/core/kqp/session_actor/kqp_session_actor.cpp
+++ b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
@@ -1287,7 +1287,7 @@ public:
 
         auto executerActor = CreateKqpExecuter(std::move(request), Settings.Database,
                 (QueryState && QueryState->UserToken) ? TMaybe<TString>(QueryState->UserToken) : Nothing(),
-                RequestCounters, Settings.Service.GetAggregationConfig(), Settings.Service.GetExecuterDelayToRetryMs());
+                RequestCounters, Settings.Service.GetAggregationConfig(), Settings.Service.GetExecuterRetriesConfig());
 
         auto exId = RegisterWithSameMailbox(executerActor);
         LOG_D("Created new KQP executer: " << exId);
diff --git a/ydb/core/protos/config.proto b/ydb/core/protos/config.proto
index 037b7e033d..3b185fe9b4 100644
--- a/ydb/core/protos/config.proto
+++ b/ydb/core/protos/config.proto
@@ -1204,6 +1204,12 @@ message TTableServiceConfig {
         optional double AggregationJoinThreadsKff = 36 [default = 1];
     }
 
+    message TExecuterRetriesConfig {
+        optional uint32 MinDelayToRetryMs = 1 [default = 10];
+        optional uint32 MaxDelayToRetryMs = 2 [default = 400];
+        optional uint32 MaxRetryNumber = 3 [default = 10];
+    }
+
     optional uint32 QueryLimitBytes = 1;
     optional uint32 ParametersLimitBytes = 2;
     optional uint32 SessionsLimitPerNode = 3;
@@ -1236,7 +1242,7 @@ message TTableServiceConfig {
     optional TAggregationConfig AggregationConfig = 29;
     optional bool EnableKqpScanQueryStreamLookup = 30 [default = false];
     optional bool EnableKqpDataQueryStreamLookup = 31 [default = false];
-    optional uint32 ExecuterDelayToRetryMs = 32 [default = 400];
+    optional TExecuterRetriesConfig ExecuterRetriesConfig = 32;
     optional bool EnableKqpDataQueryStreamPointLookup = 33 [default = false];
 };