[
https://issues.apache.org/jira/browse/SPARK-55038?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18054680#comment-18054680
]
Feng Zhang commented on SPARK-55038:
------------------------------------
Thanks [~andresouprayane] for looking into this.
I made a mistake in the code snippet in the original bug report (it was
actually a valid case). I have updated it to reflect the bug with array_agg
(not collect_set).
You should be able to reproduce it now with the updated code. Sorry for the
confusing.
> 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(arr1, v)) AS array_agg_cnt,
> (SELECT COUNT(*) FROM t WHERE array_contains(arr2, v)) AS
> collect_set_cnt,
> arr1,
> arr2
> FROM (
> SELECT
> array_agg(DISTINCT v) AS arr1,
> collect_set(v) AS arr2
> FROM t
> )
> """).show()
> Expected: Both counts = 3
> Actual: array_agg_cnt = 0, collect_set_cnt = 3
> {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]
