Coverage Report

Created: 2026-10-12 17:33

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exprs/aggregate/aggregate_function_state_combine.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
20
#include <utility>
21
22
#include "exprs/aggregate/aggregate_function.h"
23
24
namespace doris {
25
const static std::string AGG_COMBINE_SUFFIX = "_combine";
26
27
class AggregateStateCombine final : public IAggregateFunctionHelper<AggregateStateCombine> {
28
public:
29
    AggregateStateCombine(AggregateFunctionPtr function, const DataTypes& argument_types_,
30
                          DataTypePtr return_type)
31
72
            : IAggregateFunctionHelper(argument_types_),
32
72
              _function(std::move(function)),
33
72
              _return_type(std::move(return_type)) {}
34
35
    static AggregateFunctionPtr create(AggregateFunctionPtr function,
36
                                       const DataTypes& argument_types_,
37
72
                                       const DataTypePtr& return_type) {
38
72
        if (function == nullptr) {
39
0
            return nullptr;
40
0
        }
41
72
        return std::make_shared<AggregateStateCombine>(function, argument_types_, return_type);
42
72
    }
43
44
0
    void set_version(const int version_) override {
45
0
        IAggregateFunctionHelper::set_version(version_);
46
0
        _function->set_version(version_);
47
0
    }
48
49
962
    void create(AggregateDataPtr __restrict place) const override { _function->create(place); }
50
51
    void destroy_vec(AggregateDataPtr __restrict place,
52
0
                     const size_t num_rows) const noexcept override {
53
0
        _function->destroy_vec(place, num_rows);
54
0
    }
55
56
0
    String get_name() const override { return _function->get_name() + AGG_COMBINE_SUFFIX; }
57
58
0
    DataTypePtr get_return_type() const override { return _return_type; }
59
60
    void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
61
0
             Arena& arena) const override {
62
0
        _function->add(place, columns, row_num, arena);
63
0
    }
64
65
    void add_batch(size_t batch_size, AggregateDataPtr* places, size_t place_offset,
66
2
                   const IColumn** columns, Arena& arena, bool agg_many) const override {
67
2
        _function->add_batch(batch_size, places, place_offset, columns, arena, agg_many);
68
2
    }
69
70
    void add_batch_selected(size_t batch_size, AggregateDataPtr* places, size_t place_offset,
71
0
                            const IColumn** columns, Arena& arena) const override {
72
0
        _function->add_batch_selected(batch_size, places, place_offset, columns, arena);
73
0
    }
74
75
    void add_batch_single_place(size_t batch_size, AggregateDataPtr place, const IColumn** columns,
76
3
                                Arena& arena) const override {
77
3
        _function->add_batch_single_place(batch_size, place, columns, arena);
78
3
    }
79
80
    void add_batch_range(size_t batch_begin, size_t batch_end, AggregateDataPtr place,
81
0
                         const IColumn** columns, Arena& arena, bool has_null) override {
82
0
        _function->add_batch_range(batch_begin, batch_end, place, columns, arena, has_null);
83
0
    }
84
85
    void add_range_single_place(int64_t partition_start, int64_t partition_end, int64_t frame_start,
86
                                int64_t frame_end, AggregateDataPtr place, const IColumn** columns,
87
                                Arena& arena, UInt8* use_null_result,
88
0
                                UInt8* could_use_previous_result) const override {
89
0
        _function->add_range_single_place(partition_start, partition_end, frame_start, frame_end,
90
0
                                          place, columns, arena, use_null_result,
91
0
                                          could_use_previous_result);
92
0
    }
93
94
0
    void reset(AggregateDataPtr place) const override { _function->reset(place); }
95
96
    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
97
0
               Arena& arena) const override {
98
0
        _function->merge(place, rhs, arena);
99
0
    }
100
101
    void merge_vec(const AggregateDataPtr __restrict* __restrict places, size_t offset,
102
                   ConstAggregateDataPtr __restrict rhs, Arena& arena,
103
0
                   const size_t num_rows) const override {
104
0
        _function->merge_vec(places, offset, rhs, arena, num_rows);
105
0
    }
106
107
    void merge_vec_selected(const AggregateDataPtr __restrict* __restrict places, size_t offset,
108
                            ConstAggregateDataPtr __restrict rhs, Arena& arena,
109
0
                            const size_t num_rows) const override {
110
0
        _function->merge_vec_selected(places, offset, rhs, arena, num_rows);
111
0
    }
112
113
0
    void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
114
0
        _function->serialize(place, buf);
115
0
    }
116
117
    void serialize_vec(const std::vector<AggregateDataPtr>& places, size_t offset,
118
0
                       BufferWritable& buf, const size_t num_rows) const override {
119
0
        _function->serialize_vec(places, offset, buf, num_rows);
120
0
    }
121
122
    void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset,
123
0
                             MutableColumnPtr& dst, const size_t num_rows) const override {
124
0
        _function->serialize_to_column(places, offset, dst, num_rows);
125
0
    }
126
127
    void serialize_without_key_to_column(ConstAggregateDataPtr __restrict place,
128
1
                                         IColumn& to) const override {
129
1
        _function->serialize_without_key_to_column(place, to);
130
1
    }
131
132
    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
133
0
                     Arena& arena) const override {
134
0
        _function->deserialize(place, buf, arena);
135
0
    }
136
137
    void deserialize_vec(AggregateDataPtr places, const ColumnString* column, Arena& arena,
138
0
                         size_t num_rows) const override {
139
0
        _function->deserialize_vec(places, column, arena, num_rows);
140
0
    }
141
142
1.05k
    bool needs_deserialize_and_merge_scratch() const override {
143
1.05k
        return _function->needs_deserialize_and_merge_scratch();
144
1.05k
    }
145
146
    void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t offset,
147
                                   AggregateDataPtr rhs, const IColumn* column, Arena& arena,
148
192
                                   const size_t num_rows) const override {
149
192
        _function->deserialize_and_merge_vec(places, offset, rhs, column, arena, num_rows);
150
192
    }
151
152
    void deserialize_and_merge_vec_selected(const AggregateDataPtr* places, size_t offset,
153
                                            AggregateDataPtr rhs, const IColumn* column,
154
320
                                            Arena& arena, const size_t num_rows) const override {
155
320
        _function->deserialize_and_merge_vec_selected(places, offset, rhs, column, arena, num_rows);
156
320
    }
157
158
    void deserialize_and_merge(AggregateDataPtr __restrict place, AggregateDataPtr __restrict rhs,
159
0
                               BufferReadable& buf, Arena& arena) const override {
160
0
        _function->deserialize_and_merge(place, rhs, buf, arena);
161
0
    }
162
163
    void deserialize_from_column_row(AggregateDataPtr __restrict place, const IColumn& column,
164
180
                                     size_t row, Arena& arena) const override {
165
180
        DCHECK_LT(row, column.size());
166
180
        _function->deserialize_from_column_row(place, column, row, arena);
167
180
    }
168
169
    void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict place,
170
                                                 const IColumn& column, size_t begin, size_t end,
171
1
                                                 Arena& arena) const override {
172
1
        _function->deserialize_and_merge_from_column_range(place, column, begin, end, arena);
173
1
    }
174
175
5
    void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
176
5
        if (to.empty()) {
177
4
            _function->serialize_without_key_to_column(place, to);
178
4
            DORIS_CHECK_EQ(to.size(), 1);
179
4
            return;
180
4
        }
181
182
1
        auto serialized_column = _function->create_serialize_column();
183
1
        _function->serialize_without_key_to_column(place, *serialized_column);
184
1
        DORIS_CHECK_EQ(serialized_column->size(), 1);
185
1
        to.insert_from(*serialized_column, 0);
186
1
    }
187
188
    void insert_result_into_vec(const std::vector<AggregateDataPtr>& places, const size_t offset,
189
2
                                IColumn& to, const size_t num_rows) const override {
190
2
        if (to.empty()) {
191
2
            auto mutable_to = to.assert_mutable();
192
2
            _function->serialize_to_column(places, offset, mutable_to, num_rows);
193
2
            DORIS_CHECK_EQ(to.size(), num_rows);
194
2
            return;
195
2
        }
196
197
0
        auto serialized_column = _function->create_serialize_column();
198
0
        _function->serialize_to_column(places, offset, serialized_column, num_rows);
199
0
        DORIS_CHECK_EQ(serialized_column->size(), num_rows);
200
0
        to.insert_range_from(*serialized_column, 0, num_rows);
201
0
    }
202
203
    void streaming_agg_serialize_to_column(const IColumn** columns, MutableColumnPtr& dst,
204
1
                                           const size_t num_rows, Arena& arena) const override {
205
1
        _function->streaming_agg_serialize_to_column(columns, dst, num_rows, arena);
206
1
    }
207
208
962
    void destroy(AggregateDataPtr __restrict place) const noexcept override {
209
962
        _function->destroy(place);
210
962
    }
211
212
0
    bool is_trivial() const override { return _function->is_trivial(); }
213
214
1.21k
    size_t size_of_data() const override { return _function->size_of_data(); }
215
216
1.66k
    size_t align_of_data() const override { return _function->align_of_data(); }
217
218
0
    void check_input_columns_type(const IColumn** columns) const override {
219
0
        _function->check_input_columns_type(columns);
220
0
    }
221
222
1
    MutableColumnPtr create_serialize_column() const override {
223
1
        return _function->create_serialize_column();
224
1
    }
225
226
1
    DataTypePtr get_serialized_type() const override { return _function->get_serialized_type(); }
227
228
0
    bool supported_incremental_mode() const override {
229
0
        return _function->supported_incremental_mode();
230
0
    }
231
232
    void execute_function_with_incremental(int64_t partition_start, int64_t partition_end,
233
                                           int64_t frame_start, int64_t frame_end,
234
                                           AggregateDataPtr place, const IColumn** columns,
235
                                           Arena& arena, bool previous_is_nul, bool end_is_nul,
236
                                           bool has_null, UInt8* use_null_result,
237
0
                                           UInt8* could_use_previous_result) const override {
238
0
        _function->execute_function_with_incremental(
239
0
                partition_start, partition_end, frame_start, frame_end, place, columns, arena,
240
0
                previous_is_nul, end_is_nul, has_null, use_null_result, could_use_previous_result);
241
0
    }
242
243
0
    void set_query_context(QueryContext* context) override {
244
0
        _function->set_query_context(context);
245
0
    }
246
247
0
    bool is_blockable() const override { return _function->is_blockable(); }
248
249
private:
250
    AggregateFunctionPtr _function;
251
    DataTypePtr _return_type;
252
};
253
254
} // namespace doris