ClickHouse

1
#include <Processors/QueryPlan/ExpressionStep.h>
2
#include <Processors/Transforms/ExpressionTransform.h>
3
#include <QueryPipeline/QueryPipelineBuilder.h>
4
#include <Processors/Transforms/JoiningTransform.h>
5
#include <Interpreters/ExpressionActions.h>
6
#include <IO/Operators.h>
7
#include <Interpreters/JoinSwitcher.h>
8
#include <Common/JSONBuilder.h>
9

10
namespace DB
11
{
12

13
static ITransformingStep::Traits getTraits(const ActionsDAGPtr & actions, const Block & header, const SortDescription & sort_description)
14
{
15
    return ITransformingStep::Traits
16
    {
17
        {
18
            .returns_single_stream = false,
19
            .preserves_number_of_streams = true,
20
            .preserves_sorting = actions->isSortingPreserved(header, sort_description),
21
        },
22
        {
23
            .preserves_number_of_rows = !actions->hasArrayJoin(),
24
        }
25
    };
26
}
27

28
ExpressionStep::ExpressionStep(const DataStream & input_stream_, const ActionsDAGPtr & actions_dag_)
29
    : ITransformingStep(
30
        input_stream_,
31
        ExpressionTransform::transformHeader(input_stream_.header, *actions_dag_),
32
        getTraits(actions_dag_, input_stream_.header, input_stream_.sort_description))
33
    , actions_dag(actions_dag_)
34
{
35
}
36

37
void ExpressionStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings)
38
{
39
    auto expression = std::make_shared<ExpressionActions>(actions_dag, settings.getActionsSettings());
40

41
    pipeline.addSimpleTransform([&](const Block & header)
42
    {
43
        return std::make_shared<ExpressionTransform>(header, expression);
44
    });
45

46
    if (!blocksHaveEqualStructure(pipeline.getHeader(), output_stream->header))
47
    {
48
        auto convert_actions_dag = ActionsDAG::makeConvertingActions(
49
                pipeline.getHeader().getColumnsWithTypeAndName(),
50
                output_stream->header.getColumnsWithTypeAndName(),
51
                ActionsDAG::MatchColumnsMode::Name);
52
        auto convert_actions = std::make_shared<ExpressionActions>(convert_actions_dag, settings.getActionsSettings());
53

54
        pipeline.addSimpleTransform([&](const Block & header)
55
        {
56
            return std::make_shared<ExpressionTransform>(header, convert_actions);
57
        });
58
    }
59
}
60

61
void ExpressionStep::describeActions(FormatSettings & settings) const
62
{
63
    String prefix(settings.offset, settings.indent_char);
64
    auto expression = std::make_shared<ExpressionActions>(actions_dag);
65
    expression->describeActions(settings.out, prefix);
66
}
67

68
void ExpressionStep::describeActions(JSONBuilder::JSONMap & map) const
69
{
70
    auto expression = std::make_shared<ExpressionActions>(actions_dag);
71
    map.add("Expression", expression->toTree());
72
}
73

74
void ExpressionStep::updateOutputStream()
75
{
76
    output_stream = createOutputStream(
77
        input_streams.front(), ExpressionTransform::transformHeader(input_streams.front().header, *actions_dag), getDataStreamTraits());
78

79
    if (!getDataStreamTraits().preserves_sorting)
80
        return;
81

82
    FindAliasForInputName alias_finder(actions_dag);
83
    const auto & input_sort_description = getInputStreams().front().sort_description;
84
    for (size_t i = 0, s = input_sort_description.size(); i < s; ++i)
85
    {
86
        String alias;
87
        const auto & original_column = input_sort_description[i].column_name;
88
        const auto * alias_node = alias_finder.find(original_column);
89
        if (alias_node)
90
            output_stream->sort_description[i].column_name = alias_node->result_name;
91
    }
92
}
93

94
}
95