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
|
#include <Processors/QueryPlan/ITransformingStep.h>
#include <QueryPipeline/QueryPipelineBuilder.h>
namespace DB
{
ITransformingStep::ITransformingStep(DataStream input_stream, Block output_header, Traits traits, bool collect_processors_)
: transform_traits(std::move(traits.transform_traits))
, collect_processors(collect_processors_)
, data_stream_traits(std::move(traits.data_stream_traits))
{
input_streams.emplace_back(std::move(input_stream));
output_stream = createOutputStream(input_streams.front(), std::move(output_header), data_stream_traits);
}
DataStream ITransformingStep::createOutputStream(
const DataStream & input_stream,
Block output_header,
const DataStreamTraits & stream_traits)
{
DataStream output_stream{.header = std::move(output_header)};
output_stream.has_single_port = stream_traits.returns_single_stream
|| (input_stream.has_single_port && stream_traits.preserves_number_of_streams);
if (stream_traits.preserves_sorting)
{
output_stream.sort_description = input_stream.sort_description;
output_stream.sort_scope = input_stream.sort_scope;
}
return output_stream;
}
QueryPipelineBuilderPtr ITransformingStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings & settings)
{
if (collect_processors)
{
QueryPipelineProcessorsCollector collector(*pipelines.front(), this);
transformPipeline(*pipelines.front(), settings);
processors = collector.detachProcessors();
}
else
transformPipeline(*pipelines.front(), settings);
return std::move(pipelines.front());
}
void ITransformingStep::describePipeline(FormatSettings & settings) const
{
IQueryPlanStep::describePipeline(processors, settings);
}
}
|
