Fix erroneous finish on TDqInputMergeBlockStreamValue (ydb-platform#8834)

Author: nepal

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