[yt provider] Don't run operation input if sections have nodes to calculate (ydb-platform#11947)

Author: rvu1024

ydb/library/yql/providers/yt/provider/yql_yt_datasink_exec.cpp

@@ -93,7 +93,7 @@ class TYtDataSinkExecTransformer : public TExecTransformerBase {
                 TYtMerge::CallableName(),
                 TYtMapReduce::CallableName(),
             },
-            RequireAllOf({TYtTransientOpBase::idx_World, TYtTransientOpBase::idx_Input}),
+            RequireForTransientOp(),
             Hndl(&TYtDataSinkExecTransformer::HandleOutputOp<true>)
         );
         AddHandler(
@@ -104,7 +104,7 @@ class TYtDataSinkExecTransformer : public TExecTransformerBase {
             RequireFirst(),
             Hndl(&TYtDataSinkExecTransformer::HandleOutputOp<true>)
         );
-        AddHandler({TYtReduce::CallableName()}, RequireAllOf({TYtTransientOpBase::idx_World, TYtTransientOpBase::idx_Input}), Hndl(&TYtDataSinkExecTransformer::HandleReduce));
+        AddHandler({TYtReduce::CallableName()}, RequireForTransientOp(), Hndl(&TYtDataSinkExecTransformer::HandleReduce));
         AddHandler({TYtOutput::CallableName()}, RequireFirst(), Pass());
         AddHandler({TYtPublish::CallableName()}, RequireAllOf({TYtPublish::idx_World, TYtPublish::idx_Input}), Hndl(&TYtDataSinkExecTransformer::HandlePublish));
         AddHandler({TYtDropTable::CallableName()}, RequireFirst(), Hndl(&TYtDataSinkExecTransformer::HandleDrop));
@@ -123,6 +123,21 @@ class TYtDataSinkExecTransformer : public TExecTransformerBase {
         TExecTransformerBase::Rewind();
     }
 
+    static TExecTransformerBase::TPrerequisite RequireForTransientOp() {
+        return [] (const TExprNode::TPtr& input) {
+            auto status = RequireChild(*input, TYtTransientOpBase::idx_World);
+            // We have to run input only if it has no settings to calculate.
+            // Otherwise, we first of all wait world completion.
+            // Then begins node execution, which run settings calculation.
+            // And after that, starts input execution
+            // See YQL-19303
+            if (!HasNodesToCalculate(input->ChildPtr(TYtTransientOpBase::idx_Input))) {
+                status = status.Combine(RequireChild(*input, TYtTransientOpBase::idx_Input));
+            }
+            return status;
+        };
+    }
+
 private:
     static void PushHybridStats(const TYtState::TPtr& state, TStringBuf statName, TStringBuf opName, const TStringBuf& folderName = "") {
         with_lock(state->StatisticsMutex) {
@@ -189,6 +204,10 @@ class TYtDataSinkExecTransformer : public TExecTransformerBase {
             return CalculateNodes(State_, input, cluster, needCalc, ctx);
         }
 
+        if (auto opInput = op.Maybe<TYtTransientOpBase>().Input()) {
+            YQL_ENSURE(opInput.Ref().GetState() == TExprNode::EState::ExecutionComplete);
+        }
+
         auto outSection = op.Output();
 
         size_t outWithoutName = 0;