You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by da...@apache.org on 2025/05/14 07:46:31 UTC
(doris) branch branch-3.0 updated: [feature](agg function) support corr_welford agg function #49712 (#50274)
This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new f3a52874d04 [feature](agg function) support corr_welford agg function #49712 (#50274)
f3a52874d04 is described below
commit f3a52874d04d1c9cb2f71d6fd5e5bc2ac595bbeb
Author: Mryange <ya...@selectdb.com>
AuthorDate: Wed May 14 15:46:22 2025 +0800
[feature](agg function) support corr_welford agg function #49712 (#50274)
In the past, due to performance considerations, the implementation of
corr was relatively simple, avoiding division operations except in the
final step.
However, this approach resulted in larger errors, so corr_welford was
introduced to improve the accuracy of the calculations.
```
0.7605339114107809 corr_result
0.7605339114107733 corr_welford_result
```
---
.../aggregate_function_corr.cpp | 94 ++++++---------------
...function_corr.cpp => aggregate_function_corr.h} | 89 ++++++++++++++++---
.../aggregate_function_simple_factory.cpp | 2 +
.../doris/catalog/BuiltinAggregateFunctions.java | 2 +
.../expressions/functions/agg/CorrWelford.java | 83 ++++++++++++++++++
.../visitor/AggregateFunctionVisitor.java | 5 ++
.../agg_function/test_corr_welford.out | Bin 0 -> 343 bytes
.../agg_function/test_corr_welford.groovy | 92 ++++++++++++++++++++
8 files changed, 285 insertions(+), 82 deletions(-)
diff --git a/be/src/vec/aggregate_functions/aggregate_function_corr.cpp b/be/src/vec/aggregate_functions/aggregate_function_corr.cpp
index cdaab6e086f..cbc32960611 100644
--- a/be/src/vec/aggregate_functions/aggregate_function_corr.cpp
+++ b/be/src/vec/aggregate_functions/aggregate_function_corr.cpp
@@ -15,78 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-#include "vec/aggregate_functions/aggregate_function.h"
-#include "vec/aggregate_functions/aggregate_function_binary.h"
-#include "vec/aggregate_functions/aggregate_function_simple_factory.h"
-#include "vec/core/types.h"
+#include "aggregate_function_corr.h"
+#include "vec/aggregate_functions/aggregate_function_simple_factory.h"
namespace doris::vectorized {
-template <typename T>
-struct CorrMoment {
- T m0 {};
- T x1 {};
- T y1 {};
- T xy {};
- T x2 {};
- T y2 {};
-
- void add(T x, T y) {
- ++m0;
- x1 += x;
- y1 += y;
- xy += x * y;
- x2 += x * x;
- y2 += y * y;
- }
-
- void merge(const CorrMoment& rhs) {
- m0 += rhs.m0;
- x1 += rhs.x1;
- y1 += rhs.y1;
- xy += rhs.xy;
- x2 += rhs.x2;
- y2 += rhs.y2;
- }
-
- void write(BufferWritable& buf) const {
- write_binary(m0, buf);
- write_binary(x1, buf);
- write_binary(y1, buf);
- write_binary(xy, buf);
- write_binary(x2, buf);
- write_binary(y2, buf);
- }
-
- void read(BufferReadable& buf) {
- read_binary(m0, buf);
- read_binary(x1, buf);
- read_binary(y1, buf);
- read_binary(xy, buf);
- read_binary(x2, buf);
- read_binary(y2, buf);
- }
-
- T get() const {
- // avoid float error(silent nan) when x or y is constant
- if (m0 * x2 <= x1 * x1 || m0 * y2 <= y1 * y1) [[unlikely]] {
- return 0;
- }
- return (m0 * xy - x1 * y1) / sqrt((m0 * x2 - x1 * x1) * (m0 * y2 - y1 * y1));
- }
-
- static String name() { return "corr"; }
-
- void reset() {
- m0 = {};
- x1 = {};
- y1 = {};
- xy = {};
- x2 = {};
- y2 = {};
- }
-};
-
AggregateFunctionPtr create_aggregate_corr_function(const std::string& name,
const DataTypes& argument_types,
const bool result_is_nullable,
@@ -100,4 +33,27 @@ void register_aggregate_functions_corr(AggregateFunctionSimpleFactory& factory)
factory.register_function_both("corr", create_aggregate_corr_function);
}
+AggregateFunctionPtr create_aggregate_corr_welford_function(const std::string& name,
+ const DataTypes& argument_types,
+ const bool result_is_nullable,
+ const AggregateFunctionAttr& attr) {
+ assert_binary(name, argument_types);
+
+ WhichDataType which0(remove_nullable(argument_types[0]));
+ WhichDataType which1(remove_nullable(argument_types[1]));
+
+ if (!which0.is_float64() || !which1.is_float64()) {
+ throw doris::Exception(ErrorCode::INTERNAL_ERROR,
+ "Aggregate function {} only support double", name);
+ }
+
+ return creator_without_type::create<
+ AggregateFunctionBinary<StatFunc<double, double, CorrMomentWelford>>>(
+ argument_types, result_is_nullable);
+}
+
+void register_aggregate_functions_corr_welford(AggregateFunctionSimpleFactory& factory) {
+ factory.register_function_both("corr_welford", create_aggregate_corr_welford_function);
+}
+
} // namespace doris::vectorized
diff --git a/be/src/vec/aggregate_functions/aggregate_function_corr.cpp b/be/src/vec/aggregate_functions/aggregate_function_corr.h
similarity index 53%
copy from be/src/vec/aggregate_functions/aggregate_function_corr.cpp
copy to be/src/vec/aggregate_functions/aggregate_function_corr.h
index cdaab6e086f..eba98891a02 100644
--- a/be/src/vec/aggregate_functions/aggregate_function_corr.cpp
+++ b/be/src/vec/aggregate_functions/aggregate_function_corr.h
@@ -14,10 +14,8 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
-
#include "vec/aggregate_functions/aggregate_function.h"
#include "vec/aggregate_functions/aggregate_function_binary.h"
-#include "vec/aggregate_functions/aggregate_function_simple_factory.h"
#include "vec/core/types.h"
namespace doris::vectorized {
@@ -87,17 +85,82 @@ struct CorrMoment {
}
};
-AggregateFunctionPtr create_aggregate_corr_function(const std::string& name,
- const DataTypes& argument_types,
- const bool result_is_nullable,
- const AggregateFunctionAttr& attr) {
- assert_binary(name, argument_types);
- return create_with_two_basic_numeric_types<CorrMoment>(argument_types[0], argument_types[1],
- argument_types, result_is_nullable);
-}
+template <typename T>
+struct CorrMomentWelford {
+ static_assert(std::is_same_v<T, double>, "CorrMomentWelford only support double");
+ double meanX {};
+ double meanY {};
+ double c2 {};
+ double m2X {};
+ double m2Y {};
+ int64_t count {0};
+
+ void add(double x, double y) {
+ ++count;
+ double deltaX = x - meanX;
+ double deltaY = y - meanY;
+ meanX += deltaX / count;
+ meanY += deltaY / count;
+ c2 += deltaX * (y - meanY);
+ m2X += deltaX * (x - meanX);
+ m2Y += deltaY * (y - meanY);
+ }
+
+ void merge(const CorrMomentWelford& rhs) {
+ if (rhs.count == 0) return;
+ if (count == 0) {
+ *this = rhs;
+ return;
+ }
+
+ double deltaX = rhs.meanX - meanX;
+ double deltaY = rhs.meanY - meanY;
+ int64_t total_count = count + rhs.count;
+
+ meanX = (meanX * count + rhs.meanX * rhs.count) / total_count;
+ meanY = (meanY * count + rhs.meanY * rhs.count) / total_count;
+
+ c2 += rhs.c2 + deltaX * deltaY * count * rhs.count / total_count;
+ m2X += rhs.m2X + deltaX * deltaX * count * rhs.count / total_count;
+ m2Y += rhs.m2Y + deltaY * deltaY * count * rhs.count / total_count;
+
+ count = total_count;
+ }
+
+ void write(BufferWritable& buf) const {
+ write_binary(meanX, buf);
+ write_binary(meanY, buf);
+ write_binary(c2, buf);
+ write_binary(m2X, buf);
+ write_binary(m2Y, buf);
+ write_binary(count, buf);
+ }
+
+ void read(BufferReadable& buf) {
+ read_binary(meanX, buf);
+ read_binary(meanY, buf);
+ read_binary(c2, buf);
+ read_binary(m2X, buf);
+ read_binary(m2Y, buf);
+ read_binary(count, buf);
+ }
+
+ double get() const {
+ if (m2X == 0 || m2Y == 0) {
+ return 0;
+ }
+ return c2 / sqrt(m2X * m2Y);
+ }
-void register_aggregate_functions_corr(AggregateFunctionSimpleFactory& factory) {
- factory.register_function_both("corr", create_aggregate_corr_function);
-}
+ static String name() { return "corr_welford"; }
+ void reset() {
+ meanX = {};
+ meanY = {};
+ c2 = {};
+ m2X = {};
+ m2Y = {};
+ count = 0;
+ }
+};
} // namespace doris::vectorized
diff --git a/be/src/vec/aggregate_functions/aggregate_function_simple_factory.cpp b/be/src/vec/aggregate_functions/aggregate_function_simple_factory.cpp
index d4bd3d576e7..7a1981263fc 100644
--- a/be/src/vec/aggregate_functions/aggregate_function_simple_factory.cpp
+++ b/be/src/vec/aggregate_functions/aggregate_function_simple_factory.cpp
@@ -66,6 +66,7 @@ void register_aggregate_function_histogram(AggregateFunctionSimpleFactory& facto
void register_aggregate_function_map_agg(AggregateFunctionSimpleFactory& factory);
void register_aggregate_function_bitmap_agg(AggregateFunctionSimpleFactory& factory);
void register_aggregate_functions_corr(AggregateFunctionSimpleFactory& factory);
+void register_aggregate_functions_corr_welford(AggregateFunctionSimpleFactory& factory);
void register_aggregate_function_covar_pop(AggregateFunctionSimpleFactory& factory);
void register_aggregate_function_covar_samp(AggregateFunctionSimpleFactory& factory);
void register_aggregate_function_skewness(AggregateFunctionSimpleFactory& factory);
@@ -117,6 +118,7 @@ AggregateFunctionSimpleFactory& AggregateFunctionSimpleFactory::instance() {
register_aggregate_function_window_lead_lag_first_last(instance);
register_aggregate_function_HLL_union_agg(instance);
register_aggregate_functions_corr(instance);
+ register_aggregate_functions_corr_welford(instance);
register_aggregate_function_covar_pop(instance);
register_aggregate_function_covar_samp(instance);
register_aggregate_function_combinator_foreach(instance);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
index 5494a2924b5..4535ebdd02d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinAggregateFunctions.java
@@ -31,6 +31,7 @@ import org.apache.doris.nereids.trees.expressions.functions.agg.BitmapUnionInt;
import org.apache.doris.nereids.trees.expressions.functions.agg.CollectList;
import org.apache.doris.nereids.trees.expressions.functions.agg.CollectSet;
import org.apache.doris.nereids.trees.expressions.functions.agg.Corr;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CorrWelford;
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
import org.apache.doris.nereids.trees.expressions.functions.agg.CountByEnum;
import org.apache.doris.nereids.trees.expressions.functions.agg.Covar;
@@ -109,6 +110,7 @@ public class BuiltinAggregateFunctions implements FunctionHelper {
agg(CollectList.class, "collect_list", "group_array"),
agg(CollectSet.class, "collect_set", "group_uniq_array"),
agg(Corr.class, "corr"),
+ agg(CorrWelford.class, "corr_welford"),
agg(Count.class, "count"),
agg(CountByEnum.class, "count_by_enum"),
agg(Covar.class, "covar", "covar_pop"),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CorrWelford.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CorrWelford.java
new file mode 100644
index 00000000000..3b8b302be4f
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CorrWelford.java
@@ -0,0 +1,83 @@
+// 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.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
+import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.DoubleType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/**
+ * AggregateFunction 'corr_welford'. This class is generated by GenerateFunction.
+ */
+public class CorrWelford extends NullableAggregateFunction
+ implements UnaryExpression, ExplicitlyCastableSignature {
+
+ public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
+ FunctionSignature.ret(DoubleType.INSTANCE).args(DoubleType.INSTANCE, DoubleType.INSTANCE)
+ );
+
+ /**
+ * constructor with 2 argument.
+ */
+ public CorrWelford(Expression arg1, Expression arg2) {
+ this(false, arg1, arg2);
+ }
+
+ /**
+ * constructor with 3 arguments.
+ */
+ public CorrWelford(boolean distinct, Expression arg1, Expression arg2) {
+ this(distinct, false, arg1, arg2);
+ }
+
+ public CorrWelford(boolean distinct, boolean alwaysNullable, Expression arg1, Expression arg2) {
+ super("corr_welford", distinct, alwaysNullable, arg1, arg2);
+ }
+
+ /**
+ * withDistinctAndChildren.
+ */
+ @Override
+ public CorrWelford withDistinctAndChildren(boolean distinct, List<Expression> children) {
+ Preconditions.checkArgument(children.size() == 2);
+ return new CorrWelford(distinct, alwaysNullable, children.get(0), children.get(1));
+ }
+
+ @Override
+ public CorrWelford withAlwaysNullable(boolean alwaysNullable) {
+ return new CorrWelford(distinct, alwaysNullable, children.get(0), children.get(1));
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitCorrWelford(this, context);
+ }
+
+ @Override
+ public List<FunctionSignature> getSignatures() {
+ return SIGNATURES;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
index 618b0dc946f..02edca4642a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/AggregateFunctionVisitor.java
@@ -32,6 +32,7 @@ import org.apache.doris.nereids.trees.expressions.functions.agg.BitmapUnionInt;
import org.apache.doris.nereids.trees.expressions.functions.agg.CollectList;
import org.apache.doris.nereids.trees.expressions.functions.agg.CollectSet;
import org.apache.doris.nereids.trees.expressions.functions.agg.Corr;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CorrWelford;
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
import org.apache.doris.nereids.trees.expressions.functions.agg.CountByEnum;
import org.apache.doris.nereids.trees.expressions.functions.agg.Covar;
@@ -142,6 +143,10 @@ public interface AggregateFunctionVisitor<R, C> {
return visitNullableAggregateFunction(corr, context);
}
+ default R visitCorrWelford(CorrWelford corr, C context) {
+ return visitNullableAggregateFunction(corr, context);
+ }
+
default R visitCount(Count count, C context) {
return visitAggregateFunction(count, context);
}
diff --git a/regression-test/data/nereids_function_p0/agg_function/test_corr_welford.out b/regression-test/data/nereids_function_p0/agg_function/test_corr_welford.out
new file mode 100644
index 00000000000..facfc9cf580
Binary files /dev/null and b/regression-test/data/nereids_function_p0/agg_function/test_corr_welford.out differ
diff --git a/regression-test/suites/nereids_function_p0/agg_function/test_corr_welford.groovy b/regression-test/suites/nereids_function_p0/agg_function/test_corr_welford.groovy
new file mode 100644
index 00000000000..5cbde28e4c2
--- /dev/null
+++ b/regression-test/suites/nereids_function_p0/agg_function/test_corr_welford.groovy
@@ -0,0 +1,92 @@
+// 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.
+
+suite("test_corr_welford") {
+ sql """ DROP TABLE IF EXISTS test_corr_welford """
+
+ sql """ SET enable_nereids_planner=true """
+ sql """ SET enable_fallback_to_original_planner=false """
+
+ sql """
+ CREATE TABLE test_corr_welford (
+ `id` int,
+ `x` int,
+ `y` int,
+ ) ENGINE=OLAP
+ Duplicate KEY (`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 4
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+
+ // Perfect positive corr_welfordelation
+ sql """
+ insert into test_corr_welford values
+ (1, 1, 1),
+ (2, 2, 2),
+ (3, 3, 3),
+ (4, 4, 4),
+ (5, 5, 5)
+ """
+ qt_sql "select corr_welford(x,y) from test_corr_welford"
+ sql """ truncate table test_corr_welford """
+
+ // Perfect negative corr_welfordelation
+ sql """
+ insert into test_corr_welford values
+ (1, 1, 5),
+ (2, 2, 4),
+ (3, 3, 3),
+ (4, 4, 2),
+ (5, 5, 1)
+ """
+ qt_sql "select corr_welford(x,y) from test_corr_welford"
+ sql """ truncate table test_corr_welford """
+
+ // Zero corr_welfordelation
+ sql """
+ insert into test_corr_welford values
+ (1, 1, 1),
+ (2, 1, 2),
+ (3, 1, 3),
+ (4, 1, 4),
+ (5, 1, 5)
+ """
+ qt_sql "select corr_welford(x,y) from test_corr_welford"
+ sql """ truncate table test_corr_welford """
+
+ // Partial linear corr_welfordelation
+ sql """
+ insert into test_corr_welford values
+ (1, 1, 1),
+ (2, 2, 2),
+ (3, 3, 3),
+ (4, 4, 4),
+ (5, 5, 10)
+ """
+ qt_sql "select corr_welford(x,y) from test_corr_welford"
+
+ qt_sql "select corr_welford(cast(x as float),cast(y as float)) from test_corr_welford"
+
+ qt_sql_const1 "select corr_welford(x,1) from test_corr_welford"
+ qt_sql_const2 "select corr_welford(x,1e100) from test_corr_welford"
+ qt_sql_const3 "select corr_welford(x,1e-100) from test_corr_welford"
+ qt_sql_const4 "select corr_welford(1,y) from test_corr_welford"
+ qt_sql_const5 "select corr_welford(1e100,y) from test_corr_welford"
+ qt_sql_const6 "select corr_welford(1e-100,y) from test_corr_welford"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org