Yq 2068 task runner events refinement (#2122)

5 files changed, 120 insertions, 131 deletions

diff --git a/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp b/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp
index 9bb9b48064d..183891a205e 100644
--- a/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp
+++ b/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp
@@ -171,10 +171,10 @@ private:
     STFUNC(StateFuncBody) {
         switch (ev->GetTypeRewrite()) {
             hFunc(NTaskRunnerActor::TEvTaskRunFinished, OnRunFinished);
-            hFunc(NTaskRunnerActor::TEvAsyncInputPushFinished, OnAsyncInputPushFinished);
-            hFunc(NTaskRunnerActor::TEvChannelPopFinished, OnPopFinished);
+            hFunc(NTaskRunnerActor::TEvSourceDataAck, OnSourceDataAck);
+            hFunc(NTaskRunnerActor::TEvOutputChannelData, OnOutputChannelData);
             hFunc(NTaskRunnerActor::TEvTaskRunnerCreateFinished, OnTaskRunnerCreated);
-            hFunc(NTaskRunnerActor::TEvPushFinished, OnPushFinished);
+            hFunc(NTaskRunnerActor::TEvInputChannelDataAck, OnInputChannelDataAck);
             hFunc(TEvDqCompute::TEvStateRequest, OnStateRequest);
             hFunc(NTaskRunnerActor::TEvStatistics, OnStatisticsResponse);
             hFunc(NTaskRunnerActor::TEvLoadTaskRunnerFromStateDone, OnTaskRunnerLoaded);
@@ -277,13 +277,13 @@ private:
             CA_LOG_T("Can not drain channel because it is blocked by capacity. ChannelId: " << channelId
                 << ", peerState:(" << peerState.DebugString() << ")"
             );
-            auto ev = MakeHolder<NTaskRunnerActor::TEvChannelPopFinished>(channelId);
+            auto ev = MakeHolder<NTaskRunnerActor::TEvOutputChannelData>(channelId);
             Y_ABORT_UNLESS(!ev->Finished);
             Send(SelfId(), std::move(ev));  // try again, ev.Finished == false
             return;
         }
 
-        Send(TaskRunnerActorId, new NTaskRunnerActor::TEvPop(channelId, wasFinished, peerState.GetFreeMemory()));
+        Send(TaskRunnerActorId, new NTaskRunnerActor::TEvOutputChannelDataRequest(channelId, wasFinished, peerState.GetFreeMemory()));
     }
 
     void DrainAsyncOutput(ui64 outputIndex, TAsyncOutputInfoBase& sinkInfo) override {
@@ -312,7 +312,7 @@ private:
         sinkInfo.PopStarted = true;
         ProcessOutputsState.Inflight++;
         sinkInfo.FreeSpaceBeforeSend = sinkFreeSpaceBeforeSend;
-        Send(TaskRunnerActorId, new NTaskRunnerActor::TEvSinkPop(outputIndex, sinkFreeSpaceBeforeSend));
+        Send(TaskRunnerActorId, new NTaskRunnerActor::TEvSinkDataRequest(outputIndex, sinkFreeSpaceBeforeSend));
     }
 
     bool DoHandleChannelsAfterFinishImpl() override {
@@ -334,11 +334,10 @@ private:
 
         outputChannel->Finished = true;
         ProcessOutputsState.Inflight++;
-        Send(TaskRunnerActorId, MakeHolder<NTaskRunnerActor::TEvPop>(channelId, /* wasFinished = */ true, 0));  // finish channel
+        Send(TaskRunnerActorId, MakeHolder<NTaskRunnerActor::TEvOutputChannelDataRequest>(channelId, /* wasFinished = */ true, 0));  // finish channel
         DoExecute();
     }
 
-
     void TakeInputChannelData(TChannelDataOOB&& channelDataOOB, bool ack) override {
         CA_LOG_T("took input");
         NDqProto::TChannelData& channelData = channelDataOOB.Proto;
@@ -373,16 +372,12 @@ private:
             batch.RowCount(),
             watermark);
 
-        auto ev = batch.RowCount()
-            ? MakeHolder<NTaskRunnerActor::TEvPush>(
-                channelData.GetChannelId(),
-                std::move(batch),
-                finished,
-                /* pauseAfterPush = */ channelData.HasCheckpoint())
-            : MakeHolder<NTaskRunnerActor::TEvPush>(
-                channelData.GetChannelId(),
-                finished,
-                /* pauseAfterPush = */ channelData.HasCheckpoint());
+        auto ev = MakeHolder<NTaskRunnerActor::TEvInputChannelData>(
+            channelData.GetChannelId(),
+            batch.RowCount() ? std::optional{std::move(batch)} : std::nullopt,
+            finished,
+            channelData.HasCheckpoint()
+        );
 
         Send(TaskRunnerActorId, ev.Release(), 0, Cookie);
 
@@ -564,7 +559,7 @@ private:
         }
     }
 
-    void OnAsyncInputPushFinished(NTaskRunnerActor::TEvAsyncInputPushFinished::TPtr& ev) {
+    void OnSourceDataAck(NTaskRunnerActor::TEvSourceDataAck::TPtr& ev) {
         auto it = SourcesMap.find(ev->Get()->Index);
         Y_ABORT_UNLESS(it != SourcesMap.end());
         auto& source = it->second;
@@ -576,7 +571,7 @@ private:
         }
     }
 
-    void OnPopFinished(NTaskRunnerActor::TEvChannelPopFinished::TPtr& ev) {
+    void OnOutputChannelData(NTaskRunnerActor::TEvOutputChannelData::TPtr& ev) {
         if (Stat) {
             Stat->AddCounters2(ev->Get()->Sensors);
         }
@@ -693,7 +688,7 @@ private:
         return result;
     }
 
-    void OnPushFinished(NTaskRunnerActor::TEvPushFinished::TPtr& ev) {
+    void OnInputChannelDataAck(NTaskRunnerActor::TEvInputChannelDataAck::TPtr& ev) {
         auto it = TakeInputChannelDataRequests.find(ev->Cookie);
         YQL_ENSURE(it != TakeInputChannelDataRequests.end());
 
diff --git a/ydb/library/yql/dq/actors/task_runner/events.h b/ydb/library/yql/dq/actors/task_runner/events.h
index 69f6f586f5e..b382e346c4b 100644
--- a/ydb/library/yql/dq/actors/task_runner/events.h
+++ b/ydb/library/yql/dq/actors/task_runner/events.h
@@ -25,21 +25,21 @@ struct TTaskRunnerEvents {
     enum {
         EvCreate = EventSpaceBegin(NActors::TEvents::EEventSpace::ES_USERSPACE) + 20000,
         EvCreateFinished,
-        // channel->Pop -> TEvChannelPopFinished
-        EvPop,
-        EvPopFinished,
-        // channel->Push -> TEvPushFinished
-        EvPush,
-        EvPushFinished,
+
+        EvOutputChannelDataRequest,
+        EvOutputChannelData,
+
+	EvInputChannelData,
+        EvInputChannelDataAck,
+
+	// EvContinueRun -> TaskRunner->Run() -> TEvTaskRunFinished
         EvContinueRun,
-        // EvContinueRun -> TaskRunner->Run() -> TEvTaskRunFinished
         EvRunFinished,
 
-        EvAsyncInputPush,
-        EvAsyncInputPushFinished,
+        EvSourceDataAck,
 
-        EvSinkPop,
-        EvSinkPopFinished,
+        EvSinkDataRequest,
+        EvSinkData,
 
         EvLoadTaskRunnerFromState,
         EvLoadTaskRunnerFromStateDone,
@@ -90,16 +90,19 @@ struct TEvError
     TMaybe<TStatus> Status = Nothing();
 };
 
-struct TEvPop
-    : NActors::TEventLocal<TEvPop, TTaskRunnerEvents::EvPop>
+//Sent by ComputActor to TaskRunnerActor to request data from output channel.
+//Upon receiving this event, TaskRunnerActor reads data from corresponding TaskRunner's output channel,
+//collects data by chunks and then sends data chunks to the sender of this message with TEvOutputChaanelData.
+//See also notes to TEvOutpuChannelData.
+struct TEvOutputChannelDataRequest
+    : NActors::TEventLocal<TEvOutputChannelDataRequest, TTaskRunnerEvents::EvOutputChannelDataRequest>
 {
-    TEvPop() = default;
-    TEvPop(ui32 channelId, bool wasFinished, i64 toPop)
+    TEvOutputChannelDataRequest(ui32 channelId, bool wasFinished, i64 requestedSize)
         : ChannelId(channelId)
         , WasFinished(wasFinished)
-        , Size(toPop)
+        , Size(requestedSize)
     { }
-    TEvPop(ui32 channelId)
+    TEvOutputChannelDataRequest(ui32 channelId)
         : ChannelId(channelId)
         , WasFinished(false)
         , Size(0)
@@ -110,45 +113,37 @@ struct TEvPop
     const i64 Size;
 };
 
-struct TEvPush
-    : NActors::TEventLocal<TEvPush, TTaskRunnerEvents::EvPush>
+//Sent by ComputeActor to TaskRunnerActor and contains a portion on input channel data
+//Upon receiving this event, TaskRunnerActor moves data to its input channel buffer and send TEvInputChannelDataAck back to the sender
+//See also notes to TEvInputChannelDataAck
+struct TEvInputChannelData
+    : NActors::TEventLocal<TEvInputChannelData, TTaskRunnerEvents::EvInputChannelData>
 {
-    TEvPush() = default;
-    TEvPush(ui32 channelId, bool finish = true, bool pauseAfterPush = false, bool isOut = false)
-        : ChannelId(channelId)
-        , HasData(false)
-        , Finish(finish)
-        , PauseAfterPush(pauseAfterPush)
-        , IsOut(isOut)
-    { }
-    TEvPush(ui32 channelId, TDqSerializedBatch&& data, bool finish = false, bool pauseAfterPush = false)
+    TEvInputChannelData(ui32 channelId, std::optional<TDqSerializedBatch>&& data, bool finish, bool pauseAfterPush)
         : ChannelId(channelId)
-        , HasData(true)
-        , Finish(finish)
         , Data(std::move(data))
+        , Finish(finish)
         , PauseAfterPush(pauseAfterPush)
     { }
 
     const ui32 ChannelId;
-    const bool HasData = false;
-    const bool Finish = false;
-    TDqSerializedBatch Data;
-    bool PauseAfterPush = false;
-    const bool IsOut = false;
+    std::optional<TDqSerializedBatch> Data; //not const, because we want to efficiently move data out of this event on a reciever side
+    const bool Finish;
+    const bool PauseAfterPush;
 };
 
-struct TEvPushFinished
-    : NActors::TEventLocal<TEvPushFinished, TTaskRunnerEvents::EvPushFinished> {
-
-    TEvPushFinished() = default;
+//Sent by TaskRunnerActor to ComputeActor to ackonowledge input data received in TEvInputChannelData
+//See also note to TEvInputChannelData
+struct TEvInputChannelDataAck
+    : NActors::TEventLocal<TEvInputChannelDataAck, TTaskRunnerEvents::EvInputChannelDataAck> {
 
-    TEvPushFinished(ui32 channelId, ui64 freeSpace)
+    TEvInputChannelDataAck(ui32 channelId, ui64 freeSpace)
         : ChannelId(channelId)
         , FreeSpace(freeSpace)
     { }
 
-    ui32 ChannelId;
-    ui64 FreeSpace;
+    const ui32 ChannelId;
+    const ui64 FreeSpace;
 };
 
 struct TEvTaskRunnerCreate
@@ -205,7 +200,6 @@ struct TEvTaskRunnerCreateFinished
 struct TEvTaskRunFinished
     : NActors::TEventLocal<TEvTaskRunFinished, TTaskRunnerEvents::EvRunFinished>
 {
-    TEvTaskRunFinished() = default;
     TEvTaskRunFinished(
         NDq::ERunStatus runStatus,
         THashMap<ui32, i64>&& inputChannelsMap,
@@ -242,12 +236,13 @@ struct TEvTaskRunFinished
     TDuration ComputeTime;
 };
 
-struct TEvChannelPopFinished
-    : NActors::TEventLocal<TEvChannelPopFinished, TTaskRunnerEvents::EvPopFinished>
+//Sent by TaskRunnerActor to ComputeActor in response to TEvOutputChannelDataRequest.
+//Contains data read from output channel accompanied with auxiliary info.
+//See also notes to TEvOutputChannelDataRequest
+struct TEvOutputChannelData
+    : NActors::TEventLocal<TEvOutputChannelData, TTaskRunnerEvents::EvOutputChannelData>
 {
-    TEvChannelPopFinished() = default;
-
-    TEvChannelPopFinished(ui32 channelId)
+    TEvOutputChannelData(ui32 channelId)
         : Stats()
         , ChannelId(channelId)
         , Data()
@@ -255,7 +250,7 @@ struct TEvChannelPopFinished
         , Changed(false)
     { }
 
-    TEvChannelPopFinished(
+    TEvOutputChannelData(
             ui32 channelId,
             TVector<TDqSerializedBatch>&& data,
             TMaybe<NDqProto::TWatermark>&& watermark,
@@ -343,37 +338,43 @@ struct TEvContinueRun
     TVector<ui32> InputTransformIds;
 };
 
-struct TEvAsyncInputPushFinished
-    : NActors::TEventLocal<TEvAsyncInputPushFinished, TTaskRunnerEvents::EvAsyncInputPushFinished>
+//Sent by TaskRunnerActor to ComputeActor as an acknowledgement in AsyncInputPush method call
+//after data is pushed to corresponding TaskRunner's source
+struct TEvSourceDataAck
+    : NActors::TEventLocal<TEvSourceDataAck, TTaskRunnerEvents::EvSourceDataAck>
 {
-    TEvAsyncInputPushFinished() = default;
-    TEvAsyncInputPushFinished(ui64 index, i64 freeSpaceLeft)
+    TEvSourceDataAck(ui64 index, i64 freeSpaceLeft)
         : Index(index)
         , FreeSpaceLeft(freeSpaceLeft)
     { }
 
-    ui64 Index;
-    i64 FreeSpaceLeft;
+    const ui64 Index;
+    const i64 FreeSpaceLeft;
 };
 
-struct TEvSinkPop
-    : NActors::TEventLocal<TEvSinkPop, TTaskRunnerEvents::EvSinkPop>
+//Sent by ComputeActor to TaskRunnerActor to request output data from a sink.
+//Upon receiving this event, TaskRunnerActor reads data from corresponding TaskRunner's sink buffer,
+//and sends data to the sender of this message with TEvSinkData.
+//See also notes to TEvSinkData.
+struct TEvSinkDataRequest
+    : NActors::TEventLocal<TEvSinkDataRequest, TTaskRunnerEvents::EvSinkDataRequest>
 {
-    TEvSinkPop() = default;
-    TEvSinkPop(ui64 index, i64 size)
+    TEvSinkDataRequest(ui64 index, i64 size)
         : Index(index)
         , Size(size)
     { }
 
-    ui64 Index;
-    i64 Size;
+    const ui64 Index;
+    const i64 Size;
 };
 
-struct TEvSinkPopFinished
-    : NActors::TEventLocal<TEvSinkPopFinished, TTaskRunnerEvents::EvSinkPopFinished>
+//Sent by TaskRunnerActor to ComputeActor in response to TEvSinkDataRequest.
+//Contains data from TaskRunner's sink buffer
+//See also notes to TEvSinkDataRequest.
+struct TEvSinkData
+    : NActors::TEventLocal<TEvSinkData, TTaskRunnerEvents::EvSinkData>
 {
-    TEvSinkPopFinished() = default;
-    TEvSinkPopFinished(
+    TEvSinkData(
         ui64 index,
         TMaybe<NDqProto::TCheckpoint>&& checkpoint,
         i64 size,
diff --git a/ydb/library/yql/dq/actors/task_runner/task_runner_actor_local.cpp b/ydb/library/yql/dq/actors/task_runner/task_runner_actor_local.cpp
index 2cb8e72eeb0..93d38d8e066 100644
--- a/ydb/library/yql/dq/actors/task_runner/task_runner_actor_local.cpp
+++ b/ydb/library/yql/dq/actors/task_runner/task_runner_actor_local.cpp
@@ -59,9 +59,9 @@ public:
                 cFunc(NActors::TEvents::TEvPoison::EventType, TLocalTaskRunnerActor::PassAway);
                 hFunc(TEvTaskRunnerCreate, OnDqTask);
                 hFunc(TEvContinueRun, OnContinueRun);
-                hFunc(TEvPop, OnChannelPop);
-                hFunc(TEvPush, OnChannelPush);
-                hFunc(TEvSinkPop, OnSinkPop);
+                hFunc(TEvOutputChannelDataRequest, OnOutputChannelDataRequest);
+                hFunc(TEvInputChannelData, OnInputChannelData);
+                hFunc(TEvSinkDataRequest, OnSinkDataRequest);
                 hFunc(TEvLoadTaskRunnerFromState, OnLoadTaskRunnerFromState);
                 hFunc(TEvStatistics, OnStatisticsRequest);
                 default: {
@@ -253,19 +253,13 @@ private:
             ev->Cookie);
     }
 
-    void OnChannelPush(TEvPush::TPtr& ev) {
+    void OnInputChannelData(TEvInputChannelData::TPtr& ev) {
         auto guard = TaskRunner->BindAllocator();
-        auto hasData = ev->Get()->HasData;
         auto finish = ev->Get()->Finish;
         auto channelId = ev->Get()->ChannelId;
-        if (ev->Get()->IsOut) {
-            Y_ABORT_UNLESS(ev->Get()->Finish, "dont know what to do with the output channel");
-            TaskRunner->GetOutputChannel(channelId)->Finish();
-            return;
-        }
         auto inputChannel = TaskRunner->GetInputChannel(channelId);
-        if (hasData) {
-            inputChannel->Push(std::move(ev->Get()->Data));
+        if (ev->Get()->Data) {
+            inputChannel->Push(std::move(*ev->Get()->Data));
         }
         const ui64 freeSpace = inputChannel->GetFreeSpace();
         if (finish) {
@@ -278,7 +272,7 @@ private:
         // run
         Send(
             ev->Sender,
-            new TEvPushFinished(channelId, freeSpace),
+            new TEvInputChannelDataAck(channelId, freeSpace),
             /*flags=*/0,
             ev->Cookie);
     }
@@ -297,22 +291,22 @@ private:
         }
         Send(
             ParentId,
-            new TEvAsyncInputPushFinished(index, source->GetFreeSpace()),
+            new TEvSourceDataAck(index, source->GetFreeSpace()),
             /*flags=*/0,
             cookie);
     }
 
-    void OnChannelPop(TEvPop::TPtr& ev) {
+    void OnOutputChannelDataRequest(TEvOutputChannelDataRequest::TPtr& ev) {
         auto guard = TaskRunner->BindAllocator();
 
         auto channelId = ev->Get()->ChannelId;
         auto channel = TaskRunner->GetOutputChannel(channelId);
-        if (ev->Get()->WasFinished) {
+        auto wasFinished = ev->Get()->WasFinished;
+        if (wasFinished) {
             channel->Finish();
             LOG_I("output channel with id [" << channelId << "] finished prematurely");
         }
         int maxChunks = std::numeric_limits<int>::max();
-        auto wasFinished = ev->Get()->WasFinished;
         bool changed = false;
         bool isFinished = false;
         i64 remain = ev->Get()->Size;
@@ -362,7 +356,7 @@ private:
 
         Send(
             ev->Sender,
-            new TEvChannelPopFinished(
+            new TEvOutputChannelData(
                 channelId,
                 std::move(chunks),
                 std::move(watermark),
@@ -381,7 +375,7 @@ private:
         }
     }
 
-    void OnSinkPop(TEvSinkPop::TPtr& ev) {
+    void OnSinkDataRequest(TEvSinkDataRequest::TPtr& ev) {
         auto guard = TaskRunner->BindAllocator();
         auto sink = TaskRunner->GetSink(ev->Get()->Index);
 
diff --git a/ydb/library/yql/providers/dq/actors/worker_actor.cpp b/ydb/library/yql/providers/dq/actors/worker_actor.cpp
index a4588de0a19..887801f5914 100644
--- a/ydb/library/yql/providers/dq/actors/worker_actor.cpp
+++ b/ydb/library/yql/providers/dq/actors/worker_actor.cpp
@@ -142,15 +142,15 @@ private:
         cFunc(NActors::TEvents::TEvPoison::EventType, PassAway);
 
         HFunc(TEvTaskRunnerCreateFinished, OnTaskRunnerCreated);
-        HFunc(TEvChannelPopFinished, OnChannelPopFinished);
+        HFunc(TEvOutputChannelData, OnOutputChannelData);
         HFunc(TEvTaskRunFinished, OnRunFinished);
-        HFunc(TEvAsyncInputPushFinished, OnAsyncInputPushFinished);
+        HFunc(TEvSourceDataAck, OnSourceDataAck);
 
         // weird to have two events for error handling, but we need to use TEvDqFailure
         // between worker_actor <-> executer_actor, cause it transmits statistics in 'Metric' field
         HFunc(NDq::TEvDq::TEvAbortExecution, OnErrorFromPipe);  // received from task_runner_actor
         HFunc(TEvDqFailure, OnError); // received from this actor itself
-        HFunc(TEvPushFinished, OnPushFinished);
+        HFunc(TEvInputChannelDataAck, OnInputChannelDataAck);
         cFunc(TEvents::TEvWakeup::EventType, OnWakeup);
 
         hFunc(IDqComputeActorAsyncInput::TEvNewAsyncInputDataArrived, OnNewAsyncInputDataArrived);
@@ -222,7 +222,7 @@ private:
         Send(Executer, std::move(ev));
     }
 
-    void OnPushFinished(TEvPushFinished::TPtr&, const TActorContext& ctx) {
+    void OnInputChannelDataAck(TEvInputChannelDataAck::TPtr&, const TActorContext& ctx) {
         Run(ctx);
     }
 
@@ -379,10 +379,10 @@ private:
         auto& outChannel = OutputMap[ev->Sender];
         outChannel.RequestTime = now;
 
-        Send(TaskRunnerActor, new TEvPop(outChannel.ChannelId));
+        Send(TaskRunnerActor, new TEvOutputChannelDataRequest(outChannel.ChannelId, false, 0));
     }
 
-    void OnChannelPopFinished(TEvChannelPopFinished::TPtr& ev, const NActors::TActorContext& ctx) {
+    void OnOutputChannelData(TEvOutputChannelData::TPtr& ev, const NActors::TActorContext& ctx) {
         try {
             TFailureInjector::Reach("dq_fail_on_channel_pop_finished", [] { throw yexception() << "dq_fail_on_channel_pop_finished"; });
             auto outputActorId = OutChannelId2ActorId[ev->Get()->ChannelId];
@@ -451,7 +451,7 @@ private:
         }
         Y_ABORT_UNLESS(responseType == FINISH || responseType == CONTINUE);
         if (responseType == FINISH) {
-            Send(TaskRunnerActor, new TEvPush(channel.ChannelId));
+            Send(TaskRunnerActor, new TEvInputChannelData(channel.ChannelId, {}, true, false));
         } else {
             TDqSerializedBatch data;
             if (response.GetData().HasPayloadId()) {
@@ -459,7 +459,7 @@ private:
             }
             data.Proto = std::move(*response.MutableData());
             data.Proto.ClearPayloadId();
-            Send(TaskRunnerActor, new TEvPush(channel.ChannelId, std::move(data)));
+            Send(TaskRunnerActor, new TEvInputChannelData(channel.ChannelId, {std::move(data)}, false, false));
         }
     }
 
@@ -625,7 +625,7 @@ private:
                 for (auto& [index, sink] : SinksMap) {
                     const i64 sinkFreeSpaceBeforeSend = sink.Sink->GetFreeSpace();
                     if (sinkFreeSpaceBeforeSend > 0 && !sink.Finished) {
-                        Send(TaskRunnerActor, new TEvSinkPop(index, sinkFreeSpaceBeforeSend));
+                        Send(TaskRunnerActor, new TEvSinkDataRequest(index, sinkFreeSpaceBeforeSend));
                     }
                 }
                 break;
@@ -720,7 +720,7 @@ private:
         }
         SendFailure(MakeHolder<TEvDqFailure>(fatalCode, ev->Get()->Issues.ToString()));
     }
-    void OnAsyncInputPushFinished(TEvAsyncInputPushFinished::TPtr& ev, const TActorContext& ctx) {
+    void OnSourceDataAck(TEvSourceDataAck::TPtr& ev, const TActorContext& ctx) {
         auto index = ev->Get()->Index;
         auto& source = SourcesMap[index];
         source.PushStarted = false;
diff --git a/ydb/library/yql/providers/dq/task_runner_actor/task_runner_actor.cpp b/ydb/library/yql/providers/dq/task_runner_actor/task_runner_actor.cpp
index 55f844a0766..4772abfb085 100644
--- a/ydb/library/yql/providers/dq/task_runner_actor/task_runner_actor.cpp
+++ b/ydb/library/yql/providers/dq/task_runner_actor/task_runner_actor.cpp
@@ -228,10 +228,10 @@ public:
             cFunc(NActors::TEvents::TEvPoison::EventType, TTaskRunnerActor::PassAway);
             hFunc(TEvTaskRunnerCreate, OnDqTask);
             hFunc(TEvContinueRun, OnContinueRun);
-            hFunc(TEvPop, OnChannelPop);
-            hFunc(TEvPush, OnChannelPush);
-            hFunc(TEvSinkPop, OnSinkPop);
-            hFunc(TEvSinkPopFinished, OnSinkPopFinished);
+            hFunc(TEvOutputChannelDataRequest, OnOutputhannelDataRequest);
+            hFunc(TEvInputChannelData, OnInputChannelData);
+            hFunc(TEvSinkDataRequest, OnSinkDataRequest);
+            hFunc(TEvSinkData, OnSinkData);
             IgnoreFunc(TEvStatistics);
             default: {
                 auto message = TStringBuilder() << "Unexpected event: " << ev->GetTypeRewrite() << " (" << ev->GetTypeName() << ")" << " stageId: " << StageId;
@@ -362,22 +362,21 @@ private:
         Run(ev);
     }
 
-    void OnChannelPush(TEvPush::TPtr& ev) {
+    void OnInputChannelData(TEvInputChannelData::TPtr& ev) {
         auto* actorSystem = TActivationContext::ActorSystem();
         auto replyTo = ev->Sender;
         auto selfId = SelfId();
-        auto hasData = ev->Get()->HasData;
         auto finish = ev->Get()->Finish;
         auto channelId = ev->Get()->ChannelId;
         auto cookie = ev->Cookie;
         auto data = ev->Get()->Data;
-        Invoker->Invoke([hasData, selfId, cookie, finish, channelId, taskRunner=TaskRunner, data, actorSystem, replyTo, settings=Settings, stageId=StageId] () mutable {
+        Invoker->Invoke([selfId, cookie, finish, channelId, taskRunner=TaskRunner, data, actorSystem, replyTo, settings=Settings, stageId=StageId] () mutable {
             try {
                 // todo:(whcrc) finish output channel?
                 ui64 freeSpace = 0;
-                if (hasData) {
+                if (data) {
                     // auto guard = taskRunner->BindAllocator(); // only for local mode
-                    taskRunner->GetInputChannel(channelId)->Push(std::move(data));
+                    taskRunner->GetInputChannel(channelId)->Push(std::move(*data));
                     freeSpace = taskRunner->GetInputChannel(channelId)->GetFreeSpace();
                 }
                 if (finish) {
@@ -389,7 +388,7 @@ private:
                     new IEventHandle(
                         replyTo,
                         selfId,
-                        new TEvPushFinished(channelId, freeSpace),
+                        new TEvInputChannelDataAck(channelId, freeSpace),
                         /*flags=*/0,
                         cookie));
             } catch (...) {
@@ -433,7 +432,7 @@ private:
                     new IEventHandle(
                         parentId,
                         selfId,
-                        new TEvAsyncInputPushFinished(index, source->GetFreeSpace()),
+                        new TEvSourceDataAck(index, source->GetFreeSpace()),
                         /*flags=*/0,
                         cookie));
             } catch (...) {
@@ -449,7 +448,7 @@ private:
         });
     }
 
-    void OnChannelPop(TEvPop::TPtr& ev) {
+    void OnOutputhannelDataRequest(TEvOutputChannelDataRequest::TPtr& ev) {
         auto* actorSystem = TActivationContext::ActorSystem();
         auto replyTo = ev->Sender;
         auto selfId = SelfId();
@@ -477,7 +476,7 @@ private:
                     new IEventHandle(
                         replyTo,
                         selfId,
-                        new TEvChannelPopFinished(
+                        new TEvOutputChannelData(
                             channelId,
                             std::move(result.DataChunks),
                             Nothing(),
@@ -500,7 +499,7 @@ private:
         });
     }
 
-    void OnSinkPopFinished(TEvSinkPopFinished::TPtr& ev) {
+    void OnSinkData(TEvSinkData::TPtr& ev) {
         auto guard = TaskRunner->BindAllocator();
         NKikimr::NMiniKQL::TUnboxedValueBatch batch;
         auto sink = TaskRunner->GetSink(ev->Get()->Index);
@@ -517,7 +516,7 @@ private:
             ev->Get()->Changed);
     }
 
-    void OnSinkPop(TEvSinkPop::TPtr& ev) {
+    void OnSinkDataRequest(TEvSinkDataRequest::TPtr& ev) {
         auto selfId = SelfId();
         auto* actorSystem = TActivationContext::ActorSystem();
 
@@ -543,7 +542,7 @@ private:
                 }
                 auto finished = sink->IsFinished();
                 bool changed = finished || ev->Get()->Size > 0 || hasCheckpoint;
-                auto event = MakeHolder<TEvSinkPopFinished>(
+                auto event = MakeHolder<TEvSinkData>(
                     ev->Get()->Index,
                     std::move(maybeCheckpoint), size, checkpointSize, finished, changed);
                 event->Batch = std::move(batch);