YQL-17542 move TaskRunner dependent Execute to TDqSyncComputeActorBase

Author: zverevgeny

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

@@ -304,30 +304,14 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             ReportStats(TInstant::Now(), ESendStats::IfPossible);
         }
         if (Terminated) {
-            TaskRunner.Reset();
+            DoTerminateImpl();
             MemoryQuota.Reset();
             MemoryLimits.MemoryQuotaManager.reset();
         }
     }
 
-    virtual void DoExecuteImpl() {
-        auto sourcesState = GetSourcesState();
-
-        PollAsyncInput();
-        ERunStatus status = TaskRunner->Run();
-
-        CA_LOG_T("Resume execution, run status: " << status);
-
-        if (status != ERunStatus::Finished) {
-            PollSources(std::move(sourcesState));
-        }
-
-        if ((status == ERunStatus::PendingInput || status == ERunStatus::Finished) && Checkpoints && Checkpoints->HasPendingCheckpoint() && !Checkpoints->ComputeActorStateSaved() && ReadyToCheckpoint()) {
-            Checkpoints->DoCheckpoint();
-        }
-
-        ProcessOutputsImpl(status);
-    }
+    virtual void DoExecuteImpl() = 0;
+    virtual void DoTerminateImpl() {}
 
     virtual bool DoHandleChannelsAfterFinishImpl() {
         Y_ABORT_UNLESS(Checkpoints);
@@ -606,12 +590,11 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         InternalError(statusCode, TIssues({std::move(issue)}));
     }
 
+    virtual void InvalidateMeminfo() {}
+
     void InternalError(NYql::NDqProto::StatusIds::StatusCode statusCode, TIssues issues) {
         CA_LOG_E(InternalErrorLogString(statusCode, issues));
-        if (TaskRunner) {
-            TaskRunner->GetAllocator().InvalidateMemInfo();
-            TaskRunner->GetAllocator().DisableStrictAllocationCheck();
-        }
+        InvalidateMeminfo();
         State = NDqProto::COMPUTE_STATE_FAILURE;
         ReportStateAndMaybeDie(statusCode, issues);
     }

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

@@ -44,6 +44,37 @@ class TDqSyncComputeActorBase: public TDqComputeActorBase<TDerived, TComputeActo
         return inputTransformInfo.Buffer.Get();
     }
 protected:
+
+    void DoExecuteImpl() override{
+        auto sourcesState = TBase::GetSourcesState();
+
+        TBase::PollAsyncInput();
+        ERunStatus status = this->TaskRunner->Run();
+
+        CA_LOG_T("Resume execution, run status: " << status);
+
+        if (status != ERunStatus::Finished) {
+            TBase::PollSources(std::move(sourcesState));
+        }
+
+        if ((status == ERunStatus::PendingInput || status == ERunStatus::Finished) && this->Checkpoints && this->Checkpoints->HasPendingCheckpoint() && !this->Checkpoints->ComputeActorStateSaved() && TBase::ReadyToCheckpoint()) {
+            this->Checkpoints->DoCheckpoint();
+        }
+
+        TBase::ProcessOutputsImpl(status);
+    }
+
+    void DoTerminateImpl() override {
+        this->TaskRunner.Reset();
+    }
+
+    void InvalidateMeminfo() override {
+        if (this->TaskRunner) {
+            this->TaskRunner->GetAllocator().InvalidateMemInfo();
+            this->TaskRunner->GetAllocator().DisableStrictAllocationCheck();
+        }
+    }
+
     void SaveState(const NDqProto::TCheckpoint& checkpoint, NDqProto::TComputeActorState& state) const override {
         CA_LOG_D("Save state");
         NDqProto::TMiniKqlProgramState& mkqlProgramState = *state.MutableMiniKqlProgram();