ClickHouse/dbms/src/Processors/ISimpleTransform.cpp

110 lines
2.3 KiB
C++
Raw Normal View History

#include <Processors/ISimpleTransform.h>
namespace DB
{
2019-08-03 11:02:40 +00:00
ISimpleTransform::ISimpleTransform(Block input_header_, Block output_header_, bool skip_empty_chunks_)
: IProcessor({std::move(input_header_)}, {std::move(output_header_)})
, input(inputs.front())
, output(outputs.front())
2019-08-03 11:02:40 +00:00
, skip_empty_chunks(skip_empty_chunks_)
{
}
ISimpleTransform::Status ISimpleTransform::prepare()
{
2019-02-07 18:51:53 +00:00
/// Check can output.
2019-02-07 18:51:53 +00:00
if (output.isFinished())
{
2019-02-07 18:51:53 +00:00
input.close();
return Status::Finished;
}
2019-02-07 18:51:53 +00:00
if (!output.canPush())
{
2019-02-07 18:51:53 +00:00
input.setNotNeeded();
return Status::PortFull;
}
/// Output if has data.
if (transformed)
{
output.pushData(std::move(current_data));
transformed = false;
}
2019-03-25 16:58:59 +00:00
/// Stop if don't need more data.
if (no_more_data_needed)
{
input.close();
output.finish();
return Status::Finished;
}
2019-02-07 18:51:53 +00:00
/// Check can input.
if (!has_input)
{
2019-02-07 18:51:53 +00:00
if (input.isFinished())
{
output.finish();
return Status::Finished;
}
input.setNeeded();
if (!input.hasData())
return Status::NeedData;
current_data = input.pullData();
2019-02-07 18:51:53 +00:00
has_input = true;
2019-06-19 18:30:02 +00:00
if (current_data.exception)
{
/// Skip transform in case of exception.
has_input = false;
transformed = true;
/// No more data needed. Exception will be thrown (or swallowed) later.
input.setNotNeeded();
}
if (set_input_not_needed_after_read)
input.setNotNeeded();
}
2019-02-07 18:51:53 +00:00
/// Now transform.
return Status::Ready;
}
void ISimpleTransform::work()
{
2019-06-19 18:30:02 +00:00
if (current_data.exception)
return;
try
{
2019-06-19 18:30:02 +00:00
transform(current_data.chunk);
}
catch (DB::Exception &)
{
2019-06-19 18:30:02 +00:00
current_data.exception = std::current_exception();
transformed = true;
has_input = false;
return;
}
2019-04-08 11:28:38 +00:00
has_input = false;
2019-06-19 18:30:02 +00:00
if (!skip_empty_chunks || current_data.chunk)
transformed = true;
2019-06-19 18:30:02 +00:00
if (transformed && !current_data.chunk)
/// Support invariant that chunks must have the same number of columns as header.
2019-06-19 18:30:02 +00:00
current_data.chunk = Chunk(getOutputPort().getHeader().cloneEmpty().getColumns(), 0);
}
}