Microfix DQ Reader: do not hang on stream creation errors

MrLolthe1st · MrLolthe1st · commit 0ca47b802c1f · 2024-12-10T20:56:07.000+03:00
Microfix DQ Reader: do not hang on stream creation errors
commit_hash:6c67546e07f7e9ae5ab4a1dab703f58ba66f859b
diff --git a/yt/yql/providers/yt/comp_nodes/dq/dq_yt_rpc_reader.cpp b/yt/yql/providers/yt/comp_nodes/dq/dq_yt_rpc_reader.cpp
@@ -176,6 +176,7 @@ void TParallelFileInputState::RunNext() {
             CreateInputStream(Settings_->Requests[inputIdx]).SubscribeUnique(BIND([state = InnerState_, inputIdx](NYT::TErrorOr<NYT::NConcurrency::IAsyncZeroCopyInputStreamPtr>&& stream) {
                 if (!stream.IsOK()) {
                     state->Error = stream;
+                    state->WaitPromise.TrySet();
                     return;
                 }
                 state->RawInputs[inputIdx] = std::move(stream.Value());

Original file line number	Diff line number	Diff line change
`@@ -176,6 +176,7 @@ void TParallelFileInputState::RunNext() {`
`176`	`176`	`CreateInputStream(Settings_->Requests[inputIdx]).SubscribeUnique(BIND([state = InnerState_, inputIdx](NYT::TErrorOr<NYT::NConcurrency::IAsyncZeroCopyInputStreamPtr>&& stream) {`
`177`	`177`	`if (!stream.IsOK()) {`
`178`	`178`	`state->Error = stream;`
	`179`	`+ state->WaitPromise.TrySet();`
`179`	`180`	`return;`
`180`	`181`	`}`
`181`	`182`	`state->RawInputs[inputIdx] = std::move(stream.Value());`