Coverage Report

Created: 2026-07-16 15:06

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exprs/vectorized_agg_fn.h
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
#include <gen_cpp/Types_types.h>
20
21
#include <cstddef>
22
#include <string>
23
#include <vector>
24
25
#include "common/be_mock_util.h"
26
#include "common/status.h"
27
#include "core/block/column_with_type_and_name.h"
28
#include "core/data_type/data_type.h"
29
#include "exec/sort/sort_description.h"
30
#include "exprs/aggregate/aggregate_function.h"
31
#include "exprs/vexpr_fwd.h"
32
#include "runtime/runtime_profile.h"
33
34
namespace doris {
35
36
class RuntimeState;
37
class SlotDescriptor;
38
class ObjectPool;
39
class RowDescriptor;
40
class TExpr;
41
class TExprNode;
42
class TSortInfo;
43
44
class Arena;
45
class Block;
46
class BufferWritable;
47
class IColumn;
48
class QueryContext;
49
50
class AggFnEvaluator {
51
public:
52
    ENABLE_FACTORY_CREATOR(AggFnEvaluator);
53
420k
    MOCK_DEFINE(virtual) ~AggFnEvaluator() = default;
54
55
public:
56
    static Status create(ObjectPool* pool, const TExpr& desc, const TSortInfo& sort_info,
57
                         const bool without_key, const bool is_window_function,
58
                         AggFnEvaluator** result);
59
60
    Status prepare(RuntimeState* state, const RowDescriptor& desc,
61
                   const SlotDescriptor* intermediate_slot_desc,
62
                   const SlotDescriptor* output_slot_desc);
63
64
278k
    void set_timer(RuntimeProfile::Counter* merge_timer, RuntimeProfile::Counter* expr_timer) {
65
278k
        _merge_timer = merge_timer;
66
278k
        _expr_timer = expr_timer;
67
278k
    }
68
69
    Status open(RuntimeState* state);
70
71
    // create/destroy AGG Data
72
    void create(AggregateDataPtr place);
73
    void destroy(AggregateDataPtr place);
74
75
    // agg_function
76
    Status execute_single_add(Block* block, AggregateDataPtr place, Arena& arena);
77
78
    Status execute_batch_add(Block* block, size_t offset, AggregateDataPtr* places, Arena& arena,
79
                             bool agg_many = false);
80
81
    Status execute_batch_add_selected(Block* block, size_t offset, AggregateDataPtr* places,
82
                                      Arena& arena);
83
84
    Status streaming_agg_serialize_to_column(Block* block, MutableColumnPtr& dst,
85
                                             const size_t num_rows, Arena& arena);
86
87
    void add_range_single_place(int64_t partition_start, int64_t partition_end, int64_t frame_start,
88
                                int64_t frame_end, AggregateDataPtr place, const IColumn** columns,
89
                                Arena& arena, UInt8* use_null_result,
90
                                UInt8* could_use_previous_result);
91
92
    void execute_function_with_incremental(int64_t partition_start, int64_t partition_end,
93
                                           int64_t frame_start, int64_t frame_end,
94
                                           AggregateDataPtr place, const IColumn** columns,
95
                                           Arena& arena, bool previous_is_nul, bool end_is_nul,
96
                                           bool has_null, UInt8* use_null_result,
97
                                           UInt8* could_use_previous_result);
98
99
    void insert_result_info(AggregateDataPtr place, IColumn* column);
100
101
    void insert_result_info_vec(const std::vector<AggregateDataPtr>& place, size_t offset,
102
                                IColumn* column, const size_t num_rows);
103
104
    void insert_result_info_range(ConstAggregateDataPtr place, IColumn* column, size_t start,
105
                                  size_t end);
106
107
    void reset(AggregateDataPtr place);
108
109
15.6k
    DataTypePtr& data_type() { return _data_type; }
110
111
9.05k
    String get_name() const { return _function->get_name(); }
112
213k
    DataTypePtr get_return_type() const { return _function->get_return_type(); }
113
152k
    size_t size_of_data() const { return _function->size_of_data(); }
114
202k
    size_t align_of_data() const { return _function->align_of_data(); }
115
9.01k
    bool result_column_could_resize() const { return _function->result_column_could_resize(); }
116
8.99k
    bool supported_incremental_mode() const { return _function->supported_incremental_mode(); }
117
37.6k
    bool is_simple_count() const { return _function->is_simple_count(); }
118
13.9k
    const std::vector<bool>& always_const_argument_idx() const {
119
13.9k
        return _always_const_argument_idx;
120
13.9k
    }
121
9.04k
    const std::vector<ColumnWithTypeAndName>& always_const_arguments() const {
122
9.04k
        return _always_const_arguments;
123
9.04k
    }
124
5.38k
    void merge(AggregateDataPtr place, ConstAggregateDataPtr rhs, Arena& arena) const {
125
5.38k
        _function->merge(place, rhs, arena);
126
5.38k
    }
127
    void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset,
128
30.4k
                             MutableColumnPtr& dst, const size_t num_rows) const {
129
30.4k
        _function->serialize_to_column(places, offset, dst, num_rows);
130
30.4k
    }
131
33.2k
    void serialize_without_key_to_column(ConstAggregateDataPtr place, IColumn& to) const {
132
33.2k
        _function->serialize_without_key_to_column(place, to);
133
33.2k
    }
134
    void deserialize_and_merge_from_column(AggregateDataPtr place, const IColumn& column,
135
34.2k
                                           Arena& arena) const {
136
34.2k
        _function->deserialize_and_merge_from_column(place, column, arena);
137
34.2k
    }
138
    void deserialize_and_merge_from_column_range(AggregateDataPtr place, const IColumn& column,
139
                                                 size_t begin, size_t end, Arena& arena) const {
140
        _function->deserialize_and_merge_from_column_range(place, column, begin, end, arena);
141
    }
142
    void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t offset,
143
                                   AggregateDataPtr rhs, const IColumn* column, Arena& arena,
144
21.3k
                                   const size_t num_rows) const {
145
21.3k
        _function->deserialize_and_merge_vec(places, offset, rhs, column, arena, num_rows);
146
21.3k
    }
147
    void deserialize_and_merge_vec_selected(const AggregateDataPtr* places, size_t offset,
148
                                            AggregateDataPtr rhs, const IColumn* column,
149
25
                                            Arena& arena, const size_t num_rows) const {
150
25
        _function->deserialize_and_merge_vec_selected(places, offset, rhs, column, arena, num_rows);
151
25
    }
152
67.0k
    MutableColumnPtr create_serialize_column() const {
153
67.0k
        return _function->create_serialize_column();
154
67.0k
    }
155
66.9k
    DataTypePtr get_serialized_type() const { return _function->get_serialized_type(); }
156
    void set_query_context(QueryContext* context) { _function->set_query_context(context); }
157
158
    static std::string debug_string(const std::vector<AggFnEvaluator*>& exprs);
159
    std::string debug_string() const;
160
183k
    bool is_merge() const { return _is_merge; }
161
57.3k
    const VExprContextSPtrs& input_exprs_ctxs() const { return _input_exprs_ctxs; }
162
163
    static Status check_agg_fn_output(uint32_t key_size, const std::vector<AggFnEvaluator*>& agg_fn,
164
                                      const RowDescriptor& output_row_desc);
165
166
131k
    void set_version(const int version) { _function->set_version(version); }
167
168
    AggFnEvaluator* clone(RuntimeState* state, ObjectPool* pool);
169
170
    bool is_blockable() const;
171
172
private:
173
    const TFunction _fn;
174
175
    const bool _is_merge;
176
    // We need this flag to distinguish between the two types of aggregation functions:
177
    // 1. executed without group by key (agg function used with window function is also regarded as this type)
178
    // 2. executed with group by key
179
    const bool _without_key;
180
181
    const bool _is_window_function;
182
183
    AggFnEvaluator(const TExprNode& desc, const bool without_key, const bool is_window_function);
184
    AggFnEvaluator(AggFnEvaluator& evaluator, RuntimeState* state);
185
186
#ifdef BE_TEST
187
    AggFnEvaluator(bool is_merge, bool without_key, const bool is_window_function)
188
            : _is_merge(is_merge),
189
              _without_key(without_key),
190
              _is_window_function(is_window_function) {};
191
#endif
192
    Status _init_always_const_arguments();
193
    Status _calc_argument_columns(Block* block);
194
195
    DataTypes _argument_types_with_sort;
196
    DataTypes _real_argument_types;
197
198
    const SlotDescriptor* _intermediate_slot_desc = nullptr;
199
    const SlotDescriptor* _output_slot_desc = nullptr;
200
201
    RuntimeProfile::Counter* _merge_timer = nullptr;
202
    RuntimeProfile::Counter* _expr_timer = nullptr;
203
204
    // input context
205
    VExprContextSPtrs _input_exprs_ctxs;
206
207
    SortDescription _sort_description;
208
209
    DataTypePtr _data_type;
210
211
    AggregateFunctionPtr _function;
212
213
    std::string _expr_name;
214
215
    std::vector<const IColumn*> _agg_columns;
216
    std::vector<ColumnWithTypeAndName> _always_const_arguments;
217
    // Bool map of argument positions that must be constant by aggregate semantics.
218
    // FE checks these positions before execution.
219
    std::vector<bool> _always_const_argument_idx;
220
};
221
222
} // namespace doris