blob: ac8f2f8b7ae77150c698859169cee11f00197a75 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
#include <Processors/ISimpleTransform.h>
namespace DB
{
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())
, skip_empty_chunks(skip_empty_chunks_)
{
}
ISimpleTransform::Status ISimpleTransform::prepare()
{
/// Check can output.
if (output.isFinished())
{
input.close();
return Status::Finished;
}
if (!output.canPush())
{
input.setNotNeeded();
return Status::PortFull;
}
/// Output if has data.
if (has_output)
{
output.pushData(std::move(output_data));
has_output = false;
if (!no_more_data_needed)
return Status::PortFull;
}
/// Stop if don't need more data.
if (no_more_data_needed)
{
input.close();
output.finish();
return Status::Finished;
}
/// Check can input.
if (!has_input)
{
if (input.isFinished())
{
output.finish();
return Status::Finished;
}
input.setNeeded();
if (!input.hasData())
return Status::NeedData;
input_data = input.pullData(set_input_not_needed_after_read);
has_input = true;
if (input_data.exception)
/// No more data needed. Exception will be thrown (or swallowed) later.
input.setNotNeeded();
}
/// Now transform.
return Status::Ready;
}
void ISimpleTransform::work()
{
if (input_data.exception)
{
/// Skip transform in case of exception.
output_data = std::move(input_data);
has_input = false;
has_output = true;
return;
}
try
{
transform(input_data.chunk, output_data.chunk);
}
catch (DB::Exception &)
{
output_data.exception = std::current_exception();
has_output = true;
has_input = false;
return;
}
has_input = !needInputData();
if (!skip_empty_chunks || output_data.chunk)
has_output = true;
if (has_output && !output_data.chunk && getOutputPort().getHeader())
/// Support invariant that chunks must have the same number of columns as header.
output_data.chunk = Chunk(getOutputPort().getHeader().cloneEmpty().getColumns(), 0);
}
}
|
