blob: 0af6c90dac9b539d7ad34176728e666f5d747342 (
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
|
#include <Processors/Transforms/CopyTransform.h>
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
}
CopyTransform::CopyTransform(const Block & header, size_t num_outputs)
: IProcessor(InputPorts(1, header), OutputPorts(num_outputs, header))
{
if (num_outputs <= 1)
throw Exception(ErrorCodes::LOGICAL_ERROR, "CopyTransform expects more than 1 outputs, got {}", num_outputs);
}
IProcessor::Status CopyTransform::prepare()
{
Status status = Status::Ready;
while (status == Status::Ready)
{
status = !has_data ? prepareConsume()
: prepareGenerate();
}
return status;
}
IProcessor::Status CopyTransform::prepareConsume()
{
auto & input = getInputPort();
/// Check all outputs are finished or ready to get data.
bool all_finished = true;
for (auto & output : outputs)
{
if (output.isFinished())
continue;
all_finished = false;
}
if (all_finished)
{
input.close();
return Status::Finished;
}
/// Try get chunk from input.
if (input.isFinished())
{
for (auto & output : outputs)
output.finish();
return Status::Finished;
}
input.setNeeded();
if (!input.hasData())
return Status::NeedData;
chunk = input.pull();
has_data = true;
was_output_processed.assign(outputs.size(), false);
return Status::Ready;
}
IProcessor::Status CopyTransform::prepareGenerate()
{
bool all_outputs_processed = true;
size_t chunk_number = 0;
for (auto & output : outputs)
{
auto & was_processed = was_output_processed[chunk_number];
++chunk_number;
if (was_processed)
continue;
if (output.isFinished())
continue;
if (!output.canPush())
{
all_outputs_processed = false;
continue;
}
output.push(chunk.clone());
was_processed = true;
}
if (all_outputs_processed)
{
has_data = false;
return Status::Ready;
}
return Status::PortFull;
}
}
|
