[SPARK-48719][SQL] Fix the calculation bug of RegrSlope & `RegrInte…

…rcept` when the first parameter is null

### What changes were proposed in this pull request?

This PR aims to fix the calculation bug of `RegrSlope`&`RegrIntercept` when the first parameter is null. Regardless of whether the first parameter(y) or the second parameter(x) is null, this tuple should be filtered out.

### Why are the changes needed?

Fix bug.

### Does this PR introduce _any_ user-facing change?

Yes, the calculation changes when the first value of a tuple is null, but the value is truly correct.

### How was this patch tested?

Pass GA and test with `build/sbt "~sql/testOnly org.apache.spark.sql.SQLQueryTestSuite -- -z linear-regression.sql"`

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #47105 from wayneguow/SPARK-48719.

Authored-by: Wei Guo <guow93@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>

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

@@ -271,8 +271,14 @@ case class RegrSlope(left: Expression, right: Expression) extends DeclarativeAgg
 
   override lazy val initialValues: Seq[Expression] = covarPop.initialValues ++ varPop.initialValues
 
-  override lazy val updateExpressions: Seq[Expression] =
-    covarPop.updateExpressions ++ varPop.updateExpressions
+  override lazy val updateExpressions: Seq[Expression] = {
+    // RegrSlope only handles paris where both y and x are non-empty, so we need additional
+    // judgment for calculating VariancePop.
+    val isNull = left.isNull || right.isNull
+    covarPop.updateExpressions ++ varPop.updateExpressions.zip(varPop.aggBufferAttributes).map {
+      case (newValue, oldValue) => If(isNull, oldValue, newValue)
+    }
+  }
 
   override lazy val mergeExpressions: Seq[Expression] =
     covarPop.mergeExpressions ++ varPop.mergeExpressions
@@ -324,8 +330,14 @@ case class RegrIntercept(left: Expression, right: Expression) extends Declarativ
 
   override lazy val initialValues: Seq[Expression] = covarPop.initialValues ++ varPop.initialValues
 
-  override lazy val updateExpressions: Seq[Expression] =
-    covarPop.updateExpressions ++ varPop.updateExpressions
+  override lazy val updateExpressions: Seq[Expression] = {
+    // RegrIntercept only handles paris where both y and x are non-empty, so we need additional
+    // judgment for calculating VariancePop.
+    val isNull = left.isNull || right.isNull
+    covarPop.updateExpressions ++ varPop.updateExpressions.zip(varPop.aggBufferAttributes).map {
+      case (newValue, oldValue) => If(isNull, oldValue, newValue)
+    }
+  }
 
   override lazy val mergeExpressions: Seq[Expression] =
     covarPop.mergeExpressions ++ varPop.mergeExpressions

sql/core/src/test/resources/sql-tests/analyzer-results/linear-regression.sql.out

@@ -1,11 +1,11 @@
 -- Automatically generated by SQLQueryTestSuite
 -- !query
 CREATE OR REPLACE TEMPORARY VIEW testRegression AS SELECT * FROM VALUES
-(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35)
+(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35), (2, null, 40)
 AS testRegression(k, y, x)
 -- !query analysis
 CreateViewCommand `testRegression`, SELECT * FROM VALUES
-(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35)
+(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35), (2, null, 40)
 AS testRegression(k, y, x), false, true, LocalTempView, UNSUPPORTED, true
    +- Project [k#x, y#x, x#x]
       +- SubqueryAlias testRegression

sql/core/src/test/resources/sql-tests/inputs/linear-regression.sql

@@ -1,6 +1,6 @@
 -- Test data.
 CREATE OR REPLACE TEMPORARY VIEW testRegression AS SELECT * FROM VALUES
-(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35)
+(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35), (2, null, 40)
 AS testRegression(k, y, x);
 
 -- SPARK-37613: Support ANSI Aggregate Function: regr_count

sql/core/src/test/resources/sql-tests/results/linear-regression.sql.out

@@ -1,7 +1,7 @@
 -- Automatically generated by SQLQueryTestSuite
 -- !query
 CREATE OR REPLACE TEMPORARY VIEW testRegression AS SELECT * FROM VALUES
-(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35)
+(1, 10, null), (2, 10, 11), (2, 20, 22), (2, 25, null), (2, 30, 35), (2, null, 40)
 AS testRegression(k, y, x)
 -- !query schema
 struct<>
@@ -31,7 +31,7 @@ SELECT k, count(*), regr_count(y, x) FROM testRegression GROUP BY k
 struct<k:int,count(1):bigint,regr_count(y, x):bigint>
 -- !query output
 1	1	0
-2	4	3
+2	5	3
 
 
 -- !query
@@ -40,7 +40,7 @@ SELECT k, count(*) FILTER (WHERE x IS NOT NULL), regr_count(y, x) FROM testRegre
 struct<k:int,count(1) FILTER (WHERE (x IS NOT NULL)):bigint,regr_count(y, x):bigint>
 -- !query output
 1	0	0
-2	3	3
+2	4	3
 
 
 -- !query
@@ -99,7 +99,7 @@ SELECT k, avg(x), avg(y), regr_avgx(y, x), regr_avgy(y, x) FROM testRegression G
 struct<k:int,avg(x):double,avg(y):double,regr_avgx(y, x):double,regr_avgy(y, x):double>
 -- !query output
 1	NULL	10.0	NULL	NULL
-2	22.666666666666668	21.25	22.666666666666668	20.0
+2	27.0	21.25	22.666666666666668	20.0
 
 
 -- !query
@@ -116,15 +116,15 @@ SELECT regr_sxx(y, x) FROM testRegression
 -- !query schema
 struct<regr_sxx(y, x):double>
 -- !query output
-288.66666666666663
+288.6666666666667
 
 
 -- !query
 SELECT regr_sxx(y, x) FROM testRegression WHERE x IS NOT NULL AND y IS NOT NULL
 -- !query schema
 struct<regr_sxx(y, x):double>
 -- !query output
-288.66666666666663
+288.6666666666667
 
 
 -- !query
@@ -133,15 +133,15 @@ SELECT k, regr_sxx(y, x) FROM testRegression GROUP BY k
 struct<k:int,regr_sxx(y, x):double>
 -- !query output
 1	NULL
-2	288.66666666666663
+2	288.6666666666667
 
 
 -- !query
 SELECT k, regr_sxx(y, x) FROM testRegression WHERE x IS NOT NULL AND y IS NOT NULL GROUP BY k
 -- !query schema
 struct<k:int,regr_sxx(y, x):double>
 -- !query output
-2	288.66666666666663
+2	288.6666666666667
 
 
 -- !query
@@ -215,15 +215,15 @@ SELECT regr_slope(y, x) FROM testRegression
 -- !query schema
 struct<regr_slope(y, x):double>
 -- !query output
-0.8314087759815244
+0.8314087759815242
 
 
 -- !query
 SELECT regr_slope(y, x) FROM testRegression WHERE x IS NOT NULL AND y IS NOT NULL
 -- !query schema
 struct<regr_slope(y, x):double>
 -- !query output
-0.8314087759815244
+0.8314087759815242
 
 
 -- !query
@@ -232,15 +232,15 @@ SELECT k, regr_slope(y, x) FROM testRegression GROUP BY k
 struct<k:int,regr_slope(y, x):double>
 -- !query output
 1	NULL
-2	0.8314087759815244
+2	0.8314087759815242
 
 
 -- !query
 SELECT k, regr_slope(y, x) FROM testRegression WHERE x IS NOT NULL AND y IS NOT NULL GROUP BY k
 -- !query schema
 struct<k:int,regr_slope(y, x):double>
 -- !query output
-2	0.8314087759815244
+2	0.8314087759815242
 
 
 -- !query