[
https://issues.apache.org/jira/browse/SPARK-55038?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18054645#comment-18054645
]
André Souprayane commented on SPARK-55038:
------------------------------------------
I am not able to reproduce this issue with spark 4.0.2:
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 4.0.2-SNAPSHOT
/_/
Using Scala version 2.13.16 (OpenJDK 64-Bit Server VM, Java 17.0.16)
Type in expressions to have them evaluated.
Type :help for more information.
26/01/27 15:20:45 WARN NativeCodeLoader: Unable to load native-hadoop library
for your platform... using builtin-java classes where applicable
Spark context Web UI available at
http://992ef94a-2283-4446-87b0-861e5b987f98.internal.cloudapp.net:4040
Spark context available as 'sc' (master = local[*], app id =
local-1769527246636).
Spark session available as 'spark'.
scala> spark.conf.set("spark.sql.adaptive.enabled", "true")
|
spark.conf.set("spark.sql.objectHashAggregate.sortBased.fallbackThreshold",
"1")
| spark.sql("""
| WITH t AS (SELECT explode(array(15, 16, 17)) AS v)
| SELECT (SELECT COUNT(*) FROM t WHERE array_contains(arr, v)) AS cnt,
arr
| FROM (SELECT collect_set(v) AS arr FROM t)
| """).show()
+---+------------+
|cnt| arr|
+---+------------+
| 3|[16, 17, 15]|
+---+------------+
> AQE + sortBased aggregation produces wrong results for array_agg(DISTINCT) in
> correlated subqueries
> ---------------------------------------------------------------------------------------------------
>
> Key: SPARK-55038
> URL: https://issues.apache.org/jira/browse/SPARK-55038
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 3.5.0, 4.0.0
> Environment: - Spark 4.0.1 (also reproducible on 3.5.x)
> - Reproduced in both local[4] and distributed mode
>
> Reporter: Feng Zhang
> Priority: Critical
>
> h4. Summary
> When `spark.sql.adaptive.enabled=true`
> and`spark.sql.objectHashAggregate.sortBased.fallbackThreshold=1`, queries
> using `array_agg(DISTINCT)` in correlated subqueries produce incorrect
> results.
>
> h4. Reproducer
> {code:java}
> spark.conf.set("spark.sql.adaptive.enabled", "true")
> spark.conf.set("spark.sql.objectHashAggregate.sortBased.fallbackThreshold",
> "1") spark.sql("""
> WITH t AS (SELECT explode(array(15, 16, 17)) AS v)
> SELECT (SELECT COUNT(*) FROM t WHERE array_contains(arr, v)) AS cnt, arr
> FROM (SELECT collect_set(v) AS arr FROM t)
> """).show()
> -- Expected: cnt = 3
> -- Actual: cnt = 0
> {code}
>
> h4. Root Cause Analysis
>
> The correlated subquery is decorrelated into a hash join using the array as
> the join key.
> When array_agg(DISTINCT) (which uses collect_list(distinct)) is computed
> multiple times with sort-based aggregation + AQE, the computations produce
> arrays with the same elements but different orderings. Since array equality
> is element-by-element, the hash join fails to match.
>
> h4. Workarounds
>
> - Set spark.sql.adaptive.enabled=false, OR
> - Set spark.sql.objectHashAggregate.sortBased.fallbackThreshold to default
> (128)
>
> h4. Test Results
>
> - AQE=true, sortBased.fallbackThreshold=1: FAIL (cnt=0, expected 3)
> - AQE=false, sortBased.fallbackThreshold=1: PASS (cnt=3)
> - AQE=true, sortBased.fallbackThreshold=128 (default): PASS (cnt=3)
>
> h4. Conclusion
>
> The bug only occurs when both AQE is enabled and
> sortBased.fallbackThreshold=1.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
