Merge c73b4ce into c15923e

Author: yumkam

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

@@ -696,7 +696,13 @@ class TDqAsyncComputeActor : public TDqComputeActorBase<TDqAsyncComputeActor, TC
     }
 
     void DoExecuteImpl() override {
-        PollAsyncInput();
+        bool isFull = false;
+        if (ProcessOutputsState.Inflight == 0 && ProcessOutputsState.HasDataToSend && !ProcessOutputsState.DataWasSent && ProcessOutputsState.LastRunStatus == ERunStatus::PendingOutput) {
+            // FIXME is this condition correct?
+            // we need to stop polling when output is full and all input buffers are full
+            isFull = true;
+        }
+        PollAsyncInput(!isFull);
         if (ProcessSourcesState.Inflight == 0) {
             auto req = GetCheckpointRequest();
             CA_LOG_T("DoExecuteImpl: " << (bool) req);

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

@@ -1465,7 +1465,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;
@@ -1474,7 +1474,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);
+                }
             }
         }
 
@@ -1487,7 +1489,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);
+                }
             }
         }
     }