[Enhancement](doris-future) Support REGR_SXX_SXY_SYY aggregation functions

be/src/vec/aggregate_functions/aggregate_function_regr_union.cpp

@@ -89,5 +89,8 @@ void register_aggregate_function_regr_union(AggregateFunctionSimpleFactory& fact
     factory.register_function_both("regr_slope", create_aggregate_function_regr<RegrSlopeFunc>);
     factory.register_function_both("regr_intercept",
                                    create_aggregate_function_regr<RegrInterceptFunc>);
+    factory.register_function_both("regr_sxx", create_aggregate_function_regr<RegrSxxFunc>);
+    factory.register_function_both("regr_sxy", create_aggregate_function_regr<RegrSxyFunc>);
+    factory.register_function_both("regr_syy", create_aggregate_function_regr<RegrSyyFunc>);
 }
 } // namespace doris::vectorized

be/src/vec/aggregate_functions/aggregate_function_regr_union.h

@@ -98,6 +98,173 @@ struct AggregateFunctionRegrData {
         return slope;
     }
 };
+template <typename T>
+struct AggregateFunctionRegrSxxData {
+    using Type = T;
+    UInt64 count = 0;
+    Float64 sum_x {};
+    Float64 sum_of_x_squared {};
+
+    void write(BufferWritable& buf) const {
+        write_binary(sum_x, buf);
+        write_binary(sum_of_x_squared, buf);
+        write_binary(count, buf);
+    }
+
+    void read(BufferReadable& buf) {
+        read_binary(sum_x, buf);
+        read_binary(sum_of_x_squared, buf);
+        read_binary(count, buf);
+    }
+
+    void reset() {
+        sum_x = {};
+        sum_of_x_squared = {};
+        count = 0;
+    }
+
+    void merge(const AggregateFunctionRegrSxxData& rhs) {
+        if (rhs.count == 0) {
+            return;
+        }
+        sum_x += rhs.sum_x;
+        sum_of_x_squared += rhs.sum_of_x_squared;
+        count += rhs.count;
+    }
+
+    void add(T value_y, T value_x) {
+        sum_x += value_x;
+        sum_of_x_squared += value_x * value_x;
+        count += 1;
+    }
+
+    Float64 get_regr_sxx_result() const {
+        // count == 0
+        // The result of a query for an empty table is a null value
+        Float64 result = sum_of_x_squared - (sum_x * sum_x / count);
+        return result;
+    }
+};
+template <typename T>
+struct AggregateFunctionRegrSxyData {
+    using Type = T;
+    UInt64 count = 0;
+    Float64 sum_x {};
+    Float64 sum_y {};
+    Float64 sum_of_x_mul_y {};
+
+    void write(BufferWritable& buf) const {
+        write_binary(sum_x, buf);
+        write_binary(sum_y, buf);
+        write_binary(sum_of_x_mul_y, buf);
+        write_binary(count, buf);
+    }
+
+    void read(BufferReadable& buf) {
+        read_binary(sum_x, buf);
+        read_binary(sum_y, buf);
+        read_binary(sum_of_x_mul_y, buf);
+        read_binary(count, buf);
+    }
+
+    void reset() {
+        sum_x = {};
+        sum_y = {};
+        sum_of_x_mul_y = {};
+        count = 0;
+    }
+
+    void merge(const AggregateFunctionRegrSxyData& rhs) {
+        if (rhs.count == 0) {
+            return;
+        }
+        sum_x += rhs.sum_x;
+        sum_y += rhs.sum_y;
+        sum_of_x_mul_y += rhs.sum_of_x_mul_y;
+        count += rhs.count;
+    }
+
+    void add(T value_y, T value_x) {
+        sum_x += value_x;
+        sum_y += value_y;
+        sum_of_x_mul_y += value_x * value_y;
+        count += 1;
+    }
+
+    Float64 get_regr_sxy_result() const {
+        // count == 0
+        // The result of a query for an empty table is a null value
+        Float64 result = sum_of_x_mul_y - (sum_x * sum_y / count);
+        return result;
+    }
+};
+template <typename T>
+struct AggregateFunctionRegrSyyData {
+    using Type = T;
+    UInt64 count = 0;
+    Float64 sum_y {};
+    Float64 sum_of_y_squared {};
+
+    void write(BufferWritable& buf) const {
+        write_binary(sum_y, buf);
+        write_binary(sum_of_y_squared, buf);
+        write_binary(count, buf);
+    }
+
+    void read(BufferReadable& buf) {
+        read_binary(sum_y, buf);
+        read_binary(sum_of_y_squared, buf);
+        read_binary(count, buf);
+    }
+
+    void reset() {
+        sum_y = {};
+        sum_of_y_squared = {};
+        count = 0;
+    }
+
+    void merge(const AggregateFunctionRegrSyyData& rhs) {
+        if (rhs.count == 0) {
+            return;
+        }
+        sum_y += rhs.sum_y;
+        sum_of_y_squared += rhs.sum_of_y_squared;
+        count += rhs.count;
+    }
+
+    void add(T value_y, T value_x) {
+        sum_y += value_y;
+        sum_of_y_squared += value_y * value_y;
+        count += 1;
+    }
+
+    Float64 get_regr_syy_result() const {
+        // count == 0
+        // The result of a query for an empty table is a null value
+        Float64 result = sum_of_y_squared - (sum_y * sum_y / count);
+        return result;
+    }
+};
+template <typename T>
+struct RegrSxxFunc : AggregateFunctionRegrSxxData<T> {
+    static constexpr const char* name = "regr_sxx";
+
+    Float64 get_result() const { return this->get_regr_sxx_result(); }
+};
+
+template <typename T>
+struct RegrSxyFunc : AggregateFunctionRegrSxyData<T> {
+    static constexpr const char* name = "regr_sxy";
+
+    Float64 get_result() const { return this->get_regr_sxy_result(); }
+};
+
+template <typename T>
+struct RegrSyyFunc : AggregateFunctionRegrSyyData<T> {
+    static constexpr const char* name = "regr_syy";
+
+    Float64 get_result() const { return this->get_regr_syy_result(); }
+};
 
 template <typename T>
 struct RegrSlopeFunc : AggregateFunctionRegrData<T> {

be/src/vec/aggregate_functions/aggregate_function_simple_factory.cpp

@@ -58,6 +58,7 @@ void register_aggregate_function_window_funnel(AggregateFunctionSimpleFactory& f
 void register_aggregate_function_window_funnel_old(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_regr_union(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_retention(AggregateFunctionSimpleFactory& factory);
+void register_aggregate_function_regr_mixed(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_percentile_approx(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_orthogonal_bitmap(AggregateFunctionSimpleFactory& factory);
 void register_aggregate_function_collect_list(AggregateFunctionSimpleFactory& factory);

fe/fe-core/src/main/java/org/apache/doris/catalog/AggregateFunction.java

@@ -55,6 +55,7 @@ public class AggregateFunction extends Function {
             "approx_count_distinct", "ndv", FunctionSet.BITMAP_UNION_INT, FunctionSet.BITMAP_UNION_COUNT,
             "ndv_no_finalize", "percentile_array", "histogram",
             FunctionSet.SEQUENCE_COUNT, FunctionSet.MAP_AGG, FunctionSet.BITMAP_AGG, FunctionSet.ARRAY_AGG,
+            FunctionSet.REGR_SXX, FunctionSet.REGR_SYY, FunctionSet.REGR_SXY,
             FunctionSet.COLLECT_LIST, FunctionSet.COLLECT_SET, FunctionSet.GROUP_ARRAY_INTERSECT,
             FunctionSet.SUM0, FunctionSet.MULTI_DISTINCT_SUM0, FunctionSet.REGR_INTERCEPT, FunctionSet.REGR_SLOPE);
 

fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java

@@ -64,6 +64,9 @@
 import org.apache.doris.nereids.trees.expressions.functions.agg.QuantileUnion;
 import org.apache.doris.nereids.trees.expressions.functions.agg.RegrIntercept;
 import org.apache.doris.nereids.trees.expressions.functions.agg.RegrSlope;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RegrSxx;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RegrSxy;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RegrSyy;
 import org.apache.doris.nereids.trees.expressions.functions.agg.Retention;
 import org.apache.doris.nereids.trees.expressions.functions.agg.SequenceCount;
 import org.apache.doris.nereids.trees.expressions.functions.agg.SequenceMatch;
@@ -139,6 +142,9 @@ public class BuiltinAggregateFunctions implements FunctionHelper {
             agg(QuantileUnion.class, "quantile_union"),
             agg(RegrIntercept.class, "regr_intercept"),
             agg(RegrSlope.class, "regr_slope"),
+            agg(RegrSxx.class, "regr_sxx"),
+            agg(RegrSxy.class, "regr_sxy"),
+            agg(RegrSyy.class, "regr_syy"),
             agg(Retention.class, "retention"),
             agg(SequenceCount.class, "sequence_count"),
             agg(SequenceMatch.class, "sequence_match"),

fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java

@@ -635,6 +635,12 @@ public void addBuiltinBothScalaAndVectorized(Function fn) {
 
     public static final String REGR_SLOPE = "regr_slope";
 
+    public static final String REGR_SXX = "regr_sxx";
+
+    public static final String REGR_SXY = "regr_sxy";
+
+    public static final String REGR_SYY = "regr_syy";
+
     public static final String SEQUENCE_MATCH = "sequence_match";
 
     public static final String SEQUENCE_COUNT = "sequence_count";
@@ -708,6 +714,33 @@ private void initAggregateBuiltins() {
                     null, false, true, true, true));
         }
 
+        addBuiltin(AggregateFunction.createBuiltin(FunctionSet.REGR_SXX,
+                        Lists.newArrayList(Type.DOUBLE, Type.DOUBLE), Type.DOUBLE, Type.DOUBLE,
+                        "",
+                        "",
+                        "",
+                        null, null,
+                        "",
+                        null, false, false, false, true));
+
+        addBuiltin(AggregateFunction.createBuiltin(FunctionSet.REGR_SXY,
+                        Lists.newArrayList(Type.DOUBLE, Type.DOUBLE), Type.DOUBLE, Type.DOUBLE,
+                        "",
+                        "",
+                        "",
+                        null, null,
+                        "",
+                        null, false, false, false, true));
+
+        addBuiltin(AggregateFunction.createBuiltin(FunctionSet.REGR_SYY,
+                        Lists.newArrayList(Type.DOUBLE, Type.DOUBLE), Type.DOUBLE, Type.DOUBLE,
+                        "",
+                        "",
+                        "",
+                        null, null,
+                        "",
+                        null, false, false, false, true));
+
         // Vectorization does not need symbol any more, we should clean it in the future.
         addBuiltin(AggregateFunction.createBuiltin(FunctionSet.WINDOW_FUNNEL,
                 Lists.newArrayList(Type.BIGINT, Type.STRING, Type.DATETIME, Type.BOOLEAN),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/RegrSxx.java

@@ -0,0 +1,94 @@
+// 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.
+
+package org.apache.doris.nereids.trees.expressions.functions.agg;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.AlwaysNullable;
+import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
+import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.DoubleType;
+import org.apache.doris.nereids.types.FloatType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.SmallIntType;
+import org.apache.doris.nereids.types.TinyIntType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/** regr_sxx agg function. */
+public class RegrSxx extends AggregateFunction
+        implements BinaryExpression, ExplicitlyCastableSignature, AlwaysNullable {
+
+    public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
+            FunctionSignature.ret(DoubleType.INSTANCE).args(DoubleType.INSTANCE, DoubleType.INSTANCE),
+            FunctionSignature.ret(DoubleType.INSTANCE).args(BigIntType.INSTANCE, BigIntType.INSTANCE),
+            FunctionSignature.ret(DoubleType.INSTANCE).args(IntegerType.INSTANCE, IntegerType.INSTANCE),
+            FunctionSignature.ret(DoubleType.INSTANCE).args(SmallIntType.INSTANCE, SmallIntType.INSTANCE),
+            FunctionSignature.ret(DoubleType.INSTANCE).args(TinyIntType.INSTANCE, TinyIntType.INSTANCE),
+            FunctionSignature.ret(DoubleType.INSTANCE).args(FloatType.INSTANCE, FloatType.INSTANCE));
+
+    /**
+     * Constructor with 2 arguments.
+     */
+    public RegrSxx(Expression arg1, Expression arg2) {
+        this(false, arg1, arg2);
+    }
+
+    /**
+     * Constructor with distinct flag and 2 arguments.
+     */
+    public RegrSxx(boolean distinct, Expression arg1, Expression arg2) {
+        super("regr_sxx", distinct, arg1, arg2);
+    }
+
+    @Override
+    public void checkLegalityBeforeTypeCoercion() throws AnalysisException {
+        DataType arg0Type = left().getDataType();
+        DataType arg1Type = right().getDataType();
+        if ((!arg0Type.isNumericType() && !arg0Type.isNullType())
+                || arg0Type.isOnlyMetricType()) {
+            throw new AnalysisException("regr_sxx requires numeric for first parameter: " + toSql());
+        } else if ((!arg1Type.isNumericType() && !arg1Type.isNullType())
+                || arg1Type.isOnlyMetricType()) {
+            throw new AnalysisException("regr_sxx requires numeric for second parameter: " + toSql());
+        }
+    }
+
+    @Override
+    public RegrSxx withDistinctAndChildren(boolean distinct, List<Expression> children) {
+        Preconditions.checkArgument(children.size() == 2);
+        return new RegrSxx(distinct, children.get(0), children.get(1));
+    }
+
+    @Override
+    public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+        return visitor.visitRegrSxx(this, context);
+    }
+
+    @Override
+    public List<FunctionSignature> getSignatures() {
+        return SIGNATURES;
+    }
+}