Revert "Wait for all CAs inside Executer before shutdown (ydb-platform#7829)"

This reverts commit fe50fe3.

Author: abyss7

.gitignore

@@ -28,9 +28,6 @@ __pycache__/
 *.pb.h
 *.pb.cc
 
-# Other generated
-*.fbs.h
-
 # MacOS specific
 .DS_Store
 

ydb/core/kqp/executer_actor/kqp_data_executer.cpp

@@ -198,8 +198,6 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
     }
 
     void Finalize() {
-        YQL_ENSURE(!AlreadyReplied);
-
         if (LocksBroken) {
             YQL_ENSURE(ResponseEv->BrokenLockShardId);
             return ReplyErrorAndDie(Ydb::StatusIds::ABORTED, {});
@@ -278,7 +276,9 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
 
         ExecuterSpan.EndOk();
 
-        AlreadyReplied = true;
+        Request.Transactions.crop(0);
+        LOG_D("Sending response to: " << Target << ", results: " << ResponseEv->ResultsSize());
+        Send(Target, ResponseEv.release());
         PassAway();
     }
 
@@ -318,8 +318,6 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
             return "WaitSnapshotState";
         } else if (func == &TThis::WaitResolveState) {
             return "WaitResolveState";
-        } else if (func == &TThis::WaitShutdownState) {
-            return "WaitShutdownState";
         } else {
             return TBase::CurrentStateFuncName();
         }
@@ -2255,7 +2253,7 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
             // Volatile transactions must always use generic readsets
             VolatileTx ||
             // Transactions with topics must always use generic readsets
-            !topicTxs.empty() || 
+            !topicTxs.empty() ||
             // HTAP transactions always use generic readsets
             !evWriteTxs.empty());
 
@@ -2633,23 +2631,6 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
         }
     }
 
-    void Shutdown() override {
-        if (Planner) {
-            if (Planner->GetPendingComputeTasks().empty() && Planner->GetPendingComputeActors().empty()) {
-                LOG_I("Shutdown immediately - nothing to wait");
-                PassAway();
-            } else {
-                this->Become(&TThis::WaitShutdownState);
-                LOG_I("Waiting for shutdown of " << Planner->GetPendingComputeTasks().size() << " tasks and "
-                                                 << Planner->GetPendingComputeActors().size() << " compute actors");
-                // TODO(ilezhankin): the CA awaiting timeout should be configurable.
-                TActivationContext::Schedule(TDuration::Seconds(10), new IEventHandle(SelfId(), SelfId(), new TEvents::TEvPoison));
-            }
-        } else {
-            PassAway();
-        }
-    }
-
     void PassAway() override {
         auto totalTime = TInstant::Now() - StartTime;
         Counters->Counters->DataTxTotalTimeHistogram->Collect(totalTime.MilliSeconds());
@@ -2667,54 +2648,6 @@ class TKqpDataExecuter: public TKqpExecuterBase<TKqpDataExecuter, EExecType::Dat
         TBase::PassAway();
     }
 
-    STATEFN(WaitShutdownState) {
-        switch (ev->GetTypeRewrite()) {
-            hFunc(TEvDqCompute::TEvState, HandleShutdown);
-            hFunc(TEvInterconnect::TEvNodeDisconnected, HandleShutdown);
-            hFunc(TEvents::TEvPoison, HandleShutdown);
-            default:
-                LOG_E("Unexpected event: " << ev->GetTypeName()); // ignore all other events
-        }
-    }
-
-    void HandleShutdown(TEvDqCompute::TEvState::TPtr& ev) {
-        HandleComputeStats(ev);
-
-        if (Planner->GetPendingComputeTasks().empty() && Planner->GetPendingComputeActors().empty()) {
-            PassAway();
-        }
-    }
-
-    void HandleShutdown(TEvInterconnect::TEvNodeDisconnected::TPtr& ev) {
-        const auto nodeId = ev->Get()->NodeId;
-        LOG_N("Node has disconnected while shutdown: " << nodeId);
-
-        YQL_ENSURE(Planner);
-
-        for (const auto& task : TasksGraph.GetTasks()) {
-            if (task.Meta.NodeId == nodeId && !task.Meta.Completed) {
-                if (task.ComputeActorId) {
-                    Planner->CompletedCA(task.Id, task.ComputeActorId);
-                } else {
-                    Planner->TaskNotStarted(task.Id);
-                }
-            }
-        }
-
-        if (Planner->GetPendingComputeTasks().empty() && Planner->GetPendingComputeActors().empty()) {
-            PassAway();
-        }
-    }
-
-    void HandleShutdown(TEvents::TEvPoison::TPtr& ev) {
-        // Self-poison means timeout - don't wait anymore.
-        LOG_I("Timed out on waiting for Compute Actors to finish - forcing shutdown");
-
-        if (ev->Sender == SelfId()) {
-            PassAway();
-        }
-    }
-
 private:
     void ReplyTxStateUnknown(ui64 shardId) {
         auto message = TStringBuilder() << "Tx state unknown for shard " << shardId << ", txid " << TxId;

ydb/core/kqp/executer_actor/kqp_executer_impl.h

@@ -418,7 +418,7 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
                     auto& extraData = ExtraData[computeActor];
                     extraData.TaskId = taskId;
                     extraData.Data.Swap(state.MutableExtraData());
-                    
+
 
                     Stats->AddComputeActorStats(
                         computeActor.NodeId(),
@@ -686,7 +686,7 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
         if (statusCode == Ydb::StatusIds::INTERNAL_ERROR) {
             InternalError(issues);
         } else if (statusCode == Ydb::StatusIds::TIMEOUT) {
-            TimeoutError(ev->Sender);
+            AbortExecutionAndDie(ev->Sender, NYql::NDqProto::StatusIds::TIMEOUT, "Request timeout exceeded");
         } else {
             RuntimeError(NYql::NDq::DqStatusToYdbStatus(msg.GetStatusCode()), issues);
         }
@@ -1624,14 +1624,14 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
 protected:
     void TerminateComputeActors(Ydb::StatusIds::StatusCode code, const NYql::TIssues& issues) {
         for (const auto& task : this->TasksGraph.GetTasks()) {
-            if (task.ComputeActorId && !task.Meta.Completed) {
+            if (task.ComputeActorId) {
                 LOG_I("aborting compute actor execution, message: " << issues.ToOneLineString()
                     << ", compute actor: " << task.ComputeActorId << ", task: " << task.Id);
 
                 auto ev = MakeHolder<TEvKqp::TEvAbortExecution>(NYql::NDq::YdbStatusToDqStatus(code), issues);
                 this->Send(task.ComputeActorId, ev.Release());
             } else {
-                LOG_I("task: " << task.Id << ", does not have the CA id yet or is already complete");
+                LOG_I("task: " << task.Id << ", does not have Compute ActorId yet");
             }
         }
     }
@@ -1649,6 +1649,7 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
 
     void InternalError(const NYql::TIssues& issues) {
         LOG_E(issues.ToOneLineString());
+        TerminateComputeActors(Ydb::StatusIds::INTERNAL_ERROR, issues);
         auto issue = NYql::YqlIssue({}, NYql::TIssuesIds::UNEXPECTED, "Internal error while executing transaction.");
         for (const NYql::TIssue& i : issues) {
             issue.AddSubIssue(MakeIntrusive<NYql::TIssue>(i));
@@ -1662,13 +1663,15 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
 
     void ReplyUnavailable(const TString& message) {
         LOG_E("UNAVAILABLE: " << message);
+        TerminateComputeActors(Ydb::StatusIds::UNAVAILABLE, message);
         auto issue = NYql::YqlIssue({}, NYql::TIssuesIds::KIKIMR_TEMPORARILY_UNAVAILABLE);
         issue.AddSubIssue(new NYql::TIssue(message));
         ReplyErrorAndDie(Ydb::StatusIds::UNAVAILABLE, issue);
     }
 
     void RuntimeError(Ydb::StatusIds::StatusCode code, const NYql::TIssues& issues) {
         LOG_E(Ydb::StatusIds_StatusCode_Name(code) << ": " << issues.ToOneLineString());
+        TerminateComputeActors(code, issues);
         ReplyErrorAndDie(code, issues);
     }
 
@@ -1684,19 +1687,11 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
         ReplyErrorAndDie(status, &issues);
     }
 
-    void TimeoutError(TActorId abortSender) {
+    void AbortExecutionAndDie(TActorId abortSender, NYql::NDqProto::StatusIds::StatusCode status, const TString& message) {
         if (AlreadyReplied) {
-            LOG_E("Timeout when we already replied - not good" << Endl << TBackTrace().PrintToString() << Endl);
             return;
         }
 
-        const auto status = NYql::NDqProto::StatusIds::TIMEOUT;
-        const TString message = "Request timeout exceeded";
-
-        TerminateComputeActors(Ydb::StatusIds::TIMEOUT, message);
-
-        AlreadyReplied = true;
-
         LOG_E("Abort execution: " << NYql::NDqProto::StatusIds_StatusCode_Name(status) << "," << message);
         if (ExecuterSpan) {
             ExecuterSpan.EndError(TStringBuilder() << NYql::NDqProto::StatusIds_StatusCode_Name(status));
@@ -1706,24 +1701,34 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
 
         // TEvAbortExecution can come from either ComputeActor or SessionActor (== Target).
         if (abortSender != Target) {
-            auto abortEv = MakeHolder<TEvKqp::TEvAbortExecution>(status, message);
+            auto abortEv = MakeHolder<TEvKqp::TEvAbortExecution>(status, "Request timeout exceeded");
             this->Send(Target, abortEv.Release());
         }
 
+        AlreadyReplied = true;
         LOG_E("Sending timeout response to: " << Target);
+        this->Send(Target, ResponseEv.release());
 
-        this->Shutdown();
+        Request.Transactions.crop(0);
+        TerminateComputeActors(Ydb::StatusIds::TIMEOUT, message);
+        this->PassAway();
     }
 
     virtual void ReplyErrorAndDie(Ydb::StatusIds::StatusCode status,
         google::protobuf::RepeatedPtrField<Ydb::Issue::IssueMessage>* issues)
     {
         if (AlreadyReplied) {
-            LOG_E("Error when we already replied - not good" << Endl << TBackTrace().PrintToString() << Endl);
             return;
         }
 
-        TerminateComputeActors(status, "Terminate execution");
+        if (Planner) {
+            for (auto computeActor : Planner->GetPendingComputeActors()) {
+                LOG_D("terminate compute actor " << computeActor.first);
+
+                auto ev = MakeHolder<TEvKqp::TEvAbortExecution>(NYql::NDq::YdbStatusToDqStatus(status), "Terminate execution");
+                this->Send(computeActor.first, ev.Release());
+            }
+        }
 
         AlreadyReplied = true;
         auto& response = *ResponseEv->Record.MutableResponse();
@@ -1749,7 +1754,9 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
         ExecuterSpan.EndError(response.DebugString());
         ExecuterStateSpan.EndError(response.DebugString());
 
-        this->Shutdown();
+        Request.Transactions.crop(0);
+        this->Send(Target, ResponseEv.release());
+        this->PassAway();
     }
 
 protected:
@@ -1817,46 +1824,7 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
     }
 
 protected:
-    // Introduced separate method from `PassAway()` - to not get confused with expectations from other actors,
-    // that `PassAway()` should kill actor immediately.
-    virtual void Shutdown() {
-        PassAway();
-    }
-
     void PassAway() override {
-        YQL_ENSURE(AlreadyReplied && ResponseEv);
-
-        // Fill response stats
-        {
-            auto& response = *ResponseEv->Record.MutableResponse();
-
-            YQL_ENSURE(Stats);
-
-            ReportEventElapsedTime();
-
-            Stats->FinishTs = TInstant::Now();
-            Stats->Finish();
-
-            if (Stats->CollectStatsByLongTasks || CollectFullStats(Request.StatsMode)) {
-                response.MutableResult()->MutableStats()->ClearTxPlansWithStats();
-                for (ui32 txId = 0; txId < Request.Transactions.size(); ++txId) {
-                    const auto& tx = Request.Transactions[txId].Body;
-                    auto planWithStats = AddExecStatsToTxPlan(tx->GetPlan(), response.GetResult().GetStats());
-                    response.MutableResult()->MutableStats()->AddTxPlansWithStats(planWithStats);
-                }
-            }
-
-            if (Stats->CollectStatsByLongTasks) {
-                const auto& txPlansWithStats = response.GetResult().GetStats().GetTxPlansWithStats();
-                if (!txPlansWithStats.empty()) {
-                    LOG_N("Full stats: " << response.GetResult().GetStats());
-                }
-            }
-        }
-
-        Request.Transactions.crop(0);
-        this->Send(Target, ResponseEv.release());
-
         for (auto channelPair: ResultChannelProxies) {
             LOG_D("terminate result channel " << channelPair.first << " proxy at " << channelPair.second->SelfId());
 
@@ -1877,11 +1845,12 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
 
         if (KqpTableResolverId) {
             this->Send(KqpTableResolverId, new TEvents::TEvPoison);
+            this->Send(this->SelfId(), new TEvents::TEvPoison);
+            LOG_T("Terminate, become ZombieState");
+            this->Become(&TKqpExecuterBase::ZombieState);
+        } else {
+            IActor::PassAway();
         }
-
-        this->Send(this->SelfId(), new TEvents::TEvPoison);
-        LOG_T("Terminate, become ZombieState");
-        this->Become(&TKqpExecuterBase::ZombieState);
     }
 
     STATEFN(ZombieState) {

ydb/core/kqp/executer_actor/kqp_planner.cpp

@@ -644,17 +644,6 @@ bool TKqpPlanner::CompletedCA(ui64 taskId, TActorId computeActor) {
     return true;
 }
 
-void TKqpPlanner::TaskNotStarted(ui64 taskId) {
-    // NOTE: should be invoked only while shutting down - when node is disconnected.
-
-    auto& task = TasksGraph.GetTask(taskId);
-
-    YQL_ENSURE(!task.ComputeActorId);
-    YQL_ENSURE(!task.Meta.Completed);
-
-    PendingComputeTasks.erase(taskId);
-}
-
 TProgressStat::TEntry TKqpPlanner::CalculateConsumptionUpdate() {
     TProgressStat::TEntry consumption;
 

ydb/core/kqp/executer_actor/kqp_planner.h

@@ -76,7 +76,6 @@ class TKqpPlanner {
     std::unique_ptr<IEventHandle> AssignTasksToNodes();
     bool AcknowledgeCA(ui64 taskId, TActorId computeActor, const NYql::NDqProto::TEvComputeActorState* state);
     bool CompletedCA(ui64 taskId, TActorId computeActor);
-    void TaskNotStarted(ui64 taskId);
     TProgressStat::TEntry CalculateConsumptionUpdate();
     void ShiftConsumption();
     void Submit();

ydb/core/kqp/executer_actor/kqp_scan_executer.cpp

@@ -273,9 +273,6 @@ class TKqpScanExecuter : public TKqpExecuterBase<TKqpScanExecuter, EExecType::Sc
 public:
 
     void Finalize() {
-        YQL_ENSURE(!AlreadyReplied);
-        AlreadyReplied = true;
-
         ResponseEv->Record.MutableResponse()->SetStatus(Ydb::StatusIds::SUCCESS);
 
         LWTRACK(KqpScanExecuterFinalize, ResponseEv->Orbit, TxId, LastTaskId, LastComputeActorId, ResponseEv->ResultsSize());
@@ -284,6 +281,8 @@ class TKqpScanExecuter : public TKqpExecuterBase<TKqpScanExecuter, EExecType::Sc
             ExecuterSpan.EndOk();
         }
 
+        LOG_D("Sending response to: " << Target);
+        Send(Target, ResponseEv.release());
         PassAway();
     }