forked from alibaba/AliSQL
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathwindow_aggregator.cpp
More file actions
88 lines (74 loc) · 3.16 KB
/
window_aggregator.cpp
File metadata and controls
88 lines (74 loc) · 3.16 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
#include "duckdb/function/window/window_aggregator.hpp"
#include "duckdb/function/window/window_collection.hpp"
#include "duckdb/function/window/window_shared_expressions.hpp"
#include "duckdb/planner/expression/bound_window_expression.hpp"
namespace duckdb {
//===--------------------------------------------------------------------===//
// WindowAggregator
//===--------------------------------------------------------------------===//
WindowAggregatorState::WindowAggregatorState() : allocator(Allocator::DefaultAllocator()) {
}
WindowAggregator::WindowAggregator(const BoundWindowExpression &wexpr)
: wexpr(wexpr), aggr(wexpr), result_type(wexpr.return_type), state_size(aggr.function.state_size(aggr.function)),
exclude_mode(wexpr.exclude_clause) {
for (auto &child : wexpr.children) {
arg_types.emplace_back(child->return_type);
}
}
WindowAggregator::WindowAggregator(const BoundWindowExpression &wexpr, WindowSharedExpressions &shared)
: WindowAggregator(wexpr) {
for (auto &child : wexpr.children) {
child_idx.emplace_back(shared.RegisterCollection(child, false));
}
}
WindowAggregator::~WindowAggregator() {
}
unique_ptr<WindowAggregatorState> WindowAggregator::GetGlobalState(ClientContext &context, idx_t group_count,
const ValidityMask &) const {
return make_uniq<WindowAggregatorGlobalState>(context, *this, group_count);
}
void WindowAggregatorLocalState::Sink(WindowAggregatorGlobalState &gastate, DataChunk &sink_chunk,
DataChunk &coll_chunk, idx_t input_idx) {
}
void WindowAggregator::Sink(WindowAggregatorState &gstate, WindowAggregatorState &lstate, DataChunk &sink_chunk,
DataChunk &coll_chunk, idx_t input_idx, optional_ptr<SelectionVector> filter_sel,
idx_t filtered) {
auto &gastate = gstate.Cast<WindowAggregatorGlobalState>();
auto &lastate = lstate.Cast<WindowAggregatorLocalState>();
lastate.Sink(gastate, sink_chunk, coll_chunk, input_idx);
if (filter_sel) {
auto &filter_mask = gastate.filter_mask;
for (idx_t f = 0; f < filtered; ++f) {
filter_mask.SetValid(input_idx + filter_sel->get_index(f));
}
}
}
void WindowAggregatorLocalState::InitSubFrames(SubFrames &frames, const WindowExcludeMode exclude_mode) {
idx_t nframes = 0;
switch (exclude_mode) {
case WindowExcludeMode::NO_OTHER:
nframes = 1;
break;
case WindowExcludeMode::TIES:
nframes = 3;
break;
case WindowExcludeMode::CURRENT_ROW:
case WindowExcludeMode::GROUP:
nframes = 2;
break;
}
frames.resize(nframes, {0, 0});
}
void WindowAggregatorLocalState::Finalize(WindowAggregatorGlobalState &gastate, CollectionPtr collection) {
// Prepare to scan
if (!cursor) {
cursor = make_uniq<WindowCursor>(*collection, gastate.aggregator.child_idx);
}
}
void WindowAggregator::Finalize(WindowAggregatorState &gstate, WindowAggregatorState &lstate, CollectionPtr collection,
const FrameStats &stats) {
auto &gasink = gstate.Cast<WindowAggregatorGlobalState>();
auto &lastate = lstate.Cast<WindowAggregatorLocalState>();
lastate.Finalize(gasink, collection);
}
} // namespace duckdb