ClickHouse

1
#include <AggregateFunctions/AggregateFunctionFactory.h>
2
#include <AggregateFunctions/FactoryHelpers.h>
3

4
#include <DataTypes/DataTypesNumber.h>
5
#include <Columns/ColumnsNumber.h>
6
#include <IO/ReadHelpers.h>
7
#include <IO/WriteHelpers.h>
8
#include <AggregateFunctions/IAggregateFunction.h>
9
#include <Common/assert_cast.h>
10
#include <Common/transformEndianness.h>
11

12

13
namespace DB
14
{
15
struct Settings;
16

17
namespace ErrorCodes
18
{
19
    extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
20
    extern const int BAD_ARGUMENTS;
21
}
22

23
namespace
24
{
25

26
/** Tracks the leftmost and rightmost (x, y) data points.
27
  */
28
struct AggregateFunctionBoundingRatioData
29
{
30
    struct Point
31
    {
32
        Float64 x;
33
        Float64 y;
34
    };
35

36
    bool empty = true;
37
    Point left;
38
    Point right;
39

40
    void add(Float64 x, Float64 y)
41
    {
42
        Point point{x, y};
43

44
        if (empty)
45
        {
46
            left = point;
47
            right = point;
48
            empty = false;
49
        }
50
        else if (point.x < left.x)
51
        {
52
            left = point;
53
        }
54
        else if (point.x > right.x)
55
        {
56
            right = point;
57
        }
58
    }
59

60
    void merge(const AggregateFunctionBoundingRatioData & other)
61
    {
62
        if (empty)
63
        {
64
            *this = other;
65
        }
66
        else
67
        {
68
            if (other.left.x < left.x)
69
                left = other.left;
70
            if (other.right.x > right.x)
71
                right = other.right;
72
        }
73
    }
74

75
    void serialize(WriteBuffer & buf) const;
76
    void deserialize(ReadBuffer & buf);
77
};
78

79
template <std::endian endian>
80
inline void transformEndianness(AggregateFunctionBoundingRatioData::Point & p)
81
{
82
    DB::transformEndianness<endian>(p.x);
83
    DB::transformEndianness<endian>(p.y);
84
}
85

86
void AggregateFunctionBoundingRatioData::serialize(WriteBuffer & buf) const
87
{
88
    writeBinaryLittleEndian(empty, buf);
89

90
    if (!empty)
91
    {
92
        writeBinaryLittleEndian(left, buf);
93
        writeBinaryLittleEndian(right, buf);
94
    }
95
}
96

97
void AggregateFunctionBoundingRatioData::deserialize(ReadBuffer & buf)
98
{
99
    readBinaryLittleEndian(empty, buf);
100

101
    if (!empty)
102
    {
103
        readBinaryLittleEndian(left, buf);
104
        readBinaryLittleEndian(right, buf);
105
    }
106
}
107

108
inline void writeBinary(const AggregateFunctionBoundingRatioData::Point & p, WriteBuffer & buf)
109
{
110
    writePODBinary(p, buf);
111
}
112

113
inline void readBinary(AggregateFunctionBoundingRatioData::Point & p, ReadBuffer & buf)
114
{
115
    readPODBinary(p, buf);
116
}
117

118

119
class AggregateFunctionBoundingRatio final : public IAggregateFunctionDataHelper<AggregateFunctionBoundingRatioData, AggregateFunctionBoundingRatio>
120
{
121
private:
122
    /** Calculates the slope of a line between leftmost and rightmost data points.
123
      * (y2 - y1) / (x2 - x1)
124
      */
125
    static Float64 NO_SANITIZE_UNDEFINED getBoundingRatio(const AggregateFunctionBoundingRatioData & data)
126
    {
127
        if (data.empty)
128
            return std::numeric_limits<Float64>::quiet_NaN();
129

130
        return (data.right.y - data.left.y) / (data.right.x - data.left.x);
131
    }
132

133
public:
134
    String getName() const override
135
    {
136
        return "boundingRatio";
137
    }
138

139
    explicit AggregateFunctionBoundingRatio(const DataTypes & arguments)
140
        : IAggregateFunctionDataHelper<AggregateFunctionBoundingRatioData, AggregateFunctionBoundingRatio>(arguments, {}, std::make_shared<DataTypeFloat64>())
141
    {
142
        const auto * x_arg = arguments.at(0).get();
143
        const auto * y_arg = arguments.at(1).get();
144

145
        if (!x_arg->isValueRepresentedByNumber() || !y_arg->isValueRepresentedByNumber())
146
            throw Exception(ErrorCodes::BAD_ARGUMENTS,
147
                            "Illegal types of arguments of aggregate function {}, must have number representation.",
148
                            getName());
149
    }
150

151
    bool allocatesMemoryInArena() const override { return false; }
152

153
    void add(AggregateDataPtr __restrict place, const IColumn ** columns, const size_t row_num, Arena *) const override
154
    {
155
        /// NOTE Slightly inefficient.
156
        const auto x = columns[0]->getFloat64(row_num);
157
        const auto y = columns[1]->getFloat64(row_num);
158
        data(place).add(x, y);
159
    }
160

161
    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena *) const override
162
    {
163
        data(place).merge(data(rhs));
164
    }
165

166
    void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
167
    {
168
        data(place).serialize(buf);
169
    }
170

171
    void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
172
    {
173
        data(place).deserialize(buf);
174
    }
175

176
    void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
177
    {
178
        assert_cast<ColumnFloat64 &>(to).getData().push_back(getBoundingRatio(data(place)));
179
    }
180
};
181

182

183
AggregateFunctionPtr createAggregateFunctionRate(const std::string & name, const DataTypes & argument_types, const Array & parameters, const Settings *)
184
{
185
    assertNoParameters(name, parameters);
186
    assertBinary(name, argument_types);
187

188
    if (argument_types.size() < 2)
189
        throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
190
                        "Aggregate function {} requires at least two arguments",
191
                        name);
192

193
    return std::make_shared<AggregateFunctionBoundingRatio>(argument_types);
194
}
195

196
}
197

198
void registerAggregateFunctionRate(AggregateFunctionFactory & factory)
199
{
200
    factory.registerFunction("boundingRatio", createAggregateFunctionRate);
201
}
202

203
}
204