thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-26 09:32:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add default status to not-prepared processors.

Browse Source
This commit is contained in:
Nikolai Kochetov 2024-08-15 15:07:07 +00:00
parent d1b3bf3103
commit 4548957f5a
5 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/Processors/Executors/ExecutingGraph.cpp
View File

@ -279,7 +279,7 @@ bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue
try
{
auto & processor = *node.processor;
IProcessor::Status last_status = node.last_processor_status;
const auto last_status = node.last_processor_status;
IProcessor::Status status = processor.prepare(node.updated_input_ports, node.updated_output_ports);
node.last_processor_status = status;
@ -319,7 +319,7 @@ bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue
node.updated_input_ports.clear();
node.updated_output_ports.clear();
switch (node.last_processor_status)
switch (*node.last_processor_status)
{
case IProcessor::Status::NeedData:
case IProcessor::Status::PortFull:

2
src/Processors/Executors/ExecutingGraph.h
View File

@ -92,7 +92,7 @@ public:
std::exception_ptr exception;
/// Last state for profiling.
IProcessor::Status last_processor_status = IProcessor::Status::NeedData;
std::optional<IProcessor::Status> last_processor_status;
/// Ports which have changed their state since last processor->prepare() call.
/// They changed when neighbour processors interact with connected ports.

2
src/Processors/Executors/PipelineExecutor.cpp
View File

@ -432,7 +432,7 @@ String PipelineExecutor::dumpPipeline() const
}
}
std::vector<IProcessor::Status> statuses;
std::vector<std::optional<IProcessor::Status>> statuses;
std::vector<IProcessor *> proc_list;
statuses.reserve(graph->nodes.size());
proc_list.reserve(graph->nodes.size());

7
src/Processors/IProcessor.cpp
View File

@ -55,9 +55,12 @@ void IProcessor::dump() const
}
std::string IProcessor::statusToName(Status status)
std::string IProcessor::statusToName(std::optional<Status> status)
{
switch (status)
if (status == std::nullopt)
return "NotStarted";
switch (*status)
{
case Status::NeedData:
return "NeedData";

2
src/Processors/IProcessor.h
View File

@ -162,7 +162,7 @@ public:
ExpandPipeline,
};
static std::string statusToName(Status status);
static std::string statusToName(std::optional<Status> status);
/** Method 'prepare' is responsible for all cheap ("instantaneous": O(1) of data volume, no wait) calculations.
*