Merge fd18923 into d1f5b91

Author: yumkam

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

@@ -528,6 +528,9 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
         const bool shouldSkipData = Channels->ShouldSkipData(outputChannel.ChannelId);
         const bool hasFreeMemory = Channels->HasFreeMemoryInChannel(outputChannel.ChannelId);
         UpdateBlocked(outputChannel, !hasFreeMemory);
+        if (!hasFreeMemory) {
+            ProcessOutputsState.IsFull = true;
+        }
 
         if (!shouldSkipData && !outputChannel.EarlyFinish && !hasFreeMemory) {
             CA_LOG_T("DrainOutputChannel return because No free memory in channel, channel: " << outputChannel.ChannelId);
@@ -555,6 +558,9 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
 
         const ui32 allowedOvercommit = AllowedChannelsOvercommit();
         const i64 sinkFreeSpaceBeforeSend = sinkInfo.AsyncOutput->GetFreeSpace();
+        if (sinkFreeSpaceBeforeSend <= 0) {
+            ProcessOutputsState.IsFull = true;
+        }
 
         i64 toSend = sinkFreeSpaceBeforeSend + allowedOvercommit;
         CA_LOG_T("About to drain sink " << outputIndex
@@ -696,7 +702,7 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
     }
 
     void DoExecuteImpl() override {
-        PollAsyncInput();
+        PollAsyncInput(!ProcessOutputsState.IsFull);
         if (ProcessSourcesState.Inflight == 0) {
             auto req = GetCheckpointRequest();
             CA_LOG_T("DoExecuteImpl: " << (bool) req);
@@ -939,6 +945,10 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
             }
         }
 
+        if (!Channels->HasFreeMemoryInChannel(outputChannel.ChannelId)) {
+            ProcessOutputsState.IsFull = true;
+        }
+
         ProcessOutputsState.DataWasSent |= asyncData.Changed;
 
         ProcessOutputsState.AllOutputsFinished =
@@ -1018,6 +1028,9 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
         CA_LOG_T("Drain sink " << outputIndex
             << ". Free space decreased: " << (sinkInfo.FreeSpaceBeforeSend - sinkInfo.AsyncOutput->GetFreeSpace())
             << ", sent data from buffer: " << dataSize);
+        if (sinkInfo.AsyncOutput->GetFreeSpace() <= 0) {
+            ProcessOutputsState.IsFull = true;
+        }
 
         ProcessOutputsState.DataWasSent |= dataWasSent;
         ProcessOutputsState.AllOutputsFinished =

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

@@ -388,13 +388,13 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
     }
 
     void ProcessOutputsImpl(ERunStatus status) {
-        ProcessOutputsState.LastRunStatus = status;
-
         CA_LOG_T("ProcessOutputsState.Inflight: " << ProcessOutputsState.Inflight);
         if (ProcessOutputsState.Inflight == 0) {
             ProcessOutputsState = TProcessOutputsState();
         }
 
+        ProcessOutputsState.LastRunStatus = status;
+
         for (auto& entry : OutputChannelsMap) {
             const ui64 channelId = entry.first;
             TOutputChannelInfo& outputChannel = entry.second;
@@ -459,7 +459,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
             return;
         }
 
-        if (status != ERunStatus::Finished) {
+        if (status == ERunStatus::PendingInput) {
             for (auto& [id, inputTransform] : InputTransformsMap) {
                 if (!inputTransform.Buffer->Empty()) {
                     ContinueExecute(EResumeSource::CAPendingInput);
@@ -1462,7 +1462,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         }
     }
 
-    void PollAsyncInput() {
+    void PollAsyncInput(bool continueExecuteOnFull = true) {
         if (!Running) {
             CA_LOG_T("Skip polling inputs and sources because not running");
             return;
@@ -1471,7 +1471,9 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         CA_LOG_T("Poll inputs");
         for (auto& [inputIndex, transform] : InputTransformsMap) {
             if (auto resume = transform.PollAsyncInput(MetricsReporter, WatermarksTracker, RuntimeSettings.AsyncInputPushLimit)) {
-                ContinueExecute(*resume);
+                if (*resume != EResumeSource::CAPollAsyncNoSpace || continueExecuteOnFull) {
+                    ContinueExecute(*resume);
+                }
             }
         }
 
@@ -1484,7 +1486,9 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         CA_LOG_T("Poll sources");
         for (auto& [inputIndex, source] : SourcesMap) {
             if (auto resume =  source.PollAsyncInput(MetricsReporter, WatermarksTracker, RuntimeSettings.AsyncInputPushLimit)) {
-                ContinueExecute(*resume);
+                if (*resume != EResumeSource::CAPollAsyncNoSpace || continueExecuteOnFull) {
+                    ContinueExecute(*resume);
+                }
             }
         }
     }
@@ -2051,6 +2055,7 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         bool AllOutputsFinished = true;
         ERunStatus LastRunStatus = ERunStatus::PendingInput;
         bool LastPopReturnedNoData = false;
+        bool IsFull = false;
     };
     TProcessOutputsState ProcessOutputsState;