Merge 9d9379a into 1d0d00e

Author: zverevgeny

ydb/core/kqp/compute_actor/kqp_scan_compute_actor.cpp

@@ -156,22 +156,18 @@ void TKqpScanComputeActor::Handle(TEvScanExchange::TEvFetcherFinished::TPtr& ev)
     }
 }
 
-void TKqpScanComputeActor::PollSources(std::any prev) {
+void TKqpScanComputeActor::PollSources(ui64 prevFreeSpace) {
     if (!ScanData || ScanData->IsFinished()) {
         return;
     }
     const auto hasNewMemoryPred = [&]() {
-        if (!prev.has_value()) {
-            return false;
-        }
         const ui64 freeSpace = CalculateFreeSpace();
-        const ui64 prevFreeSpace = std::any_cast<ui64>(prev);
         return freeSpace > prevFreeSpace;
     };
     if (!hasNewMemoryPred() && ScanData->GetStoredBytes()) {
         return;
     }
-    const ui32 freeSpace = CalculateFreeSpace();
+    const ui64 freeSpace = CalculateFreeSpace();
     CA_LOG_D("POLL_SOURCES:START:" << Fetchers.size() << ";fs=" << freeSpace);
     for (auto&& i : Fetchers) {
         Send(i, new TEvScanExchange::TEvAckData(freeSpace));

ydb/core/kqp/compute_actor/kqp_scan_compute_actor.h

@@ -82,14 +82,14 @@ class TKqpScanComputeActor: public NYql::NDq::TDqSyncComputeActorBase<TKqpScanCo
             : 0ul;
     }
 
-    std::any GetSourcesState() override {
+    ui64 GetSourcesState() {
         if (!ScanData) {
             return 0;
         }
         return CalculateFreeSpace();
     }
 
-    void PollSources(std::any prev) override;
+    void PollSources(ui64 prevFreeSpace);
 
     void PassAway() override {
         if (TaskRunner) {

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

@@ -311,15 +311,15 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
     }
 
     virtual void DoExecuteImpl() {
-        auto sourcesState = GetSourcesState();
+        auto sourcesState = static_cast<TDerived*>(this)->GetSourcesState();
 
         PollAsyncInput();
         ERunStatus status = TaskRunner->Run();
 
         CA_LOG_T("Resume execution, run status: " << status);
 
         if (status != ERunStatus::Finished) {
-            PollSources(std::move(sourcesState));
+            static_cast<TDerived*>(this)->PollSources(std::move(sourcesState));
         }
 
         if ((status == ERunStatus::PendingInput || status == ERunStatus::Finished) && Checkpoints && Checkpoints->HasPendingCheckpoint() && !Checkpoints->ComputeActorStateSaved() && ReadyToCheckpoint()) {
@@ -1049,13 +1049,6 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
 protected:
     // virtual methods (TODO: replace with static_cast<TDerived*>(this)->Foo()
 
-    virtual std::any GetSourcesState() {
-        return nullptr;
-    }
-
-    virtual void PollSources(std::any /* state */) {
-    }
-
     virtual void TerminateSources(const TIssues& /* issues */, bool /* success */) {
     }
 
@@ -1071,6 +1064,15 @@ 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:
     void HandleExecuteBase(TEvDqCompute::TEvResumeExecution::TPtr&) {
         ResumeEventScheduled = false;