This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-2.1 in repository https://gitbox.apache.org/repos/asf/doris.git
commit f0a4ec5f8572854e3bb8018eb69a37bf74a1d920 Author: Gabriel <[email protected]> AuthorDate: Wed Jan 17 12:22:49 2024 +0800 [pipelineX](fix) Fix runtime filter dependency DCHECK failed (#30038) --- be/src/pipeline/pipeline_x/pipeline_x_task.cpp | 22 +++++++++------------- 1 file changed, 9 insertions(+), 13 deletions(-) diff --git a/be/src/pipeline/pipeline_x/pipeline_x_task.cpp b/be/src/pipeline/pipeline_x/pipeline_x_task.cpp index 574f4e7e5cb..1b1c04eb814 100644 --- a/be/src/pipeline/pipeline_x/pipeline_x_task.cpp +++ b/be/src/pipeline/pipeline_x/pipeline_x_task.cpp @@ -195,20 +195,16 @@ Status PipelineXTask::_open() { _dry_run = _sink->should_dry_run(_state); for (auto& o : _operators) { auto* local_state = _state->get_local_state(o->operator_id()); - for (size_t i = 0; i < 2; i++) { - auto st = local_state->open(_state); - if (st.is<ErrorCode::PIP_WAIT_FOR_RF>()) { - DCHECK(_filter_dependency); - _blocked_dep = _filter_dependency->is_blocked_by(this); - if (_blocked_dep) { - set_state(PipelineTaskState::BLOCKED_FOR_RF); - RETURN_IF_ERROR(st); - } else if (i == 1) { - CHECK(false) << debug_string(); - } - } else { - break; + auto st = local_state->open(_state); + if (st.is<ErrorCode::PIP_WAIT_FOR_RF>()) { + DCHECK(_filter_dependency); + _blocked_dep = _filter_dependency->is_blocked_by(this); + if (_blocked_dep) { + set_state(PipelineTaskState::BLOCKED_FOR_RF); + RETURN_IF_ERROR(st); } + } else { + RETURN_IF_ERROR(st); } } RETURN_IF_ERROR(_state->get_sink_local_state(_sink->operator_id())->open(_state)); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
