[SPARK-37641][SQL] Support ANSI Aggregate Function: regr_r2

### What changes were proposed in this pull request?
This PR used to support ANSI aggregate Function: `regr_r2`

**Syntax**: REGR_R2(y, x)
**Arguments**:
- **y**:The dependent variable. This must be an expression that can be evaluated to a numeric type.
- **x**:The independent variable. This must be an expression that can be evaluated to a numeric type.

**Examples**:
`select k, regr_r2(v, v2) from aggr group by k;`

| k | regr_r2(v, v2) |
|--|---------------|
| 1 |    [NULL]         |
| 2 | 0.9976905312   |

The mainstream database supports `regr_r2` show below:
**Teradata**
https://docs.teradata.com/r/756LNiPSFdY~4JcCCcR5Cw/exhFe2f_YyGqKFakYYUn2A
**Snowflake**
https://docs.snowflake.com/en/sql-reference/functions/regr_r2.html
**Oracle**
https://docs.oracle.com/en/database/oracle/oracle-database/19/sqlrf/REGR_-Linear-Regression-Functions.html#GUID-A675B68F-2A88-4843-BE2C-FCDE9C65F9A9
**DB2**
https://www.ibm.com/docs/en/db2/11.5?topic=af-regression-functions-regr-avgx-regr-avgy-regr-count
**H2**
http://www.h2database.com/html/functions-aggregate.html#regr_r2
**Postgresql**
https://www.postgresql.org/docs/8.4/functions-aggregate.html
**Sybase**
https://infocenter.sybase.com/help/index.jsp?topic=/com.sybase.help.sqlanywhere.12.0.0/dbreference/regr-r2-function.html
**Exasol**
https://docs.exasol.com/sql_references/functions/alphabeticallistfunctions/regr_function.htm

### Why are the changes needed?
`regr_r2` is very useful.

### Does this PR introduce _any_ user-facing change?
'Yes'. New feature.

### How was this patch tested?
New tests.

Closes #34894 from beliefer/SPARK-37641.

Authored-by: Jiaan Geng <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit b01d81e)
Signed-off-by: Wenchen Fan <[email protected]>

Author: beliefer

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -497,6 +497,7 @@ object FunctionRegistry {
     expression[RegrCount]("regr_count"),
     expression[RegrAvgX]("regr_avgx"),
     expression[RegrAvgY]("regr_avgy"),
+    expression[RegrR2]("regr_r2"),
 
     // string functions
     expression[Ascii]("ascii"),

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/linearRegression.scala

@@ -17,9 +17,10 @@
 
 package org.apache.spark.sql.catalyst.expressions.aggregate
 
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions.{And, Expression, ExpressionDescription, If, ImplicitCastInputTypes, IsNotNull, Literal, RuntimeReplaceableAggregate}
 import org.apache.spark.sql.catalyst.trees.BinaryLike
-import org.apache.spark.sql.types.{AbstractDataType, NumericType}
+import org.apache.spark.sql.types.{AbstractDataType, DoubleType, NumericType}
 
 @ExpressionDescription(
   usage = """
@@ -118,3 +119,34 @@ case class RegrAvgY(
       newLeft: Expression, newRight: Expression): RegrAvgY =
     this.copy(left = newLeft, right = newRight)
 }
+
+// scalastyle:off line.size.limit
+@ExpressionDescription(
+  usage = "_FUNC_(y, x) - Returns the coefficient of determination for non-null pairs in a group, where `y` is the dependent variable and `x` is the independent variable.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(y, x) FROM VALUES (1, 2), (2, 2), (2, 3), (2, 4) AS tab(y, x);
+       0.2727272727272727
+      > SELECT _FUNC_(y, x) FROM VALUES (1, null) AS tab(y, x);
+       NULL
+      > SELECT _FUNC_(y, x) FROM VALUES (null, 1) AS tab(y, x);
+       NULL
+      > SELECT _FUNC_(y, x) FROM VALUES (1, 2), (2, null), (2, 3), (2, 4) AS tab(y, x);
+       0.7500000000000001
+      > SELECT _FUNC_(y, x) FROM VALUES (1, 2), (2, null), (null, 3), (2, 4) AS tab(y, x);
+       1.0
+  """,
+  group = "agg_funcs",
+  since = "3.3.0")
+// scalastyle:on line.size.limit
+case class RegrR2(x: Expression, y: Expression) extends PearsonCorrelation(x, y, true) {
+  override def prettyName: String = "regr_r2"
+  override val evaluateExpression: Expression = {
+    val corr = ck / sqrt(xMk * yMk)
+    If(xMk === 0.0, Literal.create(null, DoubleType),
+      If(yMk === 0.0, Literal.create(1.0, DoubleType), corr * corr))
+  }
+  override protected def withNewChildrenInternal(
+      newLeft: Expression, newRight: Expression): RegrR2 =
+    this.copy(x = newLeft, y = newRight)
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/AggregateExpressionSuite.scala

@@ -18,8 +18,8 @@
 package org.apache.spark.sql.catalyst.expressions.aggregate
 
 import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
-import org.apache.spark.sql.catalyst.expressions.{Add, AttributeSet}
+import org.apache.spark.sql.catalyst.analysis.{TypeCheckResult, UnresolvedAttribute}
+import org.apache.spark.sql.catalyst.expressions.{Add, AttributeSet, Literal}
 
 class AggregateExpressionSuite extends SparkFunSuite {
 
@@ -31,4 +31,21 @@ class AggregateExpressionSuite extends SparkFunSuite {
     assert(expected == actual, s"Expected: $expected. Actual: $actual")
   }
 
+  test("test regr_r2 input types") {
+    val checkResult1 = RegrR2(Literal("a"), Literal(1d)).checkInputDataTypes()
+    assert(checkResult1.isInstanceOf[TypeCheckResult.TypeCheckFailure])
+    assert(checkResult1.asInstanceOf[TypeCheckResult.TypeCheckFailure].message
+      .contains("argument 1 requires double type, however, ''a'' is of string type"))
+    val checkResult2 = RegrR2(Literal(3.0D), Literal('b')).checkInputDataTypes()
+    assert(checkResult2.isInstanceOf[TypeCheckResult.TypeCheckFailure])
+    assert(checkResult2.asInstanceOf[TypeCheckResult.TypeCheckFailure].message
+      .contains("argument 2 requires double type, however, ''b'' is of string type"))
+    val checkResult3 = RegrR2(Literal(3.0D), Literal(Array(0))).checkInputDataTypes()
+    assert(checkResult3.isInstanceOf[TypeCheckResult.TypeCheckFailure])
+    assert(checkResult3.asInstanceOf[TypeCheckResult.TypeCheckFailure].message
+      .contains("argument 2 requires double type, however, '[0]' is of array<int> type"))
+    assert(RegrR2(Literal(3.0D), Literal(1d)).checkInputDataTypes() ===
+      TypeCheckResult.TypeCheckSuccess)
+  }
+
 }

sql/core/src/test/resources/sql-functions/sql-expression-schema.md

@@ -1,6 +1,6 @@
 <!-- Automatically generated by ExpressionsSchemaSuite -->
 ## Summary
-  - Number of queries: 384
+  - Number of queries: 385
   - Number of expressions that missing example: 12
   - Expressions missing examples: bigint,binary,boolean,date,decimal,double,float,int,smallint,string,timestamp,tinyint
 ## Schema of Built-in Functions
@@ -371,6 +371,7 @@
 | org.apache.spark.sql.catalyst.expressions.aggregate.RegrAvgX | regr_avgx | SELECT regr_avgx(y, x) FROM VALUES (1, 2), (2, 2), (2, 3), (2, 4) AS tab(y, x) | struct<regr_avgx(y, x):double> |
 | org.apache.spark.sql.catalyst.expressions.aggregate.RegrAvgY | regr_avgy | SELECT regr_avgy(y, x) FROM VALUES (1, 2), (2, 2), (2, 3), (2, 4) AS tab(y, x) | struct<regr_avgy(y, x):double> |
 | org.apache.spark.sql.catalyst.expressions.aggregate.RegrCount | regr_count | SELECT regr_count(y, x) FROM VALUES (1, 2), (2, 2), (2, 3), (2, 4) AS tab(y, x) | struct<regr_count(y, x):bigint> |
+| org.apache.spark.sql.catalyst.expressions.aggregate.RegrR2 | regr_r2 | SELECT regr_r2(y, x) FROM VALUES (1, 2), (2, 2), (2, 3), (2, 4) AS tab(y, x) | struct<regr_r2(y, x):double> |
 | org.apache.spark.sql.catalyst.expressions.aggregate.Skewness | skewness | SELECT skewness(col) FROM VALUES (-10), (-20), (100), (1000) AS tab(col) | struct<skewness(col):double> |
 | org.apache.spark.sql.catalyst.expressions.aggregate.StddevPop | stddev_pop | SELECT stddev_pop(col) FROM VALUES (1), (2), (3) AS tab(col) | struct<stddev_pop(col):double> |
 | org.apache.spark.sql.catalyst.expressions.aggregate.StddevSamp | std | SELECT std(col) FROM VALUES (1), (2), (3) AS tab(col) | struct<std(col):double> |

sql/core/src/test/resources/sql-tests/inputs/group-by.sql

@@ -244,6 +244,12 @@ SELECT regr_count(y, x) FROM testRegression WHERE x IS NOT NULL;
 SELECT k, count(*), regr_count(y, x) FROM testRegression GROUP BY k;
 SELECT k, count(*) FILTER (WHERE x IS NOT NULL), regr_count(y, x) FROM testRegression GROUP BY k;
 
+-- SPARK-37613: Support ANSI Aggregate Function: regr_r2
+SELECT regr_r2(y, x) FROM testRegression;
+SELECT regr_r2(y, x) FROM testRegression WHERE x IS NOT NULL;
+SELECT k, corr(y, x), regr_r2(y, x) FROM testRegression GROUP BY k;
+SELECT k, corr(y, x) FILTER (WHERE x IS NOT NULL), regr_r2(y, x) FROM testRegression GROUP BY k;
+
 -- SPARK-27974: Support ANSI Aggregate Function: array_agg
 SELECT
   collect_list(col),

sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part1.sql

@@ -85,7 +85,7 @@ SELECT regr_count(b, a) FROM aggtest;
 -- SELECT regr_syy(b, a) FROM aggtest;
 -- SELECT regr_sxy(b, a) FROM aggtest;
 SELECT regr_avgx(b, a), regr_avgy(b, a) FROM aggtest;
--- SELECT regr_r2(b, a) FROM aggtest;
+SELECT regr_r2(b, a) FROM aggtest;
 -- SELECT regr_slope(b, a), regr_intercept(b, a) FROM aggtest;
 SELECT covar_pop(b, a), covar_samp(b, a) FROM aggtest;
 SELECT corr(b, a) FROM aggtest;

sql/core/src/test/resources/sql-tests/inputs/udf/postgreSQL/udf-aggregates_part1.sql

@@ -85,7 +85,7 @@ SELECT regr_count(b, a) FROM aggtest;
 -- SELECT regr_syy(b, a) FROM aggtest;
 -- SELECT regr_sxy(b, a) FROM aggtest;
 SELECT regr_avgx(b, a), regr_avgy(b, a) FROM aggtest;
--- SELECT regr_r2(b, a) FROM aggtest;
+SELECT regr_r2(b, a) FROM aggtest;
 -- SELECT regr_slope(b, a), regr_intercept(b, a) FROM aggtest;
 SELECT udf(covar_pop(b, udf(a))), covar_samp(udf(b), a) FROM aggtest;
 SELECT corr(b, udf(a)) FROM aggtest;

sql/core/src/test/resources/sql-tests/results/group-by.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 95
+-- Number of queries: 99
 
 
 -- !query
@@ -877,6 +877,40 @@ struct<k:int,count(1) FILTER (WHERE (x IS NOT NULL)):bigint,regr_count(y, x):big
 2	3	3
 
 
+-- !query
+SELECT regr_r2(y, x) FROM testRegression
+-- !query schema
+struct<regr_r2(y, x):double>
+-- !query output
+0.997690531177829
+
+
+-- !query
+SELECT regr_r2(y, x) FROM testRegression WHERE x IS NOT NULL
+-- !query schema
+struct<regr_r2(y, x):double>
+-- !query output
+0.997690531177829
+
+
+-- !query
+SELECT k, corr(y, x), regr_r2(y, x) FROM testRegression GROUP BY k
+-- !query schema
+struct<k:int,corr(y, x):double,regr_r2(y, x):double>
+-- !query output
+1	NULL	NULL
+2	0.9988445981121533	0.997690531177829
+
+
+-- !query
+SELECT k, corr(y, x) FILTER (WHERE x IS NOT NULL), regr_r2(y, x) FROM testRegression GROUP BY k
+-- !query schema
+struct<k:int,corr(y, x) FILTER (WHERE (x IS NOT NULL)):double,regr_r2(y, x):double>
+-- !query output
+1	NULL	NULL
+2	0.9988445981121533	0.997690531177829
+
+
 -- !query
 SELECT
   collect_list(col),

sql/core/src/test/resources/sql-tests/results/postgreSQL/aggregates_part1.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 46
+-- Number of queries: 47
 
 
 -- !query
@@ -304,6 +304,14 @@ struct<regr_avgx(b, a):double,regr_avgy(b, a):double>
 49.5	107.94315227307379
 
 
+-- !query
+SELECT regr_r2(b, a) FROM aggtest
+-- !query schema
+struct<regr_r2(b, a):double>
+-- !query output
+0.019497798203180258
+
+
 -- !query
 SELECT covar_pop(b, a), covar_samp(b, a) FROM aggtest
 -- !query schema

sql/core/src/test/resources/sql-tests/results/udf/postgreSQL/udf-aggregates_part1.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 45
+-- Number of queries: 46
 
 
 -- !query
@@ -295,6 +295,14 @@ struct<regr_avgx(b, a):double,regr_avgy(b, a):double>
 49.5	107.94315227307379
 
 
+-- !query
+SELECT regr_r2(b, a) FROM aggtest
+-- !query schema
+struct<regr_r2(b, a):double>
+-- !query output
+0.019497798203180258
+
+
 -- !query
 SELECT udf(covar_pop(b, udf(a))), covar_samp(udf(b), a) FROM aggtest
 -- !query schema