1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
#pragma once
#include <Columns/ColumnArray.h>
#include <Columns/ColumnsNumber.h>
#include <Common/PODArray.h>
#include <Common/PODArray_fwd.h>
#include <DataTypes/DataTypeArray.h>
#include <AggregateFunctions/IAggregateFunction.h>
#include <AggregateFunctions/Moments.h>
#include <DataTypes/DataTypesNumber.h>
namespace DB
{
struct Settings;
enum class StatisticsMatrixFunctionKind
{
covarPopMatrix,
covarSampMatrix,
corrMatrix
};
template <StatisticsMatrixFunctionKind _kind>
struct AggregateFunctionVarianceMatrixData
{
using DataType = std::conditional_t<_kind == StatisticsMatrixFunctionKind::corrMatrix, CorrMoments<Float64>, CovarMoments<Float64>>;
AggregateFunctionVarianceMatrixData() = default;
explicit AggregateFunctionVarianceMatrixData(const size_t _num_args)
: num_args(_num_args)
{
data_matrix.resize_fill(num_args * (num_args + 1) / 2, DataType());
}
void add(const IColumn ** column, const size_t row_num)
{
for (size_t i = 0; i < num_args; ++i)
for (size_t j = 0; j <= i; ++j)
data_matrix[i * (i + 1) / 2 + j].add(column[i]->getFloat64(row_num), column[j]->getFloat64(row_num));
}
void merge(const AggregateFunctionVarianceMatrixData & other)
{
for (size_t i = 0; i < num_args; ++i)
for (size_t j = 0; j <= i; ++j)
data_matrix[i * (i + 1) / 2 + j].merge(other.data_matrix[i * (i + 1) / 2 + j]);
}
void serialize(WriteBuffer & buf) const
{
for (size_t i = 0; i < num_args; ++i)
for (size_t j = 0; j <= i; ++j)
data_matrix[i * (i + 1) / 2 + j].write(buf);
}
void deserialize(ReadBuffer & buf)
{
for (size_t i = 0; i < num_args; ++i)
for (size_t j = 0; j <= i; ++j)
data_matrix[i * (i + 1) / 2 + j].read(buf);
}
void insertResultInto(IColumn & to) const
{
auto & data_to = assert_cast<ColumnFloat64 &>(assert_cast<ColumnArray &>(assert_cast<ColumnArray &>(to).getData()).getData()).getData();
auto & root_offsets_to = assert_cast<ColumnArray &>(to).getOffsets();
auto & nested_offsets_to = assert_cast<ColumnArray &>(assert_cast<ColumnArray &>(to).getData()).getOffsets();
for (size_t i = 0; i < num_args; ++i)
{
for (size_t j = 0; j < num_args; ++j)
{
auto & data = i < j ? data_matrix[j * (j + 1) / 2 + i] : data_matrix[i * (i + 1) / 2 + j];
if constexpr (kind == StatisticsMatrixFunctionKind::covarPopMatrix)
data_to.push_back(data.getPopulation());
if constexpr (kind == StatisticsMatrixFunctionKind::covarSampMatrix)
data_to.push_back(data.getSample());
if constexpr (kind == StatisticsMatrixFunctionKind::corrMatrix)
data_to.push_back(data.get());
}
nested_offsets_to.push_back(nested_offsets_to.back() + num_args);
}
root_offsets_to.push_back(root_offsets_to.back() + num_args);
}
static constexpr StatisticsMatrixFunctionKind kind = _kind;
PaddedPODArray<DataType> data_matrix;
size_t num_args;
};
template <typename Data>
class AggregateFunctionVarianceMatrix final
: public IAggregateFunctionDataHelper<Data, AggregateFunctionVarianceMatrix<Data>>
{
public:
explicit AggregateFunctionVarianceMatrix(const DataTypes & argument_types_)
: IAggregateFunctionDataHelper<Data, AggregateFunctionVarianceMatrix<Data>>(argument_types_, {}, createResultType())
{}
AggregateFunctionVarianceMatrix(const IDataType &, const DataTypes & argument_types_)
: IAggregateFunctionDataHelper<Data, AggregateFunctionVarianceMatrix<Data>>(argument_types_, {}, createResultType())
{}
String getName() const override
{
if constexpr (Data::kind == StatisticsMatrixFunctionKind::covarPopMatrix)
return "covarPopMatrix";
if constexpr (Data::kind == StatisticsMatrixFunctionKind::covarSampMatrix)
return "covarSampMatrix";
if constexpr (Data::kind == StatisticsMatrixFunctionKind::corrMatrix)
return "corrMatrix";
UNREACHABLE();
}
void create(AggregateDataPtr __restrict place) const override
{
new (place) Data(this->argument_types.size());
}
static DataTypePtr createResultType()
{
return std::make_shared<DataTypeArray>(std::make_shared<DataTypeArray>(std::make_shared<DataTypeFloat64>()));
}
bool allocatesMemoryInArena() const override { return false; }
void add(AggregateDataPtr __restrict place, const IColumn ** columns, size_t row_num, Arena *) const override
{
this->data(place).add(columns, row_num);
}
void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, Arena *) const override
{
this->data(place).merge(this->data(rhs));
}
void serialize(ConstAggregateDataPtr __restrict place, WriteBuffer & buf, std::optional<size_t> /* version */) const override
{
this->data(place).serialize(buf);
}
void deserialize(AggregateDataPtr __restrict place, ReadBuffer & buf, std::optional<size_t> /* version */, Arena *) const override
{
this->data(place).deserialize(buf);
}
void insertResultInto(AggregateDataPtr __restrict place, IColumn & to, Arena *) const override
{
this->data(place).insertResultInto(to);
}
};
using AggregateFunctionCovarPopMatrix = AggregateFunctionVarianceMatrix<AggregateFunctionVarianceMatrixData<StatisticsMatrixFunctionKind::covarPopMatrix>>;
using AggregateFunctionCovarSampMatrix = AggregateFunctionVarianceMatrix<AggregateFunctionVarianceMatrixData<StatisticsMatrixFunctionKind::covarSampMatrix>>;
using AggregateFunctionCorrMatrix = AggregateFunctionVarianceMatrix<AggregateFunctionVarianceMatrixData<StatisticsMatrixFunctionKind::corrMatrix>>;
}
|
