From 10bca906d17478f5ff4f5a6786637e29cbb396e0 Mon Sep 17 00:00:00 2001
From: Andrey Neporada <aneporada@ydb.tech>
Date: Fri, 6 Sep 2024 13:53:36 +0300
Subject: [PATCH] Fix erroneous finish on TDqInputMergeBlockStreamValue (#8834)

---
 ydb/library/yql/dq/runtime/dq_input_producer.cpp | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/ydb/library/yql/dq/runtime/dq_input_producer.cpp b/ydb/library/yql/dq/runtime/dq_input_producer.cpp
index 05c80f11921b..c35e3c45c397 100644
--- a/ydb/library/yql/dq/runtime/dq_input_producer.cpp
+++ b/ydb/library/yql/dq/runtime/dq_input_producer.cpp
@@ -433,6 +433,10 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
             return CurrBlockIndex_ >= BlockLen_;
         }
 
+        bool IsFinished() const {
+            return IsFinished_;
+        }
+
         void NextRow() {
             Y_DEBUG_ABORT_UNLESS(!IsEmpty());
             ++CurrBlockIndex_;
@@ -645,7 +649,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;
@@ -657,6 +661,7 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
         }
 
         if (!OutputBlockLen_) {
+            YQL_ENSURE(AllOf(InputData_, [](const TDqInputBatch& input) { return input.IsEmpty() && input.IsFinished(); }));
             return NUdf::EFetchStatus::Finish;
         }