Revert "Yql 17542 simplify alloc in compute actor (ydb-platform#1452)"

This reverts commit 9c7d88b.

Author: lll-phill-lll

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

@@ -457,6 +457,18 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
         return inputChannel->FreeSpace;
     }
 
+    TGuard<NKikimr::NMiniKQL::TScopedAlloc> BindAllocator() override {
+        return TypeEnv->BindAllocator();
+    }
+
+    std::optional<TGuard<NKikimr::NMiniKQL::TScopedAlloc>> MaybeBindAllocator() override {
+        std::optional<TGuard<NKikimr::NMiniKQL::TScopedAlloc>> guard;
+        if (TypeEnv) {
+            guard.emplace(TypeEnv->BindAllocator());
+        }
+        return guard;
+    }
+
     void OnTaskRunnerCreated(NTaskRunnerActor::TEvTaskRunnerCreateFinished::TPtr& ev) {
         const auto& secureParams = ev->Get()->SecureParams;
         const auto& taskParams = ev->Get()->TaskParams;

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

@@ -447,7 +447,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         }
 
         {
-            auto guard = BindAllocator(); // Source/Sink could destroy mkql values inside PassAway, which requires allocator to be bound
+            auto guard = MaybeBindAllocator(); // Source/Sink could destroy mkql values inside PassAway, which requires allocator to be bound
 
             for (auto& [_, source] : SourcesMap) {
                 if (source.Actor) {
@@ -955,8 +955,12 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         TerminateSources(TIssues({TIssue(message)}), success);
     }
 
-    TGuard<NKikimr::NMiniKQL::TScopedAlloc> BindAllocator() {
-        return Guard(GetAllocator());
+    virtual TGuard<NKikimr::NMiniKQL::TScopedAlloc> BindAllocator() {
+        return TaskRunner->BindAllocator();
+    }
+
+    virtual std::optional<TGuard<NKikimr::NMiniKQL::TScopedAlloc>> MaybeBindAllocator() {
+        return TaskRunner->BindAllocator();
     }
 
     virtual bool SayHelloOnBootstrap() {
@@ -1276,7 +1280,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
                         .TypeEnv = typeEnv,
                         .HolderFactory = holderFactory,
                         .TaskCounters = TaskCounters,
-                        .Alloc = Alloc,
+                        .Alloc = TaskRunner ? Alloc : nullptr,
                         .MemoryQuotaManager = MemoryLimits.MemoryQuotaManager,
                         .SourceSettings = (!settings.empty() ? settings.at(inputIndex) : nullptr),
                         .Arena = Task.GetArena(),

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

@@ -194,9 +194,8 @@ class TDqSyncComputeActorBase: public TDqComputeActorBase<TDerived, TComputeActo
     void PrepareTaskRunner(const IDqTaskRunnerExecutionContext& execCtx) {
         YQL_ENSURE(TaskRunner);
 
-        auto guard = TBase::BindAllocator();
+        auto guard = this->TaskRunner->BindAllocator(this->MemoryQuota->GetMkqlMemoryLimit());
         auto* alloc = guard.GetMutex();
-        alloc->SetLimit(this->MemoryQuota->GetMkqlMemoryLimit());
 
         this->MemoryQuota->TrySetIncreaseMemoryLimitCallback(alloc);