Fix erroneous finish on TDqInputMergeBlockStreamValue (ydb-platform#8834

)
shnikd · Sep 6, 2024 · 4782b8a · 4782b8a
1 parent f464fbb
commit 4782b8a
Showing 1 changed file with 6 additions and 1 deletion.
diff --git a/ydb/library/yql/dq/runtime/dq_input_producer.cpp b/ydb/library/yql/dq/runtime/dq_input_producer.cpp
@@ -421,6 +421,10 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
             return CurrBlockIndex_ >= BlockLen_;
         }
 
+        bool IsFinished() const {
+            return IsFinished_;
+        }
+
         void NextRow() {
             Y_DEBUG_ABORT_UNLESS(!IsEmpty());
             ++CurrBlockIndex_;
@@ -633,7 +637,7 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
             input.NextRow();
             InputRows_.pop_back();
             if (input.IsEmpty()) {
-                auto status = input.FetchNext();
+                auto status = FetchInput(inputIndex);
                 if (status == NUdf::EFetchStatus::Yield) {
                     StartInputIndex_ = inputIndex;
                     return status;
@@ -645,6 +649,7 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
         }
 
         if (!OutputBlockLen_) {
+            YQL_ENSURE(AllOf(InputData_, [](const TDqInputBatch& input) { return input.IsEmpty() && input.IsFinished(); }));
             return NUdf::EFetchStatus::Finish;
         }