Yaoxuan Wu created SPARK-60156:
----------------------------------
Summary: Correlated scalar subquery over a DOUBLE sum returns NULL
for a row that has matches
Key: SPARK-60156
URL: https://issues.apache.org/jira/browse/SPARK-60156
Project: Spark
Issue Type: Bug
Components: Optimizer
Affects Versions: 4.1.1, 4.2.0, 3.5.8, 4.0.0, 5.0.0
Reporter: Yaoxuan Wu
{code:java}
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
spark.createDataFrame([(1, 'a', 1.0), (1, 'b', 1e16), (1, 'a', 1.0), (1, 'b',
-1e16)],
'k int, c string, x double').createOrReplaceTempView('t')
spark.sql("CREATE OR REPLACE TEMP VIEW g AS "
"SELECT k, count(DISTINCT c) AS n, sum(x) AS s FROM t GROUP BY
k")spark.table("g").collect()
# [Row(k=1, n=2, s=2.0)]spark.sql("SELECT k, n, s, (SELECT max(x) FROM t WHERE
t.x < g.s) AS m FROM g").collect()
# expected: [Row(k=1, n=2, s=2.0, m=1.0)]
# Spark 4.2.0: [Row(k=1, n=2, s=2.0, m=None)]spark.sql("SELECT k, n, s, (SELECT
count(*) FROM t WHERE t.x < g.s) AS m FROM g").collect()
# expected: [Row(k=1, n=2, s=2.0, m=3)]
# Spark 4.2.0: [Row(k=1, n=2, s=2.0, m=0)]
{code}
Same on 3.5.8, 4.0.0, 4.1.1, 4.2.0 and master, at default configs.
This is not a question of summation order. The order of a floating-point sum is
not specified, so s may be 0.0, 1.0 or 2.0. But x = -1e16 is smaller than all
of them, so under any order {{max}} is not NULL and {{count}} is at least 1.
Spark returns s = 2.0 for the row, and for 2.0 the answer is 1.0 / 3.
Optimized plan (shortened):
{code:java}
Join LeftOuter, (normalizenanandzero(coalesce(s#121, 0.0)) =
normalizenanandzero(coalesce(s#129, 0.0)) AND (isnull(s#121) = isnull(s#129)))
:- Aggregate [k#0], [k#0, count(distinct c#1) AS n#120L, sum(x#2) AS s#121]
+- Aggregate [s#129], [max(x#128) AS max(x)#125, s#129]
+- Join Inner, (x#128 < s#129)
:- Aggregate [s#121], [s#121 AS s#129]
: +- Aggregate [k#0], [sum(x#2) AS s#121]
+- Project [x#128] ... {code}
Cause: for the non-equality correlation, DecorrelateInnerQuery builds a domain
join from a second copy of the outer plan and joins the result back on equality
of s. Column pruning removes count(DISTINCT c) from the copy only. The outer
side computes sum(x) through the distinct-aggregate rewrite (2.0). The copy
uses a plain hash aggregate (0.0, the same as SELECT k, sum(x) FROM t GROUP BY
k). The join back finds no match, so the subquery value is lost.SPARK-55038 has
the same mechanism with array_agg(DISTINCT) (element order differs between the
two evaluations). Its proposed fix sorted collect_list/collect_set output,
which does not help here. Fixing decorrelation would cover both, for example by
joining back on a row identifier instead of the outer values, or by evaluating
the outer plan once.scalarSubqueryUseSingleJoin=false,
decorrelateJoinPredicate.enabled=false and adaptive.enabled=false do not change
the result. decorrelateInnerQuery.enabled=false rejects the query.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
