forked from ClickHouse/ClickHouse
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathFillingStep.cpp
More file actions
89 lines (79 loc) · 2.96 KB
/
Copy pathFillingStep.cpp
File metadata and controls
89 lines (79 loc) · 2.96 KB
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
#include <Processors/QueryPlan/FillingStep.h>
#include <Processors/QueryPlan/QueryPlanFormat.h>
#include <Processors/Transforms/FillingTransform.h>
#include <QueryPipeline/QueryPipelineBuilder.h>
#include <IO/Operators.h>
#include <Interpreters/ExpressionActions.h>
#include <Common/JSONBuilder.h>
namespace DB
{
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
}
static ITransformingStep::Traits getTraits()
{
return ITransformingStep::Traits
{
{
.returns_single_stream = true,
.preserves_number_of_streams = true,
.preserves_sorting = true,
},
{
.preserves_number_of_rows = false,
}
};
}
FillingStep::FillingStep(
SharedHeader input_header_,
SortDescription sort_description_,
SortDescription fill_description_,
InterpolateDescriptionPtr interpolate_description_,
bool use_with_fill_by_sorting_prefix_)
: ITransformingStep(input_header_, std::make_shared<const Block>(FillingTransform::transformHeader(*input_header_, sort_description_)), getTraits())
, sort_description(std::move(sort_description_))
, fill_description(std::move(fill_description_))
, interpolate_description(interpolate_description_)
, use_with_fill_by_sorting_prefix(use_with_fill_by_sorting_prefix_)
{
}
void FillingStep::transformPipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &)
{
if (pipeline.getNumStreams() != 1)
throw Exception(ErrorCodes::LOGICAL_ERROR, "FillingStep expects single input");
pipeline.addSimpleTransform([&](const SharedHeader & header, QueryPipelineBuilder::StreamType stream_type) -> ProcessorPtr
{
if (stream_type == QueryPipelineBuilder::StreamType::Totals)
return std::make_shared<FillingNoopTransform>(header, fill_description);
return std::make_shared<FillingTransform>(
header, sort_description, fill_description, std::move(interpolate_description), use_with_fill_by_sorting_prefix);
});
}
void FillingStep::describeActions(FormatSettings & settings) const
{
const String & prefix = settings.detail_prefix;
settings.out << prefix;
dumpSortDescription(sort_description, settings);
settings.out << '\n';
if (interpolate_description)
{
auto expression = std::make_shared<ExpressionActions>(interpolate_description->actions.clone());
if (!settings.compact)
expression->describeActions(settings.out, prefix);
}
}
void FillingStep::describeActions(JSONBuilder::JSONMap & map) const
{
map.add("Sort Description", explainSortDescription(sort_description));
if (interpolate_description)
{
auto expression = std::make_shared<ExpressionActions>(interpolate_description->actions.clone());
map.add("Expression", expression->toTree());
}
}
void FillingStep::updateOutputHeader()
{
output_header = std::make_shared<const Block>(FillingTransform::transformHeader(*input_headers.front(), sort_description));
}
}