Coverage Report

Created: 2026-07-09 02:45

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exprs/aggregate/aggregate_function_percentile_reservoir.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 <glog/logging.h>
21
22
#include <memory>
23
24
#include "core/column/column_vector.h"
25
#include "core/data_type/data_type_number.h"
26
#include "core/data_type/primitive_type.h"
27
#include "exprs/aggregate/aggregate_function.h"
28
#include "util/reservoir_sampler.h"
29
30
namespace doris {
31
32
class Arena;
33
class BufferReadable;
34
35
struct QuantileReservoirSampler {
36
0
    void add(const double x, const double input_level) {
37
0
        this->level = input_level;
38
0
        data.insert(x);
39
0
    }
40
41
2
    void add_batch(const double* values, size_t size, const double input_level) {
42
2
        this->level = input_level;
43
2
        data.insert_many(values, size);
44
2
    }
45
46
0
    void merge(const QuantileReservoirSampler& rhs) {
47
0
        level = rhs.level;
48
0
        data.merge(rhs.data);
49
0
    }
50
51
2
    void reset() {
52
2
        level = 0.0;
53
2
        data.clear();
54
2
    }
55
56
0
    void serialize(BufferWritable& buf) const {
57
0
        buf.write_binary(level);
58
0
        data.write(buf);
59
0
    }
60
61
0
    void deserialize(BufferReadable& buf) {
62
0
        buf.read_binary(level);
63
0
        data.read(buf);
64
0
    }
65
66
2
    double get() const {
67
        // The caller is a ConstAggregateDataPtr, but it itself is an AggregateDataPtr.
68
        // To call a non-const method here, a const_cast is required.
69
2
        return const_cast<ReservoirSampler&>(data).quantileInterpolated(this->level);
70
2
    }
71
72
private:
73
    double level = 0.0;
74
    ReservoirSampler data;
75
};
76
77
template <typename Data>
78
class AggregateFunctionPercentileReservoir final
79
        : public IAggregateFunctionDataHelper<Data, AggregateFunctionPercentileReservoir<Data>>,
80
          MultiExpression,
81
          NullableAggregateFunction {
82
public:
83
    AggregateFunctionPercentileReservoir(const DataTypes& argument_types_)
84
1
            : IAggregateFunctionDataHelper<Data, AggregateFunctionPercentileReservoir<Data>>(
85
1
                      argument_types_) {}
86
87
0
    String get_name() const override { return "percentile_reservoir"; }
88
89
0
    DataTypePtr get_return_type() const override { return std::make_shared<DataTypeFloat64>(); }
90
91
    void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
92
0
             Arena&) const override {
93
0
        auto value = assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[0])
94
0
                             .get_data()[row_num];
95
0
        auto level = assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[1])
96
0
                             .get_data()[0];
97
0
        this->data(place).add(value, level);
98
0
    }
99
100
0
    void check_input_columns_type(const IColumn** columns) const override {
101
0
        this->template check_argument_column_type<ColumnFloat64>(columns[0]);
102
0
        this->template check_argument_column_type<ColumnFloat64>(columns[1]);
103
0
    }
104
105
    void add_batch_single_place(size_t batch_size, AggregateDataPtr place, const IColumn** columns,
106
1
                                Arena&) const override {
107
1
        const auto& sources =
108
1
                assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[0]);
109
1
        const auto& levels =
110
1
                assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[1]);
111
1
        this->data(place).add_batch(sources.get_data().data(), batch_size, levels.get_data()[0]);
112
1
    }
113
114
    void add_range_single_place(int64_t partition_start, int64_t partition_end, int64_t frame_start,
115
                                int64_t frame_end, AggregateDataPtr place, const IColumn** columns,
116
                                Arena&, UInt8* use_null_result,
117
2
                                UInt8* could_use_previous_result) const override {
118
2
        frame_start = std::max<int64_t>(frame_start, partition_start);
119
2
        frame_end = std::min<int64_t>(frame_end, partition_end);
120
2
        if (frame_start < frame_end) {
121
1
            const auto& sources =
122
1
                    assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[0]);
123
1
            const auto& levels =
124
1
                    assert_cast<const ColumnFloat64&, TypeCheckOnRelease::DISABLE>(*columns[1]);
125
1
            this->data(place).add_batch(sources.get_data().data() + frame_start,
126
1
                                        frame_end - frame_start, levels.get_data()[0]);
127
1
            *use_null_result = false;
128
1
            *could_use_previous_result = true;
129
1
        } else if (!*could_use_previous_result) {
130
1
            *use_null_result = true;
131
1
        }
132
2
    }
133
134
2
    void reset(AggregateDataPtr place) const override { this->data(place).reset(); }
135
136
    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
137
0
               Arena&) const override {
138
0
        this->data(place).merge(this->data(rhs));
139
0
    }
140
141
0
    void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
142
0
        this->data(place).serialize(buf);
143
0
    }
144
145
    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
146
0
                     Arena&) const override {
147
0
        this->data(place).deserialize(buf);
148
0
    }
149
150
2
    void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
151
2
        assert_cast<ColumnFloat64&, TypeCheckOnRelease::DISABLE>(to).get_data().push_back(
152
2
                this->data(place).get());
153
2
    }
154
};
155
156
} // namespace doris