[
https://issues.apache.org/jira/browse/SPARK-60156?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yaoxuan Wu updated SPARK-60156:
-------------------------------
Labels: correctness (was: )
> 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.0.0, 3.5.8, 4.2.0, 4.1.1, 5.0.0
> Reporter: Yaoxuan Wu
> Priority: Major
> Labels: correctness
>
> {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
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
