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

Fix Pipeline stuck for INSERT SELECT FINAL

Browse Source 
INSERT SELECT FINAL where SELECT (max_threads>1) has multiple streams
but INSERT has only one (max_insert_threads==0) will add ConcatProcessor
that will stuck the pipeline in this case.
This commit is contained in:
Azat Khuzhin 2020-06-05 01:11:48 +03:00
parent 5deda4c7fd
commit 1887e343ec
2 changed files with 21 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/Processors/ConcatProcessor.cpp
View File

@ -4,6 +4,17 @@
namespace DB
{
ConcatProcessor::ConcatProcessor(const Block & header, size_t num_inputs)
: IProcessor(InputPorts(num_inputs, header), OutputPorts{header}), current_input(inputs.begin())
{
}
void ConcatProcessor::prepareInitializeInputs()
{
for (auto & input : inputs)
input.setNeeded();
}
ConcatProcessor::Status ConcatProcessor::prepare()
{
auto & output = outputs.front();
@ -42,6 +53,12 @@ ConcatProcessor::Status ConcatProcessor::prepare()
auto & input = *current_input;
if (!is_initialized)
{
prepareInitializeInputs();
is_initialized = true;
}
input.setNeeded();
if (!input.hasData())

8
src/Processors/ConcatProcessor.h
View File

@ -16,10 +16,7 @@ namespace DB
class ConcatProcessor : public IProcessor
{
public:
ConcatProcessor(const Block & header, size_t num_inputs)
: IProcessor(InputPorts(num_inputs, header), OutputPorts{header}), current_input(inputs.begin())
{
}
ConcatProcessor(const Block & header, size_t num_inputs);
String getName() const override { return "Concat"; }
@ -29,6 +26,9 @@ public:
private:
InputPorts::iterator current_input;
bool is_initialized = false;
void prepareInitializeInputs();
};
}