Coverage Report

Created: 2026-06-22 12:33

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exprs/aggregate/aggregate_function_retention.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
// This file is copied from
19
// https://github.com/ClickHouse/ClickHouse/blob/master/AggregateFunctionRetention.h
20
// and modified by Doris
21
22
#pragma once
23
24
#include <stddef.h>
25
#include <stdint.h>
26
27
#include <boost/iterator/iterator_facade.hpp>
28
#include <memory>
29
30
#include "common/exception.h"
31
#include "common/status.h"
32
#include "core/assert_cast.h"
33
#include "core/column/column.h"
34
#include "core/column/column_array.h"
35
#include "core/column/column_nullable.h"
36
#include "core/column/column_vector.h"
37
#include "core/data_type/data_type_array.h"
38
#include "core/data_type/data_type_nullable.h"
39
#include "core/data_type/data_type_number.h"
40
#include "core/types.h"
41
#include "exprs/aggregate/aggregate_function.h"
42
#include "util/var_int.h"
43
44
namespace doris {
45
class Arena;
46
class BufferReadable;
47
class BufferWritable;
48
} // namespace doris
49
50
namespace doris {
51
struct RetentionState {
52
    static constexpr size_t MAX_EVENTS = 32;
53
    uint8_t events[MAX_EVENTS] = {0};
54
55
353
    RetentionState() = default;
56
57
40
    void reset() {
58
1.32k
        for (int64_t i = 0; i < MAX_EVENTS; i++) {
59
1.28k
            events[i] = 0;
60
1.28k
        }
61
40
    }
62
63
83
    void set(int event) { events[event] = 1; }
64
65
118
    void merge(const RetentionState& other) {
66
3.89k
        for (int64_t i = 0; i < MAX_EVENTS; i++) {
67
3.77k
            events[i] |= other.events[i];
68
3.77k
        }
69
118
    }
70
71
156
    void write(BufferWritable& out) const {
72
156
        int64_t serialized_events = 0;
73
5.14k
        for (int64_t i = 0; i < MAX_EVENTS; i++) {
74
4.99k
            serialized_events |= events[i];
75
4.99k
            serialized_events <<= 1;
76
4.99k
        }
77
156
        write_var_int(serialized_events, out);
78
156
    }
79
80
117
    void read(BufferReadable& in) {
81
117
        int64_t serialized_events = 0;
82
117
        uint64_t u_serialized_events = 0;
83
117
        read_var_int(serialized_events, in);
84
117
        u_serialized_events = serialized_events;
85
86
117
        u_serialized_events >>= 1;
87
3.86k
        for (int64_t i = MAX_EVENTS - 1; i >= 0; i--) {
88
3.74k
            events[i] = (uint8_t)(1 & u_serialized_events);
89
3.74k
            u_serialized_events >>= 1;
90
3.74k
        }
91
117
    }
92
93
70
    void insert_result_into(IColumn& to, size_t events_size, const uint8_t* arg_events) const {
94
70
        auto& data_to = assert_cast<ColumnUInt8&>(to).get_data();
95
96
70
        ColumnArray::Offset64 current_offset = data_to.size();
97
70
        data_to.resize(current_offset + events_size);
98
99
70
        bool first_flag = arg_events[0];
100
70
        data_to[current_offset] = first_flag;
101
70
        ++current_offset;
102
103
219
        for (size_t i = 1; i < events_size; ++i) {
104
149
            data_to[current_offset] = (first_flag && arg_events[i]);
105
149
            ++current_offset;
106
149
        }
107
70
    }
108
};
109
110
class AggregateFunctionRetention final
111
        : public IAggregateFunctionDataHelper<RetentionState, AggregateFunctionRetention>,
112
          VarargsExpression,
113
          NullableAggregateFunction {
114
public:
115
    AggregateFunctionRetention(const DataTypes& argument_types_)
116
442
            : IAggregateFunctionDataHelper<RetentionState, AggregateFunctionRetention>(
117
442
                      argument_types_) {
118
        // RetentionState only has room for MAX_EVENTS(32) events (fixed-size events[] array,
119
        // plus an int64 serialized bitmap). More params would overflow events[] in add()/
120
        // insert_result_into() and corrupt the heap, so reject it at construction time.
121
442
        if (argument_types_.size() > RetentionState::MAX_EVENTS) {
122
1
            throw Exception(ErrorCode::INVALID_ARGUMENT,
123
1
                            "retention function can accept at most {} params, but got {}",
124
1
                            RetentionState::MAX_EVENTS, argument_types_.size());
125
1
        }
126
442
    }
127
128
31
    String get_name() const override { return "retention"; }
129
130
61
    DataTypePtr get_return_type() const override {
131
61
        return std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeUInt8>()));
132
61
    }
133
134
40
    void reset(AggregateDataPtr __restrict place) const override { this->data(place).reset(); }
135
    void add(AggregateDataPtr __restrict place, const IColumn** columns, const ssize_t row_num,
136
231
             Arena&) const override {
137
921
        for (int i = 0; i < get_argument_types().size(); i++) {
138
690
            auto event = assert_cast<const ColumnUInt8*, TypeCheckOnRelease::DISABLE>(columns[i])
139
690
                                 ->get_data()[row_num];
140
690
            if (event) {
141
83
                this->data(place).set(i);
142
83
            }
143
690
        }
144
231
    }
145
146
    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
147
118
               Arena&) const override {
148
118
        this->data(place).merge(this->data(rhs));
149
118
    }
150
151
156
    void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
152
156
        this->data(place).write(buf);
153
156
    }
154
155
    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
156
117
                     Arena&) const override {
157
117
        this->data(place).read(buf);
158
117
    }
159
160
70
    void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
161
70
        auto& to_arr = assert_cast<ColumnArray&>(to);
162
70
        auto& to_nested_col = to_arr.get_data();
163
70
        if (is_column_nullable(to_nested_col)) {
164
70
            auto col_null = reinterpret_cast<ColumnNullable*>(&to_nested_col);
165
70
            this->data(place).insert_result_into(col_null->get_nested_column(),
166
70
                                                 get_argument_types().size(),
167
70
                                                 this->data(place).events);
168
70
            col_null->get_null_map_data().resize_fill(col_null->get_nested_column().size(), 0);
169
70
        } else {
170
0
            this->data(place).insert_result_into(to_nested_col, get_argument_types().size(),
171
0
                                                 this->data(place).events);
172
0
        }
173
70
        to_arr.get_offsets().push_back(to_nested_col.size());
174
70
    }
175
};
176
} // namespace doris