This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 3a07a3dbb0ee [SPARK-58213][SQL] corr should return NULL when a column
has zero variance
3a07a3dbb0ee is described below
commit 3a07a3dbb0ee9c737713e7492e25ed7c4866c2b6
Author: StormSpirit <[email protected]>
AuthorDate: Thu Aug 20 14:19:14 2026 +0800
[SPARK-58213][SQL] corr should return NULL when a column has zero variance
### What changes were proposed in this pull request?
For `corr(x, y)` over n >= 2 valid pairs where either column has exactly
zero variance (for example a constant column), the Pearson correlation is
undefined, but the aggregate previously still evaluated `ck / sqrt(xMk * yMk)`
with a zero denominator, which threw DIVIDE_BY_ZERO under ANSI mode. This PR
adds a guard in `Corr.evaluateExpression` that returns an explicit `DoubleType`
NULL before the division when `xMk` or `yMk` is exactly 0.0 and n >= 2. The n
== 1.0 branch is kept ahead [...]
### Why are the changes needed?
For corr, a zero-variance result is undefined and should be represented as
SQL NULL instead of failing in ANSI mode or depending on the division-by-zero
fallback. Similar degenerate-input handling already exists in related
statistical aggregates: for example, `regr_r2` checks the variance accumulators
before dividing. See
[SPARK-58213](https://issues.apache.org/jira/browse/SPARK-58213).
### Does this PR introduce _any_ user-facing change?
Yes. For n >= 2 pairs where either column is constant, `corr` now always
returns NULL: in ANSI mode this previously threw DIVIDE_BY_ZERO, and in
non-ANSI mode the same NULL was produced implicitly through division by zero.
The n == 1.0 single-pair behavior is unchanged. The primitive
`DataFrameStatFunctions.corr` and MLlib `Statistics.corr` NaN contracts are not
affected.
### How was this patch tested?
Added a DataFrameAggregateSuite regression test covering constant x,
constant y, both constant, all-NULL, partial NULL, single pair, normal and
zero-correlation inputs, cross-checked across ANSI/non-ANSI and
legacy/non-legacy configurations with whole-stage codegen on and off. The
affected linear-regression, window and udf-window SQL golden files were
regenerated by targeted SQLQueryTestSuite runs. The existing
DataFrameWindowFunctionsSuite and DataFrameStatSuite pass with no test sou [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex and Claude AI
Closes #58030 from jiangxt2/fix/corr-zero-variance-null.
Authored-by: StormSpirit <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 5d5b647a2a898090946c452223d34d7e9e5fd3a4)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/catalyst/expressions/aggregate/Corr.scala | 7 ++-
.../analyzer-results/linear-regression.sql.out | 47 ++++++++++++++++++++
.../sql-tests/inputs/linear-regression.sql | 14 ++++++
.../sql-tests/results/linear-regression.sql.out | 43 +++++++++++++++++++
.../sql-tests/results/udf/udf-window.sql.out | 26 +++++------
.../resources/sql-tests/results/window.sql.out | 28 +++++-------
.../apache/spark/sql/DataFrameAggregateSuite.scala | 50 ++++++++++++++++++++++
7 files changed, 181 insertions(+), 34 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Corr.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Corr.scala
index bc78dfdf8cec..0502cf38f849 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Corr.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Corr.scala
@@ -136,7 +136,12 @@ case class Corr(
override val evaluateExpression: Expression = {
If(n === 0.0, Literal.create(null, DoubleType),
- If(n === 1.0, divideByZeroEvalResult, ck / sqrt(xMk * yMk)))
+ If(n === 1.0, divideByZeroEvalResult,
+ // For an exactly zero variance accumulator, corr is undefined. Keep
this check after
+ // n == 1.0 to preserve the legacy single-pair result, and before
division to avoid the
+ // ANSI divide-by-zero error for this case.
+ If(xMk === 0.0 || yMk === 0.0, Literal.create(null, DoubleType),
+ ck / sqrt(xMk * yMk))))
}
override def prettyName: String = "corr"
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/linear-regression.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/linear-regression.sql.out
index fa87a63e7f13..38fac614df70 100644
---
a/sql/core/src/test/resources/sql-tests/analyzer-results/linear-regression.sql.out
+++
b/sql/core/src/test/resources/sql-tests/analyzer-results/linear-regression.sql.out
@@ -433,3 +433,50 @@ Aggregate [regr_r2(cast(y#x as double), cast(k#x as
double)) AS regr_r2(y, k)#x]
+- Project [k#x, y#x, x#x]
+- SubqueryAlias testRegression
+- LocalRelation [k#x, y#x, x#x]
+
+
+-- !query
+CREATE OR REPLACE TEMPORARY VIEW testCorrConstant AS SELECT * FROM VALUES
+(1, 1), (1, 2), (1, 3) AS t(x, y)
+-- !query analysis
+CreateViewCommand `testCorrConstant`, SELECT * FROM VALUES
+(1, 1), (1, 2), (1, 3) AS t(x, y), false, true, LocalTempView, UNSUPPORTED,
true
+ +- Project [x#x, y#x]
+ +- SubqueryAlias t
+ +- LocalRelation [x#x, y#x]
+
+
+-- !query
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM testCorrConstant
+-- !query analysis
+Aggregate [corr(cast(x#x as double), cast(y#x as double)) AS corr(CAST(x AS
DOUBLE), CAST(y AS DOUBLE))#x]
++- SubqueryAlias testcorrconstant
+ +- View (`testCorrConstant`, [x#x, y#x])
+ +- Project [cast(x#x as int) AS x#x, cast(y#x as int) AS y#x]
+ +- Project [x#x, y#x]
+ +- SubqueryAlias t
+ +- LocalRelation [x#x, y#x]
+
+
+-- !query
+DROP VIEW testCorrConstant
+-- !query analysis
+DropTempViewCommand testCorrConstant, false
+
+
+-- !query
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM VALUES
+(1, 1), (1, 2), (1, 3) AS t(x, y)
+-- !query analysis
+Aggregate [corr(cast(x#x as double), cast(y#x as double)) AS corr(CAST(x AS
DOUBLE), CAST(y AS DOUBLE))#x]
++- SubqueryAlias t
+ +- LocalRelation [x#x, y#x]
+
+
+-- !query
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM VALUES
+(1, 1), (2, 1), (3, 1) AS t(x, y)
+-- !query analysis
+Aggregate [corr(cast(x#x as double), cast(y#x as double)) AS corr(CAST(x AS
DOUBLE), CAST(y AS DOUBLE))#x]
++- SubqueryAlias t
+ +- LocalRelation [x#x, y#x]
diff --git a/sql/core/src/test/resources/sql-tests/inputs/linear-regression.sql
b/sql/core/src/test/resources/sql-tests/inputs/linear-regression.sql
index a3fa6d4c4cd4..7939b2b9f4c7 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/linear-regression.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/linear-regression.sql
@@ -54,3 +54,17 @@ SELECT k, regr_intercept(y, x) FROM testRegression WHERE x
IS NOT NULL AND y IS
-- SPARK-55969: regr_r2 should treat first param as dependent variable
SELECT regr_r2(k, x) FROM testRegression where k=2;
SELECT regr_r2(y, k) FROM testRegression where k=2;
+
+-- SPARK-58213: corr should return NULL for constant columns (zero variance)
+CREATE OR REPLACE TEMPORARY VIEW testCorrConstant AS SELECT * FROM VALUES
+(1, 1), (1, 2), (1, 3) AS t(x, y);
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM testCorrConstant;
+DROP VIEW testCorrConstant;
+
+-- x-constant (y varies): corr should return NULL
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM VALUES
+(1, 1), (1, 2), (1, 3) AS t(x, y);
+
+-- y-constant (x varies): corr should return NULL
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM VALUES
+(1, 1), (2, 1), (3, 1) AS t(x, y);
diff --git
a/sql/core/src/test/resources/sql-tests/results/linear-regression.sql.out
b/sql/core/src/test/resources/sql-tests/results/linear-regression.sql.out
index 96b2aa08884e..410d7192e11d 100644
--- a/sql/core/src/test/resources/sql-tests/results/linear-regression.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/linear-regression.sql.out
@@ -290,3 +290,46 @@ SELECT regr_r2(y, k) FROM testRegression where k=2
struct<regr_r2(y, k):double>
-- !query output
NULL
+
+
+-- !query
+CREATE OR REPLACE TEMPORARY VIEW testCorrConstant AS SELECT * FROM VALUES
+(1, 1), (1, 2), (1, 3) AS t(x, y)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM testCorrConstant
+-- !query schema
+struct<corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)):double>
+-- !query output
+NULL
+
+
+-- !query
+DROP VIEW testCorrConstant
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM VALUES
+(1, 1), (1, 2), (1, 3) AS t(x, y)
+-- !query schema
+struct<corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)):double>
+-- !query output
+NULL
+
+
+-- !query
+SELECT corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)) FROM VALUES
+(1, 1), (2, 1), (3, 1) AS t(x, y)
+-- !query schema
+struct<corr(CAST(x AS DOUBLE), CAST(y AS DOUBLE)):double>
+-- !query output
+NULL
diff --git
a/sql/core/src/test/resources/sql-tests/results/udf/udf-window.sql.out
b/sql/core/src/test/resources/sql-tests/results/udf/udf-window.sql.out
index 20bb65d1d3e3..4d60e6003d6a 100644
--- a/sql/core/src/test/resources/sql-tests/results/udf/udf-window.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/udf/udf-window.sql.out
@@ -377,23 +377,17 @@ FROM testData
WINDOW w AS (PARTITION BY udf(cate) ORDER BY udf(val))
ORDER BY cate, udf(val)
-- !query schema
-struct<>
+struct<udf(val):int,cate:string,max:int,min:int,min:int,count:bigint,sum:bigint,avg:double,stddev:double,first_value:int,first_value_ignore_null:int,first_value_contain_null:int,any_value:int,any_value_ignore_null:int,any_value_contain_null:int,last_value:int,last_value_ignore_null:int,last_value_contain_null:int,rank:int,dense_rank:int,cume_dist:double,percent_rank:double,ntile:int,row_number:int,var_pop:double,var_samp:double,approx_count_distinct:bigint,covar_pop:double,corr:double,st
[...]
-- !query output
-org.apache.spark.SparkArithmeticException
-{
- "errorClass" : "DIVIDE_BY_ZERO",
- "sqlState" : "22012",
- "messageParameters" : {
- "config" : "\"spark.sql.ansi.enabled\""
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 1126,
- "stopIndex" : 1161,
- "fragment" : "corr(udf(val), udf(val_long)) OVER w"
- } ]
-}
+NULL NULL NULL NULL NULL 0 NULL NULL NULL NULL
NULL NULL NULL NULL NULL NULL NULL NULL 1 1
0.5 0.0 1 1 NULL NULL 0 NULL NULL NULL
NULL [] [] NULL NULL
+3 NULL 3 3 3 1 3 3.0 NULL NULL
3 NULL NULL 3 NULL 3 3 3 2 2
1.0 1.0 2 2 0.0 NULL 1 0.0 NULL NULL
0.0 [3] [3] NULL NULL
+NULL a NULL NULL NULL 0 NULL NULL NULL NULL
NULL NULL NULL NULL NULL NULL NULL NULL 1 1
0.25 0.0 1 1 NULL NULL 0 NULL NULL NULL
NULL [] [] NULL NULL
+1 a 1 1 1 2 2 1.0 0.0 NULL
1 NULL NULL 1 NULL 1 1 1 2 2
0.75 0.3333333333333333 1 2 0.0 0.0 1 0.0
NULL 0.0 0.0 [1,1] [1] 0.7071067811865476 -1.5
+1 a 1 1 1 2 2 1.0 0.0 NULL
1 NULL NULL 1 NULL 1 1 1 2 2
0.75 0.3333333333333333 2 3 0.0 0.0 1 0.0
NULL 0.0 0.0 [1,1] [1] 0.7071067811865476 -1.5
+2 a 2 1 1 3 4 1.3333333333333333
0.5773502691896258 NULL 1 NULL NULL 1 NULL 2
2 2 4 3 1.0 1.0 2 4
0.22222222222222224 0.33333333333333337 2 4.772185885555555E8
1.0 0.5773502691896258 0.4714045207910317 [1,1,2] [1,2]
1.1539890888012805 -0.6672217220327235
+1 b 1 1 1 1 1 1.0 NULL 1
1 1 1 1 1 1 1 1 1 1
0.3333333333333333 0.0 1 1 0.0 NULL 1 NULL
NULL NULL 0.0 [1] [1] NULL NULL
+2 b 2 1 1 2 3 1.5
0.7071067811865476 1 1 1 1 1 1 2
2 2 2 2 0.6666666666666666 0.5 1 2
0.25 0.5 2 0.0 NULL 0.7071067811865476 0.5 [1,2]
[1,2] 0.0 -2.0000000000000013
+3 b 3 1 1 3 6 2.0 1.0 1
1 1 1 1 1 3 3 3 3 3
1.0 1.0 2 3 0.6666666666666666 1.0 3
5.3687091175E8 1.0 1.0 0.816496580927726 [1,2,3] [1,2,3]
0.7057890433107311 -1.4999999999999984
-- !query
diff --git a/sql/core/src/test/resources/sql-tests/results/window.sql.out
b/sql/core/src/test/resources/sql-tests/results/window.sql.out
index a6e76c599731..daa91aab3070 100644
--- a/sql/core/src/test/resources/sql-tests/results/window.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/window.sql.out
@@ -555,23 +555,17 @@ FROM testData
WINDOW w AS (PARTITION BY cate ORDER BY val)
ORDER BY cate, val
-- !query schema
-struct<>
--- !query output
-org.apache.spark.SparkArithmeticException
-{
- "errorClass" : "DIVIDE_BY_ZERO",
- "sqlState" : "22012",
- "messageParameters" : {
- "config" : "\"spark.sql.ansi.enabled\""
- },
- "queryContext" : [ {
- "objectType" : "",
- "objectName" : "",
- "startIndex" : 1016,
- "stopIndex" : 1041,
- "fragment" : "corr(val, val_long) OVER w"
- } ]
-}
+struct<val:int,cate:string,max:int,min:int,min:int,count:bigint,sum:bigint,avg:double,stddev:double,first_value:int,first_value_ignore_null:int,first_value_contain_null:int,any_value:int,any_value_ignore_null:int,any_value_contain_null:int,last_value:int,last_value_ignore_null:int,last_value_contain_null:int,rank:int,dense_rank:int,cume_dist:double,percent_rank:double,ntile:int,row_number:int,var_pop:double,var_samp:double,approx_count_distinct:bigint,covar_pop:double,corr:double,stddev_
[...]
+-- !query output
+NULL NULL NULL NULL NULL 0 NULL NULL NULL NULL
NULL NULL NULL NULL NULL NULL NULL NULL 1 1
0.5 0.0 1 1 NULL NULL 0 NULL NULL NULL
NULL [] [] NULL NULL
+3 NULL 3 3 3 1 3 3.0 NULL NULL
3 NULL NULL 3 NULL 3 3 3 2 2
1.0 1.0 2 2 0.0 NULL 1 0.0 NULL NULL
0.0 [3] [3] NULL NULL
+NULL a NULL NULL NULL 0 NULL NULL NULL NULL
NULL NULL NULL NULL NULL NULL NULL NULL 1 1
0.25 0.0 1 1 NULL NULL 0 NULL NULL NULL
NULL [] [] NULL NULL
+1 a 1 1 1 2 2 1.0 0.0 NULL
1 NULL NULL 1 NULL 1 1 1 2 2
0.75 0.3333333333333333 1 2 0.0 0.0 1 0.0
NULL 0.0 0.0 [1,1] [1] 0.7071067811865476 -1.5
+1 a 1 1 1 2 2 1.0 0.0 NULL
1 NULL NULL 1 NULL 1 1 1 2 2
0.75 0.3333333333333333 2 3 0.0 0.0 1 0.0
NULL 0.0 0.0 [1,1] [1] 0.7071067811865476 -1.5
+2 a 2 1 1 3 4 1.3333333333333333
0.5773502691896258 NULL 1 NULL NULL 1 NULL 2
2 2 4 3 1.0 1.0 2 4
0.22222222222222224 0.33333333333333337 2 4.772185885555555E8
1.0 0.5773502691896258 0.4714045207910317 [1,1,2] [1,2]
1.1539890888012805 -0.6672217220327235
+1 b 1 1 1 1 1 1.0 NULL 1
1 1 1 1 1 1 1 1 1 1
0.3333333333333333 0.0 1 1 0.0 NULL 1 NULL
NULL NULL 0.0 [1] [1] NULL NULL
+2 b 2 1 1 2 3 1.5
0.7071067811865476 1 1 1 1 1 1 2
2 2 2 2 0.6666666666666666 0.5 1 2
0.25 0.5 2 0.0 NULL 0.7071067811865476 0.5 [1,2]
[1,2] 0.0 -2.0000000000000013
+3 b 3 1 1 3 6 2.0 1.0 1
1 1 1 1 1 3 3 3 3 3
1.0 1.0 2 3 0.6666666666666666 1.0 3
5.3687091175E8 1.0 1.0 0.816496580927726 [1,2,3] [1,2,3]
0.7057890433107311 -1.4999999999999984
-- !query
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
index 1918acab6c4e..5300c0f3ad4a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
@@ -573,6 +573,56 @@ class DataFrameAggregateSuite extends SharedSparkSession
}
}
+ testWithWholeStageCodegenOnAndOff("SPARK-58213: corr returns NULL for zero
variance") { _ =>
+ val input = Seq(
+ ("both-constant", Some(1.0), Some(1.0)),
+ ("both-constant", Some(1.0), Some(1.0)),
+ ("empty", None, None),
+ ("normal", Some(1.0), Some(1.0)),
+ ("normal", Some(2.0), Some(2.0)),
+ ("normal", Some(3.0), Some(3.0)),
+ ("partial-null", None, Some(0.0)),
+ ("partial-null", Some(1.0), Some(1.0)),
+ ("partial-null", Some(2.0), None),
+ ("partial-null", Some(3.0), Some(3.0)),
+ ("single", Some(1.0), Some(2.0)),
+ ("x-constant", Some(1.0), Some(1.0)),
+ ("x-constant", Some(1.0), Some(2.0)),
+ ("x-constant", Some(1.0), Some(3.0)),
+ ("y-constant", Some(1.0), Some(1.0)),
+ ("y-constant", Some(2.0), Some(1.0)),
+ ("y-constant", Some(3.0), Some(1.0)),
+ ("zero-correlation", Some(0.0), Some(1.0)),
+ ("zero-correlation", Some(1.0), Some(-2.0)),
+ ("zero-correlation", Some(2.0), Some(1.0))).toDF("case", "x", "y")
+
+ Seq(true, false).foreach { ansiEnabled =>
+ Seq(true, false).foreach { legacyStatisticalAggregate =>
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString,
+ SQLConf.LEGACY_STATISTICAL_AGGREGATE.key ->
+ legacyStatisticalAggregate.toString) {
+ val singleRow = if (legacyStatisticalAggregate) {
+ Row("single", Double.NaN)
+ } else {
+ Row("single", null)
+ }
+ checkAnswer(
+ input.groupBy($"case").agg(corr($"x", $"y")).orderBy($"case"),
+ Seq(
+ Row("both-constant", null),
+ Row("empty", null),
+ Row("normal", 1.0),
+ Row("partial-null", 1.0),
+ singleRow,
+ Row("x-constant", null),
+ Row("y-constant", null),
+ Row("zero-correlation", 0.0)))
+ }
+ }
+ }
+ }
+
test("null moments") {
val emptyTableData = Seq.empty[(Int, Int)].toDF("a", "b")
checkAnswer(emptyTableData.agg(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
