Improve compute actor code

Author: abyss7

ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h

@@ -515,22 +515,6 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             RuntimeSettings.TerminateHandler(success, issues);
         }
 
-        // Send final state to executer to inform about termination.
-        if (!success) {
-            auto ev = MakeHolder<TEvDqCompute::TEvState>();
-            auto& record = ev->Record;
-
-            if (RuntimeSettings.StatsMode >= NDqProto::DQ_STATS_MODE_BASIC) {
-                FillStats(record.MutableStats(), /* last */ true);
-            }
-
-            record.SetState(NDqProto::COMPUTE_STATE_FINISHED);
-            record.SetStatusCode(NDqProto::StatusIds::ABORTED);
-            record.SetTaskId(Task.GetId());
-
-            this->Send(ExecuterId, ev.Release());
-        }
-
         this->PassAway();
         Terminated = true;
     }
@@ -570,7 +554,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         }
     }
 
-    void ReportStateAndMaybeDie(NYql::NDqProto::StatusIds::StatusCode statusCode, const TIssues& issues)
+    void ReportStateAndMaybeDie(NYql::NDqProto::StatusIds::StatusCode statusCode, const TIssues& issues, bool forceTerminate = false)
     {
         auto execEv = MakeHolder<TEvDqCompute::TEvState>();
         auto& record = execEv->Record;
@@ -591,7 +575,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
 
         this->Send(ExecuterId, execEv.Release());
 
-        if (Checkpoints && State == NDqProto::COMPUTE_STATE_FINISHED) {
+        if (!forceTerminate && Checkpoints && State == NDqProto::COMPUTE_STATE_FINISHED) {
             // checkpointed CAs must not self-destroy
             return;
         }
@@ -1048,10 +1032,6 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         auto tag = (EEvWakeupTag) ev->Get()->Tag;
         switch (tag) {
             case EEvWakeupTag::TimeoutTag: {
-                auto abortEv = MakeHolder<TEvDq::TEvAbortExecution>(NYql::NDqProto::StatusIds::TIMEOUT, TStringBuilder()
-                    << "Timeout event from compute actor " << this->SelfId()
-                    << ", TxId: " << TxId << ", task: " << Task.GetId());
-
                 if (ComputeActorSpan) {
                     ComputeActorSpan.EndError(
                         TStringBuilder()
@@ -1060,10 +1040,8 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
                     );
                 }
 
-                this->Send(ExecuterId, abortEv.Release());
-
-                TerminateSources("timeout exceeded", false);
-                Terminate(false, "timeout exceeded");
+                State = NDqProto::COMPUTE_STATE_FAILURE;
+                ReportStateAndMaybeDie(NYql::NDqProto::StatusIds::TIMEOUT, {TIssue("timeout exceeded")}, true);
                 break;
             }
             case EEvWakeupTag::PeriodicStatsTag: {
@@ -1087,8 +1065,9 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         switch (lostEventType) {
             case TEvDqCompute::TEvState::EventType: {
                 CA_LOG_E("Handle undelivered TEvState event, abort execution");
-                this->TerminateSources("executer lost", false);
-                Terminate(false, "executer lost");
+
+                TerminateSources("executer lost", false);
+                Terminate(false, "executer lost"); // Executer lost - no need to report state
                 break;
             }
             default: {
@@ -1134,14 +1113,17 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             InternalError(NYql::NDqProto::StatusIds::INTERNAL_ERROR, *ev->Get()->GetIssues().begin());
             return;
         }
+
         TIssues issues = ev->Get()->GetIssues();
         CA_LOG_E("Handle abort execution event from: " << ev->Sender
             << ", status: " << NYql::NDqProto::StatusIds_StatusCode_Name(ev->Get()->Record.GetStatusCode())
             << ", reason: " << issues.ToOneLineString());
 
-        bool success = ev->Get()->Record.GetStatusCode() == NYql::NDqProto::StatusIds::SUCCESS;
-
-        this->TerminateSources(issues, success);
+        if (ev->Get()->Record.GetStatusCode() == NYql::NDqProto::StatusIds::SUCCESS) {
+            State = NDqProto::COMPUTE_STATE_FINISHED;
+        } else {
+            State = NDqProto::COMPUTE_STATE_FAILURE;
+        }
 
         if (ev->Sender != ExecuterId) {
             if (ComputeActorSpan) {
@@ -1151,7 +1133,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             NActors::TActivationContext::Send(ev->Forward(ExecuterId));
         }
 
-        Terminate(success, issues);
+        ReportStateAndMaybeDie(ev->Get()->Record.GetStatusCode(), issues, true);
     }
 
     void HandleExecuteBase(NActors::TEvInterconnect::TEvNodeDisconnected::TPtr& ev) {