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
|
#include <Processors/QueryPlan/ExpressionStep.h>
#include <Processors/Transforms/ExpressionTransform.h>
#include <QueryPipeline/QueryPipelineBuilder.h>
#include <Processors/Transforms/JoiningTransform.h>
#include <Interpreters/ExpressionActions.h>
#include <IO/Operators.h>
#include <Interpreters/JoinSwitcher.h>
#include <Common/JSONBuilder.h>
namespace DB
{
static ITransformingStep::Traits getTraits(const ActionsDAGPtr & actions, const Block & header, const SortDescription & sort_description)
{
return ITransformingStep::Traits
{
{
.returns_single_stream = false,
.preserves_number_of_streams = true,
.preserves_sorting = actions->isSortingPreserved(header, sort_description),
},
{
.preserves_number_of_rows = !actions->hasArrayJoin(),
}
};
}
ExpressionStep::ExpressionStep(const DataStream & input_stream_, const ActionsDAGPtr & actions_dag_)
: ITransformingStep(
input_stream_,
ExpressionTransform::transformHeader(input_stream_.header, *actions_dag_),
getTraits(actions_dag_, input_stream_.header, input_stream_.sort_description))
, actions_dag(actions_dag_)
{
}
void ExpressionStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings & settings)
{
auto expression = std::make_shared<ExpressionActions>(actions_dag, settings.getActionsSettings());
pipeline.addSimpleTransform([&](const Block & header)
{
return std::make_shared<ExpressionTransform>(header, expression);
});
if (!blocksHaveEqualStructure(pipeline.getHeader(), output_stream->header))
{
auto convert_actions_dag = ActionsDAG::makeConvertingActions(
pipeline.getHeader().getColumnsWithTypeAndName(),
output_stream->header.getColumnsWithTypeAndName(),
ActionsDAG::MatchColumnsMode::Name);
auto convert_actions = std::make_shared<ExpressionActions>(convert_actions_dag, settings.getActionsSettings());
pipeline.addSimpleTransform([&](const Block & header)
{
return std::make_shared<ExpressionTransform>(header, convert_actions);
});
}
}
void ExpressionStep::describeActions(FormatSettings & settings) const
{
String prefix(settings.offset, settings.indent_char);
auto expression = std::make_shared<ExpressionActions>(actions_dag);
expression->describeActions(settings.out, prefix);
}
void ExpressionStep::describeActions(JSONBuilder::JSONMap & map) const
{
auto expression = std::make_shared<ExpressionActions>(actions_dag);
map.add("Expression", expression->toTree());
}
void ExpressionStep::updateOutputStream()
{
output_stream = createOutputStream(
input_streams.front(), ExpressionTransform::transformHeader(input_streams.front().header, *actions_dag), getDataStreamTraits());
if (!getDataStreamTraits().preserves_sorting)
return;
FindAliasForInputName alias_finder(actions_dag);
const auto & input_sort_description = getInputStreams().front().sort_description;
for (size_t i = 0, s = input_sort_description.size(); i < s; ++i)
{
String alias;
const auto & original_column = input_sort_description[i].column_name;
const auto * alias_node = alias_finder.find(original_column);
if (alias_node)
output_stream->sort_description[i].column_name = alias_node->result_name;
}
}
}
|
