Merge 5520542 into d010ed3

Author: zverevgeny

ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp

@@ -505,7 +505,6 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
 
         MkqlMemoryLimit = ev->Get()->MkqlMemoryLimit;
         ProfileStats = std::move(ev->Get()->ProfileStats);
-        auto sourcesState = GetSourcesState();
         auto status = ev->Get()->RunStatus;
 
         CA_LOG_T("Resume execution, run status: " << status << " checkpoint: " << (bool) ev->Get()->ProgramState
@@ -524,10 +523,6 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
             }
         }
 
-        if (status != ERunStatus::Finished) {
-            PollSources(std::move(sourcesState));
-        }
-
         if (ev->Get()->WatermarkInjectedToOutputs && !WatermarksTracker.HasOutputChannels()) {
             ResumeInputsByWatermark(*WatermarksTracker.GetPendingWatermark());
             WatermarksTracker.PopPendingWatermark();

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 = static_cast<TDerived*>(this)->GetSourcesState();
-
-        PollAsyncInput();
-        ERunStatus status = TaskRunner->Run();
-
-        CA_LOG_T("Resume execution, run status: " << status);
-
-        if (status != ERunStatus::Finished) {
-            static_cast<TDerived*>(this)->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);
     }
@@ -1064,13 +1047,6 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         return true;
     }
 
-protected:
-    // methods that are called via static_cast<TDerived*>(this) and may be overriden by a dervied class
-    void* GetSourcesState() const {
-        return nullptr;
-    }
-    void PollSources(void* /* state */) {
-    }
 
 
 protected:

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 = static_cast<TDerived*>(this)->GetSourcesState();
+
+        TBase::PollAsyncInput();
+        ERunStatus status = this->TaskRunner->Run();
+
+        CA_LOG_T("Resume execution, run status: " << status);
+
+        if (status != ERunStatus::Finished) {
+             static_cast<TDerived*>(this)->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();
@@ -131,6 +162,15 @@ class TDqSyncComputeActorBase: public TDqComputeActorBase<TDerived, TComputeActo
     const NYql::NDq::TDqMeteringStats* GetMeteringStats() override {
         return this->TaskRunner ? this->TaskRunner->GetMeteringStats() : nullptr;
     }
+
+protected:
+    // methods that are called via static_cast<TDerived*>(this) and may be overriden by a dervied class
+    void* GetSourcesState() const {
+        return nullptr;
+    }
+    void PollSources(void* /* state */) {
+    }
+
 };
 
 } //namespace NYql::NDq