be/src/exprs/aggregate/aggregate_function_covar.h

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#pragma once

#include <glog/logging.h>

#include <boost/iterator/iterator_facade.hpp>
#include <cstddef>
#include <cstdint>
#include <memory>

#include "core/assert_cast.h"
#include "core/column/column.h"
#include "core/column/column_nullable.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_number.h"
#include "core/types.h"
#include "exprs/aggregate/aggregate_function.h"

namespace doris {
#include "common/compile_check_begin.h"

class Arena;
class BufferReadable;
class BufferWritable;
template <PrimitiveType T>
class ColumnDecimal;
template <PrimitiveType T>
class ColumnVector;

template <PrimitiveType T>
struct BaseData {
    BaseData() = default;
    virtual ~BaseData() = default;
    static DataTypePtr get_return_type() { return std::make_shared<DataTypeFloat64>(); }

    void write(BufferWritable& buf) const {
        buf.write_binary(sum_x);
        buf.write_binary(sum_y);
        buf.write_binary(sum_xy);
        buf.write_binary(count);
    }

    void read(BufferReadable& buf) {
        buf.read_binary(sum_x);
        buf.read_binary(sum_y);
        buf.read_binary(sum_xy);
        buf.read_binary(count);
    }

    void reset() {
        sum_x = 0.0;
        sum_y = 0.0;
        sum_xy = 0.0;
        count = 0;
    }

    // Cov(X, Y) = E(XY) - E(X)E(Y)
    double get_pop_result() const {
        if (count == 1) {
            return 0.0;
        }
        return sum_xy / (double)count - sum_x * sum_y / ((double)count * (double)count);
    }

    double get_samp_result() const {
        return sum_xy / double(count - 1) -
               sum_x * sum_y / ((double)(count) * ((double)(count - 1)));
    }

    void merge(const BaseData& rhs) {
        if (rhs.count == 0) {
            return;
        }
        sum_x += rhs.sum_x;
        sum_y += rhs.sum_y;
        sum_xy += rhs.sum_xy;
        count += rhs.count;
    }

    void add(const IColumn* column_x, const IColumn* column_y, size_t row_num) {
        const auto& sources_x = assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
                                            TypeCheckOnRelease::DISABLE>(*column_x);
        double source_data_x = double(sources_x.get_data()[row_num]);
        const auto& sources_y = assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
                                            TypeCheckOnRelease::DISABLE>(*column_y);
        double source_data_y = double(sources_y.get_data()[row_num]);

        sum_x += source_data_x;
        sum_y += source_data_y;
        sum_xy += source_data_x * source_data_y;
        count += 1;
    }

    double sum_x {};
    double sum_y {};
    double sum_xy {};
    int64_t count {};
};

template <PrimitiveType T>
struct PopData : BaseData<T> {
    static const char* name() { return "covar"; }

    void insert_result_into(IColumn& to) const {
        auto& col = assert_cast<ColumnFloat64&>(to);
        col.get_data().push_back(this->get_pop_result());
    }
};

template <PrimitiveType T>
struct SampData : BaseData<T> {
    static const char* name() { return "covar_samp"; }

    void insert_result_into(IColumn& to) const {
        auto& col = assert_cast<ColumnFloat64&>(to);
        if (this->count == 1 || this->count == 0) {
            col.insert_default();
        } else {
            col.get_data().push_back(this->get_samp_result());
        }
    }
};

template <typename Data>
class AggregateFunctionSampCovariance
        : public IAggregateFunctionDataHelper<Data, AggregateFunctionSampCovariance<Data>>,
          MultiExpression,
          NullableAggregateFunction {
public:
    AggregateFunctionSampCovariance(const DataTypes& argument_types_)
            : IAggregateFunctionDataHelper<Data, AggregateFunctionSampCovariance<Data>>(
                      argument_types_) {}

    String get_name() const override { return Data::name(); }

    DataTypePtr get_return_type() const override { return Data::get_return_type(); }

    void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
             Arena&) const override {
        this->data(place).add(columns[0], columns[1], row_num);
    }

    void reset(AggregateDataPtr __restrict place) const override { this->data(place).reset(); }

    void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
               Arena&) const override {
        this->data(place).merge(this->data(rhs));
    }

    void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
        this->data(place).write(buf);
    }

    void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
                     Arena&) const override {
        this->data(place).read(buf);
    }

    void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
        this->data(place).insert_result_into(to);
    }
};

#include "common/compile_check_end.h"
} // namespace doris

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 <boost/iterator/iterator_facade.hpp>
23		#include <cstddef>
24		#include <cstdint>
25		#include <memory>
26
27		#include "core/assert_cast.h"
28		#include "core/column/column.h"
29		#include "core/column/column_nullable.h"
30		#include "core/data_type/data_type_decimal.h"
31		#include "core/data_type/data_type_number.h"
32		#include "core/types.h"
33		#include "exprs/aggregate/aggregate_function.h"
34
35		namespace doris {
36		#include "common/compile_check_begin.h"
37
38		class Arena;
39		class BufferReadable;
40		class BufferWritable;
41		template <PrimitiveType T>
42		class ColumnDecimal;
43		template <PrimitiveType T>
44		class ColumnVector;
45
46		template <PrimitiveType T>
47		struct BaseData {
48	0	BaseData() = default;
49	0	virtual ~BaseData() = default;
50	0	static DataTypePtr get_return_type() { return std::make_shared<DataTypeFloat64>(); }
51
52	0	void write(BufferWritable& buf) const {
53	0	buf.write_binary(sum_x);
54	0	buf.write_binary(sum_y);
55	0	buf.write_binary(sum_xy);
56	0	buf.write_binary(count);
57	0	}
58
59	0	void read(BufferReadable& buf) {
60	0	buf.read_binary(sum_x);
61	0	buf.read_binary(sum_y);
62	0	buf.read_binary(sum_xy);
63	0	buf.read_binary(count);
64	0	}
65
66	0	void reset() {
67	0	sum_x = 0.0;
68	0	sum_y = 0.0;
69	0	sum_xy = 0.0;
70	0	count = 0;
71	0	}
72
73		// Cov(X, Y) = E(XY) - E(X)E(Y)
74	0	double get_pop_result() const {
75	0	if (count == 1) {
76	0	return 0.0;
77	0	}
78	0	return sum_xy / (double)count - sum_x * sum_y / ((double)count * (double)count);
79	0	}
80
81	0	double get_samp_result() const {
82	0	return sum_xy / double(count - 1) -
83	0	sum_x * sum_y / ((double)(count) * ((double)(count - 1)));
84	0	}
85
86	0	void merge(const BaseData& rhs) {
87	0	if (rhs.count == 0) {
88	0	return;
89	0	}
90	0	sum_x += rhs.sum_x;
91	0	sum_y += rhs.sum_y;
92	0	sum_xy += rhs.sum_xy;
93	0	count += rhs.count;
94	0	}
95
96	0	void add(const IColumn* column_x, const IColumn* column_y, size_t row_num) {
97	0	const auto& sources_x = assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
98	0	TypeCheckOnRelease::DISABLE>(*column_x);
99	0	double source_data_x = double(sources_x.get_data()[row_num]);
100	0	const auto& sources_y = assert_cast<const typename PrimitiveTypeTraits<T>::ColumnType&,
101	0	TypeCheckOnRelease::DISABLE>(*column_y);
102	0	double source_data_y = double(sources_y.get_data()[row_num]);
103
104	0	sum_x += source_data_x;
105	0	sum_y += source_data_y;
106	0	sum_xy += source_data_x * source_data_y;
107	0	count += 1;
108	0	}
109
110		double sum_x {};
111		double sum_y {};
112		double sum_xy {};
113		int64_t count {};
114		};
115
116		template <PrimitiveType T>
117		struct PopData : BaseData<T> {
118	0	static const char* name() { return "covar"; }
119
120	0	void insert_result_into(IColumn& to) const {
121	0	auto& col = assert_cast<ColumnFloat64&>(to);
122	0	col.get_data().push_back(this->get_pop_result());
123	0	}
124		};
125
126		template <PrimitiveType T>
127		struct SampData : BaseData<T> {
128	0	static const char* name() { return "covar_samp"; }
129
130	0	void insert_result_into(IColumn& to) const {
131	0	auto& col = assert_cast<ColumnFloat64&>(to);
132	0	if (this->count == 1 \|\| this->count == 0) {
133	0	col.insert_default();
134	0	} else {
135	0	col.get_data().push_back(this->get_samp_result());
136	0	}
137	0	}
138		};
139
140		template <typename Data>
141		class AggregateFunctionSampCovariance
142		: public IAggregateFunctionDataHelper<Data, AggregateFunctionSampCovariance<Data>>,
143		MultiExpression,
144		NullableAggregateFunction {
145		public:
146		AggregateFunctionSampCovariance(const DataTypes& argument_types_)
147	0	: IAggregateFunctionDataHelper<Data, AggregateFunctionSampCovariance<Data>>(
148	0	argument_types_) {} Unexecuted instantiation: _ZN5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEEC2ERKSt6vectorISt10shared_ptrIKNS_9IDataTypeEESaIS9_EE Unexecuted instantiation: _ZN5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEEC2ERKSt6vectorISt10shared_ptrIKNS_9IDataTypeEESaIS9_EE
149
150	0	String get_name() const override { return Data::name(); } Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE8get_nameB5cxx11Ev Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE8get_nameB5cxx11Ev
151
152	0	DataTypePtr get_return_type() const override { return Data::get_return_type(); } Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE15get_return_typeEv Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE15get_return_typeEv
153
154		void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num,
155	0	Arena&) const override {
156	0	this->data(place).add(columns[0], columns[1], row_num);
157	0	} Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE3addEPcPPKNS_7IColumnElRNS_5ArenaE Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE3addEPcPPKNS_7IColumnElRNS_5ArenaE
158
159	0	void reset(AggregateDataPtr __restrict place) const override { this->data(place).reset(); } Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE5resetEPc Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE5resetEPc
160
161		void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs,
162	0	Arena&) const override {
163	0	this->data(place).merge(this->data(rhs));
164	0	} Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE5mergeEPcPKcRNS_5ArenaE Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE5mergeEPcPKcRNS_5ArenaE
165
166	0	void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override {
167	0	this->data(place).write(buf);
168	0	} Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE9serializeEPKcRNS_14BufferWritableE Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE9serializeEPKcRNS_14BufferWritableE
169
170		void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf,
171	0	Arena&) const override {
172	0	this->data(place).read(buf);
173	0	} Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE11deserializeEPcRNS_14BufferReadableERNS_5ArenaE Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE11deserializeEPcRNS_14BufferReadableERNS_5ArenaE
174
175	0	void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override {
176	0	this->data(place).insert_result_into(to);
177	0	} Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_8SampDataILNS_13PrimitiveTypeE9EEEE18insert_result_intoEPKcRNS_7IColumnE Unexecuted instantiation: _ZNK5doris31AggregateFunctionSampCovarianceINS_7PopDataILNS_13PrimitiveTypeE9EEEE18insert_result_intoEPKcRNS_7IColumnE
178		};
179
180		#include "common/compile_check_end.h"
181		} // namespace doris

Coverage Report

Created: 2026-03-16 19:13