ClickHouse

1
#include <Planner/ActionsChain.h>
2

3
#include <boost/algorithm/string/split.hpp>
4
#include <boost/algorithm/string/join.hpp>
5

6
#include <IO/WriteBuffer.h>
7
#include <IO/WriteHelpers.h>
8
#include <IO/Operators.h>
9
#include <IO/WriteBufferFromString.h>
10

11
namespace DB
12
{
13

14
ActionsChainStep::ActionsChainStep(ActionsDAGPtr actions_,
15
    bool use_actions_nodes_as_output_columns_,
16
    ColumnsWithTypeAndName additional_output_columns_)
17
    : actions(std::move(actions_))
18
    , use_actions_nodes_as_output_columns(use_actions_nodes_as_output_columns_)
19
    , additional_output_columns(std::move(additional_output_columns_))
20
{
21
    initialize();
22
}
23

24
void ActionsChainStep::finalizeInputAndOutputColumns(const NameSet & child_input_columns)
25
{
26
    child_required_output_columns_names.clear();
27

28
    auto child_input_columns_copy = child_input_columns;
29

30
    std::unordered_set<std::string_view> output_nodes_names;
31
    output_nodes_names.reserve(actions->getOutputs().size());
32

33
    for (auto & output_node : actions->getOutputs())
34
        output_nodes_names.insert(output_node->result_name);
35

36
    for (const auto & node : actions->getNodes())
37
    {
38
        auto it = child_input_columns_copy.find(node.result_name);
39
        if (it == child_input_columns_copy.end())
40
            continue;
41

42
        child_input_columns_copy.erase(it);
43
        child_required_output_columns_names.insert(node.result_name);
44

45
        if (output_nodes_names.contains(node.result_name))
46
            continue;
47

48
        actions->getOutputs().push_back(&node);
49
        output_nodes_names.insert(node.result_name);
50
    }
51

52
    actions->removeUnusedActions();
53
    /// TODO: Analyzer fix ActionsDAG input and constant nodes with same name
54
    actions->projectInput();
55
    initialize();
56
}
57

58
void ActionsChainStep::dump(WriteBuffer & buffer) const
59
{
60
    buffer << "DAG" << '\n';
61
    buffer << actions->dumpDAG();
62

63
    if (!available_output_columns.empty())
64
    {
65
        buffer << "Available output columns " << available_output_columns.size() << '\n';
66
        for (const auto & column : available_output_columns)
67
            buffer << "Name " << column.name << " type " << column.type->getName() << '\n';
68
    }
69

70
    if (!child_required_output_columns_names.empty())
71
    {
72
        buffer << "Child required output columns " << boost::join(child_required_output_columns_names, ", ");
73
        buffer << '\n';
74
    }
75
}
76

77
String ActionsChainStep::dump() const
78
{
79
    WriteBufferFromOwnString buffer;
80
    dump(buffer);
81

82
    return buffer.str();
83
}
84

85
void ActionsChainStep::initialize()
86
{
87
    auto required_columns_names = actions->getRequiredColumnsNames();
88
    input_columns_names = NameSet(required_columns_names.begin(), required_columns_names.end());
89

90
    available_output_columns.clear();
91

92
    if (use_actions_nodes_as_output_columns)
93
    {
94
        std::unordered_set<std::string_view> available_output_columns_names;
95

96
        for (const auto & node : actions->getNodes())
97
        {
98
            if (available_output_columns_names.contains(node.result_name))
99
                continue;
100

101
            available_output_columns.emplace_back(node.column, node.result_type, node.result_name);
102
            available_output_columns_names.insert(node.result_name);
103
        }
104
    }
105

106
    available_output_columns.insert(available_output_columns.end(), additional_output_columns.begin(), additional_output_columns.end());
107
}
108

109
void ActionsChain::finalize()
110
{
111
    if (steps.empty())
112
        return;
113

114
    /// For last chain step there are no columns required in child nodes
115
    NameSet empty_child_input_columns;
116
    steps.back().get()->finalizeInputAndOutputColumns(empty_child_input_columns);
117

118
    Int64 steps_last_index = steps.size() - 1;
119
    for (Int64 i = steps_last_index; i >= 1; --i)
120
    {
121
        auto & current_step = steps[i];
122
        auto & previous_step = steps[i - 1];
123

124
        previous_step->finalizeInputAndOutputColumns(current_step->getInputColumnNames());
125
    }
126
}
127

128
void ActionsChain::dump(WriteBuffer & buffer) const
129
{
130
    size_t steps_size = steps.size();
131

132
    for (size_t i = 0; i < steps_size; ++i)
133
    {
134
        const auto & step = steps[i];
135
        buffer << "Step " << i << '\n';
136
        step->dump(buffer);
137

138
        buffer << '\n';
139
    }
140
}
141

142
String ActionsChain::dump() const
143
{
144
    WriteBufferFromOwnString buffer;
145
    dump(buffer);
146

147
    return buffer.str();
148
}
149

150
}
151