forked from ClickHouse/ClickHouse
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathISimpleTransform.cpp
More file actions
120 lines (96 loc) · 2.71 KB
/
Copy pathISimpleTransform.cpp
File metadata and controls
120 lines (96 loc) · 2.71 KB
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
111
112
113
114
115
116
117
118
119
120
#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::ISimpleTransform(SharedHeader input_header_, SharedHeader 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();
onFinish();
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();
onFinish();
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)
{
transform(input_data.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().empty())
/// Support invariant that chunks must have the same number of columns as header.
output_data.chunk = Chunk(getOutputPort().getHeader().cloneEmpty().getColumns(), 0);
}
}