Fix erroneous finish on TDqInputMergeBlockStreamValue (#8834)

Author: nepal

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;
         }