thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-24 00:22:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Try fix Pipeline stuck

Browse Source
This commit is contained in:
Nikolai Kochetov 2020-12-27 14:13:17 +03:00
parent b19146065f
commit 30bbba6ed5
2 changed files with 7 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/Processors/DelayedPortsProcessor.cpp
View File

@ -10,6 +10,7 @@ DelayedPortsProcessor::DelayedPortsProcessor(
, num_delayed(delayed_ports.size())
{
port_pairs.resize(num_ports);
output_to_pair.reserve(outputs.size());
for (const auto & delayed : delayed_ports)
port_pairs[delayed].is_delayed = true;
@ -24,6 +25,7 @@ DelayedPortsProcessor::DelayedPortsProcessor(
if (!port_pairs[i].is_delayed || !assert_delayed_ports_empty)
{
port_pairs[i].output_port = &*output_it;
output_to_pair.push_back(i);
++output_it;
}
}
@ -78,8 +80,9 @@ IProcessor::Status DelayedPortsProcessor::prepare(const PortNumbers & updated_in
for (const auto & output_number : updated_outputs)
{
if (!skip_delayed || !port_pairs[output_number].is_delayed)
need_data = processPair(port_pairs[output_number]) || need_data;
auto pair_num = output_to_pair[output_number];
if (!skip_delayed || !port_pairs[pair_num].is_delayed)
need_data = processPair(port_pairs[pair_num]) || need_data;
}
for (const auto & input_number : updated_inputs)

2
src/Processors/DelayedPortsProcessor.h
View File

@ -31,6 +31,8 @@ private:
size_t num_delayed;
size_t num_finished = 0;
std::vector<size_t> output_to_pair;
bool processPair(PortsPair & pair);
};