ClickHouse

1
#include <Processors/QueryPlan/ITransformingStep.h>
2
#include <QueryPipeline/QueryPipelineBuilder.h>
3

4
namespace DB
5
{
6

7
ITransformingStep::ITransformingStep(DataStream input_stream, Block output_header, Traits traits, bool collect_processors_)
8
    : transform_traits(std::move(traits.transform_traits))
9
    , collect_processors(collect_processors_)
10
    , data_stream_traits(std::move(traits.data_stream_traits))
11
{
12
    input_streams.emplace_back(std::move(input_stream));
13
    output_stream = createOutputStream(input_streams.front(), std::move(output_header), data_stream_traits);
14
}
15

16
DataStream ITransformingStep::createOutputStream(
17
    const DataStream & input_stream,
18
    Block output_header,
19
    const DataStreamTraits & stream_traits)
20
{
21
    DataStream output_stream{.header = std::move(output_header)};
22

23
    output_stream.has_single_port = stream_traits.returns_single_stream
24
                                     || (input_stream.has_single_port && stream_traits.preserves_number_of_streams);
25

26
    if (stream_traits.preserves_sorting)
27
    {
28
        output_stream.sort_description = input_stream.sort_description;
29
        output_stream.sort_scope = input_stream.sort_scope;
30
    }
31

32
    return output_stream;
33
}
34

35

36
QueryPipelineBuilderPtr ITransformingStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings & settings)
37
{
38
    if (collect_processors)
39
    {
40
        QueryPipelineProcessorsCollector collector(*pipelines.front(), this);
41
        transformPipeline(*pipelines.front(), settings);
42
        processors = collector.detachProcessors();
43
    }
44
    else
45
        transformPipeline(*pipelines.front(), settings);
46

47
    return std::move(pipelines.front());
48
}
49

50
void ITransformingStep::describePipeline(FormatSettings & settings) const
51
{
52
    IQueryPlanStep::describePipeline(processors, settings);
53
}
54

55
}
56