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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
#include <Processors/QueryPlan/JoinStep.h>
#include <QueryPipeline/QueryPipelineBuilder.h>
#include <Processors/Transforms/JoiningTransform.h>
#include <Interpreters/IJoin.h>
#include <Interpreters/TableJoin.h>
#include <IO/Operators.h>
#include <Common/JSONBuilder.h>
#include <Common/typeid_cast.h>
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
}
JoinStep::JoinStep(
const DataStream & left_stream_,
const DataStream & right_stream_,
JoinPtr join_,
size_t max_block_size_,
size_t max_streams_,
bool keep_left_read_in_order_)
: join(std::move(join_)), max_block_size(max_block_size_), max_streams(max_streams_), keep_left_read_in_order(keep_left_read_in_order_)
{
input_streams = {left_stream_, right_stream_};
output_stream = DataStream
{
.header = JoiningTransform::transformHeader(left_stream_.header, join),
};
}
QueryPipelineBuilderPtr JoinStep::updatePipeline(QueryPipelineBuilders pipelines, const BuildQueryPipelineSettings &)
{
if (pipelines.size() != 2)
throw Exception(ErrorCodes::LOGICAL_ERROR, "JoinStep expect two input steps");
if (join->pipelineType() == JoinPipelineType::YShaped)
{
auto joined_pipeline = QueryPipelineBuilder::joinPipelinesYShaped(
std::move(pipelines[0]), std::move(pipelines[1]), join, output_stream->header, max_block_size, &processors);
joined_pipeline->resize(max_streams);
return joined_pipeline;
}
return QueryPipelineBuilder::joinPipelinesRightLeft(
std::move(pipelines[0]),
std::move(pipelines[1]),
join,
output_stream->header,
max_block_size,
max_streams,
keep_left_read_in_order,
&processors);
}
bool JoinStep::allowPushDownToRight() const
{
return join->pipelineType() == JoinPipelineType::YShaped || join->pipelineType() == JoinPipelineType::FillRightFirst;
}
void JoinStep::describePipeline(FormatSettings & settings) const
{
IQueryPlanStep::describePipeline(processors, settings);
}
void JoinStep::describeActions(FormatSettings & settings) const
{
String prefix(settings.offset, ' ');
const auto & table_join = join->getTableJoin();
settings.out << prefix << "Type: " << toString(table_join.kind()) << '\n';
settings.out << prefix << "Strictness: " << toString(table_join.strictness()) << '\n';
settings.out << prefix << "Algorithm: " << join->getName() << '\n';
if (table_join.strictness() == JoinStrictness::Asof)
settings.out << prefix << "ASOF inequality: " << toString(table_join.getAsofInequality()) << '\n';
if (!table_join.getClauses().empty())
settings.out << prefix << "Clauses: " << table_join.formatClauses(table_join.getClauses(), true /*short_format*/) << '\n';
}
void JoinStep::describeActions(JSONBuilder::JSONMap & map) const
{
const auto & table_join = join->getTableJoin();
map.add("Type", toString(table_join.kind()));
map.add("Strictness", toString(table_join.strictness()));
map.add("Algorithm", join->getName());
if (table_join.strictness() == JoinStrictness::Asof)
map.add("ASOF inequality", toString(table_join.getAsofInequality()));
if (!table_join.getClauses().empty())
map.add("Clauses", table_join.formatClauses(table_join.getClauses(), true /*short_format*/));
}
void JoinStep::updateInputStream(const DataStream & new_input_stream_, size_t idx)
{
if (idx == 0)
{
input_streams = {new_input_stream_, input_streams.at(1)};
output_stream = DataStream
{
.header = JoiningTransform::transformHeader(new_input_stream_.header, join),
};
}
else
{
input_streams = {input_streams.at(0), new_input_stream_};
}
}
static ITransformingStep::Traits getStorageJoinTraits()
{
return ITransformingStep::Traits
{
{
.returns_single_stream = false,
.preserves_number_of_streams = true,
.preserves_sorting = false,
},
{
.preserves_number_of_rows = false,
}
};
}
FilledJoinStep::FilledJoinStep(const DataStream & input_stream_, JoinPtr join_, size_t max_block_size_)
: ITransformingStep(
input_stream_,
JoiningTransform::transformHeader(input_stream_.header, join_),
getStorageJoinTraits())
, join(std::move(join_))
, max_block_size(max_block_size_)
{
if (!join->isFilled())
throw Exception(ErrorCodes::LOGICAL_ERROR, "FilledJoinStep expects Join to be filled");
}
void FilledJoinStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
{
bool default_totals = false;
if (!pipeline.hasTotals() && join->getTotals())
{
pipeline.addDefaultTotals();
default_totals = true;
}
auto finish_counter = std::make_shared<JoiningTransform::FinishCounter>(pipeline.getNumStreams());
pipeline.addSimpleTransform([&](const Block & header, QueryPipelineBuilder::StreamType stream_type)
{
bool on_totals = stream_type == QueryPipelineBuilder::StreamType::Totals;
auto counter = on_totals ? nullptr : finish_counter;
return std::make_shared<JoiningTransform>(header, output_stream->header, join, max_block_size, on_totals, default_totals, counter);
});
}
void FilledJoinStep::updateOutputStream()
{
output_stream = createOutputStream(
input_streams.front(), JoiningTransform::transformHeader(input_streams.front().header, join), getDataStreamTraits());
}
}
|
